kosiew commented on code in PR #24573:
URL: https://github.com/apache/datafusion/pull/24573#discussion_r3851730660
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -1664,60 +1666,129 @@ impl MaterializingSortMergeJoinStream {
}
// Multiple source batches: map each buffered_batch_idx to a
- // contiguous source index, reserving source 0 for a null sentinel.
- let mut batch_idx_to_source: HashMap<usize, usize> = HashMap::new();
+ // contiguous source index. A null sentinel array is prepended as
+ // source 0 only when some right index is actually null (an
+ // unmatched streamed row inside an otherwise matched chunk);
+ // `interleave` walks a null buffer for *every* output row as soon as
+ // any input is nullable, so an always-present sentinel would tax the
+ // common all-matched case.
+ let needs_null_sentinel = matched_chunks
+ .iter()
+ .any(|(_, _, right)| right.null_count() > 0);
+ let source_offset = usize::from(needs_null_sentinel);
+
+ // A group spans only a handful of buffered batches, so a linear
+ // scan beats hashing here. Measured over 8192 rows in 2048 chunks,
+ // against a `HashMap<usize, usize>` built in one pass and read back
+ // in a second (what this used to do):
+ //
+ // distinct sources | hashmap | linear scan
+ // -----------------+-----------+-------------
+ // 4 | 21.5 us | 5.0 us
+ // 16 | 22.0 us | 9.4 us
+ // 32 | 22.4 us | 13.6 us
+ // 64 | 22.9 us | 23.5 us
+ // 128 | 24.1 us | 44.8 us
+ //
+ // `std::collections::HashMap` hashes with SipHash-1-3, so a single
+ // `usize` lookup costs several ns of serial latency before the probe
+ // begins, while a scan over a handful of `usize` is one L1-resident
+ // cache line with a perfectly predicted trip count. The map is also
+ // purely additive state: `source_batches` has to be built regardless
+ // (`source_data` is gathered from it), so hashing means maintaining
+ // two containers holding the same keys.
+ //
+ // The crossover is ~32 distinct sources. That bound follows from how
+ // pairs accumulate, not from any assumption about key skew:
+ //
+ // 1. `pair_streamed_row_with_group` appends exactly one pair per
+ // buffered row and re-checks `num_unfrozen_pairs() < batch_size`
+ // before each append, so at most `batch_size` pairs accumulate
+ // between two `freeze_streamed()` calls.
+ // 2. `BufferedData::scanning_advance` walks the group's rows in
+ // order, so those pairs cover a *contiguous run* of buffered
+ // rows.
+ // 3. So the distinct `buffered_batch_idx` values seen here are the
+ // batches spanned by at most `batch_size` consecutive buffered
+ // rows: `len(source_batches) <= batch_size / R + 1`, where `R`
+ // is the smallest buffered batch in that run.
+ //
+ // The assumption is therefore not "key groups are narrow" — a group
+ // of any width still only contributes `batch_size` rows per freeze —
+ // but "buffered batches are not tiny relative to `batch_size`".
+ // Exceeding 32 sources needs `R < batch_size / 31`, i.e. under ~264
+ // rows per batch at the default `batch_size` of 8192. The buffered
+ // side of a merge join is sorted input, and every operator that
+ // normally feeds it emits ~`batch_size` batches: `SortExec` chunks
+ // its output with `sort_batch_chunked(.., batch_size)`, and
+ // `FilterExec` and `RepartitionExec` each embed a
+ // `LimitedBatchCoalescer` targeting `batch_size`.
+ //
+ // If something does feed tiny batches, this degrades gradually rather
+ // than falling off a cliff, and never affects correctness: at 4
+ // sources this loop is ~13% of the cost of the `interleave` calls it
+ // feeds (3 columns, 8192 rows), so even the 128-source case above
+ // leaves `interleave` the dominant term.
let mut source_batches: Vec<usize> = Vec::new();
- for (batch_idx, _, _) in matched_chunks {
- batch_idx_to_source.entry(*batch_idx).or_insert_with(|| {
- let idx = source_batches.len() + 1;
- source_batches.push(*batch_idx);
- idx
- });
- }
-
let mut interleave_indices: Vec<(usize, usize)> =
Vec::with_capacity(total_matched_rows);
for (batch_idx, _, right) in matched_chunks {
- let source = batch_idx_to_source[batch_idx];
- for i in 0..right.len() {
- if right.is_null(i) {
- interleave_indices.push((0, 0));
- } else {
- interleave_indices.push((source, right.value(i) as usize));
+ let source = match source_batches.iter().position(|b| b ==
batch_idx) {
Review Comment:
I think the performance concern from the previous review still applies here.
`source_batches.iter().position(...)` does a linear scan for every matched
chunk, so source-index construction becomes quadratic in the number of distinct
buffered batches in a freeze.
The bound described above is `len(source_batches) <= batch_size / R + 1`,
but `R` can validly be 1 if a child emits tiny batches. With the default
`batch_size` of 8192, that can mean up to roughly 8192 sources and about 33
million comparisons in one freeze.
Since `SortMergeJoinExec` accepts arbitrary `ExecutionPlan` children, the
batching behavior of the common in-tree producers is not a contract we can rely
on here.
Could we retain the `HashMap` lookup, or use the linear scan as a
small-source fast path and fall back to a map once the source count crosses the
measured crossover?
--
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]
