mbutrovich commented on code in PR #27278: URL: https://github.com/apache/datafusion/pull/27278#discussion_r4526275268
########## 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: > 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 9d916ab47 (`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. -- 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]
