ulysses-you opened a new pull request, #57181: URL: https://github.com/apache/spark/pull/57181
### What changes were proposed in this pull request? This PR lets adaptive query execution (AQE) convert a `SortMergeJoinExec` into a `ShuffledHashJoinExec` on the **physical plan**, so the conversion can see through non-shuffle operators (aggregate, project, filter, window, left-existence join) that sit between the join and its input shuffle. Concretely: 1. **New rule `ReplaceSortMergeJoinToShuffledHashJoin`** (in `queryStagePreparationRules`, after `ReplaceHashWithSortAgg`). Once the join's input shuffles have materialized, it reaches each side's `ShuffleQueryStageExec` through the safelist of non-shuffle operators above it and, if a build side's per-partition sizes all fit `spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold`, rewrites the join to a shuffled hash join. The swap is shuffle-free (both are `ShuffledJoin` with the same distribution/partitioning); only the child sorts become unnecessary. Because a shuffled hash join loses the sort merge join's output ordering, `EnsureRequirements` is re-run so any ordering an ancestor still needs is re-established. Gated by `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default `false`). 2. **`SimpleCostEvaluator` gains an optional local-sort cost** (`spark.sql.adaptive.costEvaluator.countLocalSort.enabled`, default `false`). The cost is packed as `[-numSkewJoins | numShuffles | numLocalSorts]`, so local sorts are a lowest-priority tiebreaker below shuffle count (and skew-join count). AQE's cost comparison then adopts the converted plan only when it does not add local sorts elsewhere. 3. `preferShuffledHashJoin` is lifted into `JoinSelectionHelper` and shared by `DynamicJoinSelection` and the new rule. Class hierarchy note: `SortMergeJoinExec` and `ShuffledHashJoinExec` both extend `ShuffledJoin`, which is why the swap needs no new shuffle. `ShuffledHashJoinExec` (via `HashJoin.outputOrdering`) only preserves the streamed side's ordering, whereas `SortMergeJoinExec` (inner) keeps both sides'; the local-sort cost handles cases where that difference forces a new sort upstream. ### Why are the changes needed? `DynamicJoinSelection` already prefers a shuffled hash join over a sort merge join, but it works by adding a join hint on the **logical** plan and only fires when the join child is a shuffle stage directly. When non-inflating operators (e.g. an aggregate, or a filter from a `HAVING`) sit between the join and its input shuffle, the hint is never added and the join stays a sort merge join even though a shuffled hash join would be cheaper. Doing the selection on the physical plan — where each side's input shuffle is an explicit materialized `ShuffleQueryStageExec` — removes that restriction and reuses the correct runtime statistics regardless of the operators above the shuffle. ### Does this PR introduce _any_ user-facing change? Yes, two new configurations, both documented in `docs/sql-performance-tuning.md`: - `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default `false`) - `spark.sql.adaptive.costEvaluator.countLocalSort.enabled` (default `false`) Both default to `false`, so there is no behavior change unless a user opts in. ### How was this patch tested? New unit tests in `AdaptiveQueryExecSuite`: - Converting a sort merge join to a shuffled hash join when an aggregate / `HAVING` filter sits above the shuffle. - The converted plan stays valid when an ancestor needs the join's ordering (`EnsureRequirements` re-adds the sort). - `SimpleCostEvaluator` orders plans by skew joins, then shuffles, then local sorts (including a real skew-join node). - With `countLocalSort` enabled, a conversion that would add a local sort (both sides are sort aggregates, a parent window partitions by the build-side key) is rejected and the sort merge join is kept. Full `AdaptiveQueryExecSuite` passes. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: 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]
