[SPARK-60458][SQL] Optimize skewed joins per join instead of all-or-nothing by ulysses-you · Pull Request #59703 · apache/spark

ulysses-you · 2026-10-22T09:34:12Z

What changes were proposed in this pull request?

OptimizeSkewedJoin applied every split, validated the whole plan once, and returned the plan unchanged when that validation failed. One join whose split broke a requirement therefore cost every skewed join in the plan the optimization.

This PR makes the strategy per join. The all-at-once attempt stays as the fast path (one candidate pass, one ValidateRequirements). When it fails and spark.sql.adaptive.forceOptimizeSkewedJoin is off, applySkewJoinsSatisfyingRequirements re-applies the candidates one at a time in the bottom-up order the plan is visited: a join keeps its split when the plan holding the kept splits and this one still satisfies every requirement. The internal spark.sql.adaptive.skewJoin.perJoinFallback.enabled (default true) turns that fallback off and restores the previous all-or-nothing behavior.

Implementation notes:

collectSplitSkewJoins gathers the candidates once, so the fallback reuses the splits the fast path built.
Candidates are matched by node identity: structurally equal joins in different places are separate candidates.
applySkewJoins walks top-down. A broadcast-hash join only constrains its streamed side, so another candidate can sit in its build side; replacing the outer join first keeps the inner candidate matchable, and the inner split is then applied inside the replacement.
Each trial is rebuilt from the original plan, so acceptance does not depend on how an earlier rebuild kept node instances.
The public OptimizeSkewedJoin.optimizeSkewJoin is removed. Its "split every join" behavior now lives in the fast path, and it had no in-tree caller.
The force path and the disabled path are unchanged.

CoalesceShufflePartitions changes too: it grouped a union's children into one coalesce group whenever the stamped decision said the union passes a layout through. That decision outlives the layout, so with one branch split the whole group gives up coalescing. The rule now asks the union what it currently reports as well, so its children coalesce independently once the union can only concatenate.

Why are the changes needed?

The rule was all-or-nothing. Two failure classes make the whole-plan validation fail:

a split introduces an extra shuffle, because the consumer above the join needs the rows on its key, and
an ancestor loses a partitioning claim, e.g. a UnionExec that was passing a branch's layout through reports UnknownPartitioning once that branch's join is split.

In both cases only the affected join is unsound; the other joins can keep their split. Plans where several joins share a stage used to get no optimization at all, and now they get the sound ones. The coalescing change keeps the stages under such a union coalesced, which the mixed split would otherwise have given up for every stage in the group.

Does this PR introduce any user-facing change?

Yes, in plan shape: an AQE query whose skewed joins previously all went unoptimized can now split the ones whose plan still validates, and the stages under a union that loses its layout stay coalesced. Results are unchanged, since a split preserves them and every accepted subset is checked with ValidateRequirements. spark.sql.adaptive.forceOptimizeSkewedJoin keeps its meaning.

How was this patch tested?

Three new tests in AdaptiveQueryExecSuite:

SPARK-60458: skip only the skewed join whose split the plan cannot take: a skewed join under a GROUP BY is rejected while the join feeding the union directly is split; the rejected join's two reads are asserted coalesced, and the result is checked to be exactly one aggregated row plus 334000 un-aggregated rows. An arm with the fallback config off asserts neither join is split.
SPARK-60458: keep the layout a union passes through: both union branches report the join's hash layout and a window above depends on it, so neither join is split; removing the window keeps both splits. Both arms check the 668000 result rows and their value/rank distributions. Both arms also pass on the previous rule, and the first is a guard that the per-join fallback keeps no split the claim cannot take.
SPARK-60458: split a join nested in a broadcast join's build side: a plan-level test that nests a sort-merge join in a broadcast join's build side and checks both splits survive; it fails on the previous bottom-up walk.

The third case of SPARK-37652 now expects 1 of 3 joins optimized, up from 0.

Commands:

build/sbt 'sql/testOnly *AdaptiveQueryExecSuite'
build/sbt sql/scalastyle

Full suite 149/149, scalastyle clean.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (deepseek-flash[1m])

🤖 Generated with Claude Code

