comphead opened a new pull request, #27278: URL: https://github.com/apache/datafusion/pull/27278
## Which issue does this PR close? - Part of #20910. ## Rationale for this change An Inner `SortMergeJoinExec` with 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_sales` to `inventory` join has this shape. Each `catalog_sales` row matches every `inventory` row of its item, and `inv_quantity_on_hand < cs_quantity` keeps about 5% of the pairs. DataFusion Comet runs Spark's sort-merge joins on `SortMergeJoinExec`, and there this join makes q72 about three times slower than in Spark ([datafusion-comet#4614](https://github.com/apache/datafusion-comet/issues/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 in `freeze_streamed`, mostly in `take` and `interleave` of 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 keep the original pairing loop. - **Filtering while pairing.** `pair_streamed_row_with_group_filtered` filters a run of at least `FILTERED_RUN_MIN_ROWS` (256) buffered rows as it pairs the run with a streamed row. The filter reads the run as a zero-copy 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. A run in a spilled buffered batch is paired unfiltered and filtered at freeze. - **Filtering before materializing at freeze.** `freeze_inner_filtered` evaluates the filter for the pairs of the other chunks on just the columns it reads, keeps the pairs that pass, and materializes the output columns for those only. - **Keeping the old order when it is cheaper.** If the filter keeps most pairs and reads most of the output columns, gathering its columns first costs more than it saves. `filter_first_pays` decides from the share `p` of pairs that passed in earlier freezes: filtering first wins while `p < 2(k - f - 1)/k`, for `k` output columns and `f` filter columns. Otherwise `freeze_inner_filtered_whole` materializes the batch and then filters it, as before. - `materialize_right_columns` and the `fetch_right_columns_*` helpers take the buffered columns to gather. - New `sort_merge_join` benchmark cases for Inner joins with a filter. They are in a separate first commit, so that they can be run against `main`. Two choices are estimates rather than measurements, and I would welcome input on them: the 256-row threshold for filtering while pairing, and the weights in the `filter_first_pays` cost model, which counts a gathered value as 1 and a copied one as 1/2. ## What is the testing strategy for this PR? New tests in `joins/sort_merge_join/tests.rs`: - `join_inner_early_filtering_matches_hash_join` compares the join with `HashJoinExec`. Its buffered side has groups of 600 rows split across 250-row batches, which are filtered while pairing, and groups of one or two rows, which are filtered at freeze. The filter columns contain NULLs. It runs at batch sizes 1, 7, 300 and 8192 with three filters: `c1 < c2`, a filter whose columns alternate between the two inputs, and a filter that nearly every pair passes, which takes the whole-batch path after the first freeze. It also checks that the output stays in streamed order. - `join_inner_early_filtering_column_free_filter` checks that a filter that reads no columns still runs on long runs and at freeze. - `join_inner_early_filtering_with_spilled_buffered_batches` forces the buffered batches to spill, so that long runs are paired unfiltered and filtered at freeze, and compares the result with the join without spilling. Benchmarks: `cargo bench -p datafusion-physical-plan --features test_utils --bench sort_merge_join`, `main` (`14b3edd9a`) against this PR on an Apple M5 Max. The two builds ran alternately for five rounds, and the table shows the medians. | Benchmark | `main` | This PR | Change | |---|---:|---:|---:| | `inner_1to1/100000` | 1.14 ms | 1.15 ms | +1.0% | | `inner_1to10/100000` | 7.21 ms | 7.24 ms | +0.3% | | `inner_1to100/20000` | 13.18 ms | 13.14 ms | -0.3% | | `left_1to100_filtered/20000` | 34.75 ms | 34.49 ms | -0.7% | | `inner_1to10_filtered/100000` (new) | 9.77 ms | 5.30 ms | -45.7% | | `inner_1to100_filtered/20000` (new) | 19.00 ms | 9.54 ms | -49.8% | | `inner_1to10_high_pass_filtered/100000` (new) | 9.22 ms | 7.27 ms | -21.1% | | `inner_1to10_high_pass_filtered_narrow/100000` (new) | 3.44 ms | 3.50 ms | +1.9% | | `inner_1to1000_selective_filtered/20000` (new) | 13.51 ms | 1.66 ms | -87.7% | | `left_1to1_unmatched/100000` | 1.15 ms | 1.15 ms | -0.4% | | `left_semi_1to10/100000` | 482 µs | 479 µs | -0.5% | | `left_anti_partial/100000` | 601 µs | 599 µs | -0.4% | | `inner_group_spans_buffered_batches/512` | 2.21 ms | 2.22 ms | +0.4% | | `inner_group_spans_buffered_batches/64` | 2.97 ms | 3.01 ms | +1.1% | | `inner_group_spans_buffered_batches/8` | 10.07 ms | 10.23 ms | +1.6% | `inner_1to1000_selective_filtered` is q72's shape. The cases that this PR does not change, such as `left_semi_1to10` and `left_anti_partial`, moved by up to 2% between rounds, so the unfiltered cases are unchanged within noise. In Comet, with this change backported to DataFusion 55.2.0, q72 at TPC-DS scale factor 10 in Spark local mode with 8 cores took 12.0 to 12.2 s, against 42.9 to 49.1 s before and 13.8 s in Spark. The join's own time went from 270 s to 40 s. TPC-DS q95, whose `<>` self-join keeps 83% of its pairs, took 0.83 to 0.90 s both before and after. ## Are there any user-facing changes? No. Inner sort-merge joins with a filter return the same rows in the same order. -- 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]
