[AURON #2540] Support residual conditions in native shuffled hash joins

native-engine/auron-planner/src/planner.rs

@@ -235,16 +235,7 @@ impl PhysicalPlanner {
                 let join_type =
                     protobuf::JoinType::try_from(hash_join.join_type).expect("invalid JoinType");
                 let join_type = JoinType::from(join_type);
-                // Spark residual join conditions are evaluated after join key
-                // matching. Keep the native implementation limited to inner
-                // joins for now so filtering matched pairs cannot change outer
-                // join null-extension semantics.
                 let join_filter = self.parse_join_filter(hash_join.filter.as_ref())?;
-                if join_filter.is_some() && join_type != JoinType::Inner {
-                    return Err(proto_error(
-                        "hash join filter is only supported for inner join",
-                    ));
-                }
 
                 let build_side =
                     protobuf::JoinSide::try_from(hash_join.build_side).expect("invalid BuildSide");

spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/joins/auron/plan/NativeShuffledHashJoinExecProvider.scala

@@ -135,6 +135,15 @@ case object NativeShuffledHashJoinExecProvider {
         HashJoin.rewriteKeyExpr(exprs)
 
       override def outputOrdering: Seq[SortOrder] = {
+        import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftAnti, LeftOuter, LeftSemi, RightOuter}
+
+        // Spark 3.1 HashJoin rejects these build sides; native SHJ supports them without ordering.
+        (joinType, buildSide) match {
+          case (LeftOuter | LeftSemi | LeftAnti | _: ExistenceJoin, JoinBuildLeft) =>
+            return Nil
+          case (RightOuter, JoinBuildRight) => return Nil
+          case _ =>
+        }
         val sparkBuildSide = buildSide match {
           case JoinBuildLeft => org.apache.spark.sql.catalyst.optimizer.BuildLeft
           case JoinBuildRight => org.apache.spark.sql.catalyst.optimizer.BuildRight

spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronShuffledHashJoinConditionSuite.scala

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron
+
+import org.apache.spark.sql.catalyst.expressions.LessThan
+import org.apache.spark.sql.catalyst.plans.ExistenceJoin
+import org.apache.spark.sql.execution.auron.plan.NativeShuffledHashJoinBase
+
+import org.apache.auron.util.AuronTestUtils
+
+class AuronShuffledHashJoinConditionSuite extends AuronJoinConditionTestBase {
+  private def withShuffledJoin(force: Boolean = false)(f: => Unit): Unit = {
+    withSparkConf("spark.auron.forceShuffledHashJoin" -> force.toString) {
+      withSQLConf(
+        "spark.sql.adaptive.enabled" -> "false",
+        "spark.sql.autoBroadcastJoinThreshold" -> "-1",
+        "spark.sql.shuffle.partitions" -> "2") {
+        withJoinInputs(f)
+      }
+    }
+  }
+
+  for ((joinType, buildAlias) <- Seq(
+      ("inner", "l"),
+      ("inner", "r"),
+      ("left outer", "r"),
+      ("right outer", "l"),
+      ("left semi", "r"),
+      ("left anti", "r"))) {
+    test(s"SHJ $joinType building $buildAlias evaluates residual conditions") {
+      withShuffledJoin() {
+        checkJoinPlan(
+          s"""
+            |SELECT /*+ SHUFFLE_HASH($buildAlias) */ *
+            |FROM condition_left l $joinType JOIN condition_right r
+            |ON l.k = r.k AND l.v < r.v
+            |""".stripMargin,
+          _.isInstanceOf[NativeShuffledHashJoinBase])
+      }
+    }
+  }
+
+  for (buildAlias <- Seq("l", "r")) {
+    test(s"SHJ full outer residual condition building $buildAlias") {
+      if (AuronTestUtils.isSparkV32OrGreater) {
+        withShuffledJoin() {
+          checkJoinPlan(
+            s"""
+              |SELECT /*+ SHUFFLE_HASH($buildAlias) */ *
+              |FROM condition_left l FULL OUTER JOIN condition_right r
+              |ON l.k = r.k AND l.v < r.v
+              |""".stripMargin,
+            _.isInstanceOf[NativeShuffledHashJoinBase])
+        }
+      }
+    }
+  }
+
+  for (joinType <- Seq("left outer", "left semi", "left anti")) {
+    test(s"forced SMJ to SHJ conversion preserves $joinType residual condition") {
+      withShuffledJoin(force = true) {
+        checkJoinPlan(
+          s"""
+            |SELECT /*+ MERGE(l, r) */ *
+            |FROM condition_left l $joinType JOIN condition_right r
+            |ON l.k = r.k AND l.v < r.v
+            |""".stripMargin,
+          _.isInstanceOf[NativeShuffledHashJoinBase])
+      }
+    }
+  }
+
+  test("SHJ existence condition uses columns outside its output") {
+    withShuffledJoin(force = true) {
+      // OR preserves an ExistenceJoin without projecting EXISTS, unsupported by Spark 3.0.
+      for (predicate <- Seq("EXISTS", "NOT EXISTS")) {
+        checkJoinPlan(
+          s"""
+            |SELECT l.k FROM condition_left l
+            |WHERE l.k = 3 OR $predicate (
+            |  SELECT 1 FROM condition_right r WHERE l.k = r.k AND l.v < r.v)
+            |""".stripMargin,
+          {
+            case join: NativeShuffledHashJoinBase =>
+              join.productIterator.exists(_.isInstanceOf[ExistenceJoin]) &&
+              join.expressions.exists(_.isInstanceOf[LessThan])
+            case _ => false
+          })
+      }
+    }
+  }
+
+  for (force <- Seq(false, true)) {
+    test(s"SHJ residual condition respects configuration with force=$force") {
+      withShuffledJoin(force) {
+        withSQLConf("spark.auron.enable.native.join.condition" -> "false") {
+          val hint = if (force) "MERGE(l, r)" else "SHUFFLE_HASH(r)"
+          checkJoinPlan(
+            s"""
+              |SELECT /*+ $hint */ *
+              |FROM condition_left l LEFT OUTER JOIN condition_right r
+              |ON l.k = r.k AND l.v < r.v
+              |""".stripMargin,
+            _.isInstanceOf[NativeShuffledHashJoinBase],
+            native = false)
+        }
+      }
+    }
+  }
+}

spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronShuffledHashJoinOrderingSuite.scala

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.auron.join.JoinBuildSides.{JoinBuildLeft, JoinBuildRight, JoinBuildSide}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Ascending, AttributeReference, GreaterThan, SortOrder}
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter, Inner, JoinType, LeftAnti, LeftOuter, LeftSemi, RightOuter}
+import org.apache.spark.sql.execution.LeafExecNode
+import org.apache.spark.sql.execution.joins.auron.plan.NativeShuffledHashJoinExecProvider
+import org.apache.spark.sql.types.{BooleanType, IntegerType}
+
+class AuronShuffledHashJoinOrderingSuite extends SparkFunSuite {
+  private val leftKey = AttributeReference("lk", IntegerType)()
+  private val rightKey = AttributeReference("rk", IntegerType)()
+  private case class OrderedInput(key: AttributeReference) extends LeafExecNode {
+    override def output: Seq[AttributeReference] = Seq(key)
+    override def outputOrdering: Seq[SortOrder] = Seq(SortOrder(key, Ascending))
+    override protected def doExecute(): RDD[InternalRow] =
+      throw new UnsupportedOperationException("ordering test does not execute rows")
+  }
+  private val left = OrderedInput(leftKey)
+  private val right = OrderedInput(rightKey)
+  private val existence = ExistenceJoin(AttributeReference("exists", BooleanType)())
+
+  private def ordering(joinType: JoinType, buildSide: JoinBuildSide): Seq[SortOrder] = {
+    NativeShuffledHashJoinExecProvider
+      .provide(
+        left,
+        right,
+        Seq(leftKey),
+        Seq(rightKey),
+        joinType,
+        Some(GreaterThan(leftKey, rightKey)),
+        buildSide,
+        isSkewJoin = false)
+      .outputOrdering
+  }
+
+  for ((joinType, buildSide) <- Seq(
+      (LeftOuter, JoinBuildLeft),
+      (LeftSemi, JoinBuildLeft),
+      (LeftAnti, JoinBuildLeft),
+      (existence, JoinBuildLeft),
+      (RightOuter, JoinBuildRight))) {
+    test(s"Spark 3.1 forced $joinType building $buildSide has no ordering guarantee") {
+      if (org.apache.spark.SPARK_VERSION.startsWith("3.1.")) {
+        assert(ordering(joinType, buildSide).isEmpty)
+      }
+    }
+  }
+
+  test("Spark 3.1 retains valid probe-side ordering") {
+    if (org.apache.spark.SPARK_VERSION.startsWith("3.1.")) {
+      for (joinType <- Seq(Inner, LeftOuter, LeftSemi, LeftAnti, existence)) {
+        assert(ordering(joinType, JoinBuildRight) == left.outputOrdering)
+      }
+      for (joinType <- Seq(Inner, RightOuter)) {
+        assert(ordering(joinType, JoinBuildLeft) == right.outputOrdering)
+      }
+      assert(ordering(FullOuter, JoinBuildLeft).isEmpty)
+      assert(ordering(FullOuter, JoinBuildRight).isEmpty)
+    }
+  }
+}

spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala

@@ -48,8 +48,6 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction
 import org.apache.spark.sql.catalyst.expressions.aggregate.Final
 import org.apache.spark.sql.catalyst.expressions.aggregate.Partial
-import org.apache.spark.sql.catalyst.plans.InnerLike
-import org.apache.spark.sql.catalyst.plans.JoinType
 import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
 import org.apache.spark.sql.catalyst.plans.physical.Partitioning
 import org.apache.spark.sql.catalyst.plans.physical.RangePartitioning
@@ -599,7 +597,7 @@ object AuronConverters extends Logging {
           JoinBuildRight
         }
 
-      validateNativeInnerJoinCondition(joinType, condition)
+      validateNativeJoinCondition(condition)
       return Shims.get.createNativeShuffledHashJoinExec(
         addRenameColumnsExec(convertToNative(left.children(0))),
         addRenameColumnsExec(convertToNative(right.children(0))),
@@ -652,7 +650,7 @@ object AuronConverters extends Logging {
         "condition" -> condition,
         "buildSide" -> buildSide))
     try {
-      validateNativeInnerJoinCondition(joinType, condition)
+      validateNativeJoinCondition(condition)
       Shims.get.createNativeShuffledHashJoinExec(
         addRenameColumnsExec(convertToNative(left)),
         addRenameColumnsExec(convertToNative(right)),
@@ -700,15 +698,6 @@ object AuronConverters extends Logging {
     }
   }
 
-  private def validateNativeInnerJoinCondition(
-      joinType: JoinType,
-      condition: Option[Expression]): Unit = {
-    validateNativeJoinCondition(condition)
-    assert(
-      condition.isEmpty || joinType.isInstanceOf[InnerLike],
-      "join condition is not supported")
-  }
-
   private def validateNativeJoinCondition(condition: Option[Expression]): Unit = {
     condition.foreach { expr =>
       assert(

spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala

@@ -28,7 +28,6 @@ import org.apache.spark.sql.auron.NativeRDD
 import org.apache.spark.sql.auron.NativeSupports
 import org.apache.spark.sql.auron.join.JoinBuildSides.{JoinBuildLeft, JoinBuildRight, JoinBuildSide}
 import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.InnerLike
 import org.apache.spark.sql.catalyst.plans.JoinType
 import org.apache.spark.sql.catalyst.plans.RightOuter
 import org.apache.spark.sql.execution.BinaryExecNode
@@ -97,7 +96,6 @@ abstract class NativeShuffledHashJoinBase(
   protected def rewriteKeyExprToLong(exprs: Seq[Expression]): Seq[Expression]
 
   // check whether native converting is supported
-  assert(condition.isEmpty || joinType.isInstanceOf[InnerLike], "join condition is not supported")
   nativeSchema
   nativeJoinOn
   nativeJoinType