[dwsmith1983]

Which issue does this PR close?

Closes #6922.

Rationale for this change

A native left outer join building on the right, or right outer join building on the left, keeps the streamed side's order only when the serde declares it. When Comet can't verify that order, it declares none but still converts the join, while Spark has already dropped the sort above. Unmatched rows then land after matched ones, as in both repros in #6922.

What changes are included in this PR?

• Such an outer hash join falls back to Spark, with a reason, when its streamed order can't be checked.
• Only an order coming from the JVM (cache, checkpoint, Spark leaf) or a nested loop join counts; chains inside one native plan stay native.
• An outer join over a nested loop join also falls back, which gains little until Native broadcast nested loop joins report the streamed side's order without keeping it #6898 is fixed.
• The compatibility and contributor guides describe when the join falls back.

How are these changes tested?

New tests in CometJoinSuite and CometInMemoryCacheSuite, with AQE on and off, cover shuffled and broadcast joins over sorted checkpoints and caches: each falls back and keeps every partition in order. Other cases check that chained outer joins within one native plan stay native and keep their order.

[sunchao]

Reviewed the full five-file diff from d834665aeab2fcbe812f638ec9f3637b248d38c4 to ae849893e33a5642d3ce469ffa002c9875a99502. The PR is not a draft. The snapshot and live discussion contained no reviews, conversation comments, inline comments or review threads.

No introduced P1/P2 issues found within this review.

Summary

• Prior state and problem: An outer hash join could omit an unverifiable streamed-order declaration but still execute natively. DataFusion then moved unmatched rows, violating an ordering that Spark had already used to remove a sort. Checkpoints and certain cached joins exposed this.
• Design approach: Fall back for ordered LeftOuter/BuildRight and RightOuter/BuildLeft joins when the order cannot be verified across a JVM boundary or nested-loop join. Retain native execution for eligible chains within one native plan.
• Correctness: Traced both hash-join converters, cache and AQE traversal, broadcast fallback handling, and DataFusion’s ordered-probe behavior. countInPlan = true preserves the previous serialization check, so the change does not introduce new trusted ordering declarations. All 19 focused tests passed, including all six added regressions.
• Compatibility analysis: Compared Spark HashJoin, ShuffledHashJoinExec, InMemoryTableScanExec and RemoveRedundantSorts sources for 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Their relevant ordering guarantees support this guard. Compared the affected converter with branch-1.1 at 89c729a9697467e8f8d89e3eadf1797a75e2328c: moving the affected queries to Spark is an intended correctness change. No new configuration or protocol compatibility issue was identified.
• Key design decisions: Resetting countInPlan at caches and scan wrappers prevents an ordering inferred inside another native plan from being trusted across that boundary. Exempting in-plan hash joins avoids unnecessarily disabling chains whose ordering DataFusion already sees.
• Implementation sketch: keepsStreamedOrder gates the new early fallback. The existing recursive predicate gains a mode parameter, while the declaration path retains its previous behavior. Tests cover checkpoints, both cache formats, AQE, native inner joins, windows, filters and nested-loop joins. Both relevant documentation pages are updated.
• Performance: The additional traversal occurs during planning for ordered outer-join candidates, with no new per-row native work. Affected joins intentionally incur Spark execution and transition costs. The passing chain tests confirm that representative native pipelines remain eligible. No unintended performance regression meeting the P1/P2 bar was established.
• Design: Keeping the guard in the shared hash-join converter applies the policy consistently to broadcast and shuffled joins. Falling back also avoids falsely declaring an order that downstream aggregation could trust.
• Abstraction & complexity: Reusing one traversal with a named Boolean separates fallback eligibility from declaration safety without adding another ordering framework. The boundary resets and paired regression tests make that distinction concrete. No necessary simplification meeting the reporting bar was identified.
• Behavioral changes worth calling out: Ordered checkpoints and unverifiable cached outputs now cause the outer join to execute in Spark. Eligible native window/filter/join chains remain native. Falling back above a nested-loop join does not repair rows that join already reordered, as documented in existing issue #6898.
• Suggested improvements: No additional code changes meeting the P1/P2 reporting bar were identified. Exact-head CI and broader Spark SQL validation remain outstanding.

Routed skills: Used .ai/skills/review-comet-pr/SKILL.md and the operator contributor guide. No sibling skill applies to this operator-only planning change.

Exact-head CI: Comet CI (run), Check PR Title and CodeQL report action_required. The labeling check passed. There is no successful build/test CI verdict for the reviewed SHA.

Validation: Native build passed with --no-default-features. JVM production and test compilation passed using a temporary JDK 21 and cached Maven dependencies. On Spark 4.1.3, the focused outer-hash-join selection ran 19 tests across CometJoinSuite and CometInMemoryCacheSuite: 19 passed, none failed, aborted or skipped. Other Spark profiles were checked against source but not executed. Full Spark SQL suites, optional HDFS/Delta features and performance benchmarks were not run. Maven formatting/style checks were skipped. Project source remained unchanged.

