Repository navigation
Conversation
Adds `sort_merge_join` benchmark cases for Inner joins with a filter: - `inner_1to10_filtered` and `inner_1to100_filtered`, in which every streamed row matches 10 or 100 buffered rows and `left.data < right.data` keeps about half of the pairs. - `inner_1to10_high_pass_filtered`, in which `left.data <> right.data` keeps 90% of the pairs, and its `_narrow` variant over just the `key` and `data` columns, where the filter reads half of the output columns. - `inner_1to1000_selective_filtered`, in which every streamed row matches 1000 buffered rows and `left.data > right.data` keeps about 5% of the 2M pairs. This is the shape of TPC-DS q72's `catalog_sales` to `inventory` join.
An Inner sort-merge join with a filter materialized every column of every pair that matched on the join keys, and only then evaluated the filter and dropped the pairs that failed it. When a key matches many rows on each side and the filter keeps few of the pairs, most of the join's time went into copying rows that were then dropped. An Inner join with a filter now evaluates it before materializing: - A run of at least 256 buffered rows is filtered as it is paired with a streamed row. The filter reads the run as a slice of its buffered batch and the streamed row's values repeated, and only the pairs that pass are recorded, in chunks marked as filtered. - The pairs of shorter runs are filtered at freeze on just the columns the filter reads, and the output columns are materialized for the pairs that pass. - When earlier freezes kept most of their pairs and the filter reads most of the output columns, the freeze materializes the batch whole and filters it, as before. Outer joins keep the deferred filtering they need for null-joined rows, and joins without a filter keep the original pairing loop.
- Drop the `streamed_side` field: only an Inner join filters early, so the streamed side is always the left input. - Compute the buffered columns the filter reads once, not on every freeze. - Repeat the streamed row with `take` instead of a `ScalarValue` round trip. - Make `materialize_right_columns` generic over `Borrow<MatchedChunk>`, so callers pass chunks without collecting a `Vec<&MatchedChunk>` and kept chunks are borrowed rather than cloned. - Reuse `evaluate_join_filter` in the outer join path. - Build one `FilterPredicate` for both index arrays in `MatchedChunk::filter`. - Fold the column-free filter test into the hash join comparison test.
The always-false column-free filter case makes the hash join oracle return no batches, so take the schema as a parameter instead of reading it from the first batch.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Heads up: this is an AI-assisted review pass. I've checked the points below against e2f4657, but I may have missed context or misjudged the intent, so please push back on anything that doesn't hold up.
Thanks @comphead, this is a big win for q72's shape, and the description explains the design well. Two things I particularly like: chunks filtered while pairing are marked, so the freeze never runs the filter twice, and the FILTERED_CHUNK_KEY sentinel leaves chunk_for, the 1:1 hot path, unchanged.
What I checked:
-
Results: a throwaway test ran 1,800 Inner joins against 58325af and again against 378d525, whose join code e2f4657 keeps. Each compared the join's output, rows and order, with the unfiltered join's output filtered row by row. The cases covered:
- groups of up to 1,400 buffered rows, in buffered batches of 1 to 3,255 rows
batch_sizefrom 1 to 8192- six filters:
c1 < c2, your interleaved and high-pass ones, one that keeps about 2% of the pairs, and column-freetrueandfalse - spilling, in a quarter of the joins
Every output matched. The test also catches each of the mutations listed in the inline comment on the test inputs.
-
Benchmarks: your numbers reproduce on an M3 Max. These are medians of 5 alternating rounds of saved binaries, 9d916ab (
mainplus your bench commit) against 378d525:Benchmark mainThis PR inner_1to10_filtered/10000011.97 ms 6.48 ms inner_1to100_filtered/2000023.32 ms 11.77 ms inner_1to10_high_pass_filtered/10000011.55 ms 9.21 ms inner_1to10_high_pass_filtered_narrow/1000004.22 ms 4.27 ms inner_1to1000_selective_filtered/2000018.00 ms 2.96 ms The other ten cases moved by 1.6% or less.
-
Tests: the 499 SMJ tests pass on e2f4657. Thanks for the quick
prepare_record_batches_for_cmpfix.
Before approving, I'd like to see:
- The tests reaching the code that filters while pairing. None of them do at the moment (inline), and changing one number fixes it.
- A look at freezes made of many small chunks when the filter keeps most pairs. These are now slower than on
main: 1.19× with 8-row buffered batches (inline). It's narrow, but this PR introduces it.
On the two estimates you asked about: my measurements for the 256-row threshold are inline. In short, filtering while pairing only pays off for selective filters, on runs of about 512 rows or more. For the cost model, the small-chunk freezes in item 2 are the only cases I found where it picks the slower path.
| 40, | ||
| |k| if k % 3 == 0 { 600 } else { (k % 3) as usize }, | ||
| 2, | ||
| 250, |
There was a problem hiding this comment.
A run never crosses a buffered batch, so with 250-row batches every run is shorter than FILTERED_RUN_MIN_ROWS (256). That means no test reaches pair_filtered_run, its fallback for spilled batches, or a freeze holding chunks that were filtered while pairing. I added probes to those branches, and none of them fired in any of the 499 SMJ tests.
These mutations all pass the tests today:
pair_filtered_runignores the filter result and keeps every pair.start_filtered_chunksets the chunk key toSome(buffered_batch_idx)rather thanFILTERED_CHUNK_KEY, so later unfiltered pairs land in the filtered chunk.freeze_inner_filtereddrops the chunks that were filtered while pairing.pair_filtered_runreturnsOk(Some(0))for a spilled batch, which drops the run.
With 300 here, all of those branches run, and each mutation makes join_inner_early_filtering_matches_hash_join or the spill test fail. Could you change it to 300? Deriving it from FILTERED_RUN_MIN_ROWS (made pub(super)) would keep the tests on this path if the threshold changes.
There was a problem hiding this comment.
Confirmed: scanning_remaining stops at the end of the scanning batch, so no run in these inputs is longer than 250 rows. Filtering while pairing is removed now, so I probed the freeze branches instead. The tests reach:
- the whole-batch path, through the range check and through the cost model
- filtering first, with one and with several buffered batches, each with partial and full passes
Three mutations each make the tests fail: dropping the kept-offset advance, skipping the whole-batch filter, and reversing the kept chunks. Skipping the whole-batch filter used to pass, because the high-pass filter kept every pair. It is now c2 <> c1 OR b2 < b1, which rejects the pairs with a NULL or equal c.
| /// it pairs the run with a streamed row. Each such evaluation has a fixed cost | ||
| /// that a long run amortizes. Shorter runs are paired unfiltered, and the | ||
| /// freeze filters a whole batch of their pairs at once. | ||
| const FILTERED_RUN_MIN_ROWS: usize = 256; |
There was a problem hiding this comment.
Since you asked about this threshold, I measured it with Inner joins of 16 streamed rows and R buffered rows per key, 2M pairs each, using the bench's key/data/payload schema. The filters keep 5%, 50% or 95% of the pairs. The table shows this PR's time divided by the time of the same build with the threshold at usize::MAX, which filters everything at freeze (median of 5 rounds):
| R | 5% kept | 50% kept | 95% kept |
|---|---|---|---|
| 256 | 1.17 | 1.14 | 1.19 |
| 512 | 0.87 | 1.05 | 1.13 |
| 2048 | 0.70 | 1.00 | 1.11 |
So filtering while pairing only wins for selective filters, and only from about 512 rows. A lower threshold is worse: at 64, R = 128 with 5% kept took 1.75× as long as filtering at freeze. Everything in the table is still faster than main (0.13× to 0.96×), so this is speed left on the table rather than a regression.
Would you consider raising the threshold to 512, and filtering while pairing only while the share of pairs that pass stays low? The pairing-time evaluations could feed the same counts that filter_first_pays uses. inner_1to1000_selective_filtered would keep its win.
There was a problem hiding this comment.
Would you consider raising the threshold to 512, and filtering while pairing only while the share of pairs that pass stays low?
Your table makes me want to go one step further and drop filtering while pairing from this PR. Filtering at freeze is as fast or faster in 7 of its 9 cells. It is slower only for selective filters on runs of 512 rows or more (0.87 and 0.70).
There is also a case where filtering while pairing is slower than main. I ran the shape of inner_1to1000_selective_filtered (20 keys, 100 streamed and 1,000 buffered rows per key) with left.data > right.data keeping 90% of the pairs at random. The baseline is 9d916ab (main plus the bench commit). These are medians of 3 alternating rounds on Apple silicon with thin LTO, and speedup is main time divided by PR time:
| Schema per side | main |
This PR | Speedup |
|---|---|---|---|
key, data, payload |
19.51 ms | 15.49 ms | 1.26x |
key, data |
6.06 ms | 7.12 ms | 0.85x |
The bench code is in the comment on filter_first_pays.
The second pairing loop, the 256-row threshold, FILTERED_CHUNK_KEY, the filtered flag on chunks, the Cow handling, and the fallback for spilled batches exist only for filtering while pairing. As your other thread found, no test reaches that path yet.
@comphead, could you rerun q72 in Comet with filtering at freeze only? Setting FILTERED_RUN_MIN_ROWS to usize::MAX is enough for the measurement. If that keeps most of the drop from 270 s to 40 s, I'd ship filtering at freeze here, and filtering while pairing can come as its own PR if q72 shows it's needed.
There was a problem hiding this comment.
Thanks for measuring this. Your table matches what I see. Rather than tune the threshold, 8012102 follows @mbutrovich's suggestion and drops filtering while pairing. The numbers are in the top-level comment. I'll post the q72 result from Comet with filtering at freeze only once I have it.
| if count == len { | ||
| kept.push(Cow::Borrowed(chunk)); | ||
| } else if count > 0 { | ||
| kept.push(Cow::Owned(chunk.filter(&mask)?)); |
There was a problem hiding this comment.
When a key group spans many small buffered batches, a freeze holds about one chunk per buffered batch. This line then builds a FilterBuilder and runs two filter kernels for every chunk that partly passes. I tried the shape of inner_group_spans_buffered_batches with left.data < right.data keeping 95% of the pairs. Against main, this PR takes 1.19× as long with 8-row buffered batches (14.8 to 17.7 ms) and 1.05× with 64-row ones.
Filtering the concatenated indices of the unfiltered chunks once, and giving each partly passing chunk a slice of the result, brings those to 1.11× and 0.94×. It also helps when the filter keeps 5% of the pairs: 0.89× to 0.82× with 8-row batches, and 0.51× to 0.43× with 64-row ones. The diff is below. It passes the SMJ tests and the differential test from the review summary.
The rest of the 8-row slowdown might need filter_first_pays to count chunks, so that a freeze of many tiny chunks that keep most pairs takes the whole-batch path.
Diff against e2f4657
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
index 8f0b5b1ce..5f222db4b 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
@@ -79,25 +79,6 @@ impl MatchedChunk {
fn len(&self) -> usize {
self.streamed_indices.len()
}
-
- /// Keeps the pairs whose bit is set in `mask`.
- fn filter(&self, mask: &BooleanBuffer) -> Result<Self> {
- let predicate = FilterBuilder::new(&BooleanArray::new(mask.clone(), None))
- .optimize()
- .build();
- Ok(Self {
- buffered_batch_idx: self.buffered_batch_idx,
- streamed_indices: as_uint64_array(
- &predicate.filter(&self.streamed_indices)?,
- )?
- .clone(),
- buffered_indices: as_uint64_array(
- &predicate.filter(&self.buffered_indices)?,
- )?
- .clone(),
- filtered: self.filtered,
- })
- }
}
/// Represents a chunk of joined data from streamed and buffered side
@@ -2067,30 +2048,44 @@ impl MaterializingSortMergeJoinStream {
})
.collect::<Result<Vec<_>>>()?;
self.freeze_filter_evaluated += num_rows;
- Some(evaluate_join_filter(
- filter,
- filter.schema(),
- columns,
- num_rows,
- )?)
+ let passed =
+ evaluate_join_filter(filter, filter.schema(), columns, num_rows)?;
+ // Filter the index arrays of all unfiltered chunks at once; each partly
+ // passing chunk below takes a slice of the result.
+ let predicate = FilterBuilder::new(&BooleanArray::new(passed.clone(), None))
+ .optimize()
+ .build();
+ let buffered_indices =
+ concat_indices(unfiltered.iter().map(|c| &c.buffered_indices))?;
+ let kept_streamed =
+ as_uint64_array(&predicate.filter(&streamed_indices)?)?.clone();
+ let kept_buffered =
+ as_uint64_array(&predicate.filter(&buffered_indices)?)?.clone();
+ Some((passed, kept_streamed, kept_buffered))
};
// Keep the pairs that passed, in order.
let mut kept: Vec<Cow<MatchedChunk>> = Vec::with_capacity(matched_chunks.len());
let mut kept_rows = 0;
let mut offset = 0;
+ let mut kept_offset = 0;
for chunk in matched_chunks {
let len = chunk.len();
match (&passed, chunk.filtered) {
- (Some(passed), false) => {
- let mask = passed.slice(offset, len);
+ (Some((passed, kept_streamed, kept_buffered)), false) => {
+ let count = passed.slice(offset, len).count_set_bits();
offset += len;
- let count = mask.count_set_bits();
if count == len {
kept.push(Cow::Borrowed(chunk));
} else if count > 0 {
- kept.push(Cow::Owned(chunk.filter(&mask)?));
+ kept.push(Cow::Owned(MatchedChunk {
+ buffered_batch_idx: chunk.buffered_batch_idx,
+ streamed_indices: kept_streamed.slice(kept_offset, count),
+ buffered_indices: kept_buffered.slice(kept_offset, count),
+ filtered: false,
+ }));
}
+ kept_offset += count;
self.freeze_filter_passed += count;
kept_rows += count;
}There was a problem hiding this comment.
Applied in 8012102, thanks. The freeze filters the concatenated index arrays once, and each chunk takes a slice of the result. Medians against 9d916ab, as a ratio to main, with both builds filtering at freeze only:
| Case | chunk by chunk | all at once |
|---|---|---|
| 8-row batches, 95% kept | 1.18 | 1.10 |
| 64-row batches, 95% kept | 1.07 | 1.01 |
| 8-row batches, 5% kept | 0.98 | 0.90 |
| 64-row batches, 5% kept | 0.68 | 0.57 |
I left the remaining 1.10x with 8-row batches. Counting chunks in filter_first_pays would need a per-chunk cost that I'd have to guess, and buffered batches that small should be rare once input is coalesced.
There was a problem hiding this comment.
The test's 600-row groups are split into 250-row batches, but the pairing threshold is 256. Does any test actually hit the filtered pairing path, including its spilled-batch branch?
There was a problem hiding this comment.
No, as @andygrove found too: no run here is longer than 250 rows. 8012102 removes that path, and the tests now reach every branch of the freeze path (details in andygrove's thread).
There was a problem hiding this comment.
What happens if the join filter is volatile or stateful? Early evaluation changes when and how often it runs.
There was a problem hiding this comment.
Each pair is still evaluated exactly once, as on main. Either the filter runs before materializing, or it runs on the whole batch, and no pair is evaluated twice or skipped. What changes is which batch of pairs an evaluation sees, and that already depends on batch_size on main. column_free_early_filter checks that a filter reading no columns still runs for every pair.
There was a problem hiding this comment.
The tests compare sorted results and only assert key order. Does anything verify exact row order within a key?
There was a problem hiding this comment.
Nothing did. In 8012102, assert_streamed_order asserts that (b1, b2) ascends. These are the row numbers of the two inputs, so the check covers the order within a key. Reversing the kept chunks now fails it.
There was a problem hiding this comment.
Why does the pairing path use only a fixed 256-row threshold while the freeze path uses a cost model? Could high-pass wide joins regress?
There was a problem hiding this comment.
The fixed threshold belonged to filtering while pairing, which 8012102 removes. Its high-pass regression was real: up to 1.5x slower than main with a narrow schema (table in the top-level comment). The freeze path keeps the cost model and adds a range check for 1:1 joins.
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @comphead. The q72 profile makes a clear case for filtering pairs before materializing them.
In the context of #26257, I'm concerned that this PR complicates a stream that #23976 refactored for simplicity, and does it for a degenerate case in Comet: a many-to-many join whose filter rejects most pairs. I'd like to make sure we're optimizing for the common case, joins where each streamed row matches one or a few buffered rows, and that this PR is the smallest change that fixes q72.
| /// Whether filtering the pairs of a freeze before materializing them is cheaper | ||
| /// than materializing every column of every pair and filtering the batch. | ||
| /// | ||
| /// Filtering first gathers the filter's `f` columns for every pair, filters the | ||
| /// pairs' two index arrays, and gathers all `k` output columns for the share | ||
| /// `p` of pairs that pass. The other way gathers the `k` columns for every | ||
| /// pair and copies the share `p` of them to filter the batch. Counting a | ||
| /// gathered value as 1 and a copied one as 1/2, filtering first is cheaper | ||
| /// while `f + 1 + k·p < k + k·p/2`, that is while `p < 2·(k - f - 1)/k`. `p` is | ||
| /// the share that passed in the earlier freezes, or 0 before the first. | ||
| fn filter_first_pays(&self, filter: &JoinFilter) -> bool { | ||
| let k = self.schema.fields().len() as f64; | ||
| let f = filter.column_indices().len() as f64; | ||
| let p = if self.freeze_filter_evaluated == 0 { | ||
| 0.0 | ||
| } else { | ||
| self.freeze_filter_passed as f64 / self.freeze_filter_evaluated as f64 | ||
| }; | ||
| p < 2.0 * (k - f - 1.0) / k |
There was a problem hiding this comment.
This model assumes that the whole-batch path gathers all k columns for every pair. When a freeze's indices form a contiguous range, main slices the columns instead (materialize_left_columns and the buffered side). That is the usual case in a 1:1 join where every row matches, and there the whole-batch path costs little more than filter_record_batch. Filtering first keeps the slices only when every pair of a chunk passes. Otherwise chunk.filter leaves gaps in the indices, and every output column goes through take. With the bench schema (k = 6, f = 2), the threshold is 2(6 - 2 - 1)/6 = 1.0, so every freeze below 100% pass filters first.
I measured 1:1 Inner joins of 100K rows per side, with the bench's three-column schema and left.data < right.data, against 9d916ab. These are medians of 3 alternating rounds on Apple silicon with thin LTO:
| Case | main |
This PR | Speedup |
|---|---|---|---|
| 1:1, no filter (unchanged path) | 1.403 ms | 1.433 ms | 0.98x |
| 1:1, 100% pass | 1.424 ms | 1.530 ms | 0.93x |
| 1:1, 90% pass, every 10th row fails | 1.692 ms | 2.018 ms | 0.84x |
| 1:1, 90% pass, random | 1.788 ms | 2.037 ms | 0.88x |
| 1:1, 50% pass, random | 1.745 ms | 1.876 ms | 0.93x |
| 1:1, 5% pass, random | 1.483 ms | 1.590 ms | 0.93x |
| N:1, 10 streamed rows per key, 90% pass | 1.531 ms | 1.522 ms | 1.01x |
| N:1, 10 streamed rows per key, 50% pass | 1.452 ms | 1.349 ms | 1.08x |
The 100% case keeps the slices and is still 7% slower, against 2% for the unfiltered control. Every streamed row of these joins goes through pair_streamed_row_with_group_filtered, so dropping filtering while pairing may remove that part. The 90% cases, which lose the slices, drop further to 0.84x and 0.88x.
Could the freeze take the whole-batch path when its streamed and buffered indices are both contiguous? That checks the freeze at hand, so it adds no state. A 1:1 high-pass case in benches/sort_merge_join.rs would cover it.
Bench cases (appended to benches/sort_merge_join.rs)
fn mix(i: u64) -> u64 {
let mut z = i.wrapping_add(0x9E37_79B9_7F4A_7C15);
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
/// Batches of 8192 rows with the given (already key-sorted) keys and data.
fn degen_batches(keys: Vec<i64>, data: Vec<i64>, narrow: bool) -> (SchemaRef, Vec<RecordBatch>) {
let s = if narrow { narrow_schema() } else { schema() };
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(Int64Array::from(keys)),
Arc::new(Int64Array::from(data.clone())),
];
if !narrow {
let payload: Vec<String> = (0..data.len()).map(|i| format!("val_{i}")).collect();
cols.push(Arc::new(StringArray::from(payload)));
}
let batch = RecordBatch::try_new(Arc::clone(&s), cols).unwrap();
let mut out = vec![];
let mut off = 0;
while off < batch.num_rows() {
let len = (batch.num_rows() - off).min(8192);
out.push(batch.slice(off, len));
off += len;
}
(s, out)
}
fn bench_degenerate(c: &mut Criterion) {
let rt = Runtime::new().unwrap();
let mut group = c.benchmark_group("smj_degen");
let n: usize = 100_000;
// 1:1, unique keys on both sides, every streamed row matches.
// left.data < right.data where right.data = t, left.data in 0..100.
for (name, filtered, t, random) in [
("1to1_unfiltered", false, 0, true),
("1to1_pass100", true, 100, true),
("1to1_pass90_regular", true, 90, false),
("1to1_pass90_random", true, 90, true),
("1to1_pass50_random", true, 50, true),
("1to1_pass5_random", true, 5, true),
] {
let keys: Vec<i64> = (0..n as i64).collect();
let ldata: Vec<i64> = (0..n as u64)
.map(|i| if random { (mix(i) % 100) as i64 } else { (i % 10 * 10) as i64 })
.collect();
let (s, left) = degen_batches(keys.clone(), ldata, false);
let (_, right) = degen_batches(keys, vec![t; n], false);
group.bench_function(BenchmarkId::new(name, n), |b| {
b.iter(|| {
do_join_with_filter(
make_exec(&left, &s),
make_exec(&right, &s),
datafusion_common::JoinType::Inner,
filtered.then(|| data_filter(Operator::Lt)),
&rt,
)
})
});
}
// N:1 (FK to PK): 10 streamed rows per key, one buffered row per key.
for (name, t) in [("nto1_pass90_random", 90), ("nto1_pass50_random", 50)] {
let lkeys: Vec<i64> = (0..n as i64).map(|i| i / 10).collect();
let ldata: Vec<i64> = (0..n as u64).map(|i| (mix(i) % 100) as i64).collect();
let (s, left) = degen_batches(lkeys, ldata, false);
let (_, right) = degen_batches((0..(n / 10) as i64).collect(), vec![t; n / 10], false);
group.bench_function(BenchmarkId::new(name, n), |b| {
b.iter(|| {
do_join_with_filter(
make_exec(&left, &s),
make_exec(&right, &s),
datafusion_common::JoinType::Inner,
Some(data_filter(Operator::Lt)),
&rt,
)
})
});
}
// 1:1000, 20 keys, 100 streamed and 1000 buffered rows per key (2M pairs).
// left.data = 90 > right.data in 0..100 keeps 90% of pairs, at random.
for (name, narrow) in [("1to1000_pass90", false), ("1to1000_pass90_narrow", true)] {
let keys = 20i64;
let lkeys: Vec<i64> = (0..keys * 100).map(|i| i / 100).collect();
let rkeys: Vec<i64> = (0..keys * 1000).map(|i| i / 1000).collect();
let rdata: Vec<i64> = (0..(keys * 1000) as u64).map(|i| (mix(i) % 100) as i64).collect();
let (s, left) = degen_batches(lkeys, vec![90; (keys * 100) as usize], narrow);
let (_, right) = degen_batches(rkeys, rdata, narrow);
group.bench_function(BenchmarkId::new(name, keys * 1000), |b| {
b.iter(|| {
do_join_with_filter(
make_exec(&left, &s),
make_exec(&right, &s),
datafusion_common::JoinType::Inner,
Some(data_filter(Operator::Gt)),
&rt,
)
})
});
}
group.finish();
}
criterion_group!(benches, bench_smj, bench_degenerate);There was a problem hiding this comment.
+1 to the contiguous check. Two more things the model leaves out, in case it stays: it counts columns rather than bytes, so a string payload weighs the same as an Int64, and p is a lifetime average that pairing-time evaluations never feed. Neither needs solving here, but the doc comment should say so.
There was a problem hiding this comment.
Thanks, reproduced and fixed in 8012102. The freeze takes the whole-batch path when both sides' indices span a range. The check looks only at the first and last index, because the slicing path checks the full range again anyway. A full scan here kept 1:1 joins about 5% slower than main. I also added inner_1to1_filtered to the bench. Medians as a ratio to main (9d916ab):
| Case | e2f4657 | 8012102 |
|---|---|---|
| 1:1, no filter | 1.03 | 1.00 |
| 1:1, 100% pass | 1.09 | 0.99 |
| 1:1, 90% pass, every 10th row fails | 1.20 | 0.98 |
| 1:1, 90% pass, random | 1.17 | 0.98 |
| 1:1, 50% pass, random | 1.09 | 0.99 |
| 1:1, 5% pass, random | 1.09 | 0.99 |
inner_1to1_filtered (new) |
n/a | 0.99 |
@viirya, I added both caveats to the filter_first_pays doc: it counts columns rather than bytes, and p is a lifetime average. Since pairing no longer evaluates the filter, every evaluation now feeds p.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
All 499 existing sort-merge join tests passed locally. I also ran two additional SortMergeJoinExec tests, one with narrow dictionary keys and one with the streamed input held open. Both passed on the merge base and failed on this head. Setting only FILTERED_RUN_MIN_ROWS to usize::MAX on this head made both pass again. The inline comments include the inputs and results, along with separate findings on index allocation and the lifetime selectivity counters.
| let streamed_columns = | ||
| materialize_left_columns(&self.streamed_batch.batch, &streamed_indices)?; | ||
| let buffered_columns = | ||
| self.materialize_right_columns(&kept, kept_rows, &self.buffered_columns)?; |
There was a problem hiding this comment.
A valid Dictionary<Int8, Utf8> payload can now fail here with DictionaryKeyOverflowError. I reproduced it with batch_size = 256, one streamed row, and two 256-row buffered batches for the same key. Each batch uses 100 distinct strings, disjoint between batches, with dictionary key row % 100. A Boolean filter keeps the first 150 rows of each batch. The base returns all 300 rows. This path puts the first 150 survivors and 106 from the next batch into the same output. interleave now needs 200 distinct dictionary keys, which Int8 cannot encode.
Could we preserve the candidate-batch boundary when filtering during pairing, or split the output when the combined dictionary cannot be represented? This two-batch case would make a small regression test.
There was a problem hiding this comment.
I don't think this is specific to this PR. A 256-row batch with 200 distinct values cannot be encoded with Int8 keys by any path. The base passes with these inputs because each freeze leaves 150 rows, which is over half of batch_size, so the coalescer passes them through without concatenating. On main, would 128-row buffered batches (two chunks per freeze, through interleave) or a filter keeping 100 rows per batch (through the coalescer's concat) fail the same way?
There was a problem hiding this comment.
@viirya is right that main already fails this way. On 9d916ab, with this schema and batch_size = 256, both variants fail with Dictionary key bigger than the key type:
- four 128-row buffered batches, keeping every row, which goes through
interleave - two 256-row batches keeping 100 rows each, which goes through the coalescer's
concat
With filtering while pairing removed in 8012102, your exact input returns all 300 rows, as on main.
| None | ||
| }; | ||
| match filtered { | ||
| Some(appended) => room -= appended, |
There was a problem hiding this comment.
Counting only survivors against room can make the join wait for more streamed input even when it already has a full batch to emit. I tested with batch_size = 512, one buffered batch of 512 rows for a single key, and two one-row streamed batches whose predicates keep 200 and 360 rows respectively. I held the streamed child open after its second batch with BarrierExec.
The base emits 512 rows before the barrier is released. This head emits zero, then returns all 560 after release. The coalescer holds the first 200 and the remaining 360 are still pair indices when the join polls the child again. A downstream LIMIT 512 can therefore wait on input it does not need. Could we trigger the freeze before polling the child when the accumulated rows can complete an output batch?
There was a problem hiding this comment.
main already polls the streamed child while holding fewer than batch_size unfrozen pairs: an unfiltered join with 200 and then 360 pairs from two one-row streamed batches would hold 560 rows the same way. The bound is still about 2 × batch_size. Freezing before the poll when the coalescer plus unfrozen pairs reach batch_size would help both cases, so it may fit better as its own change.
There was a problem hiding this comment.
This came with filtering while pairing, which 8012102 removes. The join again freezes once batch_size candidate pairs accumulate, as on main. As @viirya noted, freezing before polling the child when the coalesced rows and the unfrozen pairs can fill a batch would help main as well, so I'd keep that as a separate change.
| /// that [`Self::chunk_for`] compares to [`FILTERED_CHUNK_KEY`], so that the next | ||
| /// unfiltered pair starts a chunk of its own. | ||
| fn start_filtered_chunk(&mut self, buffered_batch_idx: usize, batch_size: usize) { | ||
| let capacity = batch_size - self.num_output_rows; |
There was a problem hiding this comment.
Each newly encountered buffered batch reserves the entire remaining output capacity for both index vectors, even when the filter keeps a single row. With one streamed row, 1,024 same-key buffered batches of 256 rows, one survivor per batch, and batch_size = 8192, a probe using the base and head chunk methods measured peak live index-vector capacity of 2.06 MiB on the base and 120 MiB here. The head has just 16 KiB of actual indices. The base freezes every 8,192 candidates, while all 1,024 filtered chunks stay live together here.
append_filtered_pairs already has count before starting the chunk. Passing it as the initial capacity reduced the same probe to 16 KiB and still lets a reused chunk grow.
There was a problem hiding this comment.
This came with filtering while pairing. 8012102 removes it along with start_filtered_chunk.
| pub buffered_filter_columns: Vec<usize>, | ||
| /// How many pairs the Inner join's filter has evaluated at freeze so far | ||
| /// (see [`Self::filter_first_pays`]) | ||
| pub freeze_filter_evaluated: usize, |
There was a problem hiding this comment.
These two counters total pairs across the life of a partition. On wasm32, 65,536 rows with one key on each side yield 2³² candidate pairs even if the filter rejects every row. Splitting the buffered input into 128-row batches takes the freeze path for every pair, so freeze_filter_evaluated overflows. Checked builds panic, and in release the wrapped denominator can make the estimated pass rate exceed 1. Could both totals use u64, widening the per-freeze increments?
There was a problem hiding this comment.
Done in 8012102: both counters are u64, and the per-freeze increments are widened.
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
Thanks @comphead. Filtering before materializing is the right fix for q72, and I found no correctness problem in the result or its order.
I agree with @mbutrovich on scope. Filtering at freeze is the broadly useful half. Filtering while pairing brings most of the new state (second pairing loop, threshold, sentinel key, filtered flag, spill fallback), wins only for selective filters on runs of 512+ rows, and is not reached by any test yet. Could you measure q72 with FILTERED_RUN_MIN_ROWS = usize::MAX? If most of the gain stays, I'd land freeze-time filtering here and the pairing part separately.
For the freeze path, the 1:1 slowdown should be fixed before merge, since that is the common shape for a filtered inner join. Taking the whole-batch path when both index arrays are contiguous is exactly what main runs today, so it cannot be slower than main there. A 1:1 filtered inner case in the bench would keep it visible.
On two of the reported regressions, I think both are existing behavior reached by different inputs rather than something new here; details inline. The capacity and u64 counter fixes look right to me.
Smaller points inline.
| }; | ||
|
|
||
| // Keep the pairs that passed, in order. | ||
| let mut kept: Vec<Cow<MatchedChunk>> = Vec::with_capacity(matched_chunks.len()); |
There was a problem hiding this comment.
MatchedChunk::clone only bumps the refcounts of its two index buffers, so Vec<MatchedChunk> with a plain clone would do here. That drops Cow, and materialize_right_columns can go back to taking &[MatchedChunk] instead of being generic over Borrow. With @andygrove's diff above, MatchedChunk::filter goes away as well.
There was a problem hiding this comment.
Done in 8012102. Cow, MatchedChunk::filter and the Borrow bound are gone, and materialize_right_columns takes &[MatchedChunk] again. With andygrove's change, the kept chunks are slices of the filtered index arrays, so no clone is needed.
| }) | ||
| .collect(); | ||
| assert!( | ||
| keys.is_sorted(), |
There was a problem hiding this comment.
This checks only that keys ascend, so a reorder within a key passes. b1 and b2 are row numbers on each side, so an Inner join's output should ascend on (b1, b2). Asserting that covers the order within a key too.
There was a problem hiding this comment.
Done in 8012102: it asserts that (b1, b2) ascends.
Address review feedback: - Drop filtering while pairing. It wins only for selective filters on runs of 512+ rows and regresses narrow high-pass joins. This removes the run threshold, `FILTERED_CHUNK_KEY`, the `filtered` flag, `Cow` and the spill fallback. - Take the whole-batch path when both sides' indices span a range, as in a 1:1 join, where the columns materialize as slices. - Filter the concatenated index arrays once and slice them per chunk. - Count the cost model's pairs in `u64`, and document its limits. - Assert the order within a key, and make the high-pass test filter reject some pairs so that the whole-batch filter is covered. - Add a filtered 1:1 Inner join benchmark.
|
Thanks @andygrove, @vinycalves, @mbutrovich, @sunchao and @viirya. 8012102 takes the scope suggestion: the join now filters at freeze only. Filtering while pairing is gone, and with it the run threshold, These are local medians of 3 alternating rounds on an Apple M5 Max, against 9d916ab (
Filtering while pairing was up to 1.4x faster for selective filters on runs of 512 rows or more. It was also up to 1.5x slower than The inline replies cover the rest. |
|
Sounds good for me. |
Which issue does this PR close?
Rationale for this change
An Inner
SortMergeJoinExecwith a join filter materializes every column of every pair of rows that match on the join keys, and only then evaluates the filter and drops the pairs that fail it. When a key matches many rows on both sides and the filter keeps few of the pairs, nearly all of the join's time goes into copying rows that are then dropped.TPC-DS q72's
catalog_salestoinventoryjoin has this shape. Eachcatalog_salesrow matches everyinventoryrow of its item, andinv_quantity_on_hand < cs_quantitykeeps about 5% of the pairs. DataFusion Comet runs Spark's sort-merge joins onSortMergeJoinExec, and there this join makes q72 about three times slower than in Spark (datafusion-comet#4614). A CPU profile of that join in Comet at TPC-DS scale factor 10, on DataFusion 55.2.0 (before #26277), put 53% of the stage's CPU time infreeze_streamed, mostly intakeandinterleaveof output columns, and 26% in recording pairs. Evaluating the filter itself took about 1%.What changes are included in this PR?
The changes apply to Inner joins with a filter. Outer joins keep the deferred filtering that their null-joined rows need, and joins without a filter are unchanged.
freeze_inner_filteredgathers only the columns the filter reads and evaluates the filter on them. It then filters the pairs' index arrays once, gives each chunk of pairs a slice of the result, and materializes the output columns for the pairs that passed.freeze_inner_filtered_wholematerializes the batch and then filters it, as before. It runs in two cases:spans_rangechecks only the first and last index, because the slicing path checks the range again.filter_first_payssays gathering the filter's columns first costs more than it saves. It decides from the sharepof pairs that passed in earlier freezes: filtering first wins whilep < 2(k - f - 1)/k, forkoutput columns andffilter columns. The model counts columns rather than bytes, andpis a lifetime average.materialize_right_columnsand thefetch_right_columns_*helpers take the buffered columns to gather.sort_merge_joinbenchmark cases for Inner joins with a filter. All butinner_1to1_filteredare in a separate first commit, so that they can be run againstmain.An earlier revision also filtered long runs of buffered rows while pairing them with a streamed row. Review measurements showed it won only for selective filters on runs of 512 rows or more, and was up to 1.5x slower than
mainfor narrow joins whose filter keeps most pairs, so it was dropped from this PR.What is the testing strategy for this PR?
New tests in
joins/sort_merge_join/tests.rs:join_inner_early_filtering_matches_hash_joincompares the join withHashJoinExecat batch sizes 1, 7, 300 and 8192. Its buffered side has groups of 600 rows split across 250-row batches, which freeze across several buffered batches, and groups of one or two rows. The filter columns contain NULLs. It runs five filters:c1 < c2, which keeps about half of the pairsIt also checks that the output ascends on both inputs' row numbers, so the order within a key is kept. The single-pair freezes at batch size 1 cover the range path.
join_inner_early_filtering_with_spilled_buffered_batchesforces the buffered batches to spill and compares the result with the join without spilling.prepare_record_batches_for_cmptakes the schema, so that it handles a join that returns no batches.Benchmarks:
cargo bench -p datafusion-physical-plan --features test_utils --bench sort_merge_join,mainplus the bench commit (9d916ab) against 8012102, with thin LTO on an Apple M5 Max. The two builds ran alternately for five rounds, and the table shows the medians.maininner_1to1/100000inner_1to1_filtered/100000(new) *inner_1to10/100000*inner_1to100/20000left_1to100_filtered/20000inner_1to10_filtered/100000(new)inner_1to100_filtered/20000(new)inner_1to10_high_pass_filtered/100000(new)inner_1to10_high_pass_filtered_narrow/100000(new)inner_1to1000_selective_filtered/20000(new)left_1to1_unmatched/100000left_semi_1to10/100000left_anti_partial/100000inner_group_spans_buffered_batches/512inner_group_spans_buffered_batches/64inner_group_spans_buffered_batches/8* Medians of a separate seven-round run. In the five-round run these two read +3.7% and +5.8%, and in earlier runs
inner_1to10, whose code path this PR does not change, ranged from -7% to +6%.inner_1to1000_selective_filteredis q72's shape. The cases that this PR does not change, such asleft_semi_1to10andleft_anti_partial, are unchanged within noise.The one shape that is still slower than
mainis a key group spread over very small buffered batches with a filter that keeps nearly every pair: 1.10x with 8-row batches and 1.01x with 64-row ones at 95% kept.In Comet, the earlier revision, which also filtered while pairing, backported to DataFusion 55.2.0, ran q72 at TPC-DS scale factor 10 in Spark local mode with 8 cores in 12.0 to 12.2 s. That compares with 42.9 to 49.1 s before and 13.8 s in Spark. The q72 result for this revision is to come.
Are there any user-facing changes?
No. Inner sort-merge joins with a filter return the same rows in the same order.