[SPARK-58099][SQL] Remove the local sort dangling below the shuffle added by skew join optimization

[ulysses-you]

What changes were proposed in this pull request?

This PR enhances RemoveRedundantSorts to remove a dangling local sort -- a local SortExec sitting directly below a shuffle that neither consumes nor exposes an ordering (both its requiredChildOrdering and outputOrdering are empty). Such a shuffle destroys the child ordering, so the sort has no effect and is dead.

To make this effective for the skew-join case, RemoveRedundantSorts is moved after OptimizeSkewedJoin in the AQE queryStagePreparationRules (AdaptiveSparkPlanExec), and the same reorder is applied to the non-AQE QueryExecution.preparations to keep the two rule chains consistent. In both chains RemoveRedundantSorts now runs after DisableUnnecessaryBucketedScan (see the correctness note below).

Why are the changes needed?

Consider a stage with a ShuffledHashJoin feeding a SortMergeJoin on the same key, where the SHJ output partitioning already satisfies the SMJ requirement (no exchange in between, only the SMJ's local sort). When the SHJ is skewed and skew-join optimization is applied, an extra shuffle is inserted between the two joins. EnsureRequirements wraps that shuffle on top of the existing Sort -> SHJ, and adds a new sort above the shuffle for the SMJ. The original local sort is then left dangling right below the newly inserted shuffle, computed in the SHJ's stage for nothing.

The current RemoveRedundantSorts cannot remove it: it only strips a sort whose child already satisfies the ordering, whereas this sort is dead because it sits directly under an ordering-destroying shuffle -- a different kind of redundancy. Running before OptimizeSkewedJoin also means it never sees the added shuffle.

Does this PR introduce any user-facing change?

No. It only removes a redundant sort operator from the physical plan (and fixes the latent ordering-loss hazard described above).

How was this patch tested?

New UT SPARK-58099 in AdaptiveQueryExecSuite reproduces the SHJ->SMJ skew-join scenario: with the rule disabled it asserts a local sort dangles below the added shuffle, and with the rule enabled it asserts the sort is removed while both joins remain skew joins and results are unchanged.

New UTs in RemoveRedundantSortsSuite (both AE / non-AE variants):

• remove local sort dangling below a shuffle that does not require ordering exercises the new removal branch directly on a physical plan (the dangling-sort shape cannot be produced from SQL/DataFrame because the logical EliminateSorts strips such a local sort before physical planning), covering local-sort removal, global-sort preservation, and the rule-disabled case.
• keep local sort satisfied only by a bucketed scan output ordering is a regression test for the reorder-vs-DisableUnnecessaryBucketedScan correctness hazard; it fails with the old rule order and passes with the new one.

Also verified RemoveRedundantSortsSuite (both AE / non-AE variants), DisableUnnecessaryBucketedScanSuite, and the full AdaptiveQueryExecSuite pass.

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

Generated-by: Claude Code (Claude Opus 4.8)

[ulysses-you]

cc @viirya @cloud-fan @LuciferYang thank you

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits.
Small, correct, well-motivated enhancement. The new RemoveRedundantSorts case is result-preserving, the rule reorder is deliberate and safe, and the test exercises the branch end-to-end.

No findings.

Verification

Traced the Shuffle(SortExec(local, child)) -> Shuffle(child) rewrite: a shuffle repartitions rows across the network, so dropping an input within-partition sort cannot change the output row multiset or partition assignment (equivalent for empty / single / many rows, NULLs, and duplicates). The one case where within-partition determinism matters -- round-robin / null-aware partitioning (SPARK-23207) -- is handled by the shuffle's own internal local sort in prepareShuffleDependency, never via an external SortExec child, so it is unaffected. The guards (requiredChildOrdering.head.isEmpty && outputOrdering.isEmpty) exclude any shuffle that needs or exposes an ordering, and .head is safe since a shuffle is a UnaryExecNode. The reorder to run after OptimizeSkewedJoin does not break any rule that previously ran after RemoveRedundantSorts (RemoveRedundantWindowGroupLimits, DisableUnnecessaryBucketedScan, OptimizeSkewedJoin are all insensitive to whether a redundant sort was already removed).

[viirya]

The change looks correct to me. I verified the tricky interactions: the global = false restriction keeps the rule away from the GlobalLimit <- Shuffle(SinglePartition) <- ... <- Sort(global) shape that InsertSortForLimitAndOffset relies on; removing a local sort under a round-robin repartition doesn't affect retry determinism since the shuffle's internal sortBeforeRepartition sort is applied regardless of the child ordering; and moving RemoveRedundantSorts after OptimizeSkewedJoin can't regress skew detection, because the sorts its SMJ pattern needs sit above shuffle stages whose outputOrdering is Nil, so the old position could never remove them anyway.

A few comments:

• Could you add a test for the new removal kind to RemoveRedundantSortsSuite? Something cheap like sortWithinPartitions + repartition (or SORT BY under DISTRIBUTE BY) would exercise the new pattern through both the AE and non-AE variants of that suite. Right now the non-AQE side of this change (both the new case and the rule reorder in QueryExecution.preparations) has no test backing it -- the new UT only covers the AQE skew-join path.

• In QueryExecution.preparations, the existing comment ("RemoveRedundantSorts and RemoveRedundantWindowGroupLimits needs to be added after EnsureRequirements...") now sits above RemoveRedundantWindowGroupLimits while the rule it names has moved two lines down, and unlike the AQE side there's no comment explaining the new position. A one-liner noting it must stay after DisableUnnecessaryBucketedScan and mirror the AQE rule order would help.

• On that note, I think the non-AQE reorder is more than a consistency cleanup. With the old order there seems to be a latent correctness hazard: for a query like SELECT * FROM bucketed_sorted_table SORT BY sort_col with no interesting-partition operator above, RemoveRedundantSorts could remove the sort based on the bucketed scan's outputOrdering, and then DisableUnnecessaryBucketedScan disables the bucketed scan and the ordering is silently lost. Running RemoveRedundantSorts after DisableUnnecessaryBucketedScan closes that hole. Worth confirming, and if it's real, mentioning it in the PR description and maybe adding a regression test.

• Tiny nit: the new class doc could mention why the outputOrdering.isEmpty guard is there (a custom ShuffleExchangeLike that preserves the child ordering must keep the sort), so both conditions in the code have a matching rationale in the doc.

[ulysses-you]

@viirya thanks for the careful review! Addressed all four points in the followup commit:

1. Bucketed-scan hazard confirmed and it's real. Under the old order, with spark.sql.legacy.bucketedTableScan.outputOrdering=true a query like SELECT * FROM bucketed_sorted_table SORT BY sort_col (no interesting-partition operator above) has its sort stripped by RemoveRedundantSorts based on the scan's outputOrdering, then DisableUnnecessaryBucketedScan disables the scan and the ordering is silently lost. Running RemoveRedundantSorts after DisableUnnecessaryBucketedScan closes it. Added regression test keep local sort satisfied only by a bucketed scan output ordering (fails on the old order, passes now) and mentioned it in the PR description.

2. RemoveRedundantSortsSuite coverage. Note your suggested sortWithinPartitions + repartition / SORT BY under DISTRIBUTE BY shapes can't reach the new branch: the logical EliminateSorts strips a local sort sitting below a repartition before physical planning, so the dangling-sort shape is physical-plan-only. I added a test that applies the rule directly to a physical plan (local-sort removal, global-sort preservation, rule-disabled), plus the bucketed-scan test above -- both run through the AE and non-AE variants.

3. QueryExecution.preparations comment updated: the stale comment no longer names the moved rule, and there's a one-liner noting RemoveRedundantSorts sits after DisableUnnecessaryBucketedScan to mirror the AQE order.

4. Class doc now explains the outputOrdering.isEmpty guard: a custom ShuffleExchangeLike that preserves child ordering reports a non-empty outputOrdering, so its local sort is not dead and must be kept.