…othing `OptimizeSkewedJoin` applied every split and validated the whole plan once. When that failed, it returned the plan unchanged, so a single join whose split broke a requirement cost every skewed join in the plan the optimization. When the all-at-once attempt fails, apply the splits one at a time in the bottom-up order the plan is visited: a join keeps its split when the plan holding the kept splits and this one still satisfies every requirement. `spark.sql.adaptive.forceOptimizeSkewedJoin` keeps its meaning, and the common case still costs one validation. Candidates are matched by node identity, since structurally equal joins in different places are separate candidates, and each trial is rebuilt from the original plan so acceptance does not depend on how an earlier rebuild kept node instances. Assisted-by: deepseek-flash[1m]

spark-test

Review summary

The change does what it describes for the common shapes. Each accepted subset is still checked with the same ValidateRequirements (including the required-distribution variant) the previous rule used, so the fallback adds no shuffle and leaves results unchanged. The new tests pin which join is split, not just how many.

One behavioral gap is in candidate substitution. Matching by node identity inside transformUp drops the split of a broadcast-hash join whose build side contains another skewed-join candidate. This only applies when the local shuffle reader is disabled. The previous rule split both joins in that shape, and the scaladoc's no-nesting claim does not hold for the BHJ arms. Separately, one new test comment refers to 'Base', which will not mean anything after merge. A question about the fallback's planning cost on plans with many skewed joins follows.

Cross-cutting observations

The dropped broadcast-join split and the planning-cost question both come from how applySkewJoins replays the collected candidates: a whole-plan rebuild with a linear identity lookup at every node, repeated once per candidate in the fallback. Reworking that matching to fix the dropped split is a natural moment to switch the lookup to an identity map, which removes one factor of the candidate count at no behavior cost.
Separation: The matching fix is needed whatever the answer on candidate counts, and the cost question still needs workload input after it, so this shared code path is context and does not merge their ownership or repair scope.

Findings

2 total: 0 P0, 0 P1, 0 P2, 2 P3.

Nit (P3)

Identity matching drops the split of a broadcast-hash join whose build side holds another candidate — sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala:282 — see inline.
Test comment refers to 'Base', which has no referent after merge — sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:1991 — see inline.

Decision challenges

Is the per-candidate fallback's driver cost bounded for plans with many skewed joins?

Verified: when the all-at-once split fails and force is off, applySkewJoinsSatisfyingRequirements rebuilds the whole plan from the original once per candidate (applySkewJoins(plan, accepted :+ candidate)) and runs a full ValidateRequirements on each trial. Each rebuild does a linear candidates.find at every node, and the loop ends with one more full rebuild. With K candidates and N nodes that is roughly O(K^2 * N) per re-optimization, which runs after every stage completion, where the previous give-up path did one validation. It is reached whenever a split breaks a shared requirement, e.g. a window or aggregate over a wide UNION ALL of skewed join branches, even when nothing is kept in the end. Not verified: whether supported workloads reach candidate counts where this matters. Executor behavior is unaffected. Can you confirm that the 'handful of skewed joins' assumption holds for the workloads you expect? If not, two changes would remove one factor of K with no behavior change: an identity map (java.util.IdentityHashMap) instead of the linear find, and returning the last accepted trial instead of rebuilding it. A candidate cap or an early bail-out would bound the rest.

Generated by Omnigent on Databricks

`applySkewJoins` matched candidates by node identity while walking bottom-up. A sort-merge or shuffled-hash candidate has shuffle stages as children and contains nothing else, but a broadcast-hash join only constrains its streamed side, so another candidate can sit in its build side. Walking bottom-up replaced the inner join first, which rebuilt the outer join into an instance the matcher no longer knew; the outer split was dropped, and its prebuilt replacement would have carried the original build subtree anyway. Walk top-down instead: the outer join is replaced while it is still the node the plan holds, and the inner split is then applied inside the replacement. The new plan-level test nests a sort-merge join in a broadcast join's build side and checks that both splits survive. Assisted-by: deepseek-flash[1m]

ulysses-you · 2026-10-22T11:59:07Z

cc @cloud-fan @peter-toth @dongjoon-hyun thank you

dongjoon-hyun · 2026-10-22T12:33:48Z

Thank you for pinging me, @ulysses-you .

dongjoon-hyun

