viirya commented on code in PR #25584: URL: https://github.com/apache/datafusion/pull/25584#discussion_r4300497399
########## datafusion/physical-plan/src/joins/sort_merge_join/semi_anti_summary.rs: ########## @@ -0,0 +1,254 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Exact existence checks for one cross-side column comparison. +//! +//! A buffered key group witnesses `outer <[=] inner` iff it does so against +//! the inner maximum, and `outer >[=] inner` iff it does so against the minimum. +//! For `!=`, two distinct non-null values match every non-null outer value. +//! Complex expressions and floating-point types retain ordinary evaluation. + +use std::sync::Arc; + +use arrow::array::{Array, AsArray, BooleanArray, RecordBatch}; +use arrow::compute::or; +use arrow::datatypes::{DataType, Schema}; +use datafusion_common::{JoinSide, Result, ScalarValue}; +use datafusion_expr::{Accumulator, ColumnarValue, Operator}; +use datafusion_functions_aggregate_common::min_max::{MaxAccumulator, MinAccumulator}; +use datafusion_physical_expr::expressions::{BinaryExpr, Column, NotExpr}; +use datafusion_physical_expr_common::datum::apply_cmp; + +use crate::joins::utils::{JoinFilter, boolean_mask_from_filter}; + +/// A recognized comparison, normalized to `outer OP inner`. +#[derive(Debug)] +pub(super) struct SemiAntiComparison { + pub(super) outer_column: usize, + inner_column: usize, + op: Operator, +} + +/// Cached once per buffered key group and reused across outer batches. +#[derive(Debug)] +pub(super) struct GroupSummary { + extreme: ScalarValue, + // A not-equal comparison needs both min and max. + other: Option<ScalarValue>, +} + +impl GroupSummary { + /// The retained extrema plus scalar clones and Arrow scalar buffers used + /// by `apply_cmp`. Admission used the source buffer as an upper bound; + /// once reduction finishes, only these actual scalar sizes are needed. + pub(super) fn memory_size(&self) -> usize { + self.extreme + .size() + .max(self.other.as_ref().map_or(0, ScalarValue::size)) Review Comment: For `!=` both extrema are retained, so this should be the sum rather than the max. That's moot if the reservation is resized to `extreme.size() + other.size()` after summarizing, as suggested above. ########## datafusion/physical-plan/benches/sort_merge_join.rs: ########## @@ -248,5 +255,362 @@ fn bench_smj(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, bench_smj); +/// Compare exact base and candidate builds with identical pre-sorted inputs. +/// SQL versions also measure sorting and the optimizer's join selection; these +/// cases isolate residual semi/anti joins, including output collection. +fn bench_semi_anti_filter(c: &mut Criterion) { + let mut group = c.benchmark_group("sort_merge_join_semi_anti_filter"); + + // Vary group count, rows per side, selectivity, and value type. Early-match + // cases measure the cost of summarizing when the first inner row suffices; + // the one-probe case gives the summary no reuse within the group. + for (name, groups, probe_rows, inner_rows, early_match, op, kind) in [ + ( + "not_equal_no_match", + 1, + 256, + 4096, + false, + Operator::NotEq, + JoinType::LeftAnti, + ), + ( + "not_equal_early_match", + 16, + 64, + 128, + true, + Operator::NotEq, + JoinType::LeftSemi, + ), + // The first probe fails, but the second inner row matches every + // outer row. This exposes unnecessary full-group work after a miss. + ( + "not_equal_second_witness", + 16, + 64, + 4096, + true, + Operator::NotEq, + JoinType::LeftSemi, + ), + ( + "small_group_no_match", + 4096, + 2, + 3, + false, + Operator::NotEq, + JoinType::LeftAnti, + ), + ( + "small_group_four_no_match", + 4096, + 4, + 4, + false, + Operator::NotEq, + JoinType::LeftAnti, + ), + ( + "small_group_seven_no_match", + 4096, + 7, + 7, + false, + Operator::NotEq, + JoinType::LeftAnti, + ), + ( + "singleton_early_match", + 8192, + 1, + 1, + true, + Operator::Lt, + JoinType::LeftSemi, + ), + ( + "range_no_match", + 1, + 256, + 4096, + false, + Operator::Gt, + JoinType::LeftSemi, + ), + ( + "range_early_match_one_probe", Review Comment: This covers the singleton-outer case only when the first inner row is a witness. Could you add the opposite direction, e.g. 16 groups × 1 outer × 4096 inner with no match on a `LeftAnti`? That's the case the `outer_group_len > 1` gate gives up, and it would show whether dropping the gate is worth it. ########## datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs: ########## @@ -707,6 +721,127 @@ impl BitwiseSortMergeJoinStream { } } + /// Try initial probes or an exact summary, updating the matched bits. + /// Return true when this outer slice needs no ordinary filter evaluation. + /// Keep this work separate from the asynchronous spill and batch traversal. + #[inline(never)] + fn try_apply_summary( + &mut self, + outer_slice: &RecordBatch, + summary: &mut Option<Box<CachedSummary>>, + allow_summary: &mut Option<bool>, + ) -> Result<bool> { + let outer_group_start = self.outer_offset; + let outer_group_len = outer_slice.num_rows(); + let filter = self.filter.as_ref().unwrap(); + + // Probe a small prefix before paying for min/max. This cost heuristic + // preserves early witnesses without adding an unconditional inner + // scan. A single outer row also stays on the ordinary path, which + // already visits each inner row at most once. + if summary.is_none() + && outer_group_len > 1 Review Comment: The reasoning here undercounts the ordinary path's cost. It does visit each inner row at most once, but each visit is a full `evaluate_filter_for_inner_row` call against the outer slice, not one comparison. With a unique outer key against a large inner group and no witness (the usual `NOT EXISTS` shape when the outer side is keyed), that's N filter evaluations, where the summary would do a single min/max pass and one comparison. Once the early-witness cost is bounded (e.g. the doubling-prefix scan), the cost follows the witness position regardless of outer slice length. Could `outer_group_len > 1` be dropped here, along with the matching condition at line 882? -- 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]
