dwsmith1983 commented on code in PR #26793:
URL: https://github.com/apache/datafusion/pull/26793#discussion_r4443251591
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -3217,28 +3241,39 @@ 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: a
+ // slice at least as long as the longest of those arrays keeps that
+ // cost linear in the build rows. No slice is longer than the build
side.
+ let chunk_rows = left_values
+ .iter()
+ .map(|values| unsliced_hash_len(&values.to_data()))
+ .fold(max_batch_rows, usize::max)
+ .min(num_rows);
Review Comment:
Thanks @jayzhan211, good catch. Concat merges the per-batch dictionaries, so
with slices only as long as the merged dictionary, every slice rehashed about
as many values as it had rows. c0ee1c5ff sizes slices at 16 times the shared
values, capped at 2^20 rows unless the values alone are longer, so a large
merged dictionary can't grow the hashes buffer and prepared scratch to the
whole build side. Keys without a dictionary or list view still use the largest
batch. One side effect: with dictionary keys, prepared mode now reserves
scratch for the longer slice (up to about 8 MiB per partition at the cap).
Your case locally (release, median `build_time` of 9 runs, 256 x 8192 rows,
a separate 32-value dictionary per batch): main 7.93 ms, 8a4222e9a 10.28 ms,
now 7.89 ms. With 1024 values per batch (262k merged, where the cap applies):
11.64 / 14.36 / 12.92 ms. When every build row has its own dictionary value (2M
merged), both heads are about 6% over main, since the whole build side is a
single slice there.
`join_keeps_build_order_with_dictionary_keys` now fills from two slices that
cross a batch boundary, since the old data fit in one slice under the new rule.
Hashing shared dictionary values once across `create_hashes` calls is
#26780. Its hash join part (#26815) was closed after the benchmark showed no
gain, so I kept this a local sizing rule.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -3234,18 +3232,19 @@ fn build_in_memory(
}
// The maximum is known: avoid geometric growth and its excess
capacity.
let mut hashes_buffer = Vec::with_capacity(max_batch_rows);
- let mut offset = num_rows;
-
- // Update the hashmap starting from the last batch, so rows with
smaller
- // indices end up on top of the chains. Each batch is indexed at its
- // position in the forward concatenation below.
- for batch in batches.iter().rev() {
- offset -= batch.num_rows();
+ // `max_batch_rows` is 0 when the build side is empty
+ let chunk_rows = max_batch_rows.max(1);
+
+ // Update the hashmap from slices of at most `max_batch_rows` rows,
+ // starting from the last one, so rows with smaller indices end up on
+ // top of the chains. Each slice is indexed at its offset in `batch`.
+ for offset in (0..num_rows).step_by(chunk_rows).rev() {
+ let chunk = batch.slice(offset, chunk_rows.min(num_rows - offset));
Review Comment:
@2010YOUY01 following up on the issue I offered: it already exists as #26780
(hash shared dictionary values once and reuse them across `create_hashes` calls
and operators), so I didn't file a new one. Its hash join part, #26815, was
closed after its benchmark showed no gain; for this PR the per-slice cost is
now bounded by slice sizing instead (c0ee1c5ff, numbers in my reply on the
`chunk_rows` thread).
--
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]
