From 387bef11242cb9d192082c7b5c95c929d7aeca09 Mon Sep 17 00:00:00 2001 From: huhengrui Date: Tue, 7 Jul 2026 19:02:33 -0700 Subject: [PATCH 1/4] Enable BuildLeft for LeftSemi ShuffledHashJoin Velox supports kRightSemiFilter (mapped from Substrait JOIN_TYPE_RIGHT_SEMI) 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 (TPC-DS and TPC-H at 1TB and 6TB) with the current Velox version. 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. Also fix the post-join projection index for LeftSemi/LeftAnti with BuildLeft (exchangeTable=true) in JoinUtils.createProjectRelPostJoinRel. The physical output layout is backend-specific: - Velox (kRightSemiFilter / kAnti): outputs only the build side (= original left) columns, indexed from 0. - ClickHouse: keeps a concatenated [streamed ++ build] layout, indexed from streamedOutput.size. Introduce a new BackendSettingsApi capability flag semiAntiBuildLeftOutputsBuildOnly (default false) and dispatch on it, so CH's existing path is untouched and only Velox picks up the new index. 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 Co-authored-by: Guo Wangyang Co-authored-by: Lipeng Zhu Signed-off-by: huhengrui --- .../gluten/backendsapi/velox/VeloxBackend.scala | 11 ++++++++--- .../gluten/backendsapi/BackendSettingsApi.scala | 13 +++++++++++++ .../org/apache/gluten/execution/JoinUtils.scala | 12 ++++++++++-- 3 files changed, 31 insertions(+), 5 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..c866a48d690 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,13 +508,18 @@ 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 } } } + + override def semiAntiBuildLeftOutputsBuildOnly: Boolean = true + override def supportHashBuildJoinTypeOnRight: JoinType => Boolean = { t => if (super.supportHashBuildJoinTypeOnRight(t)) { 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..621bc728c3b 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,19 @@ trait BackendSettingsApi { case _ => false } + /** + * Whether the backend's LeftSemi/LeftAnti join with BuildLeft (exchangeTable=true) outputs only + * the build-side columns indexed from 0. + * + * Velox's kRightSemiFilter / kAnti in this shape produces exactly the build (= original left) + * columns and nothing from the streamed side. The Substrait-level result projection must + * therefore index left columns from 0 rather than from `streamedOutput.size` (which is what a + * concatenated [streamed ++ build] output would require). + * + * ClickHouse currently keeps the concatenated layout, so it must return 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 _ => From 43ae866108e3faf1f10420476753de91c06119a6 Mon Sep 17 00:00:00 2001 From: huhengrui Date: Wed, 8 Jul 2026 07:23:09 -0700 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. Rationale: 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. Reusing the same knobs keeps the two decisions in sync. 3. LeftSemiBuildLeftGuardRule is a QueryStagePrepRule (Velox backend only) that inspects both SortMergeJoinExec and ShuffledHashJoinExec LeftSemi shapes and tags the join with RewriteJoin.ForceShjBuildRightTag when skew is detected. Both shapes are covered because Spark's JoinSelection and Gluten's RewriteJoin can each produce either form before the rule runs. 4. RewriteJoin.getSmjBuildSide honors the tag by returning BuildRight and propagates it to the resulting ShuffledHashJoinExec. 5. OffloadJoin.getShjBuildSide returns BuildRight early when the tag is set, bypassing getOptimalBuildSide. Non-skewed LeftSemi joins remain BuildLeft via getOptimalBuildSide, so the wins from the previous commit are preserved on well-shaped queries. Co-authored-by: Claude Co-authored-by: Guo Wangyang Co-authored-by: Lipeng Zhu Signed-off-by: huhengrui --- .../backendsapi/velox/VeloxRuleApi.scala | 6 + .../LeftSemiBuildLeftGuardRule.scala | 91 +++++++++++++ .../offload/OffloadSingleNodeRules.scala | 10 ++ .../columnar/rewrite/RewriteJoin.scala | 21 ++- .../columnar/util/ShuffleSkewDetector.scala | 127 ++++++++++++++++++ 5 files changed, 254 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..25dde8085e8 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,12 @@ object VeloxRuleApi { * injected through [[injectLegacy]]. */ private def injectSpark(injector: SparkInjector): Unit = { + // QueryStagePrepRule: fires after each AQE shuffle stage materializes, so mapStats are + // available. Guards LeftSemi BuildLeft against unsafe conditions (right-side too small, + // insufficient right/left ratio, or probe-side partition skew) by tagging the join to + // force BuildRight. + 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..34434e4b7d5 --- /dev/null +++ b/backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala @@ -0,0 +1,91 @@ +/* + * 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 unsafe and, if so, sets + * [[RewriteJoin.ForceShjBuildRightTag]] 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 flips build back to the right so Velox builds the (skewed) side per + * partition. + * + * Distribution 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 judgement = buildSkewJudgement() + plan.transformUp { + case smj: SortMergeJoinExec if smj.joinType == LeftSemi => + maybeTag(smj, smj.right, "SMJ", judgement) + smj + case shj: ShuffledHashJoinExec if shj.joinType == LeftSemi => + maybeTag(shj, shj.right, "SHJ", judgement) + shj + } + } + + private def maybeTag( + join: SparkPlan, + rightSide: SparkPlan, + kind: String, + judgement: ShuffleSkewDetector.SkewJudgement): Unit = { + if (join.getTagValue(RewriteJoin.ForceShjBuildRightTag).getOrElse(false)) { + // Already tagged (e.g. by RewriteJoin propagating from a tagged SMJ). Nothing to do. + return + } + val result = ShuffleSkewDetector.analyze(rightSide, judgement) + if (result.isSkewed) { + logInfo( + s"LeftSemiBuildLeftGuardRule: right-side partition skew on LeftSemi $kind " + + s"(totalBytes=${result.totalBytes}, max=${result.maxBytes}, " + + s"median=${result.medianBytes}); forcing BuildRight.") + join.setTagValue(RewriteJoin.ForceShjBuildRightTag, 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..ed7a4f3cb3b 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,11 +140,20 @@ object OffloadJoin { return BuildLeft } + // Safety override: honor a ForceShjBuildRight tag set by a QueryStagePrepRule (or propagated + // from an SMJ rewritten upstream). The rule already inspected shuffle stats and decided + // BuildLeft is unsafe (probe-side skew, right-side too small, or insufficient ratio). + // This check must precede the optimizeBuildSide gate so the safety tag is never bypassed. + if (shj.getTagValue(RewriteJoin.ForceShjBuildRightTag).getOrElse(false)) { + 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. return shj.buildSide } + shj.logicalLink .flatMap { case join: Join => Some(getOptimalBuildSide(join)) 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..edeeccbf68a 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,21 @@ 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 { + + /** + * Tag set by a QueryStagePrepRule when a LeftSemi join must fall back to BuildRight. Consumed by + * [[getSmjBuildSide]] here (SMJ -> SHJ rewrite path) and by + * [[org.apache.gluten.extension.columnar.offload.OffloadJoin.getShjBuildSide]] (offload). + */ + val ForceShjBuildRightTag: TreeNodeTag[Boolean] = + TreeNodeTag[Boolean]("org.apache.gluten.RewriteJoin.ForceShjBuildRight") + override def isRewritable(plan: SparkPlan): Boolean = { plan match { case _: SortMergeJoinExec => true @@ -34,6 +44,10 @@ object RewriteJoin extends RewriteSingleNode with JoinSelectionHelper { } private def getSmjBuildSide(join: SortMergeJoinExec): Option[BuildSide] = { + // Honor an explicit ForceShjBuildRight tag from a QueryStagePrepRule. + if (join.getTagValue(ForceShjBuildRightTag).getOrElse(false)) { + return if (canBuildShuffledHashJoinRight(join.joinType)) Some(BuildRight) else None + } val leftBuildable = canBuildShuffledHashJoinLeft(join.joinType) val rightBuildable = canBuildShuffledHashJoinRight(join.joinType) if (!leftBuildable && !rightBuildable) { @@ -62,7 +76,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 +85,11 @@ object RewriteJoin extends RewriteSingleNode with JoinSelectionHelper { smj.left, smj.right, smj.isSkewJoin) + // Propagate the tag so OffloadJoin also honors it. + if (smj.getTagValue(ForceShjBuildRightTag).getOrElse(false)) { + shj.setTagValue(ForceShjBuildRightTag, 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..c7cfcb9b1f3 --- /dev/null +++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala @@ -0,0 +1,127 @@ +/* + * 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 + +/** + * Utility for reading materialized shuffle-stage statistics and detecting partition skew. + * + * Distribution thresholds (factor, per-partition byte floor) come from Spark's own AQE + * `spark.sql.adaptive.skewJoin.*` configs -- the definition of "a shuffle partition is skewed" is + * the same one AQE uses to decide skew splits, so reusing it keeps the two decisions in sync. + * Callers layer their own [[SkewJudgement]] on top to add rule-specific total-size gating. + */ +object ShuffleSkewDetector { + + /** + * Per-rule thresholds. Distribution knobs (factor, partitionThresholdBytes) mirror Spark + * AQE and would normally be passed in from `SQLConf.get`. `minTotalBytes` is the caller's + * total-size floor: shuffles below this are cheap in absolute terms and are declared + * not-skewed regardless of ratio. Pass `0L` to disable the total-size gate. + */ + case class SkewJudgement( + factor: Double, + partitionThresholdBytes: Long, + minTotalBytes: Long) + + /** + * Walk the unary child chain from `side` and return the first (outermost) + * [[ShuffleQueryStageExec]] encountered. This avoids picking a nested/inner shuffle stage when + * multiple stages exist under a join side in complex AQE plans. + */ + private def findShuffleStage(side: SparkPlan): Option[ShuffleQueryStageExec] = { + side match { + case s: ShuffleQueryStageExec => Some(s) + case p if p.children.length == 1 => findShuffleStage(p.children.head) + case _ => None + } + } + + /** + * Analyze the shuffle stage under `side` and decide whether it is skewed under the given + * judgement. Returns a [[Result]] carrying both the decision and the raw stats, so callers + * can log consistently. + */ + def analyze(side: SparkPlan, judgement: SkewJudgement): Result = { + val stageOpt = findShuffleStage(side) + if (stageOpt.isEmpty) { + return Result(hasStage = false, materialized = false, isSkewed = false) + } + val stage = stageOpt.get + if (!stage.isMaterialized) { + return Result(hasStage = true, materialized = false, isSkewed = false) + } + stage.mapStats match { + case None => + Result(hasStage = true, materialized = true, isSkewed = false) + case Some(ms) => + val bytes = ms.bytesByPartitionId + if (bytes.isEmpty) { + return Result(hasStage = true, materialized = true, isSkewed = false) + } + val nonEmpty = bytes.filter(_ > 0) + val totalBytes = bytes.foldLeft(0L)(_ + _) + if (nonEmpty.isEmpty) { + return Result( + hasStage = true, + materialized = true, + isSkewed = false, + totalBytes = totalBytes, + totalPartitions = bytes.length) + } + val sorted = nonEmpty.sorted + val median = sorted(sorted.length / 2) + val max = bytes.max + + val bigEnoughTotal = totalBytes >= judgement.minTotalBytes + val partitionSkewed = median > 0 && + max.toDouble > judgement.factor * median && + max > judgement.partitionThresholdBytes + + Result( + hasStage = true, + materialized = true, + isSkewed = bigEnoughTotal && partitionSkewed, + totalBytes = totalBytes, + nonEmptyPartitions = nonEmpty.length, + totalPartitions = bytes.length, + medianBytes = median, + maxBytes = max + ) + } + } + + /** + * Result of [[analyze]]. Fields default to zero when the corresponding stage / stats are + * unavailable, so callers can log unconditionally. + */ + 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 b244bd1b4963d341d4a9e3db07ae46aab1af9bc1 Mon Sep 17 00:00:00 2001 From: huhengrui Date: Wed, 8 Jul 2026 07:24:39 -0700 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 | 8 +- .../LeftSemiBuildLeftGuardRule.scala | 102 +++++-- docs/Configuration.md | 259 +++++++++--------- .../apache/gluten/config/GlutenConfig.scala | 47 ++++ .../columnar/util/ShuffleSkewDetector.scala | 27 +- 5 files changed, 284 insertions(+), 159 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 c866a48d690..611b53f545c 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 @@ -511,8 +511,12 @@ object VeloxBackendSettings extends BackendSettingsApi { 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 + // Requires both opt-in config AND AQE enabled: the runtime guard rule + // (LeftSemiBuildLeftGuardRule) only fires under AQE, so without AQE there is + // no safety net against skew/size regressions. + 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 34434e4b7d5..86e1bb4633a 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 @@ -28,54 +29,109 @@ import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoin import org.apache.spark.sql.internal.SQLConf /** - * QueryStagePrepRule that fires after each AQE shuffle stage completes, i.e. once `mapStats` - * are available. + * 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 unsafe and, if so, sets - * [[RewriteJoin.ForceShjBuildRightTag]] on the join node. The tag is later consumed by: + * For a LeftSemi join (either [[SortMergeJoinExec]] or already-rewritten [[ShuffledHashJoinExec]]) + * it decides whether BuildLeft is unsafe and, if so, sets [[RewriteJoin.ForceShjBuildRightTag]] 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 flips build back to the right so Velox builds the (skewed) side per - * partition. + * Three reasons to reject BuildLeft (OR-ed together): + * - right-side shuffle too small (`< minRightBytes`): BuildLeft is only profitable at scale; on + * small shuffles the plan is fragile and BuildLeft can regress badly. + * - right-side partition skew: AQE `OptimizeSkewedJoin` cannot split the probe side of a + * ShuffledHashJoin, so a skewed probe materializes as straggler tasks. + * - insufficient size ratio: BuildLeft carries a fixed streamed-probe overhead on the right side. + * When right/left is small (the two sides are close in size), that overhead outweighs the + * hash-build win. Only when right >> left is BuildLeft profitable. * - * Distribution thresholds reuse Spark's `spark.sql.adaptive.skewJoin.*` so we don't fight - * AQE's own definition of skew. + * Skew thresholds reuse Spark's `spark.sql.adaptive.skewJoin.*` so we don't fight AQE's own + * definition of skew. The size ratio and right-side floor are Gluten-owned knobs + * (`minRightToLeftRatio` and `minRightBytes`). */ case class LeftSemiBuildLeftGuardRule(session: SparkSession) extends Rule[SparkPlan] with Logging { override def apply(plan: SparkPlan): SparkPlan = { - val judgement = buildSkewJudgement() - plan.transformUp { + // Fast path: if the feature is disabled, avoid the transformUp entirely. + 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", judgement) - smj + maybeTag(smj, smj.left, smj.right, "SMJ", skewJudgement, minRatio, minRightBytes) case shj: ShuffledHashJoinExec if shj.joinType == LeftSemi => - maybeTag(shj, shj.right, "SHJ", judgement) - shj + maybeTag(shj, shj.left, shj.right, "SHJ", skewJudgement, minRatio, minRightBytes) + case _ => } + plan } private def maybeTag( join: SparkPlan, + leftSide: SparkPlan, rightSide: SparkPlan, kind: String, - judgement: ShuffleSkewDetector.SkewJudgement): Unit = { + skewJudgement: ShuffleSkewDetector.SkewJudgement, + minRatio: Double, + minRightBytes: Long): Unit = { if (join.getTagValue(RewriteJoin.ForceShjBuildRightTag).getOrElse(false)) { // Already tagged (e.g. by RewriteJoin propagating from a tagged SMJ). Nothing to do. return } - val result = ShuffleSkewDetector.analyze(rightSide, judgement) - if (result.isSkewed) { - logInfo( + + // Step 1 (cheap, O(N) fold): right-side totalBytes. Force BuildRight when stats are + // unavailable -- without concrete byte counts we cannot verify safety. + val rightTotalBytes = ShuffleSkewDetector.totalBytes(rightSide) match { + case Some(bytes) => bytes + case None => + join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true) + return + } + if (rightTotalBytes < minRightBytes) { + logDebug( + s"LeftSemiBuildLeftGuardRule: right-side too small on LeftSemi $kind " + + s"(rightBytes=$rightTotalBytes < minRightBytes=$minRightBytes); " + + "forcing BuildRight.") + join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true) + return + } + + // Step 2 (cheap, O(N) fold): ratio guard. Force BuildRight when left stats unavailable. + val leftTotalBytes = ShuffleSkewDetector.totalBytes(leftSide) match { + case Some(bytes) => bytes + case None => + join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true) + 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); forcing BuildRight.") + join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true) + return + } + } + + // Step 3 (expensive, O(N log N) sort for median): skew analysis. Only reached when + // both size and ratio passed, so on regressive shapes (which we saw dominate q14a-like + // plans) the sort is skipped. + val rightStats = ShuffleSkewDetector.analyze(rightSide, skewJudgement) + if (rightStats.isSkewed) { + logDebug( s"LeftSemiBuildLeftGuardRule: right-side partition skew on LeftSemi $kind " + - s"(totalBytes=${result.totalBytes}, max=${result.maxBytes}, " + - s"median=${result.medianBytes}); forcing BuildRight.") + s"(totalBytes=${rightStats.totalBytes}, max=${rightStats.maxBytes}, " + + s"median=${rightStats.medianBytes}); forcing BuildRight.") join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true) } } diff --git a/docs/Configuration.md b/docs/Configuration.md index 0e4bc90b0c8..6d6d9db5d5c 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`); on smaller shuffles the plan may regress. Enable explicitly for TB-scale workloads where the LeftSemi joins are known to hit the profitable region. | +| spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightBytes | 🔄 Dynamic | 10GB | Minimum right-side shuffle total bytes for LeftSemi BuildLeft to activate. Below this size the optimizer's default decision is kept (typically BuildRight via SortMergeJoin). Aligned with Spark's own spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold (10GB), which defines the size above which a costly optimization pass is worth doing. On smaller scales (e.g. TPC-DS 1TB) LeftSemi BuildLeft has been observed to regress q14a/q14b by 10x; the 10GB gate keeps the patch dormant on those workloads. | +| spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightToLeftRatio | 🔄 Dynamic | 10.0 | Minimum right/left shuffle-total-bytes ratio required for LeftSemi BuildLeft. BuildLeft only wins when the left side is much smaller than the right, so the streamed-probe overhead on the right is amortized by the hash-build savings on the left. When right/left is below this ratio, that overhead outweighs the savings and BuildLeft regresses. Empirically on TPC-DS at 6TB, BuildLeft is profitable for q95 (ratio 57x), q94 (26x), q14a/b (5300x); it regresses when the ratio is <10x. 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/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 93fb3888e99..ab2910d2c42 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,44 @@ 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`); on smaller shuffles the plan may " + + "regress. Enable explicitly for TB-scale workloads where the LeftSemi joins are " + + "known to hit the profitable region.") + .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. " + + "Below this size the optimizer's default decision is kept (typically BuildRight " + + "via SortMergeJoin). Aligned with Spark's own " + + "spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold (10GB), " + + "which defines the size above which a costly optimization pass is worth doing. " + + "On smaller scales (e.g. TPC-DS 1TB) LeftSemi BuildLeft has been observed to " + + "regress q14a/q14b by 10x; the 10GB gate keeps the patch dormant on those " + + "workloads.") + .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. " + + "BuildLeft only wins when the left side is much smaller than the right, so the " + + "streamed-probe overhead on the right is amortized by the hash-build savings on " + + "the left. When right/left is below this ratio, that overhead outweighs the " + + "savings and BuildLeft regresses. Empirically on TPC-DS at 6TB, BuildLeft is " + + "profitable for q95 (ratio 57x), q94 (26x), q14a/b (5300x); it regresses when " + + "the ratio is <10x. 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/util/ShuffleSkewDetector.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala index c7cfcb9b1f3..a10545dd224 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 @@ -30,10 +30,10 @@ import org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec object ShuffleSkewDetector { /** - * Per-rule thresholds. Distribution knobs (factor, partitionThresholdBytes) mirror Spark - * AQE and would normally be passed in from `SQLConf.get`. `minTotalBytes` is the caller's - * total-size floor: shuffles below this are cheap in absolute terms and are declared - * not-skewed regardless of ratio. Pass `0L` to disable the total-size gate. + * Per-rule thresholds. Distribution knobs (factor, partitionThresholdBytes) mirror Spark AQE and + * would normally be passed in from `SQLConf.get`. `minTotalBytes` is the caller's total-size + * floor: shuffles below this are cheap in absolute terms and are declared not-skewed regardless + * of ratio. Pass `0L` to disable the total-size gate. */ case class SkewJudgement( factor: Double, @@ -53,10 +53,25 @@ object ShuffleSkewDetector { } } + /** + * Cheap probe: return the total shuffle bytes for the stage under `side`, skipping the sort / + * median / max computation that [[analyze]] does. Returns `None` when the stage is absent, not + * yet materialized, or materialized but `mapStats` is unavailable (stats unknown). Returns + * `Some(bytes)` only when a concrete byte count can be computed. + */ + def totalBytes(side: SparkPlan): Option[Long] = { + findShuffleStage(side) match { + case None => None + case Some(stage) => + if (!stage.isMaterialized) return None + stage.mapStats.map(ms => ms.bytesByPartitionId.foldLeft(0L)(_ + _)) + } + } + /** * Analyze the shuffle stage under `side` and decide whether it is skewed under the given - * judgement. Returns a [[Result]] carrying both the decision and the raw stats, so callers - * can log consistently. + * judgement. Returns a [[Result]] carrying both the decision and the raw stats, so callers can + * log consistently. */ def analyze(side: SparkPlan, judgement: SkewJudgement): Result = { val stageOpt = findShuffleStage(side) From 44b9b5fa6392fe09835f0236032aa3cf038b07b5 Mon Sep 17 00:00:00 2001 From: huhengrui Date: Fri, 10 Jul 2026 10:34:41 +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 --- .../velox/VeloxAdaptiveQueryExecSuite.scala | 161 ++++++++++++++++++ 1 file changed, 161 insertions(+) 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 6f1272b7f77..839e5b8115c 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 @@ -1502,6 +1502,167 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT } } + testGluten("LeftSemi BuildLeft guard: enabled with large ratio chooses BuildLeft") { + withTempView("big", "small") { + // big: 1000 rows across 10 partitions, small: 10 rows across 5 partitions. + // This creates a large right/left ratio so BuildLeft should activate. + 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", + // Set minRightBytes very high so guard forces BuildRight + 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") { + // small: 10 rows, medium: 50 rows. Right is larger so unguarded getOptimalBuildSide + // would pick BuildLeft (left is smaller). But ratio = 50/10 = 5x < minRatio(10.0), + // so the ratio guard forces BuildRight. + 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", + // Ratio ~5x won't meet this threshold + 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", + // Feature disabled (default) + 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") { + // small: 10 rows — will be the left (build) side candidate. + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + // large_skewed: 1000 rows, heavily skewed (900 rows key=1). Right side is much larger + // than left, so unguarded getOptimalBuildSide would pick BuildLeft (left is smaller). + // But the right-side skew should trigger the skew guard and force BuildRight. + 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", + // Set skew thresholds very low so the hot partition triggers skew detection + 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")