[dwsmith1983]

Which issue does this PR close?

Closes #6770.
Closes #6898.

#7555 should merge first: it restores the same order for windows that stay in Spark, and this PR's Spark fallback for windows relies on it.

Rationale for this change

With spark.comet.exec.forceShuffledHashJoin on, sort-merge joins were turned into hash joins after Spark had removed the sorts they made redundant, and nothing put them back. Sorting within partitions on the join key returned unsorted rows, and a partitioned insert keyed on it failed with FileAlreadyExistsException.

What changes are included in this PR?

• A planner strategy picks the shuffled hash join where Spark would plan a sort-merge join, so Spark places the sorts itself.
• Broadcasts, joins with strategy hints, unsortable or collated keys, and oversized or unsupported build sides are left to Spark. The build side follows AQE's PREFER_SHUFFLE_HASH hint.
• The native broadcast nested loop join falls back when Spark reports an output order for it (a sorted streamed side: native range, sorted cache, sorted bucketed scan, sort or sort-merge join output), since it emits rows build row by build row.
• Hash joins of the types that keep their probe order declare the streamed side's order natively, so a native window above sees it. The join dynamic filter still reaches the reader below.
• A native window over an order Comet can't verify, such as a checkpoint's or an outer hash join's, restores the whole order its input reports, led by its own keys, unless the input already has it. Spark planned the operators above the window around that order, so a second window ordered by more keys, or a sort Spark dropped as redundant, still sees it. The window falls back to Spark when that order has a key Comet cannot serialize or a collated string key. This complements fix: declare the input order of a native window over a sorted cache #7256, which declares the order when Comet can trust it.
• Behavior change: a join hint such as MERGE now wins over the force config.
• A join that the make_date rule keeps in Spark now runs as Spark's shuffled hash join, under the same build size limit, where main leaves it as a sort-merge join.
• Plan-only mode shows Spark's sort-merge joins, and joins with a streaming side are left to Spark's planner.

branch-1.1 and branch-1.2 have the same rewrite, so a backport needs this strategy rather than a cherry-pick.

How are these changes tested?

New CometJoinSuite tests, with AQE off and on, check that sorts and partitioned inserts on the join key keep their order. Others cover hints, broadcasts, skew, dynamic join filters, collated keys, plan-only mode, the AQE build side hint, nested loop joins over ordered inputs and a native window over an inner join of a sorted cache, and the existing size limit tests now run against the strategy. Two run the forced join over a sorted cache and a sorted localCheckpoint and check that both outer join shapes keep the (k, v) order. Two more run a native window over a forced or hinted inner join of a sorted checkpoint and directly over the checkpoint. Native tests cover the declared order on inner joins and the dynamic filter crossing it. window_input_order.rs covers when the window adds the sort, the order it restores, the planning errors for an input not ordered on its partition keys, and that its time is reported. CometWindowExecSuite runs stacked windows and a dropped sortWithinPartitions over sorted checkpoints at 2,000 and 200,000 rows, descending and nulls-last orders, a float key with -0.0 and 0.0, a partition key the input reports through an alias or the other key of an inner join, and both fallbacks. A forced join with a make_date condition checks that it stays a Spark shuffled hash join with the make_date fallback reason.

[dwsmith1983]

@andygrove I'd suggest this for backport-1.1. branch-1.1 has the same RewriteJoin, so with spark.comet.exec.forceShuffledHashJoin on, 1.1.0 drops the same sorts: sortWithinPartitions and SORT BY on the join key return unsorted partitions, and a partitioned insert keyed on the join key fails with FileAlreadyExistsException. The branch has no #6384 size limit, so the backport would bring the strategy over without the size check rather than cherry-pick this as is.

[sunchao]

Reviewed the full 10-file diff from 412468e2817207eff0c1907d1c4aac8c3f06680d to 4ffdeba6c9089e0ccf09443f028766eb63aa543c. The PR remains non-draft. No introduced P1/P2 issues found within this review. The existing backport discussion contains no unresolved substantive blocker.

Summary

