diff --git a/native-engine/auron-planner/src/planner.rs b/native-engine/auron-planner/src/planner.rs index 24921d849..e50e66024 100644 --- a/native-engine/auron-planner/src/planner.rs +++ b/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"); diff --git a/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/joins/auron/plan/NativeShuffledHashJoinExecProvider.scala b/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/joins/auron/plan/NativeShuffledHashJoinExecProvider.scala index 44bb1ba76..6075ec2e9 100644 --- a/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/execution/joins/auron/plan/NativeShuffledHashJoinExecProvider.scala +++ b/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 diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronShuffledHashJoinConditionSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronShuffledHashJoinConditionSuite.scala new file mode 100644 index 000000000..fd91a1091 --- /dev/null +++ b/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) + } + } + } + } +} diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronShuffledHashJoinOrderingSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronShuffledHashJoinOrderingSuite.scala new file mode 100644 index 000000000..718608a4e --- /dev/null +++ b/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) + } + } +} diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala index e6cc43f40..1625be0a7 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala +++ b/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( diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala b/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala index deb826f73..98c284f5e 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeShuffledHashJoinBase.scala +++ b/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