Thank you for the PR, @ulysses-you. I reviewed the latest commit, 351074c, and left 9 inline comments. The fallback keeps only subsets that pass the same ValidateRequirements check as before, and I found no plan where it changes results. Item 1 is the one trade-off I found: in the shape of the first new test, keeping one union branch's split turns partition coalescing off for the whole union. Here is a summary, ordered by priority.

Performance

OptimizeSkewedJoin L345: when the fallback keeps one union branch's split and rejects another, CoalesceShufflePartitions skips coalescing for every stage under that union, which were coalesced before this PR.

API and design

OptimizeSkewedJoin L344: there is no internal conf to get the all-or-nothing behavior back, so the only way out is turning skew join handling off entirely.

Tests

AdaptiveQueryExecSuite L1943: aggregateJoins.size == 1 holds only because CombineAdjacentAggregation merges the partial and final aggregates over the join.
AdaptiveQueryExecSuite L1951: the skewed-read check re-implements AQEShuffleReadExec.hasSkewedPartition.
AdaptiveQueryExecSuite L1939: the inline collects re-implement findTopLevelAggregate and findTopLevelSortMergeJoin.
AdaptiveQueryExecSuite L1934: each query runs once more to check rows that runAdaptiveAndVerifyResult already compared with the non-AQE answer.

Docs and comments

OptimizeSkewedJoin L276: three new doc comments are inaccurate or stale (L215, L260 and L276).
PR description: it predates 351074c, so it lists two new tests and 148/148 and does not mention the top-down walk (posted on AdaptiveQueryExecSuite L2016).

Cleanup

OptimizeSkewedJoin L361: SkewJoinCandidate.joinType repeats join.joinType, and the only log that reads it cannot tell two skipped joins apart.

Item 1 is the most important, and item 2 is a design question. Item 8 only needs the PR description, which becomes the commit message.

dongjoon-hyun · 2026-10-22T14:12:50Z

+      stripSkewJoinChildWrappers(ensureRequirements.apply(optimized))
    } else {
-      plan
+      applySkewJoinsSatisfyingRequirements(plan, candidates)


When this fallback keeps one union branch's split and rejects another, CoalesceShufflePartitions skips coalescing for every stage under that union, while before this PR nothing there was split and all of them were coalesced. Take the first new test's query. StampUnionDecisions runs before this rule (AdaptiveSparkPlanExec.scala:155 and 175), when both branches still report hashpartitioning(key1), so the union is stamped non-plain. Here join2's split is kept and join1's is rejected, so the union then reports UnknownPartitioning and unionRDDs concatenates. But childrenNeedCompatiblePartitioning reads the stamp (case u: UnionExec => !u.isPlainUnion at CoalesceShufflePartitions.scala:189, which reads it since SPARK-59122). So join1's two stages, which carry no partition specs, and join2's two skew reads form one coalesce group, and coalescePartitionsWithSkew logs "Could not apply partition coalescing because of missing MapOutputStatistics or shuffle partition specs." and gives up (ShufflePartitionsUtil.scala:135-140). Results are unchanged, but join1's branch and join2's non-skewed partitions now run at the full shuffle partition count, while join1, still unsplit, keeps its straggler. The third case of SPARK-37652 takes the same path. Before this PR, a union could already reach this with a split join next to a branch that was never skewed, but now the shape this PR targets reaches it too. Only master and branch-4.x are affected, since branch-4.2 and earlier have case _: UnionExec => false there. I traced this by reading the code and did not run it. Could CoalesceShufflePartitions group a union's children only while the union still reports a real partitioning, e.g. case u: UnionExec => !u.isPlainUnion && !u.outputPartitioning.isInstanceOf[UnknownPartitioning] as its comment at CoalesceShufflePartitions.scala:185-188 describes, here or in a separate JIRA, with the first test asserting that join1's children get an AQEShuffleReadExec?

Fixed in 98db648. CoalesceShufflePartitions.childrenNeedCompatiblePartitioning now also reads u.outputPartitioning, so a union whose layout a split took away coalesces its children independently instead of giving up for the whole group. The condition short-circuits on plain unions, so they do not pay the extra rawPartitioning. The first test asserts the rejected join's two reads are coalesced; with the old predicate it fails with List() had size 0 instead of expected size 2.

dongjoon-hyun · 2026-10-22T14:12:50Z

-        case SkewJoinChildWrapper(child) => child
-      }
+      stripSkewJoinChildWrappers(ensureRequirements.apply(optimized))
    } else {


Minor: This branch used to return plan, and nothing turns the new fallback off. Every AQE query whose all-at-once split fails validation with spark.sql.adaptive.forceOptimizeSkewedJoin off now gets a different plan at runtime, with trade-offs such as the coalescing one at L345. If the fallback misbehaves, the only way back is spark.sql.adaptive.skewJoin.enabled=false, which also drops the splits the old rule made, and spark.sql.adaptive.optimizer.excludedRules cannot exclude this rule because it filters only AQEOptimizer's logical rules (AQEOptimizer.scala:50-61). Neighboring rules have such a switch, e.g. spark.sql.execution.combineAdjacentAggregation, and SPARK-52921 added spark.sql.unionOutputPartitioning for a plan-shape change like this one. Could an internal conf, e.g. spark.sql.adaptive.skewJoin.perJoinFallback.enabled with default true and version 4.4.0, guard this branch and return plan when it is off?

Added in 98db648: spark.sql.adaptive.skewJoin.perJoinFallback.enabled, .internal(), default true, version("4.4.0"). Turning it off returns plan, as this branch did before the PR, and the first test runs its query with the config off to assert neither join is split.

dongjoon-hyun · 2026-10-22T14:12:51Z

+          case agg: BaseAggregateExec =>
+            collect(agg) { case j: SortMergeJoinExec => j }
+        }.flatten
+        assert(aggregateJoins.size == 1)


