viirya commented on code in PR #25491:
URL: https://github.com/apache/datafusion/pull/25491#discussion_r4084592883
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -3104,18 +3211,28 @@ async fn collect_left_input(
&& !left_values.is_empty()
&& left_values[0].logical_null_count() > 0;
+ if prepared {
+ drop(batches);
+ let retained = RecordBatchMemoryCounter::new().count_batch(&batch);
+ let allowance = input_bytes + copy_bytes;
+ debug_assert!(retained <= allowance);
Review Comment:
**Blocking.** This needs to go back to returning an error.
`debug_assert!` compiles out in release, and the workspace sets no
`overflow-checks` in `[profile.release]`, so when `retained > allowance` the
next line computes `allowance - retained` as a wrapped `usize` and
`MemoryReservation::shrink` panics:
```rust
// execution/src/memory_pool/mod.rs
.unwrap_or_else(|prev| {
panic!("Cannot free the capacity {capacity} out of allocated size
{prev}")
});
```
That is worse than both the previous revision and what I suggested — a panic
inside the operator rather than a `Result` the caller can handle.
I agree with your reasoning for keeping this a hard failure, so I am not
asking for my earlier `try_grow` fallback. Just restoring the previous form
covers it:
```rust
if retained > allowance {
return internal_err!(
"Prepared hash-join concat exceeded its admitted copy bound"
);
}
reservation.shrink(allowance - retained);
```
Worth keeping as a real check rather than debug-only, because the three
quantities are computed under three different rules: `input_bytes` deduplicates
shared buffers by pointer (`RecordBatchMemoryCounter::count_buffer_memory_size`
inserts on `data_ptr().addr()`), `prepared_copy_bytes` sums each batch's
`get_sliced_size()` independently, and `retained` measures the concat output's
`capacity()`. This assertion is the thing that would catch an arrow-rs
allocation change, which is precisely when you want it compiled in.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2944,7 +3037,21 @@ async fn collect_left_input(
// Arc is used instead of Box to allow sharing with
SharedBuildAccumulator for hash map pushdown
let mut hashmap = new_join_hashmap(num_rows, &mut reservation,
&metrics)?;
- let mut hashes_buffer = Vec::new();
+ let scratch_reservation = reservation.new_empty();
+ if prepared {
+ // Allow one logical null mask per key plus the combined mask.
+ let masks = if null_equality ==
NullEquality::NullEqualsNothing {
+ on_left.len() + 1
+ } else {
+ 0
+ };
+ scratch_reservation.try_grow(
+ max_batch_rows * size_of::<u64>()
+ + (max_batch_rows.div_ceil(8) + 64) * masks,
Review Comment:
Non-blocking, but worth a note in the description: the overflow guards came
out along with the refactor. `prepared.rs` had 8 `checked_add`/`checked_mul`
calls and now has none, and this expression was previously wrapped in a
`"Prepared hash-join scratch size overflow"` error.
Overflow here needs absurd row counts, so I am not claiming it is reachable
— the flat `on_left.len() + 1` mask formula is a clear improvement over
modelling `update_hash`'s temporaries, and I would not want to trade that back.
Just flagging that the change is silent; if it was deliberate, saying so in the
PR description would save the next reader the same diff archaeology.
--
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]
