comphead commented on code in PR #26228:
URL: https://github.com/apache/datafusion/pull/26228#discussion_r4380348191
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -5014,6 +4998,123 @@ mod tests {
TestMemoryExec::try_new_exec(&[vec![batch.clone(), batch]], schema,
None).unwrap()
}
+ /// Unmatched build rows of a `LeftAnti` join are emitted in build index
+ /// order, so the concatenated build batch must keep the input batch order.
Review Comment:
This test is the only regression test for #26226, so I would keep it even if
the PR is reworked around concat-first. To avoid it reading as an ordering
promise (#26235 is still open), the doc could say what it really checks:
```rust
/// The hash map and array map paths must emit unmatched build rows of a
/// `LeftAnti` join in the same order. Build order is not a plan-level
/// guarantee (see #26235).
```
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -3275,21 +3267,13 @@ fn build_in_memory(
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 {
Review Comment:
With `reverse` gone, this block and the one in the `ArrayMap` branch (L3220)
are now identical. Hoisting a single concat above `array_map_key_range` is now
a pure move, which is jayzhan211's concat-first suggestion. The hash loop could
then walk `batch.slice(offset, len)` chunks in reverse, so `hashes_buffer`
stays at `max_batch_rows` and the separate offset bookkeeping goes away. One
edge case: `max_batch_rows` is 0 for an empty build, so a
`step_by(max_batch_rows)` loop would need a guard.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -5014,6 +4998,123 @@ mod tests {
TestMemoryExec::try_new_exec(&[vec![batch.clone(), batch]], schema,
None).unwrap()
}
+ /// Unmatched build rows of a `LeftAnti` join are emitted in build index
+ /// order, so the concatenated build batch must keep the input batch order.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn join_left_anti_keeps_build_order_across_batches(
+ batch_size: usize,
+ use_perfect_hash_join_as_possible: bool,
+ ) -> Result<()> {
+ let task_ctx = prepare_task_ctx(batch_size,
use_perfect_hash_join_as_possible);
+ let batch1 = build_table_i32(
+ ("a1", &vec![1, 2]),
+ ("b1", &vec![10, 20]),
+ ("c1", &vec![7, 8]),
+ );
+ let batch2 = build_table_i32(
+ ("a1", &vec![3, 4]),
+ ("b1", &vec![3000, 4000]),
+ ("c1", &vec![9, 10]),
+ );
+ let schema = batch1.schema();
+ let left =
+ TestMemoryExec::try_new_exec(&[vec![batch1, batch2]], schema,
None).unwrap();
+ let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2",
&vec![1]));
+ let on = vec![(
+ Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
+ )];
+
+ let (columns, batches, metrics) = join_collect(
+ left,
+ right,
+ on,
+ &JoinType::LeftAnti,
+ NullEquality::NullEqualsNothing,
+ task_ctx,
+ )
+ .await?;
+
+ assert_eq!(columns, vec!["a1", "b1", "c1"]);
+ allow_duplicates! {
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+------+----+
+ | a1 | b1 | c1 |
+ +----+------+----+
+ | 1 | 10 | 7 |
+ | 2 | 20 | 8 |
+ | 3 | 3000 | 9 |
+ | 4 | 4000 | 10 |
+ +----+------+----+
+ ");
+ }
+ assert_phj_used(&metrics, use_perfect_hash_join_as_possible);
+
+ Ok(())
+ }
+
+ /// Build rows sharing a key across build batches are matched in build
+ /// order, which relies on the hash map chains putting smaller build
+ /// indices first.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn join_inner_keeps_build_order_for_duplicate_keys_across_batches(
Review Comment:
This overlaps `join_inner_one_two_parts_left_randomly_ordered`, which
already pins duplicate-key match order across two build batches (`3` before
`2`, `0` before `1`). That test uses the default config, so with a key range of
1 it should only hit the `ArrayMap` path. Parameterizing it with
`hash_join_exec_configs` and `prepare_task_ctx` would cover the hash map path
without adding a test that passes both before and after the change.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs:
##########
@@ -689,6 +689,56 @@ async fn prepared_build_matches_collect_left() ->
Result<()> {
Ok(())
}
+/// A prepared hash-map build keeps its rows in input batch order, and each
+/// hash-map index still points at the row with its key.
+#[tokio::test]
+async fn prepared_build_keeps_build_order_across_batches() -> Result<()> {
Review Comment:
I think this test can go:
- Prepared builds only support inner joins, so the row layout of
`prepared.build.batch` is not observable through any join output. The first
assertion pins an internal detail.
- The join half is covered by `prepared_build_matches_collect_left`. Its
`[None, 0, 0, -4096, 4096]` case is split into 2-row batches and the key range
forces the hash map. It compares against the ordinary build, so wrong index
offsets would show up there.
If you keep it, the `ConfigOptions` override is unnecessary. Keys `10..4000`
over 4 rows already fail the default density check, so the existing
`prepare(&base, batches, pool)` helper is enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