Minor: This holds only because CombineAdjacentAggregation merges the partial and final aggregates over join1 into one Complete aggregate. Without that merge, both aggregates sit above join1, and the nested collect at L1939-1942 returns join1 once for each. So with the internal spark.sql.execution.combineAdjacentAggregation off, or after any change that stops that merge, this fails although the rule did the right thing. I checked this by reading the code and did not run it. Could this take only the topmost aggregate, e.g. collectFirst(adaptivePlan) { case agg: BaseAggregateExec => findTopLevelSortMergeJoin(agg) }.getOrElse(Nil), or dedupe by identity?

Changed in 98db648 to collect { case agg: BaseAggregateExec => findTopLevelSortMergeJoin(agg) }.flatten.distinct. Verified by running the test with spark.sql.execution.combineAdjacentAggregation=false: the previous form failed, the new form passes.

dongjoon-hyun · 2026-10-22T14:12:51Z

+        assert(unionJoins.head.isSkewJoin)
+        assert(collect(unionJoins.head) {
+          case r: AQEShuffleReadExec => r
+        }.exists(_.partitionSpecs.exists(_.isInstanceOf[PartialReducerPartitionSpec])))


Nit: This is AQEShuffleReadExec.hasSkewedPartition (AQEShuffleReadExec.scala:144-145), which this suite already uses, e.g. at L2605 and L2660. Could this be }.exists(_.hasSkewedPartition))?

Switched to hasSkewedPartition in 98db648.

dongjoon-hyun · 2026-10-22T14:12:51Z

