This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new dff5fd7b89 [GLUTEN-12921][VL] Support full outer
BroadcastNestedLoopJoin with guarded rewrite (#12922)
dff5fd7b89 is described below
commit dff5fd7b896358e7f965ed77db71c8d23fbbdf0e
Author: WangGuangxin <[email protected]>
AuthorDate: Wed Sep 23 09:36:08 2026 +0800
[GLUTEN-12921][VL] Support full outer BroadcastNestedLoopJoin with guarded
rewrite (#12922)
---
.../gluten/backendsapi/velox/VeloxRuleApi.scala | 1 +
.../org/apache/gluten/config/VeloxConfig.scala | 14 +
.../VeloxBroadcastNestedLoopJoinRewriteRule.scala | 239 +++++++++++++++++
docs/velox-configuration.md | 1 +
.../BroadcastNestedLoopJoinExecTransformer.scala | 13 +
.../execution/ColumnarBroadcastExchangeExec.scala | 8 +-
...utenBroadcastNestedLoopJoinFullOuterSuite.scala | 298 +++++++++++++++++++++
7 files changed, 573 insertions(+), 1 deletion(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
index 997d66b47e..0937d5d065 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
@@ -93,6 +93,7 @@ object VeloxRuleApi {
// Legacy: Pre-transform rules.
injector.injectPreTransform(_ => RemoveTransitions)
+ injector.injectPreTransform(_ => VeloxBroadcastNestedLoopJoinRewriteRule())
injector.injectPreTransform(_ => PushDownInputFileExpression.PreOffload)
injector.injectPreTransform(c => FallbackOnANSIMode.apply(c.session))
injector.injectPreTransform(c => FallbackMultiCodegens.apply(c.session))
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
index 5d377a9368..eadd105e8c 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
@@ -73,6 +73,9 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) {
def enableBroadcastBuildOncePerExecutor: Boolean =
getConf(VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR)
+ def broadcastNestedLoopJoinFullOuterRewriteThreshold: Long =
+ getConf(VELOX_BROADCAST_NESTED_LOOP_JOIN_FULL_OUTER_REWRITE_THRESHOLD)
+
def veloxBroadcastHashTableBuildTargetBytes: Long =
getConf(COLUMNAR_VELOX_BROADCAST_HASH_TABLE_BUILD_TARGET_BYTES)
@@ -252,6 +255,17 @@ object VeloxConfig extends ConfigRegistry {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("32MB")
+ val VELOX_BROADCAST_NESTED_LOOP_JOIN_FULL_OUTER_REWRITE_THRESHOLD =
+ buildConf(
+
"spark.gluten.sql.columnar.backend.velox.broadcastNLJ.fullOuterRewriteThreshold")
+ .doc(
+ "Maximum per-side plan size in bytes for rewriting a full outer
broadcast nested loop " +
+ "join into a left outer join and an existence join followed by
union. The rewrite is " +
+ "applied only when both sides have known statistics and each side is
at or below this " +
+ "threshold. Set to -1 to disable the rewrite.")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("10MB")
+
val COLUMNAR_VELOX_ASYNC_TIMEOUT_ON_TASK_STOPPING =
buildStaticConf("spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping")
.doc("Timeout in milliseconds when waiting for runtime-scoped async work
to finish during" +
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/extension/VeloxBroadcastNestedLoopJoinRewriteRule.scala
b/backends-velox/src/main/scala/org/apache/gluten/extension/VeloxBroadcastNestedLoopJoinRewriteRule.scala
new file mode 100644
index 0000000000..2d61c9be57
--- /dev/null
+++
b/backends-velox/src/main/scala/org/apache/gluten/extension/VeloxBroadcastNestedLoopJoinRewriteRule.scala
@@ -0,0 +1,239 @@
+/*
+ * 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.gluten.extension
+
+import org.apache.gluten.config.VeloxConfig
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, Expression, Literal, Not}
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter,
LeftOuter}
+import org.apache.spark.sql.catalyst.plans.logical.Join
+import org.apache.spark.sql.catalyst.plans.physical.IdentityBroadcastMode
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.{FilterExec, ProjectExec, SparkPlan,
UnionExec}
+import org.apache.spark.sql.execution.adaptive.BroadcastQueryStageExec
+import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
BroadcastExchangeLike, ReusedExchangeExec}
+import org.apache.spark.sql.execution.joins.BroadcastNestedLoopJoinExec
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * Rewrites `BroadcastNestedLoopJoinExec(FullOuter)` into a union of two
nested-loop joins that
+ * Velox already supports natively:
+ * 1. left outer join to produce matches plus unmatched streamed-side rows
+ * 2. an existence join, with the original broadcast side streamed, to
identify its unmatched rows
+ */
+case class VeloxBroadcastNestedLoopJoinRewriteRule() extends Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = {
+ val threshold =
VeloxConfig.get.broadcastNestedLoopJoinFullOuterRewriteThreshold
+ if (threshold < 0) {
+ plan
+ } else {
+ plan.transformUp {
+ case bnlj: BroadcastNestedLoopJoinExec
+ if bnlj.joinType == FullOuter &&
+ childDeterministic(bnlj.left) &&
+ childDeterministic(bnlj.right) &&
+ shouldRewriteFullOuter(
+ bnlj,
+ threshold) && conditionOffloadable(bnlj) &&
broadcastSideRelocatable(bnlj) =>
+ rewriteFullOuter(bnlj)
+ }
+ }
+ }
+
+ // Query stages are leaf nodes, so inspect the encapsulated broadcast plan
as well.
+ private def childDeterministic(plan: SparkPlan): Boolean =
+ plan.deterministic && (plan match {
+ case stage: BroadcastQueryStageExec => childDeterministic(stage.plan)
+ case _ => true
+ })
+
+ /**
+ * The rewrite reuses the original broadcast side in two roles at once:
[[rewriteFullOuter]] keeps
+ * it as the build (broadcast) side of `branchA`, while
[[buildUnmatchedBroadcastSide]] calls
+ * [[unwrapBroadcast]] on it and consumes the unwrapped subtree as a normal
STREAMED input of
+ * `branchB`. That is only safe when the broadcast side can be cleanly
re-materialized as a
+ * partitioned plan. Reject the rewrite otherwise, e.g. for the MERGE
cardinality-check join (`ON
+ * t.pk > s.pk` with an `autoBroadcastJoinThreshold = -1` broadcast of a
reused `Union` source):
+ * there the broadcast side does not unwrap to a plain partitioned subtree,
so after the rewrite a
+ * `ColumnarBroadcastExchangeExec` ends up in `branchB`'s streamed slot and
is executed via
+ * `ColumnarInputAdapter.doExecuteColumnar -> executeColumnar()`, which the
broadcast exchange
+ * does not support, crashing with `[INTERNAL_ERROR] ... has column support
mismatch`.
+ *
+ * A broadcast side is considered relocatable only when:
+ * - it is an exclusively-owned broadcast, i.e. NOT a
[[ReusedExchangeExec]] (a reused/shared
+ * exchange must not be turned into a streamed input); and
+ * - its unwrapped payload does not itself contain a nested broadcast,
which would otherwise
+ * leak into the streamed position of `branchB`.
+ */
+ private def broadcastSideRelocatable(bnlj: BroadcastNestedLoopJoinExec):
Boolean = {
+ val broadcastSide = bnlj.buildSide match {
+ case BuildLeft => bnlj.left
+ case BuildRight => bnlj.right
+ }
+ isCleanRelocatableBroadcast(broadcastSide)
+ }
+
+ private def isCleanRelocatableBroadcast(plan: SparkPlan): Boolean = plan
match {
+ case stage: BroadcastQueryStageExec =>
isCleanRelocatableBroadcast(stage.plan)
+ case _: ReusedExchangeExec => false
+ case exchange: BroadcastExchangeLike => !containsBroadcast(exchange.child)
+ case _ => false
+ }
+
+ private def containsBroadcast(plan: SparkPlan): Boolean =
+ plan.exists {
+ case _: BroadcastExchangeLike => true
+ case _: BroadcastQueryStageExec => true
+ case _: ReusedExchangeExec => true
+ case _ => false
+ }
+
+ private def shouldRewriteFullOuter(
+ bnlj: BroadcastNestedLoopJoinExec,
+ threshold: Long): Boolean = {
+ bnlj.logicalLink.collect {
+ case join: Join =>
+ val leftSize = join.left.stats.sizeInBytes
+ val rightSize = join.right.stats.sizeInBytes
+ leftSize >= 0 && rightSize >= 0 && leftSize <= threshold && rightSize
<= threshold
+ }.getOrElse(false)
+ }
+
+ private def conditionOffloadable(bnlj: BroadcastNestedLoopJoinExec): Boolean
=
+ bnlj.condition.exists {
+ cond =>
+ cond.references.exists(bnlj.left.outputSet.contains) &&
+ cond.references.exists(bnlj.right.outputSet.contains)
+ }
+
+ private def rewriteFullOuter(bnlj: BroadcastNestedLoopJoinExec): SparkPlan =
{
+ val matchesAndStreamedUnmatched = bnlj.buildSide match {
+ case BuildRight =>
+ projectToOutput(
+ BroadcastNestedLoopJoinExec(
+ bnlj.left,
+ bnlj.right,
+ BuildRight,
+ LeftOuter,
+ bnlj.condition),
+ bnlj.output)
+ case BuildLeft =>
+ projectToOutput(
+ BroadcastNestedLoopJoinExec(
+ bnlj.right,
+ bnlj.left,
+ BuildRight,
+ LeftOuter,
+ bnlj.condition),
+ bnlj.output)
+ }
+
+ val unmatchedBroadcastRows = bnlj.buildSide match {
+ case BuildRight =>
+ buildUnmatchedBroadcastSide(
+ unmatchedSide = bnlj.right,
+ otherSide = bnlj.left,
+ unmatchedSideIsLeft = false,
+ condition = bnlj.condition,
+ output = bnlj.output)
+ case BuildLeft =>
+ buildUnmatchedBroadcastSide(
+ unmatchedSide = bnlj.left,
+ otherSide = bnlj.right,
+ unmatchedSideIsLeft = true,
+ condition = bnlj.condition,
+ output = bnlj.output)
+ }
+
+ val union = UnionExec(Seq(matchesAndStreamedUnmatched,
unmatchedBroadcastRows))
+ ProjectExec(
+ union.output.zip(bnlj.output).map {
+ case (childAttr, targetAttr) =>
+ Alias(childAttr, targetAttr.name)(
+ exprId = targetAttr.exprId,
+ qualifier = targetAttr.qualifier,
+ explicitMetadata = Some(targetAttr.metadata))
+ },
+ union
+ )
+ }
+
+ private def buildUnmatchedBroadcastSide(
+ unmatchedSide: SparkPlan,
+ otherSide: SparkPlan,
+ unmatchedSideIsLeft: Boolean,
+ condition: Option[Expression],
+ output: Seq[Attribute]): SparkPlan = {
+ val unmatchedSideBase = unwrapBroadcast(unmatchedSide)
+ val otherSideBase = unwrapBroadcast(otherSide)
+ val existsAttr =
+ AttributeReference("__gluten_bnlj_exists", BooleanType, nullable =
false)()
+ val unmatchedJoin = BroadcastNestedLoopJoinExec(
+ unmatchedSideBase,
+ ensureBroadcast(otherSideBase),
+ BuildRight,
+ ExistenceJoin(existsAttr),
+ condition)
+ val unmatchedOnly = FilterExec(Not(existsAttr), unmatchedJoin)
+ val projected = if (unmatchedSideIsLeft) {
+ output.zipWithIndex.map {
+ case (targetAttr, idx) if idx < unmatchedSideBase.output.size =>
+ aliasTo(unmatchedSideBase.output(idx), targetAttr)
+ case (targetAttr, _) =>
+ nullAliasFor(targetAttr)
+ }
+ } else {
+ output.zipWithIndex.map {
+ case (targetAttr, idx) if idx < otherSide.output.size =>
+ nullAliasFor(targetAttr)
+ case (targetAttr, idx) =>
+ aliasTo(unmatchedSideBase.output(idx - otherSide.output.size),
targetAttr)
+ }
+ }
+ ProjectExec(projected, unmatchedOnly)
+ }
+
+ private def unwrapBroadcast(plan: SparkPlan): SparkPlan = plan match {
+ case stage: BroadcastQueryStageExec => unwrapBroadcast(stage.plan)
+ case reused: ReusedExchangeExec => unwrapBroadcast(reused.child)
+ case exchange: BroadcastExchangeLike => exchange.child
+ case other => other
+ }
+
+ private def ensureBroadcast(plan: SparkPlan): SparkPlan = plan match {
+ case exchange: BroadcastExchangeLike => exchange
+ case other => BroadcastExchangeExec(IdentityBroadcastMode, other)
+ }
+
+ private def projectToOutput(child: SparkPlan, output: Seq[Attribute]):
ProjectExec = {
+ val sourceByExprId = child.output.map(attr => attr.exprId -> attr).toMap
+ ProjectExec(
+ output.map(targetAttr => aliasTo(sourceByExprId(targetAttr.exprId),
targetAttr)),
+ child)
+ }
+
+ private def aliasTo(childAttr: Expression, targetAttr: Attribute): Alias = {
+ Alias(childAttr, targetAttr.name)(
+ exprId = targetAttr.exprId,
+ qualifier = targetAttr.qualifier,
+ explicitMetadata = Some(targetAttr.metadata))
+ }
+
+ private def nullAliasFor(targetAttr: Attribute): Alias =
+ aliasTo(Literal.create(null, targetAttr.dataType), targetAttr)
+}
diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md
index 3ab3220ad8..82b7848bc8 100644
--- a/docs/velox-configuration.md
+++ b/docs/velox-configuration.md
@@ -16,6 +16,7 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinPct
| 🔄 Dynamic | 90 | If partial aggregation aggregationPct
greater than this value, partial aggregation may be early abandoned. Note: this
option only works when flushable partial aggregation is enabled. Ignored when
spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false.
[...]
| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinRows
| 🔄 Dynamic | 100000 | If partial aggregation input rows
number greater than this value, partial aggregation may be early abandoned.
Note: this option only works when flushable partial aggregation is enabled.
Ignored when
spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false.
[...]
| spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping
| âš“ Static | 30000ms | Timeout in milliseconds when waiting
for runtime-scoped async work to finish during teardown.
[...]
+|
spark.gluten.sql.columnar.backend.velox.broadcastNLJ.fullOuterRewriteThreshold
| 🔄 Dynamic | 10MB | Maximum per-side plan size in bytes for
rewriting a full outer broadcast nested loop join into a left outer join and an
existence join followed by union. The rewrite is applied only when both sides
have known statistics and each side is at or below this threshold. Set to -1 to
disable the rewrite.
[...]
| spark.gluten.sql.columnar.backend.velox.cacheEnabled
| âš“ Static | false | Enable Velox cache, default off. It's
recommended to enablesoft-affinity as well when enable velox cache.
[...]
| spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct
| âš“ Static | 0 | Set prefetch cache min pct for velox
file scan
[...]
| spark.gluten.sql.columnar.backend.velox.checkUsageLeak
| âš“ Static | true | Enable check memory usage leak.
[...]
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
index eb7cb0df6c..c5812b3993 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
@@ -17,6 +17,7 @@
package org.apache.gluten.execution
import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.extension.columnar.transition.ConventionReq
import org.apache.gluten.metrics.MetricsUpdater
import org.apache.gluten.substrait.{JoinParams, SubstraitContext}
import org.apache.gluten.utils.SubstraitUtil
@@ -51,6 +52,18 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
override def leftKeys: Seq[Expression] = Nil
override def rightKeys: Seq[Expression] = Nil
+ override def requiredChildConvention(): Seq[ConventionReq] = {
+ val batchReq =
+ ConventionReq.ofBatch(
+
ConventionReq.BatchType.Is(BackendsApiManager.getSettings.primaryBatchType))
+ buildSide match {
+ case BuildLeft =>
+ Seq(ConventionReq.any, batchReq)
+ case BuildRight =>
+ Seq(batchReq, ConventionReq.any)
+ }
+ }
+
private lazy val substraitJoinType: NestedLoopJoinRel.JoinType =
SubstraitUtil.toNestedLoopJoinSubstrait(joinType)
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala
index c4f6c3d504..654fa5bc5a 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala
@@ -18,7 +18,7 @@ package org.apache.spark.sql.execution
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.execution.ValidatablePlan
-import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.extension.columnar.transition.{Convention,
ConventionReq}
import org.apache.gluten.metrics.GlutenTimeMetric
import org.apache.gluten.sql.shims.SparkShimLoader
@@ -134,6 +134,12 @@ case class ColumnarBroadcastExchangeExec(mode:
BroadcastMode, child: SparkPlan)
override def rowType(): Convention.RowType = Convention.RowType.None
+ override def requiredChildConvention(): Seq[ConventionReq] = {
+ Seq(
+ ConventionReq.ofBatch(
+
ConventionReq.BatchType.Is(BackendsApiManager.getSettings.primaryBatchType)))
+ }
+
override def doCanonicalize(): SparkPlan = {
val canonicalized =
BackendsApiManager.getSparkPlanExecApiInstance.doCanonicalizeForBroadcastMode(mode)
diff --git
a/gluten-ut/test/src/test/scala/org/apache/spark/sql/execution/GlutenBroadcastNestedLoopJoinFullOuterSuite.scala
b/gluten-ut/test/src/test/scala/org/apache/spark/sql/execution/GlutenBroadcastNestedLoopJoinFullOuterSuite.scala
new file mode 100644
index 0000000000..da58a5b0d6
--- /dev/null
+++
b/gluten-ut/test/src/test/scala/org/apache/spark/sql/execution/GlutenBroadcastNestedLoopJoinFullOuterSuite.scala
@@ -0,0 +1,298 @@
+/*
+ * 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.spark.sql.execution
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{BroadcastNestedLoopJoinExecTransformer,
SortMergeJoinExecTransformer}
+import org.apache.gluten.utils.BackendTestUtils
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, GlutenQueryTest, Row}
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter}
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec,
AdaptiveSparkPlanHelper}
+import org.apache.spark.sql.execution.joins.BroadcastNestedLoopJoinExec
+import org.apache.spark.sql.functions.rand
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+import scala.reflect.ClassTag
+
+/**
+ * Tests for the Velox full outer `BroadcastNestedLoopJoinExec` rewrite.
+ *
+ * The full outer BNLJ rewrite is a Velox backend feature, hence each test is
guarded with
+ * `assumeVeloxBackend()` so the ClickHouse backend skips them.
+ */
+class GlutenBroadcastNestedLoopJoinFullOuterSuite
+ extends GlutenQueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+ import testImplicits._
+
+ // This shared suite is also compiled when the Velox backend is not on the
classpath.
+ private val fullOuterRewriteThresholdKey =
+
"spark.gluten.sql.columnar.backend.velox.broadcastNLJ.fullOuterRewriteThreshold"
+
+ // Disable the forced shuffled hash join rewrite so explicit join hints
retain their semantics.
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.plugins", "org.apache.gluten.GlutenPlugin")
+ .set("spark.default.parallelism", "1")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "1024MB")
+ .set("spark.ui.enabled", "false")
+ .set(GlutenConfig.GLUTEN_UI_ENABLED.key, "false")
+ .set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
+ }
+
+ private def assumeVeloxBackend(): Unit =
assume(BackendTestUtils.isVeloxBackendLoaded())
+
+ private def materializePlan(df: Dataset[_]): SparkPlan = {
+ val materializedDf = df.toDF()
+ val executedPlan = materializedDf.queryExecution.executedPlan
+ executedPlan.execute()
+ stripAQEPlan(executedPlan match {
+ case adaptivePlan: AdaptiveSparkPlanExec => adaptivePlan.executedPlan
+ case otherPlan => otherPlan
+ })
+ }
+
+ private def assertPlanCount[T <: SparkPlan: ClassTag](
+ df: Dataset[_],
+ expectedCount: Int): Unit = {
+ val targetClass = implicitly[ClassTag[T]].runtimeClass
+ val plan = materializePlan(df)
+ val matchedNodes = plan.collect {
+ case node if targetClass.isInstance(node) => node
+ }
+ assert(
+ matchedNodes.size === expectedCount,
+ s"Expected $expectedCount ${targetClass.getSimpleName} node(s), but
found " +
+ s"${matchedNodes.size}:\n" + plan.treeString
+ )
+ }
+
+ private def assertNoSparkFullOuterBNLJ(df: Dataset[_]): SparkPlan = {
+ val plan = materializePlan(df)
+ val rawFullOuterBnljs = plan.collect {
+ case bnlj: BroadcastNestedLoopJoinExec if bnlj.joinType == FullOuter =>
bnlj
+ }
+ assert(
+ rawFullOuterBnljs.isEmpty,
+ s"Expected rewritten/supported final plan without raw Spark FullOuter " +
+ s"BroadcastNestedLoopJoinExec, but found ${rawFullOuterBnljs.size}:\n"
+
+ plan.treeString
+ )
+ plan
+ }
+
+ private def assertSupportedFullOuterPlan(df: Dataset[_]): Unit = {
+ val plan = assertNoSparkFullOuterBNLJ(df)
+ val nativeBnljCount = plan.collect { case _:
BroadcastNestedLoopJoinExecTransformer => 1 }.size
+ val nativeSmjCount = plan.collect { case _: SortMergeJoinExecTransformer
=> 1 }.size
+ assert(
+ nativeBnljCount + nativeSmjCount > 0,
+ s"Expected a supported native full outer plan after rewrite/planning,
but found neither " +
+ s"${classOf[BroadcastNestedLoopJoinExecTransformer].getSimpleName} nor
" +
+ s"${classOf[SortMergeJoinExecTransformer].getSimpleName}:\n" +
+ plan.treeString
+ )
+ }
+
+ private def assertNativeExistenceJoin(df: Dataset[_]): Unit = {
+ val plan = materializePlan(df)
+ val existenceJoinCount = plan.collect {
+ case bnlj: BroadcastNestedLoopJoinExecTransformer =>
+ bnlj.joinType match {
+ case ExistenceJoin(_) => 1
+ case _ => 0
+ }
+ }.sum
+ assert(
+ existenceJoinCount === 1,
+ s"Expected exactly one native ExistenceJoin in the rewritten plan, but
found " +
+ s"$existenceJoinCount:\n${plan.treeString}"
+ )
+ }
+
+ test("Full outer BroadcastNestedLoopJoinExec should be rewritten into
supported stages") {
+ assumeVeloxBackend()
+ val df1 = spark.range(4).select($"id".as("k1"))
+ val df2 = spark.range(3).select($"id".as("k2"))
+
+ Seq(true, false).foreach {
+ aqeEnabled =>
+ Seq(true, false).foreach {
+ codegenEnabled =>
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqeEnabled.toString,
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key ->
codegenEnabled.toString,
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key ->
Long.MaxValue.toString,
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "false"
+ ) {
+ val fullOuterJoin = df1.hint("broadcast").join(df2, $"k1" <
$"k2", "full_outer")
+ assertNoSparkFullOuterBNLJ(fullOuterJoin)
+ assertPlanCount[BroadcastNestedLoopJoinExecTransformer](
+ fullOuterJoin,
+ expectedCount = 2)
+ assertNativeExistenceJoin(fullOuterJoin)
+ checkAnswer(
+ fullOuterJoin,
+ Seq(
+ Row(0, 1),
+ Row(0, 2),
+ Row(1, 2),
+ Row(2, null),
+ Row(3, null),
+ Row(null, 0)))
+ }
+ }
+ }
+ }
+
+ test(
+ "Full outer BNLJ rewrite should use existence join for high-cardinality
matches") {
+ assumeVeloxBackend()
+ val left = (Seq.fill(100)(1) :+ 2).toDF("k1")
+ val right = (Seq.fill(100)(1) :+ 0).toDF("k2")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "false"
+ ) {
+ val fullOuterJoin = left.join(right.hint("broadcast"), $"k1" <= $"k2",
"full_outer")
+ assertNoSparkFullOuterBNLJ(fullOuterJoin)
+ assertNativeExistenceJoin(fullOuterJoin)
+ assert(fullOuterJoin.count() === 10002)
+ }
+ }
+
+ test("Full outer BNLJ rewrite should be disabled by a negative threshold") {
+ assumeVeloxBackend()
+ val left = spark.range(4).select($"id".as("k1"))
+ val right = spark.range(3).select($"id".as("k2"))
+
+ withSQLConf(
+ fullOuterRewriteThresholdKey -> "-1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString
+ ) {
+ val fullOuterJoin = left.hint("broadcast").join(right, $"k1" < $"k2",
"full_outer")
+ val plan = materializePlan(fullOuterJoin)
+ assert(
+ plan.exists {
+ case bnlj: BroadcastNestedLoopJoinExec if bnlj.joinType == FullOuter
=> true
+ case _ => false
+ },
+ s"Expected the original full outer BNLJ when the rewrite is
disabled:\n${plan.treeString}"
+ )
+ }
+ }
+
+ test("Full outer BNLJ rewrite should skip nondeterministic children") {
+ assumeVeloxBackend()
+ val stable = Seq(0.5).toDF("stable")
+ val nondeterministic = spark.range(1).select(rand().as("random"))
+
+ withSQLConf(
+ fullOuterRewriteThresholdKey -> Long.MaxValue.toString,
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false"
+ ) {
+ val joins = Seq(
+ nondeterministic
+ .join(stable.hint("broadcast"), $"random" < $"stable", "full_outer"),
+ stable
+ .join(nondeterministic.hint("broadcast"), $"stable" < $"random",
"full_outer")
+ )
+
+ joins.foreach {
+ join =>
+ val plan = materializePlan(join)
+ assert(
+ plan.exists {
+ case bnlj: BroadcastNestedLoopJoinExec if bnlj.joinType ==
FullOuter => true
+ case _ => false
+ },
+ s"Expected the original full outer BNLJ for a nondeterministic
child:\n" +
+ plan.treeString
+ )
+ }
+ }
+ }
+
+ test(
+ "Full outer BroadcastNestedLoopJoin rewrite should preserve null semantics
for equals") {
+ assumeVeloxBackend()
+ val df1 = Seq[java.lang.Integer](null, 1, 2, null).toDF("k1")
+ val df2 = Seq[java.lang.Integer](null, 1, 3, null).toDF("k2")
+
+ Seq(true, false).foreach {
+ codegenEnabled =>
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegenEnabled.toString,
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "false"
+ ) {
+ val fullOuterJoin = df1.hint("broadcast").join(df2, $"k1" === $"k2",
"full_outer")
+ assertSupportedFullOuterPlan(fullOuterJoin)
+ checkAnswer(
+ fullOuterJoin,
+ Seq(
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(1, 1),
+ Row(2, null),
+ Row(null, 3)))
+ }
+ }
+ }
+
+ test(
+ "Full outer BNLJ rewrite should preserve null semantics for null-safe
equals") {
+ assumeVeloxBackend()
+ val df1 = Seq[java.lang.Integer](null, 1, 2, null).toDF("k1")
+ val df2 = Seq[java.lang.Integer](null, 1, 3, null).toDF("k2")
+
+ Seq(true, false).foreach {
+ codegenEnabled =>
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegenEnabled.toString,
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "false"
+ ) {
+ val fullOuterJoin = df1.hint("broadcast").join(df2, $"k1" <=> $"k2",
"full_outer")
+ assertSupportedFullOuterPlan(fullOuterJoin)
+ checkAnswer(
+ fullOuterJoin,
+ Seq(
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(1, 1),
+ Row(2, null),
+ Row(null, 3)))
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
