comphead commented on code in PR #27278:
URL: https://github.com/apache/datafusion/pull/27278#discussion_r4540667880
##########
datafusion/physical-plan/src/joins/sort_merge_join/tests.rs:
##########
@@ -8395,3 +8399,322 @@ async fn
left_join_with_partial_filter_preserves_streamed_order() -> Result<()>
);
Ok(())
}
+
+// ============================================================================
+// Early filtering of Inner joins
+// ============================================================================
+
+/// Rows for the early-filtering tests, sorted by the key column `names[0]`:
key
+/// `k` of `0..keys` has `rows_per_key(k)` rows. `names[1]` is the row number
and
+/// `names[2]` a pseudo-random value in `0..1000` that is NULL in every 37th
row.
+/// The rows are split into batches of `rows_per_batch`.
+fn early_filtering_batches(
+ names: [&str; 3],
+ keys: i32,
+ rows_per_key: impl Fn(i32) -> usize,
+ seed: u64,
+ rows_per_batch: usize,
+) -> (Vec<RecordBatch>, SchemaRef) {
+ let mut key_values = vec![];
+ let mut row_numbers = vec![];
+ let mut values = vec![];
+ let mut state = seed;
+ for key in 0..keys {
+ for _ in 0..rows_per_key(key) {
+ state = state
+ .wrapping_mul(6364136223846793005)
+ .wrapping_add(1442695040888963407);
+ let row = key_values.len() as i32;
+ key_values.push(key);
+ row_numbers.push(row);
+ values.push((row % 37 != 0).then_some(((state >> 33) % 1000) as
i32));
+ }
+ }
+ let schema = Arc::new(Schema::new(vec![
+ Field::new(names[0], DataType::Int32, false),
+ Field::new(names[1], DataType::Int32, false),
+ Field::new(names[2], DataType::Int32, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(key_values)),
+ Arc::new(Int32Array::from(row_numbers)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap();
+ let batches = (0..batch.num_rows())
+ .step_by(rows_per_batch)
+ .map(|offset| batch.slice(offset, rows_per_batch.min(batch.num_rows()
- offset)))
+ .collect();
+ (batches, schema)
+}
+
+/// Inputs for the early-filtering tests. Every third key has a group of 600
+/// buffered rows, long enough to be filtered as it is paired, and the 250-row
+/// buffered batches split each such group. The other keys have groups of one
or
+/// two buffered rows, whose pairs are filtered at freeze. Keys repeat up to
four
+/// times on the streamed side.
+fn early_filtering_inputs()
+-> ((Vec<RecordBatch>, SchemaRef), (Vec<RecordBatch>, SchemaRef)) {
+ let left =
+ early_filtering_batches(["a1", "b1", "c1"], 40, |k| 1 + (k % 4) as
usize, 1, 7);
+ let right = early_filtering_batches(
+ ["a2", "b2", "c2"],
+ 40,
+ |k| if k % 3 == 0 { 600 } else { (k % 3) as usize },
+ 2,
+ 250,
Review 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`.
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -47,10 +50,63 @@ use
datafusion_physical_expr_common::physical_expr::PhysicalExprRef;
use futures::StreamExt;
+/// The shortest run of buffered rows that an Inner join with a filter filters
as
+/// 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;
Review Comment:
Thanks for measuring this. Your table matches what I see. Rather than tune
the threshold, 8012102c4 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.
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -1771,33 +2008,215 @@ impl MaterializingSortMergeJoinStream {
Ok(())
}
- /// Materializes right-side columns across all matched chunks.
+ /// Freezes the matched pairs of an Inner join with a filter. The filter
runs
+ /// on just the columns it reads, for the pairs of the chunks that
+ /// [`Self::pair_filtered_run`] did not filter already, and only the pairs
that
+ /// pass have their output columns materialized. A selective filter over a
+ /// many-to-many join rejects most pairs, so materializing every column of
+ /// every pair before filtering would spend most of the freeze on rows that
+ /// are then dropped. When no chunk was filtered already and earlier
freezes
+ /// kept enough of their pairs, [`Self::filter_first_pays`] says the
opposite,
+ /// and the batch is materialized whole and then filtered.
+ fn freeze_inner_filtered(
+ &mut self,
+ matched_chunks: &[MatchedChunk],
+ total_matched_rows: usize,
+ ) -> Result<()> {
+ let Some(filter) = &self.filter else {
+ return internal_err!("early filtering requires a join filter");
+ };
+ debug_assert_eq!(
+ self.join_type,
+ JoinType::Inner,
+ "only an Inner join filters early, so the streamed side is the
left one"
+ );
+
+ // Evaluate the filter over the pairs of the chunks not filtered yet.
+ let unfiltered: Vec<&MatchedChunk> =
+ matched_chunks.iter().filter(|c| !c.filtered).collect();
+ if unfiltered.len() == matched_chunks.len() &&
!self.filter_first_pays(filter) {
+ return self.freeze_inner_filtered_whole(matched_chunks,
total_matched_rows);
+ }
+ let passed = if unfiltered.is_empty() {
+ None
+ } else {
+ let num_rows = unfiltered.iter().map(|c| c.len()).sum();
+ let streamed_indices =
+ concat_indices(unfiltered.iter().map(|c|
&c.streamed_indices))?;
+ let mut buffered_arrays = self
+ .materialize_right_columns(
+ &unfiltered,
+ num_rows,
+ &self.buffered_filter_columns,
+ )?
+ .into_iter();
+ let columns = filter
+ .column_indices()
+ .iter()
+ .map(|column_index| match column_index.side {
+ JoinSide::Left => take_column(
+ self.streamed_batch.batch.column(column_index.index),
+ &streamed_indices,
+ ),
+ JoinSide::Right => buffered_arrays.next().ok_or_else(|| {
+ internal_datafusion_err!("missing buffered filter
column")
+ }),
+ JoinSide::None => {
+ internal_err!("Unexpected JoinSide::None in filter")
+ }
+ })
+ .collect::<Result<Vec<_>>>()?;
+ self.freeze_filter_evaluated += num_rows;
+ Some(evaluate_join_filter(
+ filter,
+ filter.schema(),
+ columns,
+ num_rows,
+ )?)
+ };
+
+ // 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;
+ for chunk in matched_chunks {
+ let len = chunk.len();
+ match (&passed, chunk.filtered) {
+ (Some(passed), false) => {
+ let mask = passed.slice(offset, len);
+ 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)?));
Review Comment:
Applied in 8012102c4, thanks. The freeze filters the concatenated index
arrays once, and each chunk takes a slice of the result. Medians against
9d916ab47, 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.
##########
datafusion/physical-plan/src/joins/sort_merge_join/tests.rs:
##########
Review Comment:
No, as @andygrove found too: no run here is longer than 250 rows. 8012102c4
removes that path, and the tests now reach every branch of the freeze path
(details in andygrove's thread).
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
Review 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.
--
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]
