Skip to content

Commit c0ee1c5

Browse files
dwsmith1983dwsmith1983
committed
perf: amortize shared dictionary hashing over longer build slices
Concatenating the build side merges per-batch dictionaries, and every slice hashes the whole merged values array. With slices only as long as those values, that doubled the hashing work when each batch carried its own small dictionary. Slices are now 16 times the shared values, capped at 2^20 rows unless the values alone are longer, so the hashes buffer and prepared scratch stay bounded. Keys without a dictionary or list view keep slices of the largest batch.
1 parent 8a4222e commit c0ee1c5

2 files changed

Lines changed: 137 additions & 49 deletions

File tree

‎datafusion/physical-plan/src/joins/hash_join/exec.rs‎

Lines changed: 132 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,39 @@ fn unsliced_hash_len(data: &ArrayData) -> usize {
119119
.fold(own, usize::max)
120120
}
121121

122+
/// A build slice is this many times longer than the shared values beneath a
123+
/// dictionary or list view that it rehashes, so rehashing them adds at most
124+
/// about 1/16 to the hashing work.
125+
const SHARED_VALUES_SLICE_FACTOR: usize = 16;
126+
127+
/// The headroom from [`SHARED_VALUES_SLICE_FACTOR`] stops at this many rows
128+
/// (8 MiB of hashes), so a large merged dictionary cannot grow the hashes
129+
/// buffer and scratch space to the whole build side. Past it only the
130+
/// rehashing overhead grows, and it stays linear in the build rows.
131+
const MAX_HASH_CHUNK_ROWS: usize = 1 << 20;
132+
133+
/// The number of rows in each slice the hash map is filled from.
134+
///
135+
/// A slice is never shorter than the largest build batch or the longest
136+
/// shared values under any key, which keeps hashing linear in the build rows.
137+
/// The overhead ratio is per key column, so the longest key counts, not the sum.
138+
/// Beyond that, a slice is [`SHARED_VALUES_SLICE_FACTOR`] times the shared
139+
/// values, up to [`MAX_HASH_CHUNK_ROWS`], and never longer than the build side.
140+
fn hash_chunk_rows(
141+
left_values: &[ArrayRef],
142+
max_batch_rows: usize,
143+
num_rows: usize,
144+
) -> usize {
145+
let shared = left_values
146+
.iter()
147+
.map(|values| unsliced_hash_len(&values.to_data()))
148+
.max()
149+
.unwrap_or(0);
150+
let base = max_batch_rows.max(shared);
151+
let amortized = base.max(shared.saturating_mul(SHARED_VALUES_SLICE_FACTOR));
152+
amortized.min(base.max(MAX_HASH_CHUNK_ROWS)).min(num_rows)
153+
}
154+
122155
/// The longest array anywhere in `data`, including `data` itself.
123156
fn subtree_max_len(data: &ArrayData) -> usize {
124157
data.child_data()
@@ -704,10 +737,11 @@ impl From<&HashJoinExec> for HashJoinExecBuilder {
704737
/// hash table is updated by iterating the build-side rows in reverse order, which keeps rows
705738
/// with smaller indices "on the top" of hash table. The build-side batches are first
706739
/// concatenated in their input order. The hash table is then filled from slices of the
707-
/// concatenated batch, at most as long as the largest build-side batch (or the longest
708-
/// values under a dictionary or list view in a key, when that is longer), starting
709-
/// from the last slice. Each slice is inserted with the offset of its first row in
710-
/// the concatenated batch.
740+
/// concatenated batch, as long as the largest build-side batch. When a key has values
741+
/// under a dictionary or list view, which every slice hashes in full, slices grow to 16
742+
/// times the length of those values, capped at 2^20 rows unless the values alone are
743+
/// longer. Filling starts from the last slice, and each slice is inserted with the offset
744+
/// of its first row in the concatenated batch.
711745
///
712746
/// Example of build phase for 3 record batches, filled from slices of 3 rows:
713747
///
@@ -3242,14 +3276,11 @@ fn build_in_memory(
32423276
let mut hashmap = new_join_hashmap(num_rows, &mut reservation, metrics)?;
32433277

32443278
// Every slice shares everything beneath a dictionary or list view,
3245-
// at any depth of a key, which `create_hashes` hashes on each call: a
3246-
// slice at least as long as the longest of those arrays keeps that
3247-
// cost linear in the build rows. No slice is longer than the build side.
3248-
let chunk_rows = left_values
3249-
.iter()
3250-
.map(|values| unsliced_hash_len(&values.to_data()))
3251-
.fold(max_batch_rows, usize::max)
3252-
.min(num_rows);
3279+
// at any depth of a key, which `create_hashes` hashes on each call:
3280+
// slices `SHARED_VALUES_SLICE_FACTOR` times longer than those values
3281+
// (up to `MAX_HASH_CHUNK_ROWS`, never shorter than the values or the
3282+
// largest batch) keep that cost small.
3283+
let chunk_rows = hash_chunk_rows(&left_values, max_batch_rows, num_rows);
32533284

32543285
let scratch_reservation = reservation.new_empty();
32553286
if prepared {
@@ -5233,9 +5264,52 @@ mod tests {
52335264
Ok(())
52345265
}
52355266

5236-
/// Build batches sharing one key dictionary that is longer than each batch
5237-
/// fill the hash map from slices longer than the batches, which must join
5238-
/// as if the build side were a single batch.
5267+
/// A key with shared values gets slices `SHARED_VALUES_SLICE_FACTOR` times
5268+
/// longer than those values, up to `MAX_HASH_CHUNK_ROWS` unless the values
5269+
/// alone are longer, and never longer than the build side.
5270+
#[test]
5271+
fn hash_chunk_rows_amortizes_shared_values() -> Result<()> {
5272+
fn dict(values: i32) -> Result<ArrayRef> {
5273+
Ok(Arc::new(DictionaryArray::<Int32Type>::try_new(
5274+
Int32Array::from(vec![0]),
5275+
Arc::new(Int32Array::from_iter_values(0..values)),
5276+
)?))
5277+
}
5278+
let plain: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
5279+
let cases = [
5280+
(
5281+
vec![dict(8192)?],
5282+
8192,
5283+
2_097_152,
5284+
SHARED_VALUES_SLICE_FACTOR * 8192,
5285+
),
5286+
(vec![Arc::clone(&plain)], 8192, 2_097_152, 8192),
5287+
// The largest batch is longer than 16 times the dictionary
5288+
(vec![dict(100)?], 8192, 2_097_152, 8192),
5289+
(vec![Arc::clone(&plain), dict(8192)?], 8192, 50_000, 50_000),
5290+
// Values longer than the cap are the slice length, not 16 times them
5291+
(vec![dict(2_097_152)?], 8192, 4_194_304, 2_097_152),
5292+
(vec![dict(100_000)?], 8192, 4_194_304, MAX_HASH_CHUNK_ROWS),
5293+
// A batch longer than the cap is still the slice length
5294+
(vec![dict(1000)?], 2_097_152, 4_194_304, 2_097_152),
5295+
(vec![], 8192, 0, 0),
5296+
(vec![dict(8)?], 8192, 0, 0),
5297+
];
5298+
for (case, (keys, max_batch_rows, num_rows, expected)) in
5299+
cases.into_iter().enumerate()
5300+
{
5301+
assert_eq!(
5302+
hash_chunk_rows(&keys, max_batch_rows, num_rows),
5303+
expected,
5304+
"case {case}"
5305+
);
5306+
}
5307+
Ok(())
5308+
}
5309+
5310+
/// Build batches sharing one key dictionary fill the hash map from slices
5311+
/// longer than the batches, which must join as if the build side were a
5312+
/// single batch.
52395313
#[tokio::test]
52405314
async fn join_keeps_build_order_with_dictionary_keys() -> Result<()> {
52415315
let dict_type =
@@ -5245,38 +5319,41 @@ mod tests {
52455319
Field::new("b1", dict_type.clone(), true),
52465320
]));
52475321
// Concatenation keeps a dictionary shared by every batch as is, so the
5248-
// 12 build rows are filled from slices of 5 rows: [0, 5), [5, 10) and
5249-
// [10, 12), each crossing the 2-row batch boundaries.
5250-
let values: ArrayRef = Arc::new(StringArray::from(vec!["x", "y", "z", "w", "v"]));
5251-
let batch = |a: Vec<i32>, keys: Vec<Option<i32>>| -> Result<RecordBatch> {
5252-
let b1 = DictionaryArray::<Int32Type>::try_new(
5253-
Int32Array::from(keys),
5254-
Arc::clone(&values),
5255-
)?;
5256-
Ok(RecordBatch::try_new(
5257-
Arc::clone(&schema),
5258-
vec![Arc::new(Int32Array::from(a)), Arc::new(b1)],
5259-
)?)
5260-
};
5261-
let batches = vec![
5262-
batch(vec![1, 2], vec![Some(0), None])?,
5263-
batch(vec![3, 4], vec![Some(1), Some(2)])?,
5264-
batch(vec![5, 6], vec![Some(0), Some(1)])?,
5265-
batch(vec![7, 8], vec![Some(3), Some(0)])?,
5266-
batch(vec![9, 10], vec![None, Some(1)])?,
5267-
batch(vec![11, 12], vec![Some(0), Some(1)])?,
5268-
];
5322+
// 45 build rows, in 3-row batches over 2 dictionary values, are filled
5323+
// from slices of 32 rows: [0, 32) and [32, 45). The first boundary is
5324+
// inside the batch [30, 33), and "x" has matches in both slices.
5325+
let values: ArrayRef = Arc::new(StringArray::from(vec!["x", "y"]));
5326+
// Every fifth row has a null key, the others alternate "x" and "y"
5327+
let key = |a1: i32| (a1 % 5 != 0).then_some((a1 + 1) % 2);
5328+
let batches = (1..=45)
5329+
.step_by(3)
5330+
.map(|first| {
5331+
let a1: Vec<i32> = (first..first + 3).collect();
5332+
let b1 = DictionaryArray::<Int32Type>::try_new(
5333+
a1.iter().map(|&a1| key(a1)).collect(),
5334+
Arc::clone(&values),
5335+
)?;
5336+
Ok(RecordBatch::try_new(
5337+
Arc::clone(&schema),
5338+
vec![Arc::new(Int32Array::from(a1)), Arc::new(b1)],
5339+
)?)
5340+
})
5341+
.collect::<Result<Vec<_>>>()?;
5342+
assert_eq!(
5343+
hash_chunk_rows(&[Arc::clone(batches[0].column(1))], 3, 45),
5344+
32
5345+
);
52695346
let right_schema = Arc::new(Schema::new(vec![
52705347
Field::new("a2", DataType::Int32, false),
52715348
Field::new("b2", dict_type, true),
52725349
]));
52735350
let right = RecordBatch::try_new(
52745351
Arc::clone(&right_schema),
52755352
vec![
5276-
Arc::new(Int32Array::from(vec![10, 30])),
5353+
Arc::new(Int32Array::from(vec![10])),
52775354
Arc::new(DictionaryArray::<Int32Type>::try_new(
5278-
Int32Array::from(vec![0, 1]),
5279-
Arc::new(StringArray::from(vec!["x", "y"])),
5355+
Int32Array::from(vec![0]),
5356+
Arc::new(StringArray::from(vec!["x"])),
52805357
)?),
52815358
],
52825359
)?;
@@ -5301,13 +5378,23 @@ mod tests {
53015378
| a1 | b1 | a2 | b2 |
53025379
+----+----+----+----+
53035380
| 1 | x | 10 | x |
5304-
| 5 | x | 10 | x |
5305-
| 8 | x | 10 | x |
5381+
| 3 | x | 10 | x |
5382+
| 7 | x | 10 | x |
5383+
| 9 | x | 10 | x |
53065384
| 11 | x | 10 | x |
5307-
| 3 | y | 30 | y |
5308-
| 6 | y | 30 | y |
5309-
| 10 | y | 30 | y |
5310-
| 12 | y | 30 | y |
5385+
| 13 | x | 10 | x |
5386+
| 17 | x | 10 | x |
5387+
| 19 | x | 10 | x |
5388+
| 21 | x | 10 | x |
5389+
| 23 | x | 10 | x |
5390+
| 27 | x | 10 | x |
5391+
| 29 | x | 10 | x |
5392+
| 31 | x | 10 | x |
5393+
| 33 | x | 10 | x |
5394+
| 37 | x | 10 | x |
5395+
| 39 | x | 10 | x |
5396+
| 41 | x | 10 | x |
5397+
| 43 | x | 10 | x |
53115398
+----+----+----+----+
53125399
");
53135400
Ok(())

‎datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -233,22 +233,23 @@ pub(super) fn prepared_copy_bytes(batches: &[RecordBatch]) -> Result<usize> {
233233
.ok_or_else(overflow)
234234
}
235235

236-
/// Bound hashes and flat-key null masks for the largest input batch.
236+
/// Bound hashes and flat-key null masks for the largest slice the hash map is
237+
/// filled from.
237238
pub(super) fn prepared_scratch_bytes(
238-
max_batch_rows: usize,
239+
chunk_rows: usize,
239240
num_keys: usize,
240241
null_equality: NullEquality,
241242
) -> Result<usize> {
242243
let overflow = || exec_datafusion_err!("Prepared hash-join scratch size overflow");
243-
let hash_bytes = max_batch_rows
244+
let hash_bytes = chunk_rows
244245
.checked_mul(size_of::<u64>())
245246
.ok_or_else(overflow)?;
246247
if null_equality == NullEquality::NullEqualsNull {
247248
return Ok(hash_bytes);
248249
}
249250
// Allow one logical null mask per key plus the combined mask.
250251
let masks = num_keys.checked_add(1).ok_or_else(overflow)?;
251-
let mask_bytes = max_batch_rows
252+
let mask_bytes = chunk_rows
252253
.div_ceil(8)
253254
.checked_add(64)
254255
.and_then(|bytes| bytes.checked_mul(masks))

0 commit comments

Comments
 (0)