[dwsmith1983]

Which issue does this PR close?

• Closes Fuse operations in equal_rows_arr #12131. The fused index compare that issue asked for landed in refactor join-key equality filtering #23843; this picks up the per-chunk allocations it left behind and removes the stale note that still pointed at the old code.

Rationale for this change

Three pieces of per-chunk work on the hash join probe path repeat work whose inputs do not change within a probe batch. The multi-column key comparator was rebuilt for every output chunk, and building it scans the entire build side for negative zero when a key is a float. The integer-map path cloned both index buffers per chunk. Every output batch deep-cloned the output schema, metadata included, to wrap it in a new Arc.

What changes are included in this PR?

• The key comparator is built once per probe batch, cached on the stream, and cleared when the next batch arrives or the stream stops.
• The integer-map path takes and returns its index buffers instead of cloning them.
• build_batch_from_indices and build_batch_empty_build_side accept a SchemaRef and clone the Arc.

Two-column join on (l_orderkey, CAST(l_quantity AS DOUBLE)) over TPC-H SF1 with a 60K row build side and 17M output rows, median of five: 150 ms before, 108 ms after; HashJoinExec elapsed_compute 147 ms to 104 ms. Low-fanout float joins, the hj benchmark queries 23 to 25, and TPC-H q7 and q9 stay within run-to-run noise.

What is the testing strategy for this PR?

New tests in datafusion/physical-plan/src/joins/hash_join: join_inner_multi_key_float_zero_and_nulls_across_chunks, join_inner_multi_key_across_probe_batches (fails if the per-batch comparator reset is removed), and join_inner_array_map_across_chunks. Existing join tests and the join sqllogictest files pass unchanged.

Are there any user-facing changes?

No.

[ShayanGho]

@dwsmith1983 Thank you for working on this.

Checked this against apache/datafusion#12131 and the earlier work in #23843. The comparator caching and buffer reuse cover the remaining repeated setup and allocations. All 564 hash-join tests passed locally, and the cache lifetime and buffer ownership look correct. I haven't verified the benchmark numbers.

@Dandandan could you please approve the workflow run?

Side note, apache/datafusion#12131 is still assigned to LeslieKid from 2024.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 91.26214% with 9 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.61%. Comparing base (4bb9f8d) to head (02024b8).
⚠️ Report is 1 commits behind head on main.

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

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #25658   +/-   ##
=======================================
  Coverage   82.61%   82.61%           
=======================================
  Files        1147     1147           
  Lines      444733   444819   +86     
  Branches   444733   444819   +86     
=======================================
+ Hits       367407   367495   +88     
+ Misses      55010    55005    -5     
- Partials    22316    22319    +3     

☔ 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]

Thanks @ShayanGho for checking it against #12131 and #23843.

@jayzhan211 @adriangb could you take a look when you have time? It builds the key comparator once per probe batch and drops two per-chunk clones; the multi-column float join in the description goes from 150 ms to 108 ms.

[jayzhan211]

Thanks @dwsmith1983 , here is a suggestion

[jayzhan211]

The comparator is only valid for one probe batch, but it lives on the stream and relies on hand-placed = None resets (fetch_probe_batch, limit path). Any future path that installs a new ProcessProbeBatchState without a reset silently drops matches; removing the fetch_probe_batch reset fails join_inner_multi_key_across_probe_batches. Storing it on ProcessProbeBatchState ties its lifetime to the batch, removes both resets, and frees the probe arrays when the batch finishes instead of at the next fetch. I checked this locally: clippy is clean and the joins::hash_join tests pass.

-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub(super) enum HashJoinStreamState {

-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub(super) struct ProcessProbeBatchState {
     ...
     matched_probe_idx: Option<u32>,
+    /// Key comparator for this batch, built on first use and reused by every
+    /// chunk
+    key_comparator: Option<JoinKeyComparator>,
 }

                         matched_probe_idx: None,
+                        key_comparator: None,
                     });

-                &mut self.probe_key_comparator,
+                &mut state.key_comparator,

