sunchao commented on code in PR #25584:
URL: https://github.com/apache/datafusion/pull/25584#discussion_r4520227288
##########
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
+ && *allow_summary.get_or_insert_with(|| {
+ self.inner_key_buffer
+ .iter()
+ .map(RecordBatch::num_rows)
+ .sum::<usize>()
+ >= 7
+ })
+ {
+ if let Some(comparison) = SemiAntiComparison::try_new(
Review Comment:
Fixed in 9608e5bd01. The stream constructor recognizes the comparison once.
Each key checks only the spill state and the minimum group size before using
that descriptor.
##########
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
+ && *allow_summary.get_or_insert_with(|| {
+ self.inner_key_buffer
+ .iter()
+ .map(RecordBatch::num_rows)
+ .sum::<usize>()
+ >= 7
+ })
+ {
+ if let Some(comparison) = SemiAntiComparison::try_new(
+ filter,
+ self.outer_is_left,
+ self.outer.schema().as_ref(),
+ self.inner.schema().as_ref(),
+ )? {
+ let mut remaining = 2;
+ // A NULL outer operand cannot match any inner value.
+ let matchable = outer_group_len
+ - outer_slice.column(comparison.outer_column).null_count();
+ if matchable == 0 {
+ return Ok(true);
+ }
+ for inner_batch in &self.inner_key_buffer {
+ let len = remaining.min(inner_batch.num_rows());
+ for row in 0..len {
+ let result = comparison.evaluate_inner_row(
+ inner_batch,
+ row,
+ outer_slice,
+ )?;
+ let values = result.values();
+ apply_bitwise_binary_op(
+ self.matched.as_slice_mut(),
+ outer_group_start,
+ values.inner().as_slice(),
+ values.offset(),
+ outer_group_len,
+ |a, b| a | b,
+ );
+ let matched_count = UnalignedBitChunk::new(
+ self.matched.as_slice(),
+ outer_group_start,
+ outer_group_len,
+ )
+ .count_ones();
+ if matched_count == matchable {
+ return Ok(true);
+ }
+ }
+ remaining -= len;
+ if remaining == 0 {
+ break;
+ }
+ }
+ let bytes = comparison
Review Comment:
Fixed in 9608e5bd01. The completed summary calls try_resize with the size of
its compact Arrow scalars. The original buffered rows remain available if the
resize fails. Group completion releases the cached charge. Added bounded-pool
coverage for a skewed string group whose extrema exceed its prorated charge.
##########
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:
Removed the outer-slice length gate in 9608e5bd01. A singleton uses the same
two probes and doubling-prefix scan as a larger outer slice. A full summary can
serve subsequent singleton batches. The new test checks the active reservation
and its release at the next key.
##########
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:
Added singleton_no_match in the benchmark-only parent 41088840c6. It runs
LeftAnti with 16 groups, one outer row and 4,096 inner rows per group, for both
Int64 and Utf8. It checks the result count and zero spills before timing.
##########
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:
Fixed in 9608e5bd01. The retained reservation includes the allocation size
of each owned Arrow scalar plus the boxed summary. A new test measures the two
string scalars separately and checks that the live reservation covers both.
--
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]
