[VL] support build left when Leftsemi appears#12513
Open
hhr293 wants to merge 4 commits into
Open
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" |
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 apache#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 <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>
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 <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>
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>
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 forces BuildRight when any of the three
conditions above fail. The whole feature is behind an opt-in switch and disabled by
default.
Commits
[VL] Enable BuildLeft for LeftSemi ShuffledHashJoin
Adds
LeftSemi => truetosupportHashBuildJoinTypeOnLeftand fixes thebackend-specific post-join projection index in
JoinUtilsvia a newBackendSettingsApi.leftSemiBuildLeftOutputsBuildOnlycapability flag(Velox = true, ClickHouse = false).
[VL] Guard LeftSemi BuildLeft against probe-side partition skew
New shared utility
ShuffleSkewDetector+ new Velox-only QueryStagePrepRuleLeftSemiBuildLeftGuardRulethat tags LeftSemi joins withRewriteJoin.ForceShjBuildRightTagwhen the right-side shuffle is skewed(per Spark's own AQE
skewedPartitionFactor/skewedPartitionThresholdInBytes).RewriteJoinandOffloadJoinhonor the tag to fall back to BuildRight.[VL] 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, mirrors Spark'sapplicationSideScanSizeThreshold)spark.gluten.sql.columnar.shuffledHashJoin.leftSemi.buildLeft.minRightToLeftRatio(double, default 10.0)Behavior when disabled
With
leftSemi.buildLeft.enabled=false(the default),VeloxBackend.supportHashBuildJoinTypeOnLeftrejects
LeftSemioutright and the guard rule short-circuits toplanwithout walking it,so behavior is byte-identical to a Gluten 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)