+        assert(rows.count(_ == Row(0L, 334000L)) == 1)
+        assert(rows.count(_ == Row(0L, 1L)) == 334000)
+
+        val aggregateJoins = collect(adaptivePlan) {


Nit: These lines are findTopLevelAggregate(adaptivePlan).flatMap(findTopLevelSortMergeJoin) (L181-185 and L157-161), and collect(result) { case j: SortMergeJoinExec => j } at L2038 is findTopLevelSortMergeJoin(result). L1945 and L1976 already use the helper. Could both use the helpers, here in the form my item 3 suggests?

Both now use findTopLevelAggregate and findTopLevelSortMergeJoin in 98db648.

dongjoon-hyun · 2026-10-22T14:12:51Z

+        // key2 is always 0, so only the 334 rows of key1 = 0 join, each against all 1000 rows of
+        // the other side: the aggregated branch collapses them into one row, the other keeps one
+        // row per joined row.
+        val rows = sql(query).collect()


Nit: runAdaptiveAndVerifyResult already collects the AQE answer and compares it with the non-AQE one (L119-123), and this rule changes only the AQE plan, so a wrong split already fails there. This line, L1999 and L2008 run each query once more and collect 334,001 or 668,000 rows to the driver in a @SlowSQLTest suite, against constants derived by hand. Could these three blocks go, keeping the plan assertions?

Kept on purpose. They pin the rows to constants instead of to the non-adaptive run, so a regression that affects both engines still fails; they add about 1 s per test.

dongjoon-hyun · 2026-10-22T14:12:51Z

+   * so structurally equal joins in different places are distinct candidates. A subset may be
+   * passed, keeping the others out of the result.
+   *
+   * The walk is top-down: a sort-merge or shuffled-hash candidate has shuffle stages as children


Nit: A sort-merge candidate's children are the SortExecs at L223-224, with the stages under them; only a shuffled-hash candidate has stages as children. Two more new doc comments are off: L260 still says "the bottom-up order a transformUp visits them", though this file no longer uses transformUp (the order is foreachUp's), and L215, "The candidate plan stands for, when it is ...", does not say that it returns None otherwise. Could these read "a shuffled-hash candidate has shuffle stages as children and a sort-merge one has sorts over them, so neither contains another candidate", "in the bottom-up order foreachUp visits them", and "Returns the candidate for plan when it is ..., and None otherwise"?

Reworded in 98db648: the candidate doc states the None case, the collection says foreachUp, and the walk says a shuffled-hash candidate has shuffle stages as children while a sort-merge one has sorts over them.

dongjoon-hyun · 2026-10-22T14:12:51Z

