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

Describe the bug

On the hash map build path, HashJoinExec concatenates the build batches in reverse batch order. Rows keep their order inside each batch, but whole batches come out last to first.

• Any build-side row emitted in index order (for example an anti or semi join's unmatched build rows in CollectLeft mode) loses the build side's input order across batches.
• The array map (perfect hash) path concatenates forward and keeps the order. That path is skipped only when keys are sparse: range >= perfect_hash_join_small_build_threshold (1024) and density <= perfect_hash_join_min_key_density (0.15). So whether the output keeps input order depends on the key distribution.

Why it's reversed

JoinHashMap chains are LIFO. To keep a key's earlier rows on top of its chain, update_hash runs over the batches last to first, and offsets are assigned in that same order. The concatenated batch then has to be reversed too so the indices line up. The doc comment above HashJoinExec explains this with a 3-batch diagram.

To Reproduce

A LeftAnti HashJoinExec in CollectLeft mode, with two build batches (keys [10, 20] and [3000, 4000]) and a probe side that matches none of them, returns [3000, 4000, 10, 20].

Expected behavior

[10, 20, 3000, 4000], the build side's input order, regardless of which map the build uses.

Proposed fix

Suggested by @andygrove in apache/datafusion-comet#6810 (comment):

1. Keep inserting batches last to first, so the per-key chain order is unchanged.
2. Give each batch the offset of its start position in forward order: batch k gets the total row count of batches 0..k.
3. Concatenate forward.

This preserves both the chain order and the input order, and the result no longer depends on which path the build takes.

Tests to add: the repro above asserting [10, 20, 3000, 4000], and a variant with duplicate keys across batches for an inner join, checking that matches still come back in build order.

Additional context

Code on main (6663e43):

• update_hash over batches.iter().rev() and concat_build_batches(..., true, ...):

  datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

  Lines 3383 to 3412 in 6663e43

  	// Updating hashmap starting from the last batch
  	for batch in batches.iter().rev() {
  	hashes_buffer.clear();
  	hashes_buffer.resize(batch.num_rows(), 0);
  	update_hash(
  	&on_left,
  	batch,
  	&mut *hashmap,
  	offset,
  	&random_state,
  	&mut hashes_buffer,
  	0,
  	true,
  	null_equality,
  	)?;
  	offset += batch.num_rows();
  	}
  	// Merge all batches into a single batch, so we can directly index into the arrays
  	let batch = if prepared {
  	concat_batches(&schema, batches.iter().rev())?
  	} else {
  	concat_build_batches(
  	&schema,
  	std::mem::take(&mut batches),
  	true,
  	input_bytes,
  	&mut reservation,
  	&metrics,
  	)?

• The reverse flag in concat_build_batches:

  datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

  Lines 3180 to 3184 in 6663e43

  	let batch = if reverse {
  	concat_batches(schema, batches.iter().rev())?
  	} else {
  	concat_batches(schema, batches.iter())?
  	};

• Doc comment and diagram:

  datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

  Lines 860 to 895 in 6663e43

  	/// When using the standard `JoinHashMap`, hash join uses LIFO data structure as a hash table,
  	/// and in order to retain original build-side input order while obtaining data during probe phase,
  	/// hash table is updated by iterating batch sequence in reverse order -- it allows to
  	/// keep rows with smaller indices "on the top" of hash table, and still maintain
  	/// correct indexing for concatenated build-side data batch.
  	///
  	/// Example of build phase for 3 record batches:
  	///
  	///
  	/// ```text
  	///
  	/// Original build-side data Inserting build-side values into hashmap Concatenated build-side batch
  	/// ┌───────────────────────────┐
  	/// hashmap.insert(row-hash, row-idx + offset) │ idx │
  	/// ┌───────┐ │ ┌───────┐ │
  	/// │ Row 1 │ 1) update_hash for batch 3 with offset 0 │ │ Row 6 │ 0 │
  	/// Batch 1 │ │ - hashmap.insert(Row 7, idx 1) │ Batch 3 │ │ │
  	/// │ Row 2 │ - hashmap.insert(Row 6, idx 0) │ │ Row 7 │ 1 │
  	/// └───────┘ │ └───────┘ │
  	/// │ │
  	/// ┌───────┐ │ ┌───────┐ │
  	/// │ Row 3 │ 2) update_hash for batch 2 with offset 2 │ │ Row 3 │ 2 │
  	/// │ │ - hashmap.insert(Row 5, idx 4) │ │ │ │
  	/// Batch 2 │ Row 4 │ - hashmap.insert(Row 4, idx 3) │ Batch 2 │ Row 4 │ 3 │
  	/// │ │ - hashmap.insert(Row 3, idx 2) │ │ │ │
  	/// │ Row 5 │ │ │ Row 5 │ 4 │
  	/// └───────┘ │ └───────┘ │
  	/// │ │
  	/// ┌───────┐ │ ┌───────┐ │
  	/// │ Row 6 │ 3) update_hash for batch 1 with offset 5 │ │ Row 1 │ 5 │
  	/// Batch 3 │ │ - hashmap.insert(Row 2, idx 6) │ Batch 1 │ │ │
  	/// │ Row 7 │ - hashmap.insert(Row 1, idx 5) │ │ Row 2 │ 6 │
  	/// └───────┘ │ └───────┘ │
  	/// │ │
  	/// └───────────────────────────┘
  	/// ```

The same code in 55.2.0 (500744d), the version Comet pins:

• Build loop and reverse concat:

  datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

  Lines 2342 to 2365 in 500744d

  	let mut offset = 0;
  	let batches_iter = batches.iter().rev();
  	// Updating hashmap starting from the last batch
  	for batch in batches_iter.clone() {
  	hashes_buffer.clear();
  	hashes_buffer.resize(batch.num_rows(), 0);
  	update_hash(
  	&on_left,
  	batch,
  	&mut *hashmap,
  	offset,
  	&random_state,
  	&mut hashes_buffer,
  	0,
  	true,
  	null_equality,
  	)?;
  	offset += batch.num_rows();
  	}
  	// Merge all batches into a single batch, so we can directly index into the arrays
  	let batch = concat_batches(&schema, batches_iter.clone())?;

• Array map skip rule:

  datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

  Lines 176 to 180 in 500744d

  	if range >= perfect_hash_join_small_build_threshold as u64
  	&& dense_ratio <= perfect_hash_join_min_key_density
  	{
  	return Ok(None);
  	}

How Comet hits it: Comet runs a null-aware anti join (NOT IN) unswapped in CollectLeft mode, so Spark's streamed side becomes DataFusion's build side. With sparse keys and more than one build batch, the output comes out in reverse batch order. Spark has already removed the sort above the join because the join reports the streamed side's order, so sortWithinPartitions / SORT BY above it can return unsorted partitions.

• Comet issue: Native hash joins lose the streamed side's order when it arrives sorted from outside the native plan datafusion-comet#6787
• fix: fall back to Spark for NOT IN over an ordered streamed side datafusion-comet#6810 restores the sort above the native join
• fix: keep a streamed side sorted outside the native plan in order through outer hash joins datafusion-comet#6835 covers the outer join shapes