Then drop the probe_key_comparator field, its initializer, and both self.probe_key_comparator = None; lines. In utils.rs:

impl Debug for JoinKeyComparator {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("JoinKeyComparator").finish_non_exhaustive()
    }
}

[dwsmith1983]

Thanks @jayzhan211, that is cleaner. Done in c6d7224: the comparator now lives on ProcessProbeBatchState, the stream field and both resets are gone, and JoinKeyComparator gets a manual Debug impl.

[adriangb]

This loop calls lookup_join_hashmap once per chunk with the same build_scope_values and probe_scope_values. Because each call gets a new &mut None, the null-aware scope path still builds the comparator again for each chunk. Can you move the slot out of the loop so that this path also builds it once?

let mut offset = (0, None);
let mut key_comparator = None;
loop {
    let (build_indices, probe_indices, next_offset) = lookup_join_hashmap(
        // ...
        &mut key_comparator,
    )?;

[dwsmith1983]

Good catch, done in 31f5ae6: the slot now lives outside the loop, so each scope lookup builds the comparator once. None of the existing null-aware tests had a multi-column scope, so I added test_null_aware_left_anti_multi_column_scope, which splits both scope lookups into several chunks at the smaller batch sizes.

[adriangb]

Optional: this take-then-give-back of the index buffers repeats the pattern inside lookup_join_hashmap. A small helper that both paths call would keep the two in step if one of them changes later.

[dwsmith1983]

I looked at sharing it, but the two paths hold the buffers for different spans. lookup_join_hashmap takes them and gives them straight back once the filtered arrays are built, while the array map path hands its arrays out as the chunk's indices and only reclaims the buffers after the output batch is done. A shared helper would wrap the two one-line conversions without making either path shorter, so I left them as they are. Happy to revisit if you feel strongly.

[adriangb]

Nit: nothing enforces this rule, so a caller that keeps the slot across different arrays gets wrong matches with no error. Can you also say here that &mut None is always safe (it only costs the comparator setup on each call)? That explains the &mut None at the symmetric hash join call site.

[dwsmith1983]

Added in 31f5ae6: the doc now says nothing checks the rule and that &mut None is always correct, at the cost of rebuilding the comparator on each call.

[jayzhan211]

Thanks @dwsmith1983 , here is a suggestion

JoinKeyComparator::new still runs normalize_float_zero over the full build key column once per probe batch, so float-key joins pay O(build rows × probe batches). On TPC-H SF1 with a 1.5M-row build side (query below), join_time is ~91 ms on this PR. It drops to ~23 ms if build-side normalization is skipped, which matches the same join on a BIGINT key (~21 ms). Fine to handle in a follow-up.

SET datafusion.execution.target_partitions = 1;
EXPLAIN ANALYZE SELECT count(*)
FROM (SELECT l_orderkey AS k, CAST(l_quantity AS DOUBLE) AS q
      FROM lineitem WHERE l_orderkey < 1500000) b
JOIN (SELECT l_orderkey AS k, CAST(l_quantity AS DOUBLE) AS q FROM lineitem) p
  ON b.k = p.k AND b.q = p.q;

Suggested shape: normalize float build keys once when the build side is collected and store them in JoinLeftData next to values(). Keep them separate because the in-list dynamic filter needs the original values. Then add a JoinKeyComparator constructor that takes already-normalized left arrays. That also removes the per-batch, untracked copy of the build float column when it contains -0.0.

[dwsmith1983]

@jayzhan211 thanks, nice find. You're right that the scan for -0.0 runs over the whole build column each time the comparator is built, even when nothing needs rewriting. I'd like to keep this PR to what it has and do the build-time normalization in a follow-up: store the normalized keys in JoinLeftData beside values(), account for any copy in the build reservation, and add a constructor that takes pre-normalized left arrays. I'll use your query as the benchmark.

[dwsmith1983]

@jayzhan211 the build-side normalization is up as #25972, a draft stacked on this PR. Your query goes from 96 ms to 26 ms of join time there, about the same as the BIGINT-key control.

[jayzhan211]

Thanks @dwsmith1983 @adriangb , I will review @25972 later on