• Prior state and problem: RewriteJoin replaced sort-merge joins after Spark had removed redundant sorts. This could leave sortWithinPartitions results unordered and break partitioned inserts.
• Design approach: Select ShuffledHashJoinExec through an injected planner strategy, allowing Spark's existing requirement and sort-removal rules to operate on the correct join type.
• Correctness: Checked join extraction, residual conditions, build-side eligibility, size boundaries, broadcast deferral, hints, and ordering propagation against Spark sources. Spark's preparation and AQE replanning sequences support the approach. DataFusion's probe-order contract also matches the relevant Spark hash-join ordering claims. No reproducible P1/P2 correctness issue was identified.
• Compatibility analysis: Compared JoinSelectionHelper and SparkStrategies across Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. The shared helpers retain version-specific build-side rules. The Spark 4 shim correctly excludes LeftSingle and keys lacking binary-stable equality.
• Key design decisions: Broadcasts and explicit strategy hints defer to Spark. The existing build-size limit, smaller-side selection, and LeftSemi/ExistenceJoin restrictions remain. Declined conversions retain explain information.
• Implementation sketch: Register CometShuffledHashJoinStrategy, remove the late rewrite and ordering-repair pass, add two version-family shims, and expand join regression coverage for sorting, writes, hints, AQE, collations, and disabled modes.
• Performance: The native join algorithm is unchanged. The change removes a whole-plan rewrite and ordering-repair traversal, while adding planning-time checks per eligible join. Required output sorts restore correctness. No measured speedup or reproducible performance regression was established.
• Design: Choosing the join before Spark establishes physical requirements addresses the cause of lost sorts. It is simpler to reason about than reconstructing ordering after Spark has erased explicit sorts.
• Abstraction & complexity: The strategy keeps selection policy together, and the small shims isolate genuine Spark-version differences. Reusing Spark's selection helpers avoids duplicating its broadcast and build-side eligibility rules. No complexity concern reached the P1/P2 bar.
• Behavioral changes worth calling out: Strategy hints now win over forcing, non-binary-stable collated keys stay with Spark, plan-only reports omit forced conversion, and eligible static joins inside streaming queries can use Spark's hash join. These are documented. Comparison with branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1 confirms the intended ordering fix. The size limit and ExistenceJoin guard already existed in this PR's base.
• Suggested improvements: No additional code change meets the P1/P2 reporting bar. Obtain the pending CI and Spark SQL/profile verdicts before queueing this planner change.

Routed skills: Read AGENTS.md and applied .ai/skills/review-comet-pr/SKILL.md. No sibling subsystem skill applies to the changed files. Read the supplied discussion snapshot and checked live reviews and inline comments. No Copilot feedback was used.

Exact-head CI: Label pull requests passed. Comet CI and CodeQL both report action_required. There is no exact-head CI build or test verdict.

Validation: The exact-head Spark 4.1.3 JVM reactor test-compile passed with formatting/style checks skipped. CometConfSuite passed all 21 tests. A disposable planner harness passed 10 checks against the freshly compiled classes, covering build sides, hints, broadcast deferral, the size boundary, disabled Comet, and retention of a local sort through Spark preparation. git diff --check passed.

Validation limits: The planner harness directly exercised planJoin and Spark preparation. It did not execute native joins. Native CometJoinSuite, AQE runtime behavior, partitioned writes, streaming execution, Spark SQL suites, and other-profile builds were not independently run. No matching native build was available, and shared disk exhaustion constrained further preparation. The author's broader test claims were not treated as independently verified results. Tracked project files remain unchanged. Nothing was published.

[comphead]

Thanks @dwsmith1983. Choosing the join at planning time addresses the cause of the lost sorts, which looks better to me than repairing them afterwards. The inline comments cover reusing Spark's own strategy hint set, and trimming repeated setup and an equivalent case in the new tests.

[comphead]

Spark already keeps this exact list as JoinStrategyHint.strategies (BROADCAST, SHUFFLE_MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL, without the AQE-only hints). It is public in hints.scala on 3.4.3, 3.5.8, 4.0.1 and 4.1.3, and ResolveHints uses it to decide which hint names a user can write. Could hasStrategyHint use it, for example Seq(hint.leftHint, hint.rightHint).exists(_.exists(_.strategy.exists(JoinStrategyHint.strategies.contains)))? That drops the four constant imports and the match, and walks the hints once instead of building two intermediate collections. If Spark adds another user-facing hint later, the join would also keep honoring it without another edit here. I have not compiled the suggested form.

[dwsmith1983]

Could hasStrategyHint use it

Done. It now checks JoinStrategyHint.strategies, so the four constant imports and the match are gone. I also added a test that a side AQE marks NO_BROADCAST_HASH keeps the hash join, since that hint is outside the set and nothing covered it.

[comphead]

This repeats setup the suite already has. big and small are the same data that withChainedJoinTables builds, and the five configs here are the ones withKeptSemiJoinConf and the window test already set (they also set maxBuildSize=-1). Could this helper call withChainedJoinTables(midRows = 3000, midKeys = 100) for the tables, and could the config block become one class-level helper that takes adaptive and serves the semi join, window and sort tests? I have not run it. The unused mid table and maxBuildSize=-1 should not change these plans, since the build side is always small.

