dwsmith1983 commented on code in PR #26793:
URL: https://github.com/apache/datafusion/pull/26793#discussion_r4482043742
##########
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:
Thanks @jayzhan211, agreed. With the cap, every slice past 2^16 merged
values rehashed a growing share of the dictionary, which is the 2x I was trying
to remove in the first place. 19eb8f544 takes your suggestion: no
`MAX_HASH_CHUNK_ROWS`, slices are 16 times the shared values up to the whole
build side, and the doc and the two table rows are updated as you listed. I
also merged main (conflict with #26994, which now charges the map to its own
`map_reservation`; this branch keeps that split).
Local release `build_time` in ms, median / min of 11 runs (same harness as
before, main is the merge-base 123747e71):
| case | main | c0ee1c5ff | 19eb8f544 |
|---|---|---|---|
| 1024 x 8192, 1024 values per batch | 58.72 / 56.84 | 65.02 / 64.32 | 59.07
/ 57.90 |
| 256 x 8192, 4096 values per batch | 22.23 / 21.17 | 25.08 / 23.61 | 24.07
/ 22.64 |
| 256 x 8192, 32 values per batch | 7.94 / 7.65 | 8.05 / 7.69 | 7.95 / 7.72 |
| 256 x 8192, one shared 8192-value dictionary | 13.78 / 13.08 | 11.56 /
11.25 | 11.55 / 11.18 |
| 256 x 8192, one shared 131072-value dictionary | 44.04 / 43.50 | 11.50 /
11.01 | 11.16 / 10.96 |
Your first case is back level with main. I couldn't reproduce the gain on
the second one: here it stays about 1 ms over main on the minimum across three
reruns, even though it is a single slice now. My guess, which I haven't
profiled, is that hashing the 1M-value merged dictionary in one pass is colder
than hashing each 4096-value dictionary while it's still in cache, so that part
comes from concatenating first rather than from the slice length. My container
numbers are noisy, so treat that one with some care.
##########
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:
Done in 19eb8f544 with your suggestion: outside prepared mode the hashes
buffer is reserved when the slice is longer than a batch, and released with the
buffer once the map is filled. `join_build_hashes_buffer_is_reserved` builds 8
batches of 1024 rows on a dictionary key that all batches share (512 values, so
one 8192-row slice). It fails with "Resources exhausted" when the limit covers
the concatenated batch and the map but only half the buffer, and passes with
1.5 times the buffer. The first case passed on c0ee1c5ff.
This makes dictionary-key builds near the memory limit fail where they used
to overshoot quietly, the same as prepared mode already does. If you'd prefer,
a failed reservation could instead fall back to batch-sized slices, trading
memory for the extra hashing. I kept the plain error since it is simpler and
matches prepared mode.
--
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]