[andygrove]

Thanks @dwsmith1983. This closes the hole that #6835 left, and I like that the fix is a fallback rather than declaring an order we cannot check. The countInPlan flag keeps the order the serde sends exactly as before, so the PR cannot start declaring an order it used to withhold. I also like that chains of outer joins inside one native plan stay native, and that the tests compare against Spark and check per-partition order.

I compared against branch-1.1. That branch has none of this order logic, so an outer hash join over an ordered checkpoint, cache or nested loop join ran natively there and could return unmatched rows out of order. Moving those queries to Spark is an intended correctness fix, and I found no unintended change.

I have two questions.

1. In operators.scala, the comment on case _: CometBroadcastNestedLoopJoinExec => true only explains the one-row build case. The PR description says that with two or more build rows the fallback costs the native join and gains nothing until #6898 is fixed. Could the comment link #6898 and say this case can go once that is fixed? That way the cost has an owner.

2. The new test in CometInMemoryCacheSuite covers a broadcast LEFT JOIN only. Did you try the RightOuter with BuildLeft shape over a cache, in either cache format? If it goes through the same code path, a short note in the test would be enough.

CI has not run on this PR yet. The Comet CI job is waiting for a maintainer to approve the workflow. Since this changes when a join falls back, it also needs the run-spark-4.1-tests and run-all-spark-profiles labels, because outputOrdering differs across Spark versions and the PR tier only runs the default profile.

[dwsmith1983]

Could the comment link #6898 and say this case can go once that is fixed?

Done in 3e550f2.

Did you try the RightOuter with BuildLeft shape over a cache, in either cache format?

Yes, and it goes through the same path. The cache test now runs both shapes in both cache formats, with AQE on and off. For the right outer join, the join that writes the cache runs natively as RightOuter with BuildLeft and sends its order, and the join over the cache falls back to Spark's BroadcastHashJoin RightOuter BuildLeft, in order. Without the fallback it runs natively and puts the matched rows ahead of the unmatched ones.

[andygrove]

Thanks @dwsmith1983. Both of my questions are answered. The #6898 link is in the comment on the nested loop join case, and the cache test now covers RightOuter with BuildLeft in both cache formats.

I checked that countInPlan = false is sound. The flag only drops the check for an outer join whose probe input declared no order, and the lowest join in a chain answers the same either way. So it either falls back or declares, and a join above it can trust the order DataFusion sees. The boundary resets at the cache scans and CometScanWrapper cover the places where an order crosses into another plan.

Approving. Comet CI is still waiting for a maintainer to approve the workflows, so the Spark profile jobs should go green before this merges.

[comphead]

The two-pass walk keeps the old behavior for the declared case and the thread has settled the countInPlan soundness and the labels. Three comments.

[comphead]

This reason is shown in the explain output for a join that now stays in Spark, but "so it is not declared natively" describes the old behavior, where the join stayed native and sent no order. Could it say what happens now, for example "cannot be checked, so the join stays in Spark"? The same literal is repeated in four tests (CometInMemoryCacheSuite and three places in CometJoinSuite), so a shared constant would keep them in step.

[dwsmith1983]

Changed in 01fdf01. The reason now says "cannot be checked, so the join stays in Spark", and it lives in one constant, CometHashJoin.uncheckedStreamedOrderReason, which the four tests use.

[comphead]

This paragraph opens with "it declares nothing" and only later says the join falls back, so a reader cannot tell when "declares nothing" means native and when it means a fallback. Leading with the fallback rule and then giving the in-plan exception would be clearer. The RDD checkpoint example also reaches a native join only with spark.comet.convert.rdd.enabled=true, which defaults to false. Without it the checkpoint is a plain Spark child and the join stays in Spark with or without this change. The tests set the config, so the doc could say so.

[dwsmith1983]

Reworked in 01fdf01. The paragraph now leads with when the join falls back, notes that the checkpoint example needs spark.comet.convert.rdd.enabled=true, and then gives the exception: an order that comes only from another native hash join in the same plan keeps the join native. I narrowed that from "native operators" because a nested loop join in the same plan still makes it fall back.

[comphead]

outer hash joins over a checkpointed streamed side fall back and keep its order and an outer hash join over a native inner join over a checkpoint falls back (line 2775) repeat the same conf block and the same range(2000)...localCheckpoint setup. A small withOrderedCheckpoint { streamed => ... } helper would remove about 20 lines.

[dwsmith1983]

Done in 01fdf01 with withOrderedCheckpoint. It takes the checkpoint with AQE off, and the first test still runs both AQE modes over it.