ulysses-you opened a new pull request, #59703: URL: https://github.com/apache/spark/pull/59703
### 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. 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. - 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. ### Why are the changes needed? The rule was all-or-nothing. Two failure classes make the whole-plan validation fail: 1. a split introduces an extra shuffle, because the consumer above the join needs the rows on its key, and 2. 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. ### 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. 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? Two new tests in `AdaptiveQueryExecSuite`, each asserting both the plan shape and the expected rows: - `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 result is checked to be exactly one aggregated row plus 334000 un-aggregated rows. - `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. The first arm also passes on the base implementation, which rejects the whole plan there, so it guards that the per-join fallback accepts no split the union's claim cannot take. - 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 148/148 on the rebased head, scalastyle clean. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (deepseek-flash[1m]) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
