[VL] support build left when Leftsemi appears - #12513
Closed
hhr293 wants to merge 5 commits into
Closed
Conversation
|
Run Gluten Clickhouse CI on x86 |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR adds an opt-in optimization for the Velox backend to allow BuildLeft for LeftSemi ShuffledHashJoin when it’s beneficial, and introduces an AQE-time guard (via a QueryStagePrepRule) to force a safe fallback to BuildRight when shuffle stats indicate the optimization is likely to regress (too-small right side, insufficient size ratio, or probe-side partition skew). It also fixes backend-specific post-join projection behavior for semi/anti joins under BuildLeft.
Changes:
- Enable Velox LeftSemi SHJ BuildLeft behind a new opt-in config and AQE requirement, with a runtime guard that can tag joins to force BuildRight.
- Add shared shuffle skew detection utility and tag propagation through SMJ→SHJ rewrite and SHJ offload build-side selection.
- Fix LeftSemi/LeftAnti BuildLeft post-join output projection indexing via a backend capability flag, and add Velox AQE unit tests for the guard behavior.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala | Adds AQE tests covering enabled/disabled behavior, size/ratio gating, and probe-side skew fallback to BuildRight. |
| gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala | Introduces shuffle-stage stats inspection and skew detection utility used by the guard rule. |
| gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/RewriteJoin.scala | Adds a join tag to force BuildRight and propagates it across SMJ→SHJ rewrite. |
| gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala | Ensures SHJ build-side selection honors the force-BuildRight tag before re-optimizing build side. |
| gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala | Fixes backend-specific output projection indexing for LeftSemi/LeftAnti when BuildLeft is used. |
| gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala | Adds new configs for enabling and guarding LeftSemi BuildLeft (enable switch, min bytes, min ratio). |
| gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala | Adds backend capability flag for semi/anti BuildLeft output layout differences. |
| backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala | Adds AQE QueryStagePrepRule that inspects shuffle stats and tags unsafe LeftSemi joins to force BuildRight. |
| backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala | Injects the new QueryStagePrepRule into Velox’s Spark rule pipeline. |
| backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala | Enables LeftSemi BuildLeft support for Velox only when opt-in is enabled and AQE is on; sets the output-layout capability flag. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
hhr293
changed the title
Comment on lines
+139
to
+146
| 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 | ||
| ) | ||
| } |
Comment on lines
+1585
to
+1588
| 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" |
|
Run Gluten Clickhouse CI on x86 |
Comment on lines
+149
to
+153
| // When BuildLeft feature is enabled for LeftSemi but the guard didn't approve (no tag), | ||
| // prevent CBO from accidentally choosing BuildLeft without safety verification. | ||
| if (shj.joinType == LeftSemi && GlutenConfig.get.shjLeftSemiBuildLeftEnabled) { | ||
| return BuildRight | ||
| } |
Comment on lines
+29
to
+45
| def totalBytes(side: SparkPlan): Option[Long] = { | ||
| val stages = side.collect { case s: ShuffleQueryStageExec => s } | ||
| if (stages.isEmpty) return None | ||
| var total = 0L | ||
| var found = false | ||
| stages.foreach { | ||
| stage => | ||
| if (stage.isMaterialized) { | ||
| stage.mapStats.foreach { | ||
| ms => | ||
| total += ms.bytesByPartitionId.foldLeft(0L)(_ + _) | ||
| found = true | ||
| } | ||
| } | ||
| } | ||
| if (found) Some(total) else None | ||
| } |
Comment on lines
+48
to
+62
| val stages = side.collect { case s: ShuffleQueryStageExec => s } | ||
| if (stages.isEmpty) { | ||
| return Result(hasStage = false, materialized = false, isSkewed = false) | ||
| } | ||
| val materializedWithStats = stages.flatMap { | ||
| stage => if (stage.isMaterialized) stage.mapStats.map(ms => ms.bytesByPartitionId) else None | ||
| } | ||
| if (materializedWithStats.isEmpty) { | ||
| val anyMaterialized = stages.exists(_.isMaterialized) | ||
| return Result(hasStage = true, materialized = anyMaterialized, isSkewed = false) | ||
| } | ||
| val bytes = materializedWithStats.maxBy(_.foldLeft(0L)(_ + _)) | ||
| if (bytes.isEmpty) { | ||
| return Result(hasStage = true, materialized = true, isSkewed = false) | ||
| } |
|
Run Gluten Clickhouse CI on x86 |
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (2)
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala:45
ShuffleSkewDetector.totalBytessums MapOutputStatistics across allShuffleQueryStageExecnodes underside.collect, which can double-count bytes when the subtree contains nested/materialized stages (e.g., a top-level stage whose internal plan includes child stages). This can overestimate right/left totals and incorrectly tag joins as safe/profitable for BuildLeft.
Consider making totalBytes use only the top-level ShuffleQueryStageExec when side is a stage, and otherwise use a conservative aggregate (e.g., max of stage totals) rather than summing all stages.
def totalBytes(side: SparkPlan): Option[Long] = {
val stages = side.collect { case s: ShuffleQueryStageExec => s }
if (stages.isEmpty) return None
var total = 0L
var found = false
stages.foreach {
stage =>
if (stage.isMaterialized) {
stage.mapStats.foreach {
ms =>
total += ms.bytesByPartitionId.foldLeft(0L)(_ + _)
found = true
}
}
}
if (found) Some(total) else None
}
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala:51
ShuffleSkewDetector.analyzecollects allShuffleQueryStageExecnodes underside, so for a join child that is itself aShuffleQueryStageExec, skew detection may accidentally consider nested/child stages rather than the join-input stage. That can misclassify probe-side skew (either skipping BuildLeft unnecessarily or basing the decision on unrelated stages).
Prefer analyzing only the top-level stage when side is already a ShuffleQueryStageExec.
def analyze(side: SparkPlan, judgement: SkewJudgement): Result = {
val stages = side.collect { case s: ShuffleQueryStageExec => s }
if (stages.isEmpty) {
return Result(hasStage = false, materialized = false, isSkewed = false)
}
Velox maps LeftSemi BuildLeft to kRightSemiFilter (via Substrait JOIN_TYPE_RIGHT_SEMI) and fully supports it in both HashBuild and HashProbe. The Gluten-side restriction was added alongside a reference to Velox issue apache#9980, but that specific failure mode is no longer observable on any of the workloads we tested. Since we cannot reproduce the original issue and the enabled path exercises the same well-tested kRightSemiFilter operator that other join types already rely on, keeping the restriction is now an unnecessary loss of optimization opportunity. Enabling LeftSemi in supportHashBuildJoinTypeOnLeft lets the existing OffloadJoin.getShjBuildSide -> getOptimalBuildSide path choose the smaller input as the build side for LeftSemi joins, avoiding forcing the large fact table onto the build side. The follow-on commits add a QueryStagePrepRule guard against runtime skew and configuration knobs to gate this behavior for safe deployment. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
With LeftSemi BuildLeft enabled (previous commit), the optimizer moves the
smaller side to build. When the (now-probe) side is highly skewed, AQE's
OptimizeSkewedJoin cannot split a ShuffledHashJoin probe stage, and the
resulting stragglers can turn what would otherwise be a speedup into a
regression.
Add a skew guard that runs after each AQE shuffle stage materializes:
1. Shared utility: ShuffleSkewDetector inspects a shuffle stage's per-
partition byte distribution (via ShuffleQueryStageExec.mapStats) and
reports (isSkewed, stats) under a caller-supplied SkewJudgement
(factor, partitionThresholdBytes, minTotalBytes).
2. Detection thresholds reuse Spark's AQE definition of skew:
spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5)
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (256MB)
No Gluten-owned skew conf is introduced. Skew is a distribution
property, and Spark AQE already has a canonical definition users tune
via the AQE knobs; introducing a separate Gluten factor would produce
two truths about "what is a skewed partition" and drift from AQE.
3. LeftSemiBuildLeftGuardRule is a QueryStagePrepRule (Velox backend
only) that inspects both SortMergeJoinExec and ShuffledHashJoinExec
LeftSemi shapes and, when the right (probe) side is not skewed, tags
the join with RewriteJoin.ForceShjBuildLeftTag. Both shapes are
covered because Spark's JoinSelection and Gluten's RewriteJoin can
each produce either form before the rule runs.
4. RewriteJoin.getSmjBuildSide propagates the tag from the input SMJ to
the resulting ShuffledHashJoinExec.
5. OffloadJoin.getShjBuildSide returns BuildLeft early when the tag is
set, bypassing getOptimalBuildSide.
Non-skewed LeftSemi joins get BuildLeft from the guard; skewed ones fall
through to the standard build-side selection path.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
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 <noreply@anthropic.com>
Co-authored-by: Guo Wangyang <wangyang.guo@intel.com>
Co-authored-by: Lipeng Zhu <lipeng.zhu@intel.com>
Signed-off-by: huhengrui <hengrui.hu@intel.com>
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 <noreply@anthropic.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Guo Wangyang <wangyang.guo@intel.com> Co-authored-by: Lipeng Zhu <lipeng.zhu@intel.com> Signed-off-by: huhengrui <hengrui.hu@intel.com>
Contributor
Author
|
@zhouyuan Hi, could you help me review this pr please? |
|
Run Gluten Clickhouse CI on x86 |
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (1)
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/util/ShuffleSkewDetector.scala:41
totalBytescurrently sums MapOutputStatistics across allShuffleQueryStageExecnodes found underside. If the subtree contains multiple shuffle stages, this can overestimate the join-input shuffle size and make the guard rule’sminRightBytes/ ratio checks fire when they shouldn’t. Consider aligning this withanalyzeby using the materialized stage with the largest total bytes (or the topmost stage), instead of summing all stages.
stage.mapStats.foreach {
ms =>
total += ms.bytesByPartitionId.foldLeft(0L)(_ + _)
found = true
}
Resolve conflict in docs/Configuration.md by regenerating via dev/gen-all-config-docs.sh (spark-3.5, JDK11).
Contributor
Author
|
There is another way to reduce q95 latency just like pr12756 so close this PR |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Enable BuildLeft for LeftSemi
ShuffledHashJoinon the Velox backend, guarded by aQueryStagePrepRule and gated behind an opt-in switch.
Motivation
OffloadJoin.getShjBuildSide→getOptimalBuildSidecurrently cannot choose BuildLeftfor
LeftSemion Velox becausesupportHashBuildJoinTypeOnLeftreturnsfalseforLeftSemi. This forces the (typically larger) fact table onto the build side, producingoversized hash tables. Velox already supports
kRightSemiFiltervia SubstraitJOIN_TYPE_RIGHT_SEMI; the referenced upstream restriction (Velox issue #9980) has beenresolved. Removing the restriction lets the optimizer place the smaller side on build.
BuildLeft, however, is only profitable when:
OptimizeSkewedJoincannot split theprobe side of a
ShuffledHashJoin).To ship the optimization safely, this PR adds a
QueryStagePrepRulethat inspectsshuffle stats at AQE re-plan time and tags qualifying LeftSemi joins with
RewriteJoin.ForceShjBuildLeftTag = trueto signal that BuildLeft is safe.OffloadJoinhonors the tag: tag present → BuildLeft; tag absent (guard didn't approve)→ fall back to BuildRight. The whole feature is behind an opt-in switch and disabled by
default.
Commits
Enable BuildLeft for LeftSemi ShuffledHashJoin
Adds
LeftSemi => truetosupportHashBuildJoinTypeOnLeft(conditional on config +AQE enabled) and fixes the backend-specific post-join projection index in
JoinUtilsvia a new
BackendSettingsApi.semiAntiBuildLeftOutputsBuildOnlycapability flag(Velox = true, ClickHouse = false). Velox's
kRightSemiFilter/kAntioutputs onlybuild-side columns indexed from 0; without this flag the projection would read from
streamedOutput.size, producing wrong results.Guard LeftSemi BuildLeft against probe-side partition skew
New shared utility
ShuffleSkewDetector+ new Velox-only QueryStagePrepRuleLeftSemiBuildLeftGuardRulethat inspects shuffle stats and setsRewriteJoin.ForceShjBuildLeftTag = trueonly when all safety conditions pass.RewriteJoinpropagates the tag from SMJ → SHJ during join rewrite;OffloadJoinreads it to decide build side.Guard LeftSemi BuildLeft with an opt-in switch and size floor
Adds three Gluten configs and reorders the guard checks in ascending cost:
spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.enabled(bool, default false)spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightBytes(bytes, default 10GB)spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightToLeftRatio(double, default 10.0)Add AQE unit tests for LeftSemi BuildLeft guard rule
VeloxAdaptiveQueryExecSuite: guard rule activation / skip scenariosVeloxHashJoinSuite: LeftSemi BuildLeft output correctness, LeftSemi BuildRightregression, LeftAnti output correctness
Tag semantics
RewriteJoin.ForceShjBuildLeftTagis an opt-in allow signal:truefalseOffloadJoinforces BuildRight for LeftSemi (when feature enabled)When
leftSemi.buildLeft.enabled = false(default), the guard rule short-circuitswithout walking the plan, no tags are set, and
supportHashBuildJoinTypeOnLeftrejectsLeftSemioutright — behavior is byte-identical to a build without this patch.Configuration reference
leftSemi.buildLeft.enabledfalseleftSemi.buildLeft.minRightBytes10GBspark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold(10GB) — the same "big enough that a costly optimization is worth doing" gate.leftSemi.buildLeft.minRightToLeftRatio10.0spark.sql.shuffledHashJoinFactor, applied at build-side selection within SHJ. 10x guarantees a one order-of-magnitude size gap between the two sides so BuildLeft's hash-build savings dominate the streamed-probe overhead on the right.Benchmark results
TPC-DS and TPC-H on Velox backend, 36 executors × 8 cores, off-heap 32G, HDFS shuffle.
Patch mode:
enabled=true,minRightBytes=10GB(default),minRightToLeftRatio=10.0(default).Each cell is the mean of 3 full sweeps.
Total latency
TPC-DS 6TB impact
TPC-H 6TB impact
totalwithin 2% of baseline)