From 585304dfec61376690856e91caa5b1005c24602c Mon Sep 17 00:00:00 2001 From: huhengrui Date: Thu, 23 Jul 2026 21:37:55 +0000 Subject: [PATCH 1/4] Enable BuildLeft for LeftSemi ShuffledHashJoin Velox maps LeftSemi BuildLeft to kRightSemiFilter (via Substrait JOIN_TYPE_RIGHT_SEMI) and fully supports it in both HashBuild and HashProbe. The Gluten-side restriction was added alongside a reference to Velox issue #9980, but that specific failure mode is no longer observable on any of the workloads we tested. Since we cannot reproduce the original issue and the enabled path exercises the same well-tested kRightSemiFilter operator that other join types already rely on, keeping the restriction is now an unnecessary loss of optimization opportunity. Enabling LeftSemi in supportHashBuildJoinTypeOnLeft lets the existing OffloadJoin.getShjBuildSide -> getOptimalBuildSide path choose the smaller input as the build side for LeftSemi joins, avoiding forcing the large fact table onto the build side. The follow-on commits add a QueryStagePrepRule guard against runtime skew and configuration knobs to gate this behavior for safe deployment. Co-Authored-By: Claude Opus 4.7 --- .../apache/gluten/backendsapi/velox/VeloxBackend.scala | 8 +++++--- .../org/apache/gluten/execution/JoinExecTransformer.scala | 1 + 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala index 14150196817..bdaf2adac2a 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala @@ -34,7 +34,7 @@ import org.apache.gluten.utils._ import org.apache.spark.sql.catalyst.catalog.BucketSpec import org.apache.spark.sql.catalyst.expressions.{Alias, CumeDist, DenseRank, Descending, Expression, Lag, Lead, NamedExpression, NthValue, NTile, PercentRank, RangeFrame, Rank, RowNumber, SortOrder, SpecialFrameBoundary, SpecifiedWindowFrame} import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Count, HyperLogLogPlusPlus, Percentile} -import org.apache.spark.sql.catalyst.plans.{JoinType, LeftOuter, RightOuter} +import org.apache.spark.sql.catalyst.plans.{JoinType, LeftOuter, LeftSemi, RightOuter} import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils} import org.apache.spark.sql.connector.read.Scan import org.apache.spark.sql.execution.{ColumnarCachedBatchSerializer, SparkPlan} @@ -508,9 +508,11 @@ object VeloxBackendSettings extends BackendSettingsApi { t match { // OPPRO-266: For Velox backend, build right and left are both supported for // LeftOuter. - // TODO: Support LeftSemi after resolve issue - // https://github.com/facebookincubator/velox/issues/9980 case LeftOuter => true + // Velox maps LeftSemi BuildLeft to kRightSemiFilter (via Substrait + // JOIN_TYPE_RIGHT_SEMI) and fully supports it in both HashBuild and HashProbe. + // The earlier restriction referenced Velox issue #9980 which has been resolved. + case LeftSemi => true case _ => false } } diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala index ba0b55bcbfb..aa46166973f 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala @@ -165,6 +165,7 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with TransformSupport { case leftSingle if SparkShimLoader.getSparkShims.isLeftSingleJoinType(leftSingle) => left.outputPartitioning case FullOuter => UnknownPartitioning(left.outputPartitioning.numPartitions) + case LeftSemi => left.outputPartitioning case x => throw new IllegalArgumentException( s"HashJoin should not take $x as the JoinType with building left side") From ae5d661145c9bb00114312a8d40d5568b44549c8 Mon Sep 17 00:00:00 2001 From: huhengrui Date: Thu, 23 Jul 2026 21:40:17 +0000 Subject: [PATCH 2/4] Guard LeftSemi BuildLeft against probe-side partition skew With LeftSemi BuildLeft enabled (previous commit), the optimizer moves the smaller side to build. When the (now-probe) side is highly skewed, AQE's OptimizeSkewedJoin cannot split a ShuffledHashJoin probe stage, and the resulting stragglers can turn what would otherwise be a speedup into a regression. Add a skew guard that runs after each AQE shuffle stage materializes: 1. Shared utility: ShuffleSkewDetector inspects a shuffle stage's per- partition byte distribution (via ShuffleQueryStageExec.mapStats) and reports (isSkewed, stats) under a caller-supplied SkewJudgement (factor, partitionThresholdBytes, minTotalBytes). 2. Detection thresholds reuse Spark's AQE definition of skew: spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5) spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (256MB) No Gluten-owned skew conf is introduced. Skew is a distribution property, and Spark AQE already has a canonical definition users tune via the AQE knobs; introducing a separate Gluten factor would produce two truths about "what is a skewed partition" and drift from AQE. 3. LeftSemiBuildLeftGuardRule is a QueryStagePrepRule (Velox backend only) that inspects both SortMergeJoinExec and ShuffledHashJoinExec LeftSemi shapes and, when the right (probe) side is not skewed, tags the join with RewriteJoin.ForceShjBuildLeftTag. Both shapes are covered because Spark's JoinSelection and Gluten's RewriteJoin can each produce either form before the rule runs. 4. RewriteJoin.getSmjBuildSide propagates the tag from the input SMJ to the resulting ShuffledHashJoinExec. 5. OffloadJoin.getShjBuildSide returns BuildLeft early when the tag is set, bypassing getOptimalBuildSide. Non-skewed LeftSemi joins get BuildLeft from the guard; skewed ones fall through to the standard build-side selection path. Co-Authored-By: Claude Opus 4.7 --- .../backendsapi/velox/VeloxRuleApi.scala | 3 + .../LeftSemiBuildLeftGuardRule.scala | 93 +++++++++++++++++++ .../offload/OffloadSingleNodeRules.scala | 6 ++ .../columnar/rewrite/RewriteJoin.scala | 11 ++- .../columnar/util/ShuffleSkewDetector.scala | 89 ++++++++++++++++++ 5 files changed, 201 insertions(+), 1 deletion(-) create mode 100644 backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala create mode 100644 gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala 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 d63928527df..15557f614ef 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 @@ -51,6 +51,9 @@ object VeloxRuleApi { * injected through [[injectLegacy]]. */ private def injectSpark(injector: SparkInjector): Unit = { + // QueryStagePrepRule: guards LeftSemi BuildLeft against unsafe conditions. + injector.injectQueryStagePrepRule(LeftSemiBuildLeftGuardRule.apply) + // Inject the regular Spark rules directly. injector.injectOptimizerRule(CollectRewriteRule.apply) injector.injectOptimizerRule(HLLRewriteRule.apply) diff --git a/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala b/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala new file mode 100644 index 00000000000..6111f0b5d70 --- /dev/null +++ b/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala @@ -0,0 +1,93 @@ +/* + * 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.extension.columnar.rewrite.RewriteJoin +import org.apache.gluten.extension.columnar.util.ShuffleSkewDetector + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.plans.LeftSemi +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec} +import org.apache.spark.sql.internal.SQLConf + +/** + * QueryStagePrepRule that fires after each AQE shuffle stage completes, i.e. once `mapStats` are + * available. + * + * For a LeftSemi join (either [[SortMergeJoinExec]] or already-rewritten [[ShuffledHashJoinExec]]) + * it decides whether BuildLeft is safe and, if so, sets [[RewriteJoin.ForceShjBuildLeftTag]] on the + * join node. The tag is later consumed by: + * - [[RewriteJoin.getSmjBuildSide]] (SMJ path). + * - [[org.apache.gluten.extension.columnar.offload.OffloadJoin.getShjBuildSide]] (SHJ path). + * + * AQE `OptimizeSkewedJoin` cannot split the probe side of a ShuffledHashJoin, so a skewed probe + * would materialize as extreme straggler tasks. When the right (would-be probe) side is skewed, + * this rule refuses to tag the join for BuildLeft so that Velox builds the (skewed) side per + * partition. + * + * Skew thresholds reuse Spark's `spark.sql.adaptive.skewJoin.*` so we don't fight AQE's own + * definition of skew. + */ +case class LeftSemiBuildLeftGuardRule(session: SparkSession) + extends Rule[SparkPlan] + with Logging { + + override def apply(plan: SparkPlan): SparkPlan = { + val skewJudgement = buildSkewJudgement() + plan.foreachUp { + case smj: SortMergeJoinExec if smj.joinType == LeftSemi => + maybeTag(smj, smj.right, "SMJ", skewJudgement) + case shj: ShuffledHashJoinExec if shj.joinType == LeftSemi => + maybeTag(shj, shj.right, "SHJ", skewJudgement) + case _ => + } + plan + } + + private def maybeTag( + join: SparkPlan, + rightSide: SparkPlan, + kind: String, + skewJudgement: ShuffleSkewDetector.SkewJudgement): Unit = { + if (join.getTagValue(RewriteJoin.ForceShjBuildLeftTag).getOrElse(false)) { + return + } + + val rightStats = ShuffleSkewDetector.analyze(rightSide, skewJudgement) + if (rightStats.isSkewed) { + logDebug( + s"LeftSemiBuildLeftGuardRule: right-side partition skew on LeftSemi $kind " + + s"(totalBytes=${rightStats.totalBytes}, max=${rightStats.maxBytes}, " + + s"median=${rightStats.medianBytes}); skipping BuildLeft.") + return + } + + join.setTagValue(RewriteJoin.ForceShjBuildLeftTag, true) + } + + private def buildSkewJudgement(): ShuffleSkewDetector.SkewJudgement = { + val sqlConf = SQLConf.get + ShuffleSkewDetector.SkewJudgement( + factor = sqlConf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_FACTOR), + partitionThresholdBytes = sqlConf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD), + minTotalBytes = 0L + ) + } +} diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala index 3d844607b39..7cb500834f6 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala @@ -20,6 +20,7 @@ import org.apache.gluten.backendsapi.BackendsApiManager import org.apache.gluten.config.GlutenConfig import org.apache.gluten.execution._ import org.apache.gluten.extension.columnar.FallbackTags +import org.apache.gluten.extension.columnar.rewrite.RewriteJoin import org.apache.gluten.logging.LogLevelUtil import org.apache.gluten.sql.shims.SparkShimLoader @@ -139,6 +140,11 @@ object OffloadJoin { return BuildLeft } + // Honor a ForceShjBuildLeft tag set by the guard rule (stats verified safe for BuildLeft). + if (shj.getTagValue(RewriteJoin.ForceShjBuildLeftTag).getOrElse(false)) { + return BuildLeft + } + // Both left and right are buildable. Find out the better one. if (!GlutenConfig.get.shuffledHashJoinOptimizeBuildSide) { // User disabled build side re-optimization. Return original build side from vanilla Spark. diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/RewriteJoin.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/RewriteJoin.scala index 6d456857d73..da3947361a6 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/RewriteJoin.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/RewriteJoin.scala @@ -21,11 +21,16 @@ import org.apache.gluten.extension.columnar.offload.OffloadJoin import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide, JoinSelectionHelper} import org.apache.spark.sql.catalyst.plans.logical.Join +import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec} /** If force ShuffledHashJoin, convert [[SortMergeJoinExec]] to [[ShuffledHashJoinExec]]. */ object RewriteJoin extends RewriteSingleNode with JoinSelectionHelper { + + val ForceShjBuildLeftTag: TreeNodeTag[Boolean] = + TreeNodeTag[Boolean]("org.apache.gluten.RewriteJoin.ForceShjBuildLeft") + override def isRewritable(plan: SparkPlan): Boolean = { plan match { case _: SortMergeJoinExec => true @@ -62,7 +67,7 @@ object RewriteJoin extends RewriteSingleNode with JoinSelectionHelper { case smj: SortMergeJoinExec if GlutenConfig.get.forceShuffledHashJoin => getSmjBuildSide(smj) match { case Some(buildSide) => - ShuffledHashJoinExec( + val shj = ShuffledHashJoinExec( smj.leftKeys, smj.rightKeys, smj.joinType, @@ -71,6 +76,10 @@ object RewriteJoin extends RewriteSingleNode with JoinSelectionHelper { smj.left, smj.right, smj.isSkewJoin) + if (smj.getTagValue(ForceShjBuildLeftTag).getOrElse(false)) { + shj.setTagValue(ForceShjBuildLeftTag, true) + } + shj case _ => plan } case _ => plan diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala new file mode 100644 index 00000000000..0e363083db9 --- /dev/null +++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala @@ -0,0 +1,89 @@ +/* + * 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.columnar.util + +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec + +object ShuffleSkewDetector { + + case class SkewJudgement( + factor: Double, + partitionThresholdBytes: Long, + minTotalBytes: Long) + + def analyze(side: SparkPlan, judgement: SkewJudgement): Result = { + val stages = side.collect { case s: ShuffleQueryStageExec => s } + if (stages.isEmpty) { + return Result(hasStage = false, materialized = false, isSkewed = false) + } + val materializedWithStats = stages.flatMap { + stage => if (stage.isMaterialized) stage.mapStats.map(ms => ms.bytesByPartitionId) else None + } + if (materializedWithStats.isEmpty) { + val anyMaterialized = stages.exists(_.isMaterialized) + return Result(hasStage = true, materialized = anyMaterialized, isSkewed = false) + } + val bytes = materializedWithStats.maxBy(_.foldLeft(0L)(_ + _)) + if (bytes.isEmpty) { + return Result(hasStage = true, materialized = true, isSkewed = false) + } + val nonEmpty = bytes.filter(_ > 0) + val total = bytes.foldLeft(0L)(_ + _) + if (nonEmpty.isEmpty) { + return Result( + hasStage = true, + materialized = true, + isSkewed = false, + totalBytes = total, + totalPartitions = bytes.length) + } + val sorted = nonEmpty.sorted + val median = sorted(sorted.length / 2) + val max = bytes.max + + val bigEnoughTotal = total >= judgement.minTotalBytes + val partitionSkewed = median > 0 && + max.toDouble > judgement.factor * median && + max > judgement.partitionThresholdBytes + + Result( + hasStage = true, + materialized = true, + isSkewed = bigEnoughTotal && partitionSkewed, + totalBytes = total, + nonEmptyPartitions = nonEmpty.length, + totalPartitions = bytes.length, + medianBytes = median, + maxBytes = max + ) + } + + case class Result( + hasStage: Boolean, + materialized: Boolean, + isSkewed: Boolean, + totalBytes: Long = 0L, + nonEmptyPartitions: Int = 0, + totalPartitions: Int = 0, + medianBytes: Long = 0L, + maxBytes: Long = 0L) { + + def skewRatio: Double = + if (medianBytes > 0) maxBytes.toDouble / medianBytes.toDouble else Double.NaN + } +} From d13369096db76ab3447958b900f1b968dc3ec0a5 Mon Sep 17 00:00:00 2001 From: huhengrui Date: Thu, 23 Jul 2026 21:42:17 +0000 Subject: [PATCH 3/4] Guard LeftSemi BuildLeft with an opt-in switch and size floor Enabling LeftSemi BuildLeft (commit 1) is only a win when the shuffle is large enough that the streamed-probe overhead on the right side is amortized by the hash-build savings on the left side. On smaller shuffles the physical plan is fragile and BuildLeft can regress. Add three Gluten-owned safeguards, all evaluated in LeftSemiBuildLeftGuardRule in order of increasing cost: 1. Master opt-in switch (default OFF): spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.enabled When false, VeloxBackend.supportHashBuildJoinTypeOnLeft rejects LeftSemi outright and the follow-on JoinUtils layout gate (leftSemiBuildLeftOutputsBuildOnly) returns false, so behavior is byte-identical to a Gluten build without this patch. Rationale: this optimization is deployment-sensitive (see 2 and 3); shipping it enabled-by-default would risk regressions on workloads that do not fit the profitable region. Users opt in explicitly. 2. Right-side size floor (default 10GB): spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightBytes BuildLeft moves the large right side onto the streamed-probe path, which carries fixed per-task overheads (row iteration, hash lookup, condition eval). Those overheads dominate on small shuffles, where even a small BuildLeft mis-decision can regress a fast plan. The 10GB default is aligned with Spark's own spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold (10GB), which expresses the same idea: a table is "big enough that a costly optimization pass is worth doing" only above this size. Reusing the same threshold keeps our judgement consistent with Spark's own precedent. 3. Ratio floor (default 10.0): spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightToLeftRatio BuildLeft's win depends on the hash-build savings on the left dominating the streamed-probe overhead on the right. When right/left is small (two sides close in size), the overhead outweighs the savings and BuildLeft regresses even without skew. Default 10.0 is chosen to be safely above the boundary at which the streamed-probe overhead is amortized by hash-build savings for typical row widths and per-task shuffle sizes: at ratio 10 the left is one order of magnitude smaller than the right, which reliably places BuildLeft in the profitable region. Setting it much smaller (near 1) admits marginal cases where the two sides are comparable and the streamed overhead can dominate; setting it much larger only foregoes optimization opportunities in the wide-ratio tail without gaining safety. Conceptually analogous to Spark's spark.sql.shuffledHashJoinFactor (which likewise expresses a size- ratio threshold in physical join selection), though at a different decision point (build-side choice within SHJ rather than SHJ-vs-SMJ). Evaluation order in LeftSemiBuildLeftGuardRule (from cheapest to most expensive): (1) rightBytes < minRightBytes -- one O(N) fold, short-circuit (2) right/left < minRatio -- one more O(N) fold on left (3) skew on right side -- O(N log N) sort for median Any check tagging the join triggers ForceShjBuildRightTag and short-circuits the remaining checks. If left/right shuffle stats are not yet materialized, the rule defers -- AQE re-runs it after subsequent shuffle stages complete. When the master switch is off, the rule returns the plan unchanged without walking it, so the runtime cost of shipping the rule on all Velox jobs is negligible. Co-authored-by: Claude Co-authored-by: Guo Wangyang Co-authored-by: Lipeng Zhu Signed-off-by: huhengrui --- .../backendsapi/velox/VeloxBackend.scala | 9 +-- .../LeftSemiBuildLeftGuardRule.scala | 60 ++++++++++++------- .../apache/gluten/config/GlutenConfig.scala | 33 ++++++++++ .../offload/OffloadSingleNodeRules.scala | 7 +++ .../columnar/util/ShuffleSkewDetector.scala | 18 ++++++ 5 files changed, 100 insertions(+), 27 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala index bdaf2adac2a..2bcad8b422a 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala @@ -506,13 +506,10 @@ object VeloxBackendSettings extends BackendSettingsApi { true } else { t match { - // OPPRO-266: For Velox backend, build right and left are both supported for - // LeftOuter. case LeftOuter => true - // Velox maps LeftSemi BuildLeft to kRightSemiFilter (via Substrait - // JOIN_TYPE_RIGHT_SEMI) and fully supports it in both HashBuild and HashProbe. - // The earlier restriction referenced Velox issue #9980 which has been resolved. - case LeftSemi => true + case LeftSemi + if GlutenConfig.get.shjLeftSemiBuildLeftEnabled && + SQLConf.get.adaptiveExecutionEnabled => true case _ => false } } diff --git a/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala b/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala index 6111f0b5d70..962e55ba7b2 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala @@ -16,6 +16,7 @@ */ package org.apache.gluten.extension +import org.apache.gluten.config.GlutenConfig import org.apache.gluten.extension.columnar.rewrite.RewriteJoin import org.apache.gluten.extension.columnar.util.ShuffleSkewDetector @@ -27,35 +28,23 @@ import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec} import org.apache.spark.sql.internal.SQLConf -/** - * QueryStagePrepRule that fires after each AQE shuffle stage completes, i.e. once `mapStats` are - * available. - * - * For a LeftSemi join (either [[SortMergeJoinExec]] or already-rewritten [[ShuffledHashJoinExec]]) - * it decides whether BuildLeft is safe and, if so, sets [[RewriteJoin.ForceShjBuildLeftTag]] on the - * join node. The tag is later consumed by: - * - [[RewriteJoin.getSmjBuildSide]] (SMJ path). - * - [[org.apache.gluten.extension.columnar.offload.OffloadJoin.getShjBuildSide]] (SHJ path). - * - * AQE `OptimizeSkewedJoin` cannot split the probe side of a ShuffledHashJoin, so a skewed probe - * would materialize as extreme straggler tasks. When the right (would-be probe) side is skewed, - * this rule refuses to tag the join for BuildLeft so that Velox builds the (skewed) side per - * partition. - * - * Skew thresholds reuse Spark's `spark.sql.adaptive.skewJoin.*` so we don't fight AQE's own - * definition of skew. - */ case class LeftSemiBuildLeftGuardRule(session: SparkSession) extends Rule[SparkPlan] with Logging { override def apply(plan: SparkPlan): SparkPlan = { + val glutenConf = GlutenConfig.get + if (!glutenConf.shjLeftSemiBuildLeftEnabled) { + return plan + } val skewJudgement = buildSkewJudgement() + val minRatio = glutenConf.shjLeftSemiBuildLeftMinRightToLeftRatio + val minRightBytes = glutenConf.shjLeftSemiBuildLeftMinRightBytes plan.foreachUp { case smj: SortMergeJoinExec if smj.joinType == LeftSemi => - maybeTag(smj, smj.right, "SMJ", skewJudgement) + maybeTag(smj, smj.left, smj.right, "SMJ", skewJudgement, minRatio, minRightBytes) case shj: ShuffledHashJoinExec if shj.joinType == LeftSemi => - maybeTag(shj, shj.right, "SHJ", skewJudgement) + maybeTag(shj, shj.left, shj.right, "SHJ", skewJudgement, minRatio, minRightBytes) case _ => } plan @@ -63,13 +52,42 @@ case class LeftSemiBuildLeftGuardRule(session: SparkSession) private def maybeTag( join: SparkPlan, + leftSide: SparkPlan, rightSide: SparkPlan, kind: String, - skewJudgement: ShuffleSkewDetector.SkewJudgement): Unit = { + skewJudgement: ShuffleSkewDetector.SkewJudgement, + minRatio: Double, + minRightBytes: Long): Unit = { if (join.getTagValue(RewriteJoin.ForceShjBuildLeftTag).getOrElse(false)) { return } + val rightTotalBytes = ShuffleSkewDetector.totalBytes(rightSide) match { + case Some(bytes) => bytes + case None => return + } + if (rightTotalBytes < minRightBytes) { + logDebug( + s"LeftSemiBuildLeftGuardRule: right-side too small on LeftSemi $kind " + + s"(rightBytes=$rightTotalBytes < minRightBytes=$minRightBytes); skipping BuildLeft.") + return + } + + val leftTotalBytes = ShuffleSkewDetector.totalBytes(leftSide) match { + case Some(bytes) => bytes + case None => return + } + if (leftTotalBytes > 0) { + val ratio = rightTotalBytes.toDouble / leftTotalBytes.toDouble + if (ratio < minRatio) { + logDebug( + f"LeftSemiBuildLeftGuardRule: insufficient right/left ratio on LeftSemi $kind " + + f"(rightBytes=$rightTotalBytes / leftBytes=$leftTotalBytes " + + f"= $ratio%.1f < minRatio=$minRatio%.1f); skipping BuildLeft.") + return + } + } + val rightStats = ShuffleSkewDetector.analyze(rightSide, skewJudgement) if (rightStats.isSkewed) { logDebug( diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 93fb3888e99..a658c696460 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala @@ -115,6 +115,15 @@ class GlutenConfig(conf: SQLConf) extends GlutenCoreConfig(conf) { def shuffledHashJoinOptimizeBuildSide: Boolean = getConf(COLUMNAR_SHUFFLED_HASH_JOIN_OPTIMIZE_BUILD_SIDE) + def shjLeftSemiBuildLeftEnabled: Boolean = + getConf(COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED) + + def shjLeftSemiBuildLeftMinRightBytes: Long = + getConf(COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES) + + def shjLeftSemiBuildLeftMinRightToLeftRatio: Double = + getConf(COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO) + def forceShuffledHashJoin: Boolean = getConf(COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED) def enableColumnarSortMergeJoin: Boolean = getConf(COLUMNAR_SORTMERGEJOIN_ENABLED) @@ -982,6 +991,30 @@ object GlutenConfig extends ConfigRegistry { .booleanConf .createWithDefault(true) + val COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED = + buildConf("spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.enabled") + .doc("Whether to allow BuildLeft for LeftSemi ShuffledHashJoin. When enabled, Velox " + + "maps LeftSemi BuildLeft to kRightSemiFilter and can build a hash table on the " + + "small left side, streaming the large right side as probe. Disabled by default " + + "because the optimization is only profitable on large workloads (see " + + "`minRightBytes` and `minRightToLeftRatio`).") + .booleanConf + .createWithDefault(false) + + val COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES = + buildConf("spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightBytes") + .doc("Minimum right-side shuffle total bytes for LeftSemi BuildLeft to activate.") + .bytesConf(ByteUnit.BYTE) + .createWithDefaultString("10GB") + + val COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO = + buildConf("spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightToLeftRatio") + .doc("Minimum right/left shuffle-total-bytes ratio required for LeftSemi BuildLeft. " + + "Default 10.0 covers the observed profitable region.") + .doubleConf + .checkValue(_ >= 1.0, "ratio must be >= 1") + .createWithDefault(10.0) + val COLUMNAR_SORTMERGEJOIN_ENABLED = buildConf("spark.gluten.sql.columnar.sortMergeJoin") .doc( diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala index 7cb500834f6..078bcc42c18 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala @@ -26,6 +26,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide} +import org.apache.spark.sql.catalyst.plans.LeftSemi import org.apache.spark.sql.catalyst.plans.logical.Join import org.apache.spark.sql.execution._ import org.apache.spark.sql.execution.RDDScanTransformer @@ -145,6 +146,12 @@ object OffloadJoin { return BuildLeft } + // When BuildLeft feature is enabled for LeftSemi but the guard didn't approve (no tag), + // prevent CBO from accidentally choosing BuildLeft without safety verification. + if (shj.joinType == LeftSemi && GlutenConfig.get.shjLeftSemiBuildLeftEnabled) { + return BuildRight + } + // Both left and right are buildable. Find out the better one. if (!GlutenConfig.get.shuffledHashJoinOptimizeBuildSide) { // User disabled build side re-optimization. Return original build side from vanilla Spark. diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala index 0e363083db9..1e61a3381c9 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala @@ -26,6 +26,24 @@ object ShuffleSkewDetector { partitionThresholdBytes: Long, minTotalBytes: Long) + def totalBytes(side: SparkPlan): Option[Long] = { + val stages = side.collect { case s: ShuffleQueryStageExec => s } + if (stages.isEmpty) return None + var total = 0L + var found = false + stages.foreach { + stage => + if (stage.isMaterialized) { + stage.mapStats.foreach { + ms => + total += ms.bytesByPartitionId.foldLeft(0L)(_ + _) + found = true + } + } + } + if (found) Some(total) else None + } + def analyze(side: SparkPlan, judgement: SkewJudgement): Result = { val stages = side.collect { case s: ShuffleQueryStageExec => s } if (stages.isEmpty) { From d200063dd2182f1c960784ff6bea4a646536478b Mon Sep 17 00:00:00 2001 From: huhengrui Date: Thu, 23 Jul 2026 21:42:27 +0000 Subject: [PATCH 4/4] Add AQE unit tests for LeftSemi BuildLeft guard rule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cover the four key config/runtime scenarios: - enabled + large ratio → BuildLeft chosen - right-side below minRightBytes → guard forces BuildRight - right/left ratio below minRightToLeftRatio → guard forces BuildRight - feature disabled → BuildRight preserved (no-op) Co-Authored-By: Claude Opus 4.6 Co-authored-by: Claude Co-authored-by: Guo Wangyang Co-authored-by: Lipeng Zhu Signed-off-by: huhengrui --- .../backendsapi/velox/VeloxBackend.scala | 3 + .../gluten/execution/VeloxHashJoinSuite.scala | 101 +++++++ docs/Configuration.md | 259 +++++++++--------- .../backendsapi/BackendSettingsApi.scala | 2 + .../apache/gluten/execution/JoinUtils.scala | 12 +- .../velox/VeloxAdaptiveQueryExecSuite.scala | 148 ++++++++++ 6 files changed, 395 insertions(+), 130 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala index 2bcad8b422a..128c69490c3 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala @@ -514,6 +514,9 @@ object VeloxBackendSettings extends BackendSettingsApi { } } } + + override def semiAntiBuildLeftOutputsBuildOnly: Boolean = true + override def supportHashBuildJoinTypeOnRight: JoinType => Boolean = { t => if (super.supportHashBuildJoinTypeOnRight(t)) { diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala index c5c8d23123d..c7c5f983e31 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala @@ -22,6 +22,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.spark.SparkConf import org.apache.spark.sql.Row import org.apache.spark.sql.catalyst.expressions.AttributeReference +import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight} import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec, ColumnarSubqueryBroadcastExec, InputIteratorTransformer, SerializedHashTableBroadcastRelation} import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.execution.joins.BuildSideRelation @@ -655,4 +656,104 @@ class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite { }) } + test("LeftSemi BuildLeft output correctness") { + withSQLConf( + ("spark.sql.autoBroadcastJoinThreshold", "-1"), + ("spark.sql.adaptive.enabled", "true"), + (GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true"), + (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key, "true"), + (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key, "0"), + (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key, "1.0") + ) { + withTable("semi_left", "semi_right") { + spark.sql(""" + CREATE TABLE semi_left USING PARQUET + AS SELECT id as key, id * 10 as value FROM range(100) + """) + spark.sql(""" + CREATE TABLE semi_right USING PARQUET + AS SELECT id * 2 as key FROM range(1000) + """) + + val query = "SELECT key, value FROM semi_left WHERE key IN (SELECT key FROM semi_right)" + runQueryAndCompare(query) { + df => + val plan = df.queryExecution.executedPlan + val shjJoins = plan.collect { + case shj: ShuffledHashJoinExecTransformer => shj + } + assert(shjJoins.nonEmpty, "Should use ShuffledHashJoin") + assert( + shjJoins.exists(_.joinBuildSide == BuildLeft), + "Should have at least one LeftSemi BuildLeft join") + } + } + } + } + + test("LeftSemi BuildRight output correctness (regression)") { + withSQLConf( + ("spark.sql.autoBroadcastJoinThreshold", "-1"), + ("spark.sql.adaptive.enabled", "true"), + (GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true"), + (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key, "false") + ) { + withTable("semi_left_br", "semi_right_br") { + spark.sql(""" + CREATE TABLE semi_left_br USING PARQUET + AS SELECT id as key, id * 10 as value FROM range(100) + """) + spark.sql(""" + CREATE TABLE semi_right_br USING PARQUET + AS SELECT id * 2 as key FROM range(1000) + """) + + val query = + "SELECT key, value FROM semi_left_br WHERE key IN (SELECT key FROM semi_right_br)" + runQueryAndCompare(query) { + df => + val plan = df.queryExecution.executedPlan + val shjJoins = plan.collect { + case shj: ShuffledHashJoinExecTransformer => shj + } + assert(shjJoins.nonEmpty, "Should use ShuffledHashJoin") + assert( + shjJoins.forall(_.joinBuildSide == BuildRight), + "All LeftSemi joins should use BuildRight when feature is disabled") + } + } + } + } + + test("LeftAnti with ShuffledHashJoin output correctness") { + withSQLConf( + ("spark.sql.autoBroadcastJoinThreshold", "-1"), + ("spark.sql.adaptive.enabled", "true"), + (GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true") + ) { + withTable("anti_left", "anti_right") { + spark.sql(""" + CREATE TABLE anti_left USING PARQUET + AS SELECT id as key, id * 10 as value FROM range(100) + """) + spark.sql(""" + CREATE TABLE anti_right USING PARQUET + AS SELECT id * 2 as key FROM range(1000) + """) + + val query = + """SELECT key, value FROM anti_left a + |WHERE NOT EXISTS (SELECT 1 FROM anti_right b WHERE a.key = b.key)""".stripMargin + runQueryAndCompare(query) { + df => + val plan = df.queryExecution.executedPlan + val shjJoins = plan.collect { + case shj: ShuffledHashJoinExecTransformer => shj + } + assert(shjJoins.nonEmpty, "Should use ShuffledHashJoin") + } + } + } + } + } diff --git a/docs/Configuration.md b/docs/Configuration.md index 0e4bc90b0c8..938c4394024 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -20,134 +20,137 @@ nav_order: 15 ## Gluten configurations -| Key | Modifiability | Default | Description | -|---------------------------------------------------------------------|---------------|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| spark.gluten.costModel | 🔄 Dynamic | legacy | The class name of user-defined cost model that will be used by Gluten's transition planner. If not specified, a legacy built-in cost model will be used. The legacy cost model helps RAS planner exhaustively offload computations, and helps transition planner choose columnar-to-columnar transition over others. | -| spark.gluten.enabled | 🔄 Dynamic | true | Whether to enable gluten. Default value is true. Just an experimental property. Recommend to enable/disable Gluten through the setting for spark.plugins. | -| spark.gluten.execution.resource.expired.time | 🔄 Dynamic | 86400 | Expired time of execution with resource relation has cached. | -| spark.gluten.expression.blacklist | 🔄 Dynamic | <undefined> | A black list of expression to skip transform, multiple values separated by commas. | -| spark.gluten.loadLibFromJar | 🔄 Dynamic | false | Whether to load shared libraries from jars. | -| spark.gluten.loadLibOS | 🔄 Dynamic | <undefined> | The shared library loader's OS name. | -| spark.gluten.loadLibOSVersion | 🔄 Dynamic | <undefined> | The shared library loader's OS version. | -| spark.gluten.memory.isolation | 🔄 Dynamic | false | Enable isolated memory mode. If true, Gluten controls the maximum off-heap memory can be used by each task to X, X = executor memory / max task slots. It's recommended to set true if Gluten serves concurrent queries within a single session, since not all memory Gluten allocated is guaranteed to be spillable. In the case, the feature should be enabled to avoid OOM. | -| spark.gluten.memory.overAcquiredMemoryRatio | 🔄 Dynamic | 0.3 | If larger than 0, Velox backend will try over-acquire this ratio of the total allocated memory as backup to avoid OOM. | -| spark.gluten.memory.reservationBlockSize | 🔄 Dynamic | 8MB | Block size of native reservation listener reserve memory from Spark. | -| spark.gluten.numTaskSlotsPerExecutor | 🔄 Dynamic | -1 | Must provide default value since non-execution operations (e.g. org.apache.spark.sql.Dataset#summary) doesn't propagate configurations using org.apache.spark.sql.execution.SQLExecution#withSQLConfPropagated | -| spark.gluten.shuffleWriter.bufferSize | 🔄 Dynamic | <undefined> | -| spark.gluten.soft-affinity.duplicateReading.maxCacheItems | 🔄 Dynamic | 10000 | Enable Soft Affinity duplicate reading detection | -| spark.gluten.soft-affinity.duplicateReadingDetect.enabled | 🔄 Dynamic | false | If true, Enable Soft Affinity duplicate reading detection | -| spark.gluten.soft-affinity.enabled | 🔄 Dynamic | false | Whether to enable Soft Affinity scheduling. | -| spark.gluten.soft-affinity.min.target-hosts | 🔄 Dynamic | 1 | For on HDFS, if there are already target hosts, and then prefer to use the original target hosts to schedule | -| spark.gluten.soft-affinity.replications.num | 🔄 Dynamic | 2 | Calculate the number of the replications for scheduling to the target executors per file | -| spark.gluten.sql.adaptive.costEvaluator.enabled | ⚓ Static | true | If true, use org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator as custom cost evaluator class, else follow the configuration spark.sql.adaptive.customCostEvaluatorClass. | -| spark.gluten.sql.ansiFallback.enabled | 🔄 Dynamic | true | When true (default), Gluten will fall back to Spark when ANSI mode is enabled. When false, Gluten will attempt to execute in ANSI mode. | -| spark.gluten.sql.cacheWholeStageTransformerContext | 🔄 Dynamic | false | When true, `WholeStageTransformer` will cache the `WholeStageTransformerContext` when executing. It is used to get substrait plan node and native plan string. | -| spark.gluten.sql.collapseGetJsonObject.enabled | 🔄 Dynamic | false | Collapse nested get_json_object functions as one for optimization. | -| spark.gluten.sql.columnar.appendData | 🔄 Dynamic | true | Enable or disable columnar v2 command append data. | -| spark.gluten.sql.columnar.arrowUdf | 🔄 Dynamic | true | Enable or disable columnar arrow udf. | -| spark.gluten.sql.columnar.batchscan | 🔄 Dynamic | true | Enable or disable columnar batchscan. | -| spark.gluten.sql.columnar.broadcastExchange | 🔄 Dynamic | true | Enable or disable columnar broadcastExchange. | -| spark.gluten.sql.columnar.broadcastJoin | 🔄 Dynamic | true | Enable or disable columnar broadcastJoin. | -| spark.gluten.sql.columnar.broadcastNestedLoopJoin.enabled | 🔄 Dynamic | true | Enable or disable columnar broadcastNestedLoopJoin. | -| spark.gluten.sql.columnar.cartesianProduct.enabled | 🔄 Dynamic | true | Enable or disable columnar cartesianProduct. | -| spark.gluten.sql.columnar.cast.avg | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.coalesce | 🔄 Dynamic | true | Enable or disable columnar coalesce. | -| spark.gluten.sql.columnar.collectLimit | 🔄 Dynamic | true | Enable or disable columnar collectLimit. | -| spark.gluten.sql.columnar.collectTail | 🔄 Dynamic | true | Enable or disable columnar collectTail. | -| spark.gluten.sql.columnar.enableNestedColumnPruningInHiveTableScan | 🔄 Dynamic | true | Enable or disable nested column pruning in hivetablescan. | -| spark.gluten.sql.columnar.enableVanillaVectorizedReaders | ⚓ Static | true | Enable or disable vanilla vectorized scan. | -| spark.gluten.sql.columnar.executor.libpath | 🔄 Dynamic || The gluten executor library path. | -| spark.gluten.sql.columnar.expand | 🔄 Dynamic | true | Enable or disable columnar expand. | -| spark.gluten.sql.columnar.fallback.expressions.threshold | 🔄 Dynamic | 50 | Fall back filter/project if number of nested expressions reaches this threshold, considering Spark codegen can bring better performance for such case. | -| spark.gluten.sql.columnar.fallback.ignoreRowToColumnar | 🔄 Dynamic | true | When true, the fallback policy ignores the RowToColumnar when counting fallback number. | -| spark.gluten.sql.columnar.fallback.preferColumnar | 🔄 Dynamic | true | When true, the fallback policy prefers to use Gluten plan rather than vanilla Spark plan if the both of them contains ColumnarToRow and the vanilla Spark plan ColumnarToRow number is not smaller than Gluten plan. | -| spark.gluten.sql.columnar.filescan | 🔄 Dynamic | true | Enable or disable columnar filescan. | -| spark.gluten.sql.columnar.filter | 🔄 Dynamic | true | Enable or disable columnar filter. | -| spark.gluten.sql.columnar.force.hashagg | 🔄 Dynamic | true | Whether to force to use gluten's hash agg for replacing vanilla spark's sort agg. | -| spark.gluten.sql.columnar.forceShuffledHashJoin | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.generate | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.hashagg | 🔄 Dynamic | true | Enable or disable columnar hashagg. | -| spark.gluten.sql.columnar.hivetablescan | 🔄 Dynamic | true | Enable or disable columnar hivetablescan. | -| spark.gluten.sql.columnar.libname | 🔄 Dynamic | gluten | The gluten library name. | -| spark.gluten.sql.columnar.libpath | 🔄 Dynamic || The gluten library path. | -| spark.gluten.sql.columnar.limit | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.maxBatchSize | 🔄 Dynamic | 4096 | -| spark.gluten.sql.columnar.overwriteByExpression | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite by expression. | -| spark.gluten.sql.columnar.overwritePartitionsDynamic | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite partitions dynamic. | -| spark.gluten.sql.columnar.parquet.write.blockSize | 🔄 Dynamic | 128MB | -| spark.gluten.sql.columnar.partial.generate | 🔄 Dynamic | true | Evaluates the non-offload-able HiveUDTF using vanilla Spark generator | -| spark.gluten.sql.columnar.partial.project | 🔄 Dynamic | true | Break up one project node into 2 phases when some of the expressions are non offload-able. Phase one is a regular offloaded project transformer that evaluates the offload-able expressions in native, phase two preserves the output from phase one and evaluates the remaining non-offload-able expressions using vanilla Spark projections | -| spark.gluten.sql.columnar.physicalJoinOptimizationLevel | 🔄 Dynamic | 12 | Fallback to row operators if there are several continuous joins. | -| spark.gluten.sql.columnar.physicalJoinOptimizationOutputSize | 🔄 Dynamic | 52 | Fallback to row operators if there are several continuous joins and matched output size. | -| spark.gluten.sql.columnar.physicalJoinOptimizeEnable | 🔄 Dynamic | false | Enable or disable columnar physicalJoinOptimize. | -| spark.gluten.sql.columnar.preferStreamingAggregate | 🔄 Dynamic | true | Velox backend supports `StreamingAggregate`. `StreamingAggregate` uses the less memory as it does not need to hold all groups in memory, so it could avoid spill. When true and the child output ordering satisfies the grouping key then Gluten will choose `StreamingAggregate` as the native operator. | -| spark.gluten.sql.columnar.project | 🔄 Dynamic | true | Enable or disable columnar project. | -| spark.gluten.sql.columnar.project.collapse | 🔄 Dynamic | true | Combines two columnar project operators into one and perform alias substitution | -| spark.gluten.sql.columnar.query.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether query will fall back by counting the number of ColumnarToRow & vanilla leaf node. | -| spark.gluten.sql.columnar.range | 🔄 Dynamic | true | Enable or disable columnar range. | -| spark.gluten.sql.columnar.replaceData | 🔄 Dynamic | true | Enable or disable columnar v2 command replace data. | -| spark.gluten.sql.columnar.scanOnly | 🔄 Dynamic | false | When enabled, only scan and the filter after scan will be offloaded to native. | -| spark.gluten.sql.columnar.shuffle | 🔄 Dynamic | true | Enable or disable columnar shuffle. | -| spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled | ⚓ Static | true | If enabled, fall back to ColumnarShuffleManager when celeborn service is unavailable.Otherwise, throw an exception. | -| spark.gluten.sql.columnar.shuffle.celeborn.useRssSort | 🔄 Dynamic | true | If true, use RSS sort implementation for Celeborn sort-based shuffle.If false, use Gluten's row-based sort implementation. Only valid when `spark.celeborn.client.spark.shuffle.writer` is set to `sort`. | -| spark.gluten.sql.columnar.shuffle.codec | 🔄 Dynamic | <undefined> | By default, the supported codecs are lz4 and zstd. When spark.gluten.sql.columnar.shuffle.codecBackend=qat,the supported codecs are gzip and zstd. | -| spark.gluten.sql.columnar.shuffle.codecBackend | 🔄 Dynamic | <undefined> | -| spark.gluten.sql.columnar.shuffle.compression.threshold | 🔄 Dynamic | 100 | If number of rows in a batch falls below this threshold, will copy all buffers into one buffer to compress. | -| spark.gluten.sql.columnar.shuffle.dictionary.enabled | 🔄 Dynamic | false | Enable dictionary in hash-based shuffle. | -| spark.gluten.sql.columnar.shuffle.merge.threshold | 🔄 Dynamic | 0.25 | -| spark.gluten.sql.columnar.shuffle.partitionBufferEvictThreshold | 🔄 Dynamic | -1 | For Velox hash shuffle writer, evict partition buffers larger than this threshold after splitting an input batch. Use non-positive value to disable this feature. | -| spark.gluten.sql.columnar.shuffle.readerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for shuffle reader reading input stream from local or remote. | -| spark.gluten.sql.columnar.shuffle.realloc.threshold | 🔄 Dynamic | 0.25 | -| spark.gluten.sql.columnar.shuffle.sort.columns.threshold | 🔄 Dynamic | 100000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of columns is greater than this threshold. | -| spark.gluten.sql.columnar.shuffle.sort.deserializerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for sort-based shuffle reader deserializing raw input to columnar batch. | -| spark.gluten.sql.columnar.shuffle.sort.partitions.threshold | 🔄 Dynamic | 4000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of partitions is greater than this threshold. | -| spark.gluten.sql.columnar.shuffle.typeAwareCompress.enabled | 🔄 Dynamic | false | Enable type-aware compression (e.g. FFor for 64-bit integers) in shuffle. Not compatible with dictionary encoding; if both are enabled, type-aware compression is automatically disabled. | -| spark.gluten.sql.columnar.shuffledHashJoin | 🔄 Dynamic | true | Enable or disable columnar shuffledHashJoin. | -| spark.gluten.sql.columnar.shuffledHashJoin.optimizeBuildSide | 🔄 Dynamic | true | Whether to allow Gluten to choose an optimal build side for shuffled hash join. | -| spark.gluten.sql.columnar.smallFileThreshold | 🔄 Dynamic | 0.5 | The total size threshold of small files in table scan.To avoid small files being placed into the same partition, Gluten will try to distribute small files into different partitions when the total size of small files is below this threshold. | -| spark.gluten.sql.columnar.sort | 🔄 Dynamic | true | Enable or disable columnar sort. | -| spark.gluten.sql.columnar.sortMergeJoin | 🔄 Dynamic | true | Enable or disable columnar sortMergeJoin. This should be set with preferSortMergeJoin=false. | -| spark.gluten.sql.columnar.tableCache | ⚓ Static | true | Enable or disable columnar table cache. | -| spark.gluten.sql.columnar.tableCache.partitionStats.enabled | 🔄 Dynamic | false | When true, the Velox columnar cache serializer computes per-partition min/max/null/row-count stats and embeds them in the cached payload so that the Spark optimizer can prune whole partitions on equality / range predicates. When false (default), the serializer still writes the V3 per-column payload with empty stats so projected cache reads can lazily materialize only requested columns, while partition pruning is disabled. | -| spark.gluten.sql.columnar.takeOrderedAndProject | 🔄 Dynamic | true | -| spark.gluten.sql.columnar.union | 🔄 Dynamic | true | Enable or disable columnar union. | -| spark.gluten.sql.columnar.wholeStage.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether whole stage will fall back in AQE supported case by counting the number of ColumnarToRow & vanilla leaf node. | -| spark.gluten.sql.columnar.window | 🔄 Dynamic | true | Enable or disable columnar window. | -| spark.gluten.sql.columnar.window.group.limit | 🔄 Dynamic | true | Enable or disable columnar window group limit. | -| spark.gluten.sql.columnar.writeToDataSourceV2 | 🔄 Dynamic | true | Enable or disable columnar v2 command write to data source v2. | -| spark.gluten.sql.columnarSampleEnabled | 🔄 Dynamic | false | Disable or enable columnar sample. | -| spark.gluten.sql.columnarToRowMemoryThreshold | 🔄 Dynamic | 64MB | -| spark.gluten.sql.countDistinctWithoutExpand | 🔄 Dynamic | false | Convert Count Distinct to a UDAF called count_distinct to prevent SparkPlanner converting it to Expand+Count. WARNING: When enabled, count distinct queries will fail to fallback!!! | -| spark.gluten.sql.extendedColumnPruning.enabled | 🔄 Dynamic | true | Do extended nested column pruning for cases ignored by vanilla Spark. | -| spark.gluten.sql.fallbackRegexpExpressions | 🔄 Dynamic | false | If true, fall back all regexp expressions. There are a few incompatible cases between RE2 (used by native engine) and java.util.regex (used by Spark). User should enable this property if their incompatibility is intolerable. | -| spark.gluten.sql.fallbackUnexpectedMetadataParquet | 🔄 Dynamic | false | If enabled, Gluten will not offload scan when unexpected metadata is detected. | -| spark.gluten.sql.fallbackUnexpectedMetadataParquet.limit | 🔄 Dynamic | 10 | If supplied, metadata of `limit` number of Parquet files will be checked to determine whether to fall back to java scan. | -| spark.gluten.sql.fallbackUnexpectedMetadataParquet.samplePercentage | 🔄 Dynamic | 0.1 | The percentage of root paths to sample for metadata validation when the number of root paths is large. Value range is (0, 1.0]. 1.0 means check all paths (no sampling). A smaller value reduces validation cost for tables with many partitions. | -| spark.gluten.sql.injectNativePlanStringToExplain | 🔄 Dynamic | false | When true, Gluten will inject native plan tree to Spark's explain output. | -| spark.gluten.sql.mergeTwoPhasesAggregate.enabled | 🔄 Dynamic | true | Whether to merge two phases aggregate if there are no other operators between them. | -| spark.gluten.sql.native.bloomFilter | 🔄 Dynamic | true | -| spark.gluten.sql.native.hive.writer.enabled | 🔄 Dynamic | true | This is config to specify whether to enable the native columnar writer for HiveFileFormat. Currently only supports HiveFileFormat with Parquet as the output file type. | -| spark.gluten.sql.native.hyperLogLog.Aggregate | 🔄 Dynamic | true | -| spark.gluten.sql.native.parquet.write.blockRows | 🔄 Dynamic | 100000000 | -| spark.gluten.sql.native.union | 🔄 Dynamic | false | Enable or disable native union where computation is completely offloaded to backend. | -| spark.gluten.sql.native.writeColumnMetadataExclusionList | 🔄 Dynamic | comment | Native write files does not support column metadata. Metadata in list would be removed to support native write files. Multiple values separated by commas. | -| spark.gluten.sql.native.writer.enabled | 🔄 Dynamic | <undefined> | This is config to specify whether to enable the native columnar parquet/orc writer | -| spark.gluten.sql.orc.charType.scan.fallback.enabled | 🔄 Dynamic | true | Force fallback for orc char type scan. | -| spark.gluten.sql.pushAggregateThroughJoin.enabled | 🔄 Dynamic | false | Enables the push-aggregate-through-join optimization in Gluten. When enabled, aggregate operators may be pushed below joins during logical optimization and corresponding physical plans may be rewritten to execute the aggregation earlier. | -| spark.gluten.sql.pushAggregateThroughJoin.maxDepth | 🔄 Dynamic | 2147483647 | Maximum join traversal depth when applying the push-aggregate-through-join optimization. A value of 1 allows pushing an aggregate through a single join; larger values allow the rule to traverse and push through multiple consecutive joins. | -| spark.gluten.sql.removeNativeWriteFilesSortAndProject | 🔄 Dynamic | true | When true, Gluten will remove the vanilla Spark V1Writes added sort and project for velox backend. | -| spark.gluten.sql.rewrite.dateTimestampComparison | 🔄 Dynamic | true | Rewrite the comparision between date and timestamp to timestamp comparison.For example `from_unixtime(ts) > date` will be rewritten to `ts > to_unixtime(date)` | -| spark.gluten.sql.scan.detailedMetrics.enabled | 🔄 Dynamic | true | When true (default), Velox backend scan operators register all detailed SQL metrics. When false, only essential scan metrics are registered to reduce driver memory usage. Also enabled automatically when spark.gluten.sql.debug is true. Does not affect the ClickHouse backend. | -| spark.gluten.sql.scan.fileSchemeValidation.enabled | 🔄 Dynamic | true | When true, enable file path scheme validation for scan. Validation will fail if file scheme is not supported by registered file systems, which will cause scan operator fall back. | -| spark.gluten.sql.supported.flattenNestedFunctions | 🔄 Dynamic | and,or | Flatten nested functions as one for optimization. | -| spark.gluten.sql.text.input.empty.as.default | 🔄 Dynamic | false | treat empty fields in CSV input as default values. | -| spark.gluten.sql.text.input.max.block.size | 🔄 Dynamic | 8KB | the max block size for text input rows | -| spark.gluten.sql.validation.printStackOnFailure | 🔄 Dynamic | false | -| spark.gluten.storage.hdfsViewfs.enabled | ⚓ Static | false | If enabled, gluten will convert the viewfs path to hdfs path in scala side | -| spark.gluten.supported.hive.udfs | 🔄 Dynamic || Supported hive udf names. | -| spark.gluten.supported.python.udfs | 🔄 Dynamic || Supported python udf names. | -| spark.gluten.supported.scala.udfs | 🔄 Dynamic || Supported scala udf names. | -| spark.gluten.ui.enabled | ⚓ Static | true | Whether to enable the gluten web UI, If true, attach the gluten UI page to the Spark web UI. | +| Key | Modifiability | Default | Description | +|-----------------------------------------------------------------------------------|---------------|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| spark.gluten.costModel | 🔄 Dynamic | legacy | The class name of user-defined cost model that will be used by Gluten's transition planner. If not specified, a legacy built-in cost model will be used. The legacy cost model helps RAS planner exhaustively offload computations, and helps transition planner choose columnar-to-columnar transition over others. | +| spark.gluten.enabled | 🔄 Dynamic | true | Whether to enable gluten. Default value is true. Just an experimental property. Recommend to enable/disable Gluten through the setting for spark.plugins. | +| spark.gluten.execution.resource.expired.time | 🔄 Dynamic | 86400 | Expired time of execution with resource relation has cached. | +| spark.gluten.expression.blacklist | 🔄 Dynamic | <undefined> | A black list of expression to skip transform, multiple values separated by commas. | +| spark.gluten.loadLibFromJar | 🔄 Dynamic | false | Whether to load shared libraries from jars. | +| spark.gluten.loadLibOS | 🔄 Dynamic | <undefined> | The shared library loader's OS name. | +| spark.gluten.loadLibOSVersion | 🔄 Dynamic | <undefined> | The shared library loader's OS version. | +| spark.gluten.memory.isolation | 🔄 Dynamic | false | Enable isolated memory mode. If true, Gluten controls the maximum off-heap memory can be used by each task to X, X = executor memory / max task slots. It's recommended to set true if Gluten serves concurrent queries within a single session, since not all memory Gluten allocated is guaranteed to be spillable. In the case, the feature should be enabled to avoid OOM. | +| spark.gluten.memory.overAcquiredMemoryRatio | 🔄 Dynamic | 0.3 | If larger than 0, Velox backend will try over-acquire this ratio of the total allocated memory as backup to avoid OOM. | +| spark.gluten.memory.reservationBlockSize | 🔄 Dynamic | 8MB | Block size of native reservation listener reserve memory from Spark. | +| spark.gluten.numTaskSlotsPerExecutor | 🔄 Dynamic | -1 | Must provide default value since non-execution operations (e.g. org.apache.spark.sql.Dataset#summary) doesn't propagate configurations using org.apache.spark.sql.execution.SQLExecution#withSQLConfPropagated | +| spark.gluten.shuffleWriter.bufferSize | 🔄 Dynamic | <undefined> | +| spark.gluten.soft-affinity.duplicateReading.maxCacheItems | 🔄 Dynamic | 10000 | Enable Soft Affinity duplicate reading detection | +| spark.gluten.soft-affinity.duplicateReadingDetect.enabled | 🔄 Dynamic | false | If true, Enable Soft Affinity duplicate reading detection | +| spark.gluten.soft-affinity.enabled | 🔄 Dynamic | false | Whether to enable Soft Affinity scheduling. | +| spark.gluten.soft-affinity.min.target-hosts | 🔄 Dynamic | 1 | For on HDFS, if there are already target hosts, and then prefer to use the original target hosts to schedule | +| spark.gluten.soft-affinity.replications.num | 🔄 Dynamic | 2 | Calculate the number of the replications for scheduling to the target executors per file | +| spark.gluten.sql.adaptive.costEvaluator.enabled | ⚓ Static | true | If true, use org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator as custom cost evaluator class, else follow the configuration spark.sql.adaptive.customCostEvaluatorClass. | +| spark.gluten.sql.ansiFallback.enabled | 🔄 Dynamic | true | When true (default), Gluten will fall back to Spark when ANSI mode is enabled. When false, Gluten will attempt to execute in ANSI mode. | +| spark.gluten.sql.cacheWholeStageTransformerContext | 🔄 Dynamic | false | When true, `WholeStageTransformer` will cache the `WholeStageTransformerContext` when executing. It is used to get substrait plan node and native plan string. | +| spark.gluten.sql.collapseGetJsonObject.enabled | 🔄 Dynamic | false | Collapse nested get_json_object functions as one for optimization. | +| spark.gluten.sql.columnar.appendData | 🔄 Dynamic | true | Enable or disable columnar v2 command append data. | +| spark.gluten.sql.columnar.arrowUdf | 🔄 Dynamic | true | Enable or disable columnar arrow udf. | +| spark.gluten.sql.columnar.batchscan | 🔄 Dynamic | true | Enable or disable columnar batchscan. | +| spark.gluten.sql.columnar.broadcastExchange | 🔄 Dynamic | true | Enable or disable columnar broadcastExchange. | +| spark.gluten.sql.columnar.broadcastJoin | 🔄 Dynamic | true | Enable or disable columnar broadcastJoin. | +| spark.gluten.sql.columnar.broadcastNestedLoopJoin.enabled | 🔄 Dynamic | true | Enable or disable columnar broadcastNestedLoopJoin. | +| spark.gluten.sql.columnar.cartesianProduct.enabled | 🔄 Dynamic | true | Enable or disable columnar cartesianProduct. | +| spark.gluten.sql.columnar.cast.avg | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.coalesce | 🔄 Dynamic | true | Enable or disable columnar coalesce. | +| spark.gluten.sql.columnar.collectLimit | 🔄 Dynamic | true | Enable or disable columnar collectLimit. | +| spark.gluten.sql.columnar.collectTail | 🔄 Dynamic | true | Enable or disable columnar collectTail. | +| spark.gluten.sql.columnar.enableNestedColumnPruningInHiveTableScan | 🔄 Dynamic | true | Enable or disable nested column pruning in hivetablescan. | +| spark.gluten.sql.columnar.enableVanillaVectorizedReaders | ⚓ Static | true | Enable or disable vanilla vectorized scan. | +| spark.gluten.sql.columnar.executor.libpath | 🔄 Dynamic || The gluten executor library path. | +| spark.gluten.sql.columnar.expand | 🔄 Dynamic | true | Enable or disable columnar expand. | +| spark.gluten.sql.columnar.fallback.expressions.threshold | 🔄 Dynamic | 50 | Fall back filter/project if number of nested expressions reaches this threshold, considering Spark codegen can bring better performance for such case. | +| spark.gluten.sql.columnar.fallback.ignoreRowToColumnar | 🔄 Dynamic | true | When true, the fallback policy ignores the RowToColumnar when counting fallback number. | +| spark.gluten.sql.columnar.fallback.preferColumnar | 🔄 Dynamic | true | When true, the fallback policy prefers to use Gluten plan rather than vanilla Spark plan if the both of them contains ColumnarToRow and the vanilla Spark plan ColumnarToRow number is not smaller than Gluten plan. | +| spark.gluten.sql.columnar.filescan | 🔄 Dynamic | true | Enable or disable columnar filescan. | +| spark.gluten.sql.columnar.filter | 🔄 Dynamic | true | Enable or disable columnar filter. | +| spark.gluten.sql.columnar.force.hashagg | 🔄 Dynamic | true | Whether to force to use gluten's hash agg for replacing vanilla spark's sort agg. | +| spark.gluten.sql.columnar.forceShuffledHashJoin | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.generate | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.hashagg | 🔄 Dynamic | true | Enable or disable columnar hashagg. | +| spark.gluten.sql.columnar.hivetablescan | 🔄 Dynamic | true | Enable or disable columnar hivetablescan. | +| spark.gluten.sql.columnar.libname | 🔄 Dynamic | gluten | The gluten library name. | +| spark.gluten.sql.columnar.libpath | 🔄 Dynamic || The gluten library path. | +| spark.gluten.sql.columnar.limit | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.maxBatchSize | 🔄 Dynamic | 4096 | +| spark.gluten.sql.columnar.overwriteByExpression | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite by expression. | +| spark.gluten.sql.columnar.overwritePartitionsDynamic | 🔄 Dynamic | true | Enable or disable columnar v2 command overwrite partitions dynamic. | +| spark.gluten.sql.columnar.parquet.write.blockSize | 🔄 Dynamic | 128MB | +| spark.gluten.sql.columnar.partial.generate | 🔄 Dynamic | true | Evaluates the non-offload-able HiveUDTF using vanilla Spark generator | +| spark.gluten.sql.columnar.partial.project | 🔄 Dynamic | true | Break up one project node into 2 phases when some of the expressions are non offload-able. Phase one is a regular offloaded project transformer that evaluates the offload-able expressions in native, phase two preserves the output from phase one and evaluates the remaining non-offload-able expressions using vanilla Spark projections | +| spark.gluten.sql.columnar.physicalJoinOptimizationLevel | 🔄 Dynamic | 12 | Fallback to row operators if there are several continuous joins. | +| spark.gluten.sql.columnar.physicalJoinOptimizationOutputSize | 🔄 Dynamic | 52 | Fallback to row operators if there are several continuous joins and matched output size. | +| spark.gluten.sql.columnar.physicalJoinOptimizeEnable | 🔄 Dynamic | false | Enable or disable columnar physicalJoinOptimize. | +| spark.gluten.sql.columnar.preferStreamingAggregate | 🔄 Dynamic | true | Velox backend supports `StreamingAggregate`. `StreamingAggregate` uses the less memory as it does not need to hold all groups in memory, so it could avoid spill. When true and the child output ordering satisfies the grouping key then Gluten will choose `StreamingAggregate` as the native operator. | +| spark.gluten.sql.columnar.project | 🔄 Dynamic | true | Enable or disable columnar project. | +| spark.gluten.sql.columnar.project.collapse | 🔄 Dynamic | true | Combines two columnar project operators into one and perform alias substitution | +| spark.gluten.sql.columnar.query.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether query will fall back by counting the number of ColumnarToRow & vanilla leaf node. | +| spark.gluten.sql.columnar.range | 🔄 Dynamic | true | Enable or disable columnar range. | +| spark.gluten.sql.columnar.replaceData | 🔄 Dynamic | true | Enable or disable columnar v2 command replace data. | +| spark.gluten.sql.columnar.scanOnly | 🔄 Dynamic | false | When enabled, only scan and the filter after scan will be offloaded to native. | +| spark.gluten.sql.columnar.shuffle | 🔄 Dynamic | true | Enable or disable columnar shuffle. | +| spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled | ⚓ Static | true | If enabled, fall back to ColumnarShuffleManager when celeborn service is unavailable.Otherwise, throw an exception. | +| spark.gluten.sql.columnar.shuffle.celeborn.useRssSort | 🔄 Dynamic | true | If true, use RSS sort implementation for Celeborn sort-based shuffle.If false, use Gluten's row-based sort implementation. Only valid when `spark.celeborn.client.spark.shuffle.writer` is set to `sort`. | +| spark.gluten.sql.columnar.shuffle.codec | 🔄 Dynamic | <undefined> | By default, the supported codecs are lz4 and zstd. When spark.gluten.sql.columnar.shuffle.codecBackend=qat,the supported codecs are gzip and zstd. | +| spark.gluten.sql.columnar.shuffle.codecBackend | 🔄 Dynamic | <undefined> | +| spark.gluten.sql.columnar.shuffle.compression.threshold | 🔄 Dynamic | 100 | If number of rows in a batch falls below this threshold, will copy all buffers into one buffer to compress. | +| spark.gluten.sql.columnar.shuffle.dictionary.enabled | 🔄 Dynamic | false | Enable dictionary in hash-based shuffle. | +| spark.gluten.sql.columnar.shuffle.merge.threshold | 🔄 Dynamic | 0.25 | +| spark.gluten.sql.columnar.shuffle.partitionBufferEvictThreshold | 🔄 Dynamic | -1 | For Velox hash shuffle writer, evict partition buffers larger than this threshold after splitting an input batch. Use non-positive value to disable this feature. | +| spark.gluten.sql.columnar.shuffle.readerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for shuffle reader reading input stream from local or remote. | +| spark.gluten.sql.columnar.shuffle.realloc.threshold | 🔄 Dynamic | 0.25 | +| spark.gluten.sql.columnar.shuffle.sort.columns.threshold | 🔄 Dynamic | 100000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of columns is greater than this threshold. | +| spark.gluten.sql.columnar.shuffle.sort.deserializerBufferSize | 🔄 Dynamic | 1MB | Buffer size in bytes for sort-based shuffle reader deserializing raw input to columnar batch. | +| spark.gluten.sql.columnar.shuffle.sort.partitions.threshold | 🔄 Dynamic | 4000 | The threshold to determine whether to use sort-based columnar shuffle. Sort-based shuffle will be used if the number of partitions is greater than this threshold. | +| spark.gluten.sql.columnar.shuffle.typeAwareCompress.enabled | 🔄 Dynamic | false | Enable type-aware compression (e.g. FFor for 64-bit integers) in shuffle. Not compatible with dictionary encoding; if both are enabled, type-aware compression is automatically disabled. | +| spark.gluten.sql.columnar.shuffledHashJoin | 🔄 Dynamic | true | Enable or disable columnar shuffledHashJoin. | +| spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.enabled | 🔄 Dynamic | false | Whether to allow BuildLeft for LeftSemi ShuffledHashJoin. When enabled, Velox maps LeftSemi BuildLeft to kRightSemiFilter and can build a hash table on the small left side, streaming the large right side as probe. Disabled by default because the optimization is only profitable on large workloads (see `minRightBytes` and `minRightToLeftRatio`). | +| spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightBytes | 🔄 Dynamic | 10GB | Minimum right-side shuffle total bytes for LeftSemi BuildLeft to activate. | +| spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightToLeftRatio | 🔄 Dynamic | 10.0 | Minimum right/left shuffle-total-bytes ratio required for LeftSemi BuildLeft. Default 10.0 covers the observed profitable region. | +| spark.gluten.sql.columnar.shuffledHashJoin.optimizeBuildSide | 🔄 Dynamic | true | Whether to allow Gluten to choose an optimal build side for shuffled hash join. | +| spark.gluten.sql.columnar.smallFileThreshold | 🔄 Dynamic | 0.5 | The total size threshold of small files in table scan.To avoid small files being placed into the same partition, Gluten will try to distribute small files into different partitions when the total size of small files is below this threshold. | +| spark.gluten.sql.columnar.sort | 🔄 Dynamic | true | Enable or disable columnar sort. | +| spark.gluten.sql.columnar.sortMergeJoin | 🔄 Dynamic | true | Enable or disable columnar sortMergeJoin. This should be set with preferSortMergeJoin=false. | +| spark.gluten.sql.columnar.tableCache | ⚓ Static | true | Enable or disable columnar table cache. | +| spark.gluten.sql.columnar.tableCache.partitionStats.enabled | 🔄 Dynamic | false | When true, the Velox columnar cache serializer computes per-partition min/max/null/row-count stats and embeds them in the cached payload so that the Spark optimizer can prune whole partitions on equality / range predicates. When false (default), the serializer still writes the V3 per-column payload with empty stats so projected cache reads can lazily materialize only requested columns, while partition pruning is disabled. | +| spark.gluten.sql.columnar.takeOrderedAndProject | 🔄 Dynamic | true | +| spark.gluten.sql.columnar.union | 🔄 Dynamic | true | Enable or disable columnar union. | +| spark.gluten.sql.columnar.wholeStage.fallback.threshold | 🔄 Dynamic | -1 | The threshold for whether whole stage will fall back in AQE supported case by counting the number of ColumnarToRow & vanilla leaf node. | +| spark.gluten.sql.columnar.window | 🔄 Dynamic | true | Enable or disable columnar window. | +| spark.gluten.sql.columnar.window.group.limit | 🔄 Dynamic | true | Enable or disable columnar window group limit. | +| spark.gluten.sql.columnar.writeToDataSourceV2 | 🔄 Dynamic | true | Enable or disable columnar v2 command write to data source v2. | +| spark.gluten.sql.columnarSampleEnabled | 🔄 Dynamic | false | Disable or enable columnar sample. | +| spark.gluten.sql.columnarToRowMemoryThreshold | 🔄 Dynamic | 64MB | +| spark.gluten.sql.countDistinctWithoutExpand | 🔄 Dynamic | false | Convert Count Distinct to a UDAF called count_distinct to prevent SparkPlanner converting it to Expand+Count. WARNING: When enabled, count distinct queries will fail to fallback!!! | +| spark.gluten.sql.extendedColumnPruning.enabled | 🔄 Dynamic | true | Do extended nested column pruning for cases ignored by vanilla Spark. | +| spark.gluten.sql.fallbackRegexpExpressions | 🔄 Dynamic | false | If true, fall back all regexp expressions. There are a few incompatible cases between RE2 (used by native engine) and java.util.regex (used by Spark). User should enable this property if their incompatibility is intolerable. | +| spark.gluten.sql.fallbackUnexpectedMetadataParquet | 🔄 Dynamic | false | If enabled, Gluten will not offload scan when unexpected metadata is detected. | +| spark.gluten.sql.fallbackUnexpectedMetadataParquet.limit | 🔄 Dynamic | 10 | If supplied, metadata of `limit` number of Parquet files will be checked to determine whether to fall back to java scan. | +| spark.gluten.sql.fallbackUnexpectedMetadataParquet.samplePercentage | 🔄 Dynamic | 0.1 | The percentage of root paths to sample for metadata validation when the number of root paths is large. Value range is (0, 1.0]. 1.0 means check all paths (no sampling). A smaller value reduces validation cost for tables with many partitions. | +| spark.gluten.sql.injectNativePlanStringToExplain | 🔄 Dynamic | false | When true, Gluten will inject native plan tree to Spark's explain output. | +| spark.gluten.sql.mergeTwoPhasesAggregate.enabled | 🔄 Dynamic | true | Whether to merge two phases aggregate if there are no other operators between them. | +| spark.gluten.sql.native.bloomFilter | 🔄 Dynamic | true | +| spark.gluten.sql.native.hive.writer.enabled | 🔄 Dynamic | true | This is config to specify whether to enable the native columnar writer for HiveFileFormat. Currently only supports HiveFileFormat with Parquet as the output file type. | +| spark.gluten.sql.native.hyperLogLog.Aggregate | 🔄 Dynamic | true | +| spark.gluten.sql.native.parquet.write.blockRows | 🔄 Dynamic | 100000000 | +| spark.gluten.sql.native.union | 🔄 Dynamic | false | Enable or disable native union where computation is completely offloaded to backend. | +| spark.gluten.sql.native.writeColumnMetadataExclusionList | 🔄 Dynamic | comment | Native write files does not support column metadata. Metadata in list would be removed to support native write files. Multiple values separated by commas. | +| spark.gluten.sql.native.writer.enabled | 🔄 Dynamic | <undefined> | This is config to specify whether to enable the native columnar parquet/orc writer | +| spark.gluten.sql.orc.charType.scan.fallback.enabled | 🔄 Dynamic | true | Force fallback for orc char type scan. | +| spark.gluten.sql.pushAggregateThroughJoin.enabled | 🔄 Dynamic | false | Enables the push-aggregate-through-join optimization in Gluten. When enabled, aggregate operators may be pushed below joins during logical optimization and corresponding physical plans may be rewritten to execute the aggregation earlier. | +| spark.gluten.sql.pushAggregateThroughJoin.maxDepth | 🔄 Dynamic | 2147483647 | Maximum join traversal depth when applying the push-aggregate-through-join optimization. A value of 1 allows pushing an aggregate through a single join; larger values allow the rule to traverse and push through multiple consecutive joins. | +| spark.gluten.sql.removeNativeWriteFilesSortAndProject | 🔄 Dynamic | true | When true, Gluten will remove the vanilla Spark V1Writes added sort and project for velox backend. | +| spark.gluten.sql.rewrite.dateTimestampComparison | 🔄 Dynamic | true | Rewrite the comparision between date and timestamp to timestamp comparison.For example `from_unixtime(ts) > date` will be rewritten to `ts > to_unixtime(date)` | +| spark.gluten.sql.scan.detailedMetrics.enabled | 🔄 Dynamic | true | When true (default), Velox backend scan operators register all detailed SQL metrics. When false, only essential scan metrics are registered to reduce driver memory usage. Also enabled automatically when spark.gluten.sql.debug is true. Does not affect the ClickHouse backend. | +| spark.gluten.sql.scan.fileSchemeValidation.enabled | 🔄 Dynamic | true | When true, enable file path scheme validation for scan. Validation will fail if file scheme is not supported by registered file systems, which will cause scan operator fall back. | +| spark.gluten.sql.supported.flattenNestedFunctions | 🔄 Dynamic | and,or | Flatten nested functions as one for optimization. | +| spark.gluten.sql.text.input.empty.as.default | 🔄 Dynamic | false | treat empty fields in CSV input as default values. | +| spark.gluten.sql.text.input.max.block.size | 🔄 Dynamic | 8KB | the max block size for text input rows | +| spark.gluten.sql.validation.printStackOnFailure | 🔄 Dynamic | false | +| spark.gluten.storage.hdfsViewfs.enabled | ⚓ Static | false | If enabled, gluten will convert the viewfs path to hdfs path in scala side | +| spark.gluten.supported.hive.udfs | 🔄 Dynamic || Supported hive udf names. | +| spark.gluten.supported.python.udfs | 🔄 Dynamic || Supported python udf names. | +| spark.gluten.supported.scala.udfs | 🔄 Dynamic || Supported scala udf names. | +| spark.gluten.ui.enabled | ⚓ Static | true | Whether to enable the gluten web UI, If true, attach the gluten UI page to the Spark web UI. | ## Gluten *experimental* configurations diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala index 85b146e1956..709481d01c0 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala @@ -86,6 +86,8 @@ trait BackendSettingsApi { case _ => false } + def semiAntiBuildLeftOutputsBuildOnly: Boolean = false + def supportHashBuildJoinTypeOnRight: JoinType => Boolean = { case _: InnerLike | LeftOuter | FullOuter | LeftSemi | LeftAnti | _: ExistenceJoin => true // LeftSingle is a Spark 4.0+ join type with same semantics as LeftOuter for build side. diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala index eeb60698902..459edb96149 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala @@ -16,6 +16,7 @@ */ package org.apache.gluten.execution +import org.apache.gluten.backendsapi.BackendsApiManager import org.apache.gluten.expression.{AttributeReferenceTransformer, ExpressionConverter} import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.SubstraitContext @@ -274,8 +275,15 @@ object JoinUtils { inputBuildOutput.indices.map(ExpressionBuilder.makeSelection(_)) :+ ExpressionBuilder.makeSelection(buildOutput.size) case LeftSemi | LeftAnti => - // When the left semi/anti join support the BuildLeft - leftOutput.indices.map(idx => ExpressionBuilder.makeSelection(idx + streamedOutput.size)) + // BuildLeft (exchangeTable=true) layout is backend-specific. Velox + // (kRightSemiFilter / kAnti) outputs only the build (= original-left) columns + // indexed from 0; ClickHouse keeps a concatenated [streamed ++ build] layout. + if (BackendsApiManager.getSettings.semiAntiBuildLeftOutputsBuildOnly) { + leftOutput.indices.map(ExpressionBuilder.makeSelection(_)) + } else { + leftOutput.indices.map( + idx => ExpressionBuilder.makeSelection(idx + streamedOutput.size)) + } case LeftExistence(_) => leftOutput.indices.map(ExpressionBuilder.makeSelection(_)) case _ => diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala index 6ccddac0c62..b02a2cf360e 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala @@ -1500,6 +1500,154 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT } } + testGluten("LeftSemi BuildLeft guard: enabled with large ratio chooses BuildLeft") { + withTempView("big", "small") { + spark.sparkContext + .parallelize((1 to 1000).map(i => TestData(i % 50, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("big") + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "1", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "2.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN big ON small.c1 = big.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildLeft) + } + } + } + + testGluten("LeftSemi BuildLeft guard: right-side too small forces BuildRight") { + withTempView("big", "small") { + spark.sparkContext + .parallelize((1 to 1000).map(i => TestData(i % 50, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("big") + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "100GB", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "1.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN big ON small.c1 = big.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildRight) + } + } + } + + testGluten("LeftSemi BuildLeft guard: insufficient ratio forces BuildRight") { + withTempView("small", "medium") { + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + spark.sparkContext + .parallelize((1 to 50).map(i => TestData(i % 10, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("medium") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "1", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "10.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN medium ON small.c1 = medium.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildRight) + } + } + } + + testGluten("LeftSemi BuildLeft guard: disabled config keeps BuildRight") { + withTempView("big", "small") { + spark.sparkContext + .parallelize((1 to 1000).map(i => TestData(i % 50, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("big") + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "false", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN big ON small.c1 = big.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildRight) + } + } + } + + testGluten("LeftSemi BuildLeft guard: probe-side skew forces BuildRight") { + withTempView("small", "large_skewed") { + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + spark.sparkContext + .parallelize( + (1 to 900).map(_ => TestData(1, "hot")) ++ + (1 to 100).map(i => TestData(i % 10 + 1, i.toString)), + 10) + .toDF("c1", "c2") + .createOrReplaceTempView("large_skewed") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "1", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "1.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true", + SQLConf.SKEW_JOIN_SKEWED_PARTITION_FACTOR.key -> "2", + SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "100" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN large_skewed ON small.c1 = large_skewed.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildRight) + } + } + } + testGluten("test log level") { def verifyLog(expectedLevel: Level): Unit = { val logAppender = new LogAppender("adaptive execution")