+    }
+  }
+
+  test("SPARK-60458: split a join nested in a broadcast join's build side") {


Minor: The PR description predates this test and 351074c. It still says "Two new tests in AdaptiveQueryExecSuite" and "Full suite 148/148 on the rebased head", while this commit adds this third test and the suite now has 149 tests, and its implementation notes do not say that applySkewJoins now walks top-down so that a join nested in a broadcast-hash join's build side keeps its split. It also says only the first arm of the union test passes on the base code, but the plain arm passes there too, as the control. Could the description be updated, since it becomes the commit message?

Updated in this push: three tests, 149/149, the top-down walk, the new config and the coalescing change, and it now says both arms of the union test pass on the previous rule, with the first arm a guard.

dongjoon-hyun · 2026-10-22T14:12:51Z


+// A skewed join `OptimizeSkewedJoin` can split: the join node in the plan, its join type, and the
+// join it is replaced by, with the skewed side split.
+private case class SkewJoinCandidate(join: SparkPlan, joinType: JoinType, replacement: SparkPlan)


Nit: joinType repeats join.joinType, since all three joins extend BaseJoinExec (BaseJoinExec.scala:27-28). Its only reader is the debug log at L321-322, and the broadcast arms changed _ to a joinType binder (L243 and L249) only to fill it. That log also cannot tell joins apart: in the union test's windowed arm it prints "Skip splitting the skewed Inner join SortMergeJoin: ..." twice. Could this be SkewJoinCandidate(join: BaseJoinExec, replacement: SparkPlan), with the log printing candidate.join.simpleString(conf.maxToStringFields), which shows the keys and the join type?

Changed in 98db648: SkewJoinCandidate(join: BaseJoinExec, replacement: SparkPlan), the broadcast patterns read _ again, and the skip log prints join.simpleString(conf.maxToStringFields), so the two skipped joins read differently.

… layout away `CoalesceShufflePartitions` groups a union's children into one coalesce group when the stamped decision says the union passes a layout through. A stamped decision outlives the layout: `OptimizeSkewedJoin` keeping one branch's split takes the layout away after the stamp, and the whole group then gives up coalescing because some of its stages carry no partition specs. Ask the union what it reports now as well, so its children coalesce independently once it can only concatenate. Also: - Add the internal `spark.sql.adaptive.skewJoin.perJoinFallback.enabled`, default true, so the per-join fallback can be turned off. - Store the join as a `BaseJoinExec` in `SkewJoinCandidate` instead of duplicating its join type, and name the skipped join with `simpleString` in the debug log. - Correct the doc comments on the candidate, the collection and the walk, and use the existing test helpers for the aggregate and join lookups. Assisted-by: deepseek-flash[1m]

ulysses-you · 2026-10-23T04:25:50Z

thank you @dongjoon-hyun , addressed comments in 98db648

peter-toth

Thanks for the PR, @ulysses-you!

The fallback keeps a split only when the plan with it still passes the same ValidateRequirements check the all-at-once path uses, so I found no plan where it changes results. I ran the three new tests, SPARK-30524 and SPARK-37652 at 98db648, and all pass. With only the CoalesceShufflePartitions change reverted, the first new test fails at AdaptiveQueryExecSuite.scala:1947, so that change is pinned. A union that reports UnknownPartitioning before coalescing keeps reporting it after its children are coalesced separately, so the new grouping is safe too. What's left is small.

Minor

1. The fallback repeats the trial the fast path rejected: when every earlier candidate was kept, the last trial is the full candidate set again. With a single candidate that is the only trial. inline
2. "fails on the previous bottom-up walk" in the description: in the commit message this reads as master, whose transformUp splits both joins in that shape. The walk the test fails on is an earlier revision of this PR (r4355921588), so I'd drop that clause.
3. Conf doc misses the force path: the new conf isn't read when spark.sql.adaptive.forceOptimizeSkewedJoin is true. inline
4. Stale SkewJoinCandidate comment: it still lists "its join type", which 98db648 dropped from the case class. inline
5. The fallback-off arm passes with no joins at all: forall is true on an empty result. inline
6. Redundant isPlainUnion clause: a plain union always reports UnknownPartitioning, so the second clause alone gives the same answer. inline

peter-toth · 2026-10-23T09:44:09Z

+      plan: SparkPlan,
+      candidates: Seq[SkewJoinCandidate]): SparkPlan = {
+    var accepted = Seq.empty[SkewJoinCandidate]
+    candidates.foreach { candidate =>


Finding 1. accepted keeps the candidates in order. So when every earlier candidate has been kept, the last trial is applySkewJoins(plan, candidates), which is the optimized plan L339 just rejected. With a single candidate that is the only trial, and one skewed join under an aggregate on its key is the usual way for the fast path to fail. Such a plan now gets rebuilt and validated twice per re-optimization, and then plan comes back as before.

A probe that logs the skip fires on the GROUP BY query of SPARK-30524 (trial 1 of 1). It also fires on the third query of SPARK-37652 (trial 2 of 2), in three re-optimizations. With the change below, the three new tests, SPARK-30524 and SPARK-37652 still pass:

candidates.zipWithIndex.foreach { case (candidate, i) => // With every earlier candidate kept, the last trial is the plan the fast path rejected. val rejectedAlready = i == candidates.length - 1 && accepted.length == i if (!rejectedAlready && requirementsSatisfied(applySkewJoins(plan, accepted :+ candidate))) { accepted = accepted :+ candidate } else { logDebug(...) } } logDebug(...) if (accepted.isEmpty) plan else stripSkewJoinChildWrappers(applySkewJoins(plan, accepted))

The spark-test review already suggests reusing the last accepted trial instead of rebuilding it at L325.

peter-toth · 2026-10-23T09:44:10Z

+      .internal()
+      .doc("When true, a plan whose all-at-once skew join optimization breaks a requirement " +
+        "keeps the splits the plan can still satisfy, instead of giving up the optimization for " +
+        "every join. When false, such a plan is left unchanged.")


Finding 3. This config is only read when spark.sql.adaptive.forceOptimizeSkewedJoin is false (OptimizeSkewedJoin.scala:341-344). With the force config on, the plan gets the extra shuffles, so "such a plan is left unchanged" doesn't hold there. ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN is defined further down this object, so the key needs to be a literal here.

Suggested change

"every join. When false, such a plan is left unchanged.")

"every join. When false, such a plan is left unchanged. Only takes effect when " +

"'spark.sql.adaptive.forceOptimizeSkewedJoin' is false.")

peter-toth · 2026-10-23T09:44:10Z

+// A skewed join `OptimizeSkewedJoin` can split: the join node in the plan, its join type, and the
+// join it is replaced by, with the skewed side split.


Finding 4. This still lists "its join type", but 98db648 dropped joinType from the case class.

Suggested change

// A skewed join `OptimizeSkewedJoin` can split: the join node in the plan, its join type, and the

// join it is replaced by, with the skewed side split.

// A skewed join `OptimizeSkewedJoin` can split: the join node in the plan, and the join it is

// replaced by, with the skewed side split.

peter-toth · 2026-10-23T09:44:10Z

+        // Turning the fallback off gives the plan back to the all-or-nothing path.
+        withSQLConf(SQLConf.SKEW_JOIN_PER_JOIN_FALLBACK_ENABLED.key -> "false") {
+          val (_, withoutFallback) = runAdaptiveAndVerifyResult(query)
+          assert(findTopLevelSortMergeJoin(withoutFallback).forall(!_.isSkewJoin))


Finding 5. forall is true on an empty result, so this arm would still pass if the query stopped planning sort-merge joins. The other arms check the count first (L1942, L1951, and splitJoins at L1987). The plan with the fallback off holds two joins, and this passes at the current head:

Suggested change

assert(findTopLevelSortMergeJoin(withoutFallback).forall(!_.isSkewJoin))

val joins = findTopLevelSortMergeJoin(withoutFallback)

assert(joins.size == 2 && joins.forall(!_.isSkewJoin), withoutFallback.toString)

peter-toth · 2026-10-23T09:44:10Z

+    case u: UnionExec =>
+      !u.isPlainUnion && !u.outputPartitioning.isInstanceOf[UnknownPartitioning]


Finding 6. The isPlainUnion clause is redundant. A plain union reports super.outputPartitioning, which is UnknownPartitioning(0) (basicPhysicalOperators.scala:1377, SparkPlan.scala:163). So the second clause already returns false for it.

It doesn't save a rawPartitioning derivation either:

A stamped plain union answers outputPartitioning from the stamp, without deriving it.

An unstamped union derives it in isPlainUnion already. A non-plain one then derives it a second time in outputPartitioning.

Suggested change

case u: UnionExec =>

!u.isPlainUnion && !u.outputPartitioning.isInstanceOf[UnknownPartitioning]

case u: UnionExec => !u.outputPartitioning.isInstanceOf[UnknownPartitioning]

HyukjinKwon added the ai-review-requested Trigger AI review by Omnigent on Databricks label Oct 22, 2026

spark-test reviewed Oct 22, 2026

View reviewed changes

Comment thread sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala

Comment thread sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala Outdated

spark-test removed the ai-review-requested Trigger AI review by Omnigent on Databricks label Oct 22, 2026

dongjoon-hyun reviewed Oct 22, 2026

View reviewed changes

peter-toth reviewed Oct 23, 2026

View reviewed changes

	"every join. When false, such a plan is left unchanged.")
	"every join. When false, such a plan is left unchanged. Only takes effect when " +
	"'spark.sql.adaptive.forceOptimizeSkewedJoin' is false.")

		// A skewed join `OptimizeSkewedJoin` can split: the join node in the plan, its join type, and the
		// join it is replaced by, with the skewed side split.

	assert(findTopLevelSortMergeJoin(withoutFallback).forall(!_.isSkewJoin))
	val joins = findTopLevelSortMergeJoin(withoutFallback)
	assert(joins.size == 2 && joins.forall(!_.isSkewJoin), withoutFallback.toString)

		case u: UnionExec =>
		!u.isPlainUnion && !u.outputPartitioning.isInstanceOf[UnknownPartitioning]

	case u: UnionExec =>
	!u.isPlainUnion && !u.outputPartitioning.isInstanceOf[UnknownPartitioning]
	case u: UnionExec => !u.outputPartitioning.isInstanceOf[UnknownPartitioning]

Conversation

ulysses-you commented Oct 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

spark-test left a comment

Choose a reason for hiding this comment

Review summary

Cross-cutting observations

Findings

Nit (P3)

Decision challenges

Is the per-candidate fallback's driver cost bounded for plans with many skewed joins?

Uh oh!

Uh oh!

Uh oh!

ulysses-you commented Oct 22, 2026

Uh oh!

dongjoon-hyun commented Oct 22, 2026

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ulysses-you commented Oct 23, 2026

Uh oh!

peter-toth left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Minor

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ulysses-you commented Oct 22, 2026 •

edited

Loading

peter-toth left a comment •

edited

Loading