fix: keep build-side input order across batches in hash join

[dwsmith1983]

Which issue does this PR close?

• Closes HashJoinExec loses build-side input order across batches on the hash map path #26226.

Rationale for this change

A HashJoinExec that builds a hash map put whole build batches in reverse order, so a CollectLeft anti or semi join that returns build rows emitted them last batch first. The array map path already kept input order, so the output order depended on how sparse the join keys were. Comet relies on the build side's order for a null-aware anti join and gets unsorted partitions when keys are sparse and the build side has more than one batch.

What changes are included in this PR?

As suggested in apache/datafusion-comet#6810 (comment):

• Batches are still inserted into the hash map last to first, so each key's chain keeps its earliest row on top.
• Each batch now gets the offset of its start in forward order, and the build batches are concatenated forward, in both the ordinary and prepared build modes.
• concat_build_batches loses its reverse flag, which no caller needs anymore.
• The doc comment and diagram on HashJoinExec show the new layout.

What is the testing strategy for this PR?

New tests:

• join_left_anti_keeps_build_order_across_batches: a LeftAnti join over two build batches (keys [10, 20] and [3000, 4000]) with no matches must return 10, 20, 3000, 4000. It fails on main for every batch size on the hash map path (returns 3000, 4000, 10, 20) and passes on the array map path.
• join_inner_keeps_build_order_for_duplicate_keys_across_batches: duplicate keys across build batches still match in build order. This passes before and after; it pins the chain order so a future change can't flip it.
• prepared_build_keeps_build_order_across_batches: prepared build mode only supports inner joins, so this checks the concatenated build batch directly (10, 20, 3000, 4000) and then runs a join to confirm the hash map indices still point at the right rows. It fails on main with the build batch in 3000, 4000, 10, 20 order.

cargo test -p datafusion-physical-plan --lib and the sqllogictests pass; no existing expected output changed.

Are there any user-facing changes?

Build-side rows from a hash join (anti, semi, and unmatched rows of outer joins in CollectLeft mode) now come out in input order across build batches. No API changes.

[andygrove]

LGTM, thanks @dwsmith1983!

The offsets line up with the forward concat (num_rows is exactly the sum of the batch rows, so offset ends at 0), and the insertion order into the map is unchanged, so each probe row's matches still come back in build order. The new inner join test pins that nicely.

On performance: hashing and hash table inserts are identical, only the stored row indices change. I ran a paired benchmark of the old vs new offset schemes on the hash map path (4M-16M row builds with unique and duplicate keys, plus probe and gather) and saw no measurable difference: everything within about 1%, with confidence intervals spanning 1.0.

[andygrove]

Filed #26235 as a follow-up: now that the build-side order no longer depends on which map the build uses, HashJoinExec could report it in maintains_input_order for LeftSemi, LeftAnti and LeftMark. That way the planner (and Comet) can rely on it instead of only these tests. Not a blocker for this PR.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 66.66667% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.82%. Comparing base (6663e43) to head (5d2d39d).

Files with missing lines	Patch %	Lines
...tafusion/physical-plan/src/joins/hash_join/exec.rs	66.66%	1 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #26228      +/-   ##
==========================================
- Coverage   82.83%   82.82%   -0.01%     
==========================================
  Files        1149     1149              
  Lines      452344   452337       -7     
  Branches   452344   452337       -7     
==========================================
- Hits       374680   374659      -21     
- Misses      54864    54878      +14     
  Partials    22800    22800              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[dwsmith1983]

@alamb can you take a look at this pr?

[alamb]

Hi @dwsmith1983 -- thanks for this PR.

Is this a general purpose change or something that is comet specific? It seems like maybe others would benefit if you could show a sort by build side was eliminated or something?

I worry about adding a bunch of constrants to DataFusion for some specific usecase (Comet) that isn't used by other users of DataFusion

I am not enough of an expert in the Join code to review this PR efficient. Perhaps @2010YOUY01 or @jayzhan211 could suggest someone who might be able to review?

[dwsmith1983]

That is a fair question, @alamb.

Comet is where this came up, and as the PR stands DataFusion itself doesn't rely on the order: HashJoinExec still reports the build side order as not maintained.

The order can change on its own today, though. The array map path already returns build rows in input order, but the hash map path returned them last batch first. So the same anti or semi join comes back in a different order depending on how sparse the join keys are. With this change both paths agree. LeftSemi, LeftAnti and LeftMark then emit build rows in input order, so maintains_input_order could report that, and a sort above something like NOT EXISTS ... ORDER BY could be dropped. I can add that with an EXPLAIN test if it seems worthwhile.

@andygrove, could you weigh in from the Comet side on whether Comet should depend on this ordering, or whether it's better handled there?

[2010YOUY01]

I think this is both an optimization opportunity and a potential constraint on future features.

For example, if HJ runs out of memory, the ordering guarantee could prevent it from switching to a spilling implementation that doesn't preserve that order:

-- Example: Top-K query ordered by a column from the HJ build side

-- Optimizer exploits the join's ordering guarantee to eliminate SortExec
LimitExec(order_by = build_table, k=3)
  HashJoinExec

-- If HJ needs to spill, the ordering guarantee no longer hold.
-- The fallback would require changing the surrounding plan shape (not easily doable now)
LimitExec
  SortExec
    HashJoinExec

In general, I think we should prioritize general-purpose features over specialized optimizations. Perhaps we should first figure out how to support this kind of fallback cleanly before committing to the ordering guarantee. (though I'm also unsure if ordering promise itself is a common requirement that worth maintaining)

[jayzhan211]

Thanks @dwsmith1983 , LGTM!

[jayzhan211]

Maybe we could make this ordering guarantee configurable — based on whether spilling is enabled, or on some other config. But I'm fine with the current approach until we actually have a use case for that.

[dwsmith1983]

Thanks @2010YOUY01, that's a good point, and I agree it's a reason not to promise the order yet. To be clear, this PR doesn't make that promise: maintains_input_order still reports false for the build side, so a spilling hash join could emit rows in any order later. The change only makes the hash map path return build rows in the same order the array map path already does, so the result no longer depends on which map gets picked.

The guarantee itself is tracked in #26235. I'll add your spilling example and @jayzhan211's idea of a config there so it gets settled before anyone picks that up.

[2010YOUY01]

To be clear, this PR doesn't make that promise: maintains_input_order still reports false for the build side, so a spilling hash join could emit rows in any order later.

This PR is already making a promise with UTs asserting output order. If you only need it at operator level in comet, perhaps we can do a alternative constructor

HashJoinExec::new_with_ordered_build()

Instead of doing #26235, and make it into a plan-level optimization guarantee. @jayzhan211 suggested use config to use config to set the ordering promise, I think this might cause some extra challenge for planning/optimization.

Thank you for the invetigations 🙏🏼 @dwsmith1983

[dwsmith1983]

Fair point: the tests pin the order, so in practice it's a promise. An opt-in flag sounds right to me and fits how Comet builds the join. Since HashJoinExec already has a builder, I'd add it next to with_null_aware, something like with_preserve_build_order(true), rather than a separate constructor. The order tests would set it, and its doc would say a strategy that can't keep build order (spilling, for example) must not be used when it's on. It wouldn't touch maintains_input_order, so planning is unchanged and #26235 may not be needed.

@andygrove would that work for Comet? It would set the flag on the null-aware anti join and then drop the fallback from apache/datafusion-comet#6810 after upgrading.