[dwsmith1983]

could the config block become one class-level helper that takes adaptive and serves the semi join, window and sort tests?

Done. The semi join, window and sort tests now share one helper that takes adaptive, sets the five configs plus maxBuildSize=-1, and builds the tables with withChainedJoinTables(midRows = 3000, midKeys = 100).

[comphead]

sql(sortLossJoin).sortWithinPartitions("k") and sql(s"$sortLossJoin SORT BY k") should analyze to the same Sort with global = false over the same projection and join. In Spark 3.5.8 AstBuilder builds SORT BY as Sort(..., global = false, query) and Dataset.sortInternal builds the same node. From reading the code I expect identical plans and identical assertions here, but I have not compared the plans. The issue lists both forms, so I see why both are here. Would one of them be enough, given that the second adds a run per AQE setting without a different plan?

[dwsmith1983]

Would one of them be enough, given that the second adds a run per AQE setting without a different plan?

Yes. Both forms analyze to the same Sort with global = false, so I dropped the SORT BY test and kept the sortWithinPartitions one.

[andygrove]

Thanks @dwsmith1983. I added the backport label. I won't review yet since there are existing review comments to be addressed.

[sunchao]

Reviewed commit fddeee31fcaaed6816f1153522023caf1298226b. New P1/P2 findings are detailed inline.

[sunchao]

[P1] Preserve cached probe ordering before exposing the hash join’s ordering to Spark. With forcing enabled and AQE/broadcasts disabled, materialize a probe using .repartition(2, col("k")).sortWithinPartitions("k", "v").cache(), left-join a smaller table containing only some probe keys, then request .sortWithinPartitions("k", "v"). Selecting ShuffledHashJoinExec here lets Spark remove the final sort because it advertises the probe’s entire ordering. Comet reads the cache through native ScanExec, which carries no ordering metadata. The build-right left outer join becomes DataFusion’s Right join, whose unordered-input path appends unmatched probe rows after matched rows. The result therefore violates the requested ordering. This is introduced by the earlier selection: the exact base retains the final (k, v) sort because the original sort-merge join advertises only k. Please preserve ordering through the native input/outer join, or retain the sort/fallback until that guarantee is available, and add a cached-probe regression test.

Evidence: Spark 4.1.3 differential harness: probe range(2000).selectExpr("id % 100 as k", "id as v").repartition(2, col("k")).sortWithinPartitions("k", "v").cache() followed by materialization; build range(10).selectExpr("id * 10 as k"); left outer join followed by sortWithinPartitions("k", "v"). Applying the exact-base rewrite leaves one final Sort [k, v]; the exact-head strategy leaves none. Separately, native DataFusion HashJoinExec with build keys [1,3], ordered probe keys [0,1,2,3,4], and no ordering metadata returned [1,3,0,2,4]. Its reproduced 55.1.0 hash_join/exec.rs, hash_join/stream.rs, and joins/utils.rs are byte-identical to pinned 55.2.0. Current Comet ScanExec::new initializes EquivalenceProperties without ordering. Reproduction files and logs are in /tmp/review6785-fdde-probe-v6vo_b5b/. Full Comet native execution was not available.

[dwsmith1983]

Please preserve ordering through the native input/outer join, or retain the sort/fallback until that guarantee is available, and add a cached-probe regression test.

Confirmed. With forceShuffledHashJoin the planner now picks ShuffledHashJoinExec, which reports the cached probe's (k, v) ordering, so Spark drops the final sort. Comet reads the cache through a scan with no ordering, and DataFusion's Right join puts the unmatched probe rows after the matched ones in each batch. I reproduced it with a cached repartition(2, k).sortWithinPartitions(k, v) probe left-joined to a 5-key table. Every partition comes back out of order, with the right rows.

The same native gap is already reachable on main without this conf, through a SHUFFLE_HASH hint or a broadcast hash join. Both fail that test on main. #6835 fixes it at the source: it declares the streamed order on the probe for these two outer shapes and follows the cache scan into the cached plan. With its non-test changes on top of this branch, the forced, hinted and broadcast cases all pass.

So I'd like #6835 to land first. I'll then merge it into this branch and add the cached-probe test for the forced and hinted joins. cc @andygrove, since this puts #6785 after #6835.

[dwsmith1983]

Moved back to draft until #7555 merges. It fixes the same lost order for windows that stay in Spark, and this PR's Spark fallback for windows relies on it.

The latest push makes a native window over an order Comet can't verify restore the whole order its input reports, not only its own keys. Two stacked windows over a sorted checkpoint, the upper one ordered by more keys, numbered rows wrongly before.