jayzhan211 commented on code in PR #26793:
URL: https://github.com/apache/datafusion/pull/26793#discussion_r4480802619
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -3217,28 +3275,36 @@ fn build_in_memory(
// 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)?;
+ // Every slice shares everything beneath a dictionary or list view,
+ // at any depth of a key, which `create_hashes` hashes on each call:
+ // slices `SHARED_VALUES_SLICE_FACTOR` times longer than those values
+ // (up to `MAX_HASH_CHUNK_ROWS`, never shorter than the values or the
+ // largest batch) keep that cost small.
+ let chunk_rows = hash_chunk_rows(&left_values, max_batch_rows,
num_rows);
+
let scratch_reservation = reservation.new_empty();
if prepared {
scratch_reservation.try_grow(prepared::prepared_scratch_bytes(
- max_batch_rows,
+ chunk_rows,
on_left.len(),
null_equality,
)?)?;
}
Review Comment:
Outside prepared mode, the hashes buffer isn't reserved. That was fine while
it was one batch long, but here it is at least 8 MiB per partition once the cap
applies, and 8 bytes per build row once the slice covers the whole build. Could
you reserve it when the slice is longer than a batch, and add a test? For
example, a build with a dictionary key and several batches should hit
"Resources exhausted" under a limit that only fits without the buffer.
```suggestion
if prepared {
scratch_reservation.try_grow(prepared::prepared_scratch_bytes(
chunk_rows,
on_left.len(),
null_equality,
)?)?;
} else if chunk_rows > max_batch_rows {
// A hashes buffer longer than a batch scales with the build side
scratch_reservation.try_grow(chunk_rows * size_of::<u64>())?;
}
```
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -103,12 +103,68 @@ pub(crate) const HASH_JOIN_SEED: SeededRandomState =
const ARRAY_MAP_CREATED_COUNT_METRIC_NAME: &str = "array_map_created_count";
+/// The length of the longest array under `data` that `create_hashes` hashes
+/// in full whatever slice of `data` it gets: everything beneath a dictionary
+/// or a list view, at any depth.
+fn unsliced_hash_len(data: &ArrayData) -> usize {
+ let own = match data.data_type() {
+ DataType::Dictionary(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_) => subtree_max_len(&data.child_data()[0]),
+ _ => 0,
+ };
+ data.child_data()
+ .iter()
+ .map(unsliced_hash_len)
+ .fold(own, usize::max)
+}
+
+/// A build slice is this many times longer than the shared values beneath a
+/// dictionary or list view that it rehashes, so rehashing them adds at most
+/// about 1/16 to the hashing work.
+const SHARED_VALUES_SLICE_FACTOR: usize = 16;
+
+/// The headroom from [`SHARED_VALUES_SLICE_FACTOR`] stops at this many rows
+/// (8 MiB of hashes), so a large merged dictionary cannot grow the hashes
+/// buffer and scratch space to the whole build side. Past it only the
+/// rehashing overhead grows, and it stays linear in the build rows.
+const MAX_HASH_CHUNK_ROWS: usize = 1 << 20;
+
+/// The number of rows in each slice the hash map is filled from.
+///
+/// A slice is never shorter than the largest build batch or the longest
+/// shared values under any key, which keeps hashing linear in the build rows.
+/// The overhead ratio is per key column, so the longest key counts, not the
sum.
+/// Beyond that, a slice is [`SHARED_VALUES_SLICE_FACTOR`] times the shared
+/// values, up to [`MAX_HASH_CHUNK_ROWS`], and never longer than the build
side.
+fn hash_chunk_rows(
+ left_values: &[ArrayRef],
+ max_batch_rows: usize,
+ num_rows: usize,
+) -> usize {
+ let shared = left_values
+ .iter()
+ .map(|values| unsliced_hash_len(&values.to_data()))
+ .max()
+ .unwrap_or(0);
+ let base = max_batch_rows.max(shared);
+ let amortized =
base.max(shared.saturating_mul(SHARED_VALUES_SLICE_FACTOR));
+ amortized.min(base.max(MAX_HASH_CHUNK_ROWS)).min(num_rows)
Review Comment:
The 2^20 cap brings the slowdown back once the merged dictionary passes 2^16
values. Each of the N/2^20 slices still hashes all of it, so the extra work
grows with the build side. With 1024 batches × 8192 rows and 1024 values per
batch (about 1M merged), build time goes 50.2 ms on main → 62.2 ms here. With
256 batches × 4096 values per batch it goes 16.9 → 17.9 ms (release, min of
9–21 runs).
Dropping the cap brings both to 47.8 and 16.3 ms and keeps your win for a
dictionary shared by all batches (13.9 → 9.6 ms). `MAX_HASH_CHUNK_ROWS`, the
"capped at 2^20" sentence in the `HashJoinExec` doc, and the two cap rows in
`hash_chunk_rows_amortizes_shared_values` (which become `4_194_304` and
`1_600_000`) go away with it. The slice can then cover the whole build side, so
it needs the reservation in my other comment.
```suggestion
max_batch_rows
.max(shared.saturating_mul(SHARED_VALUES_SLICE_FACTOR))
.min(num_rows)
```
--
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]
