From b9fb06973b8e5d2aa510ca517091b7b9fa3500bb Mon Sep 17 00:00:00 2001 From: linfeng Date: Wed, 16 Sep 2026 22:21:48 +0800 Subject: [PATCH] perf: prune matched candidates in right semi/anti nested loop joins --- .../nlj/benchmarks/q18.benchmark | 12 + .../nlj/benchmarks/q19.benchmark | 12 + .../nlj/benchmarks/q20.benchmark | 28 ++ .../nlj/benchmarks/q21.benchmark | 28 ++ .../nlj/benchmarks/q22.benchmark | 29 ++ .../nlj/benchmarks/q23.benchmark | 32 ++ .../src/joins/nested_loop_join.rs | 381 ++++++++++++++++-- .../test_files/nested_loop_join_existence.slt | 226 +++++++++++ .../test_files/nested_loop_join_spill.slt | 124 ++++++ 9 files changed, 839 insertions(+), 33 deletions(-) create mode 100644 benchmarks/sql_benchmarks/nlj/benchmarks/q18.benchmark create mode 100644 benchmarks/sql_benchmarks/nlj/benchmarks/q19.benchmark create mode 100644 benchmarks/sql_benchmarks/nlj/benchmarks/q20.benchmark create mode 100644 benchmarks/sql_benchmarks/nlj/benchmarks/q21.benchmark create mode 100644 benchmarks/sql_benchmarks/nlj/benchmarks/q22.benchmark create mode 100644 benchmarks/sql_benchmarks/nlj/benchmarks/q23.benchmark create mode 100644 datafusion/sqllogictest/test_files/nested_loop_join_existence.slt diff --git a/benchmarks/sql_benchmarks/nlj/benchmarks/q18.benchmark b/benchmarks/sql_benchmarks/nlj/benchmarks/q18.benchmark new file mode 100644 index 0000000000000..1d24842a6e97d --- /dev/null +++ b/benchmarks/sql_benchmarks/nlj/benchmarks/q18.benchmark @@ -0,0 +1,12 @@ +name Q18 +group nlj + +expect_plan NestedLoopJoinExec +expect_plan join_type: RightSemi + +run +-- Q18: RIGHT SEMI 4K x 4K | no matches; the maximum sum is 4096 * 4096 - 1 +SELECT r.value +FROM range(4096) AS l +RIGHT SEMI JOIN range(4096) AS r +ON l.value * 4096 + r.value >= 4096 * 4096; diff --git a/benchmarks/sql_benchmarks/nlj/benchmarks/q19.benchmark b/benchmarks/sql_benchmarks/nlj/benchmarks/q19.benchmark new file mode 100644 index 0000000000000..c84a176d8dd43 --- /dev/null +++ b/benchmarks/sql_benchmarks/nlj/benchmarks/q19.benchmark @@ -0,0 +1,12 @@ +name Q19 +group nlj + +expect_plan NestedLoopJoinExec +expect_plan join_type: RightAnti + +run +-- Q19: RIGHT ANTI 4K x 4K | no matches; the maximum sum is 4096 * 4096 - 1 +SELECT r.value +FROM range(4096) AS l +RIGHT ANTI JOIN range(4096) AS r +ON l.value * 4096 + r.value >= 4096 * 4096; diff --git a/benchmarks/sql_benchmarks/nlj/benchmarks/q20.benchmark b/benchmarks/sql_benchmarks/nlj/benchmarks/q20.benchmark new file mode 100644 index 0000000000000..02c53b802a8b3 --- /dev/null +++ b/benchmarks/sql_benchmarks/nlj/benchmarks/q20.benchmark @@ -0,0 +1,28 @@ +name Q20 +group nlj + +expect_plan NestedLoopJoinExec +expect_plan join_type: RightSemi + +run +-- Q20: RIGHT SEMI 2K x 2K | staged matches with 260-byte filter strings +-- With batch_size=8192, candidates shrink 2048 -> 1024 -> 512 -> ... -> 16. +-- Under that setting, the multi-left-row path starts at 512 candidates. +SELECT r.value +FROM ( + SELECT repeat('x', 256) || lpad(CAST( + CASE WHEN value < 128 THEN 1024 + WHEN value < 256 THEN 1536 + WHEN value < 384 THEN 1792 + WHEN value < 512 THEN 1920 + WHEN value < 640 THEN 1984 + WHEN value < 768 THEN 2016 + ELSE 2032 END AS VARCHAR), 4, '0') AS threshold + FROM range(2048) +) AS l +RIGHT SEMI JOIN ( + SELECT value, + repeat('x', 256) || lpad(CAST(value AS VARCHAR), 4, '0') AS key + FROM range(2048) +) AS r +ON l.threshold > r.key; diff --git a/benchmarks/sql_benchmarks/nlj/benchmarks/q21.benchmark b/benchmarks/sql_benchmarks/nlj/benchmarks/q21.benchmark new file mode 100644 index 0000000000000..898fd27fff7f4 --- /dev/null +++ b/benchmarks/sql_benchmarks/nlj/benchmarks/q21.benchmark @@ -0,0 +1,28 @@ +name Q21 +group nlj + +expect_plan NestedLoopJoinExec +expect_plan join_type: RightAnti + +run +-- Q21: RIGHT ANTI 2K x 2K | staged matches with 260-byte filter strings +-- With batch_size=8192, candidates shrink 2048 -> 1024 -> 512 -> ... -> 16. +-- Under that setting, the multi-left-row path starts at 512 candidates. +SELECT r.value +FROM ( + SELECT repeat('x', 256) || lpad(CAST( + CASE WHEN value < 128 THEN 1024 + WHEN value < 256 THEN 1536 + WHEN value < 384 THEN 1792 + WHEN value < 512 THEN 1920 + WHEN value < 640 THEN 1984 + WHEN value < 768 THEN 2016 + ELSE 2032 END AS VARCHAR), 4, '0') AS threshold + FROM range(2048) +) AS l +RIGHT ANTI JOIN ( + SELECT value, + repeat('x', 256) || lpad(CAST(value AS VARCHAR), 4, '0') AS key + FROM range(2048) +) AS r +ON l.threshold > r.key; diff --git a/benchmarks/sql_benchmarks/nlj/benchmarks/q22.benchmark b/benchmarks/sql_benchmarks/nlj/benchmarks/q22.benchmark new file mode 100644 index 0000000000000..ff07e32114ff0 --- /dev/null +++ b/benchmarks/sql_benchmarks/nlj/benchmarks/q22.benchmark @@ -0,0 +1,29 @@ +name Q22 +group nlj + +expect_plan NestedLoopJoinExec +expect_plan join_type: RightSemi + +# Materialize the threshold so the filter keeps both join inputs. +# Keep the left input smaller so the default optimizer preserves RightSemi. +init +CREATE TABLE nlj_repeated_left AS +SELECT 3277 AS threshold FROM range(4096); + +init +CREATE TABLE nlj_repeated_right AS +SELECT value AS id FROM range(8192); + +run +-- Repeated 40% matches: only the first probe adds matches. +-- With one partition and batch_size=8192, 4915 survivors exceed the 4096-row +-- compaction threshold, so every left row probes the full right batch. +SELECT r.id +FROM nlj_repeated_left l +RIGHT SEMI JOIN nlj_repeated_right r ON l.threshold > r.id; + +cleanup +DROP TABLE nlj_repeated_left; + +cleanup +DROP TABLE nlj_repeated_right; diff --git a/benchmarks/sql_benchmarks/nlj/benchmarks/q23.benchmark b/benchmarks/sql_benchmarks/nlj/benchmarks/q23.benchmark new file mode 100644 index 0000000000000..2d17396a5e968 --- /dev/null +++ b/benchmarks/sql_benchmarks/nlj/benchmarks/q23.benchmark @@ -0,0 +1,32 @@ +name Q23 +group nlj + +expect_plan NestedLoopJoinExec +expect_plan join_type: RightAnti + +init +CREATE TABLE nlj_final_left AS +SELECT arrow_cast('m', 'Utf8') AS threshold; + +# Materialize 32 MiB of Utf8 strings outside the measured query. Compaction +# would copy these strings; different first bytes keep comparison cheap. +init +CREATE TABLE nlj_final_right AS +SELECT value AS id, + arrow_cast((CASE WHEN value < 4096 THEN 'a' ELSE 'z' END) + || repeat('x', 4095), 'Utf8') AS payload +FROM range(8192); + +run +-- With one partition and batch_size=8192, half the batch survives the only +-- left row. Compacting after that final probe would needlessly copy 16 MiB. +-- Return IDs only to exclude string copying from output materialization. +SELECT r.id +FROM nlj_final_left l +RIGHT ANTI JOIN nlj_final_right r ON l.threshold > r.payload; + +cleanup +DROP TABLE nlj_final_left; + +cleanup +DROP TABLE nlj_final_right; diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs index 462576b2b4f82..f5d1395d81c76 100644 --- a/datafusion/physical-plan/src/joins/nested_loop_join.rs +++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs @@ -57,6 +57,7 @@ use arrow::array::{ use arrow::buffer::BooleanBuffer; use arrow::compute::{ BatchCoalescer, concat_batches, filter, filter_record_batch, not, take, + take_record_batch, }; use arrow::datatypes::{Schema, SchemaRef}; use arrow::record_batch::RecordBatch; @@ -2124,6 +2125,174 @@ impl SpillStateActive { } } +/// Probe state for RightSemi/RightAnti within one buffered left chunk. +/// The original right batch stays in the stream for output and spill replay. +struct RightExistenceProbe { + /// Only the right columns referenced by the filter, cached across probes. + candidates: RecordBatch, + /// Filter with right column indices remapped to `candidates`. + filter: Option, + /// Candidate row -> original right row. None until the first compaction. + original_indices: Option>, + /// Matches in original right-row coordinates, for output/spill merging. + matched: BooleanBufferBuilder, + /// Matches in candidate coordinates, reset after compaction. + candidate_matched: Vec, + remaining: usize, +} + +impl RightExistenceProbe { + fn new(right: &RecordBatch, filter: Option<&JoinFilter>) -> Result { + let mut projection = Vec::new(); + let mut filter = filter.cloned(); + if let Some(filter) = &mut filter { + for column in &mut filter.column_indices { + if column.side == JoinSide::Right { + let index = projection.iter().position(|&i| i == column.index); + column.index = index.unwrap_or_else(|| { + projection.push(column.index); + projection.len() - 1 + }); + } + } + } + let rows = right.num_rows(); + let mut matched = BooleanBufferBuilder::new(rows); + matched.append_n(rows, false); + Ok(Self { + candidates: right.project(&projection)?, + filter, + original_indices: None, + matched, + candidate_matched: vec![0; rows.div_ceil(64)], + remaining: rows, + }) + } + + /// Probe the next left range, returning the number of left rows consumed. + fn probe( + &mut self, + left: &RecordBatch, + start: usize, + batch_size: usize, + ) -> Result { + let candidate_rows = self.candidates.num_rows(); + let rows_per_range = batch_size / candidate_rows; + let count = if rows_per_range > 10 { + rows_per_range + } else { + 1 + }; + let count = count.min(left.num_rows() - start); + let remaining_before = self.remaining; + + let matches = match &self.filter { + Some(filter) if count == 1 => { + apply_filter_to_row_join_batch(left, start, &self.candidates, filter)? + } + Some(filter) => { + let left_indices = UInt32Array::from_iter_values( + (start..start + count) + .flat_map(|i| std::iter::repeat_n(i as u32, candidate_rows)), + ); + let right_indices = UInt32Array::from_iter_values( + (0..count).flat_map(|_| 0..candidate_rows as u32), + ); + apply_filter_to_indices( + left, + &self.candidates, + &left_indices, + &right_indices, + filter, + )? + } + None => { + BooleanArray::new(BooleanBuffer::new_set(count * candidate_rows), None) + } + }; + if !matches.has_true() { + return Ok(count); + } + self.update_matches(matches.values()); + + if start + count < left.num_rows() + && self.remaining > 0 + && self.remaining < remaining_before + && self.remaining <= candidate_rows / 2 + { + self.compact()?; + } + Ok(count) + } + + fn update_matches(&mut self, matches: &BooleanBuffer) { + let candidate_rows = self.candidates.num_rows(); + // Each left row contributes one candidate-sized segment. BitChunks handles + // both sliced filter results and segments that start within a byte/word. + for offset in (0..matches.len()).step_by(candidate_rows) { + let segment = matches.slice(offset, candidate_rows); + let chunks = segment.bit_chunks(); + for (word_index, (previous, current)) in self + .candidate_matched + .iter_mut() + .zip(chunks.iter_padded()) + .enumerate() + { + let mut new_matches = current & !*previous; + *previous |= current; + self.remaining -= new_matches.count_ones() as usize; + // Only visit new matches, including when several left rows match + // the same candidate in this range or in subsequent probes. + while new_matches != 0 { + let candidate = + word_index * 64 + new_matches.trailing_zeros() as usize; + let original = self + .original_indices + .as_ref() + .map_or(candidate, |indices| indices[candidate]); + self.matched.set_bit(original, true); + new_matches &= new_matches - 1; + } + } + if self.remaining == 0 { + break; + } + } + } + + fn compact(&mut self) -> Result<()> { + let indices = UInt32Array::from_iter_values( + (0..self.candidates.num_rows()) + .filter(|&i| self.candidate_matched[i / 64] & (1 << (i % 64)) == 0) + .map(|i| i as u32), + ); + let original_indices = indices + .values() + .iter() + .map(|&i| { + self.original_indices + .as_ref() + .map_or(i as usize, |original| original[i as usize]) + }) + .collect(); + // Take the cached projection only when compacting, never the original + // payload or the original right columns on every probe. + self.candidates = if self.candidates.num_columns() == 0 { + create_record_batch_with_empty_schema( + self.candidates.schema(), + indices.len(), + )? + } else { + take_record_batch(&self.candidates, &indices)? + }; + self.original_indices = Some(original_indices); + self.candidate_matched.clear(); + self.candidate_matched + .resize(self.remaining.div_ceil(64), 0); + Ok(()) + } +} + pub(crate) struct NestedLoopJoinStream { // ======================================================================== // PROPERTIES: @@ -2216,6 +2385,8 @@ pub(crate) struct NestedLoopJoinStream { // For right join, keep track of matched rows in `current_right_batch` // Constructed when fetching each new incoming right batch in `FetchingRight` state. current_right_batch_matched: Option, + /// Candidate pruning for right semi/anti joins in the current left chunk. + right_existence_probe: Option, /// Memory-limited spill fallback state. See [`SpillState`] for details. spill_state: SpillState, @@ -2585,6 +2756,7 @@ impl NestedLoopJoinStream { batch_size, current_right_batch: None, current_right_batch_matched: None, + right_existence_probe: None, state: NLJState::BufferingLeft, cancelled_terminally: false, cancel_watch: None, @@ -2840,7 +3012,12 @@ impl NestedLoopJoinStream { self.current_right_batch = Some(right_batch); // Prepare right bitmap - if self.should_track_unmatched_right { + if self.should_track_unmatched_right + && !matches!( + self.join_type, + JoinType::RightSemi | JoinType::RightAnti + ) + { let zeroed_buf = BooleanBuffer::new_unset(right_batch_rows); self.current_right_batch_matched = Some(BooleanArray::new(zeroed_buf, None)); @@ -3227,6 +3404,28 @@ impl NestedLoopJoinStream { /// next state (ProbeRight) fn process_probe_batch(&mut self) -> Result { let left_data = Arc::clone(self.get_left_data()?); + if matches!(self.join_type, JoinType::RightSemi | JoinType::RightAnti) { + if self.right_existence_probe.is_none() { + let right = self.current_right_batch.as_ref().ok_or_else(|| { + internal_datafusion_err!("Right batch should be available") + })?; + self.right_existence_probe = + Some(RightExistenceProbe::new(right, self.join_filter.as_ref())?); + } + let probe = self.right_existence_probe.as_mut().unwrap(); + if probe.remaining == 0 || self.left_probe_idx >= left_data.batch().num_rows() + { + let mut probe = self.right_existence_probe.take().unwrap(); + self.current_right_batch_matched = + Some(BooleanArray::new(probe.matched.finish(), None)); + // In spill mode this completes only the current left chunk. + // EmitRightUnmatched merges these original-row matches across chunks. + return Ok(false); + } + self.left_probe_idx += + probe.probe(left_data.batch(), self.left_probe_idx, self.batch_size)?; + return Ok(true); + } let right_batch = self .current_right_batch .as_ref() @@ -3341,38 +3540,13 @@ impl NestedLoopJoinStream { // Evaluate the join filter (if any) over an intermediate batch built // using the filter's own schema/column indices. let bitmap_combined = if let Some(filter) = &self.join_filter { - // Build the intermediate batch for filter evaluation - let intermediate_batch = if filter.schema.fields().is_empty() { - // Constant predicate (e.g., TRUE/FALSE). Use an empty schema with row_count - create_record_batch_with_empty_schema( - Arc::new((*filter.schema).clone()), - total_rows, - )? - } else { - let mut filter_columns: Vec> = - Vec::with_capacity(filter.column_indices().len()); - for column_index in filter.column_indices() { - let array = if column_index.side == JoinSide::Left { - let col = left_data.batch().column(column_index.index); - take(col.as_ref(), &left_indices, None)? - } else { - let col = right_batch.column(column_index.index); - take(col.as_ref(), &right_indices, None)? - }; - filter_columns.push(array); - } - - RecordBatch::try_new(Arc::new((*filter.schema).clone()), filter_columns)? - }; - - let filter_result = filter - .expression() - .evaluate(&intermediate_batch)? - .into_array(intermediate_batch.num_rows())?; - let filter_arr = as_boolean_array(&filter_result)?; - - // Combine with null bitmap to get a unified mask - boolean_mask_from_filter(filter_arr) + apply_filter_to_indices( + left_data.batch(), + right_batch, + &left_indices, + &right_indices, + filter, + )? } else { // No filter: all pairs match BooleanArray::from(vec![true; total_rows]) @@ -3755,6 +3929,39 @@ impl NestedLoopJoinStream { // ==== Utilities ==== +/// Evaluate a Cartesian product using only the columns referenced by the filter. +fn apply_filter_to_indices( + left: &RecordBatch, + right: &RecordBatch, + left_indices: &UInt32Array, + right_indices: &UInt32Array, + filter: &JoinFilter, +) -> Result { + let columns = filter + .column_indices() + .iter() + .map(|column| { + let (batch, indices) = if column.side == JoinSide::Left { + (left, left_indices) + } else { + (right, right_indices) + }; + take(batch.column(column.index).as_ref(), indices, None) + }) + .collect::, _>>()?; + let options = RecordBatchOptions::new().with_row_count(Some(left_indices.len())); + let intermediate = RecordBatch::try_new_with_options( + Arc::clone(filter.schema()), + columns, + &options, + )?; + let result = filter + .expression() + .evaluate(&intermediate)? + .into_array(intermediate.num_rows())?; + Ok(boolean_mask_from_filter(as_boolean_array(&result)?)) +} + /// Apply the join filter between: /// (l_index th row in left buffer) x (right batch) /// Returns a bitmap, with successfully joined indices set to true @@ -5041,6 +5248,114 @@ pub(crate) mod tests { Ok(()) } + #[test] + fn right_existence_compacts_zero_column_candidates() -> Result<()> { + let right = create_record_batch_with_empty_schema(Arc::new(Schema::empty()), 5)?; + let mut probe = RightExistenceProbe::new(&right, None)?; + // A row-varying predicate can partially match even when it references + // no right columns. Retain the row count of this empty projection. + probe.update_matches(&BooleanBuffer::from(vec![true, false, true, true, false])); + probe.compact()?; + assert_eq!(probe.candidates.num_columns(), 0); + assert_eq!(probe.candidates.num_rows(), 2); + assert_eq!(probe.original_indices, Some(vec![1, 4])); + probe.update_matches(&BooleanBuffer::from(vec![false, true])); + assert_eq!(probe.remaining, 1); + assert_eq!( + probe.matched.finish().set_indices().collect::>(), + vec![0, 2, 3, 4] + ); + Ok(()) + } + + #[rstest] + fn right_existence_sliced_match_words( + #[values(63, 64, 65)] rows: usize, + ) -> Result<()> { + let right = + create_record_batch_with_empty_schema(Arc::new(Schema::empty()), rows)?; + let mut probe = RightExistenceProbe::new(&right, None)?; + // Set surrounding bits too, so failing to mask the tail or honor the + // non-byte-aligned slice would incorrectly match extra candidates. + let mut bits = vec![true; 3 + rows * 3 + 7]; + for left in 0..3 { + for candidate in 0..rows { + bits[3 + left * rows + candidate] = + candidate == 0 || candidate == rows - 1; + } + } + let mask = BooleanBuffer::from(bits).slice(3, rows * 3); + probe.update_matches(&mask); + probe.update_matches(&mask); + assert_eq!(probe.remaining, rows - 2); + assert_eq!( + probe.matched.finish().set_indices().collect::>(), + vec![0, rows - 1] + ); + Ok(()) + } + + #[rstest] + #[case(JoinType::RightSemi)] + #[case(JoinType::RightAnti)] + #[tokio::test] + async fn right_existence_stops_after_all_rows_match( + #[case] join_type: JoinType, + #[values(false, true)] with_filter: bool, + ) -> Result<()> { + let ctx = new_task_ctx(65); + let left = create_record_batch_with_empty_schema(Arc::new(Schema::empty()), 32)?; + let right = create_record_batch_with_empty_schema(Arc::new(Schema::empty()), 65)?; + let schema = right.schema(); + let filter = with_filter.then(|| { + JoinFilter::new( + Arc::new(Literal::new(ScalarValue::Boolean(Some(true)))), + vec![], + Arc::clone(&schema), + ) + }); + let left_data = JoinLeftData::new( + left, + Mutex::new(BooleanBufferBuilder::new(0)), + AtomicUsize::new(1), + MemoryConsumer::new("test").register(ctx.memory_pool()), + ); + let mut stream = NestedLoopJoinStream::new( + Arc::clone(&schema), + filter, + join_type, + Box::pin(crate::stream::RecordBatchStreamAdapter::new( + schema, + futures::stream::iter(vec![Ok(right)]), + )), + OnceFut::new(async { internal_err!("unused left input") }), + vec![], + NestedLoopJoinMetrics::new(&ExecutionPlanMetricsSet::new(), 0), + 65, + SpillState::Disabled, + ); + stream.buffered_left_data = Some(Arc::new(left_data)); + let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref()); + assert!(stream.handle_fetching_right(&mut cx).is_continue()); + assert!(stream.process_probe_batch()?); + assert_eq!(stream.left_probe_idx, 1); + assert!( + !stream.process_probe_batch()?, + "all right rows already matched" + ); + let output = stream.process_right_unmatched()?; + let rows = output.map_or(0, |batch| batch.num_rows()); + assert_eq!( + rows, + if join_type == JoinType::RightSemi { + 65 + } else { + 0 + } + ); + Ok(()) + } + #[rstest] #[tokio::test] async fn join_right_semi_with_filter( diff --git a/datafusion/sqllogictest/test_files/nested_loop_join_existence.slt b/datafusion/sqllogictest/test_files/nested_loop_join_existence.slt new file mode 100644 index 0000000000000..a9571d7e5eb63 --- /dev/null +++ b/datafusion/sqllogictest/test_files/nested_loop_join_existence.slt @@ -0,0 +1,226 @@ +# 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. + +# Candidate pruning and early stopping for right semi/anti nested loop joins. +statement ok +SET datafusion.execution.target_partitions = 1 + +statement ok +SET datafusion.optimizer.join_reordering = false + +statement ok +SET datafusion.execution.batch_size = 256 + +statement ok +CREATE TABLE existence_left AS +SELECT CASE WHEN value = 0 OR value = 1 THEN 65 + WHEN value = 2 THEN NULL + WHEN value = 3 OR value = 4 THEN 97 + WHEN value = 5 THEN 114 ELSE 127 END AS threshold +FROM range(20) + +statement ok +CREATE TABLE existence_right AS +SELECT 1000 + value AS payload, + CASE WHEN (value * 2) % 129 = 128 THEN NULL ELSE (value * 2) % 129 END AS k, + 'payload-' || value AS label +FROM range(129) + +# Multiple compactions must retain original right-row indices and payloads. +# The noncontiguous survivors shrink from 129 to 64, 32, then 15 candidates. +# With batch_size=256, 15 candidates trigger a multi-left-row range. Repeated +# thresholds must not count an already matched candidate twice; NULL never matches. +query IIII +SELECT count(*), sum(r.payload), min(r.payload), max(r.payload) +FROM existence_left l RIGHT SEMI JOIN existence_right r ON l.threshold > r.k +---- +127 135064 1000 1127 + +query IIT rowsort +SELECT r.payload, r.k, r.label +FROM existence_left l RIGHT ANTI JOIN existence_right r ON l.threshold > r.k +---- +1064 NULL payload-64 +1128 127 payload-128 + +# Assert the physical join directions as well as the SQL results. +query TT +EXPLAIN SELECT r.* +FROM existence_left l RIGHT SEMI JOIN existence_right r ON l.threshold > r.k +---- +logical_plan +01)RightSemi Join: Filter: l.threshold > r.k +02)--SubqueryAlias: l +03)----TableScan: existence_left projection=[threshold] +04)--SubqueryAlias: r +05)----TableScan: existence_right projection=[payload, k, label] +physical_plan +01)NestedLoopJoinExec: join_type=RightSemi, filter=threshold@0 > k@1 +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +query TT +EXPLAIN SELECT r.* +FROM existence_left l RIGHT ANTI JOIN existence_right r ON l.threshold > r.k +---- +logical_plan +01)RightAnti Join: Filter: l.threshold > r.k +02)--SubqueryAlias: l +03)----TableScan: existence_left projection=[threshold] +04)--SubqueryAlias: r +05)----TableScan: existence_right projection=[payload, k, label] +physical_plan +01)NestedLoopJoinExec: join_type=RightAnti, filter=threshold@0 > k@1 +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +# Repeated matches within and across left ranges must only decrement remaining +# once per right row. The last left row introduces a new match that must not be +# skipped. Cover word-aligned (64), byte-aligned (72), and unaligned (65) segments. +statement ok +SET datafusion.execution.batch_size = 1024 + +statement ok +CREATE VIEW existence_duplicates AS +SELECT CASE WHEN value = 99 THEN 2 ELSE 1 END AS threshold FROM range(100) + +query III +SELECT count(*), min(r.value), max(r.value) +FROM existence_duplicates l RIGHT SEMI JOIN range(64) r +ON l.threshold > r.value +---- +2 0 1 + +query IIII +SELECT count(*), min(r.value), max(r.value), sum(r.value) +FROM existence_duplicates l RIGHT ANTI JOIN range(64) r +ON l.threshold > r.value +---- +62 2 63 2015 + +query III +SELECT count(*), min(r.value), max(r.value) +FROM existence_duplicates l RIGHT SEMI JOIN range(72) r +ON l.threshold > r.value +---- +2 0 1 + +query IIII +SELECT count(*), min(r.value), max(r.value), sum(r.value) +FROM existence_duplicates l RIGHT ANTI JOIN range(72) r +ON l.threshold > r.value +---- +70 2 71 2555 + +query III +SELECT count(*), min(r.value), max(r.value) +FROM existence_duplicates l RIGHT SEMI JOIN range(65) r +ON l.threshold > r.value +---- +2 0 1 + +query IIII +SELECT count(*), min(r.value), max(r.value), sum(r.value) +FROM existence_duplicates l RIGHT ANTI JOIN range(65) r +ON l.threshold > r.value +---- +63 2 64 2079 + +# Skip NULL/FALSE-only ranges before a later range matches every candidate. +# Duplicate matches in that range must not decrement remaining twice. +statement ok +CREATE VIEW existence_range_matches AS +SELECT CASE WHEN value < 32 THEN NULL + WHEN value < 64 THEN 0 ELSE 65 END AS threshold +FROM range(96) + +query III +SELECT count(*), sum(r.value), max(r.value) +FROM existence_range_matches l RIGHT SEMI JOIN range(65) r +ON l.threshold > r.value +---- +65 2080 64 + +query I +SELECT count(*) +FROM existence_range_matches l RIGHT ANTI JOIN range(65) r +ON l.threshold > r.value +---- +0 + +# Reset remaining, both match bitmaps, and the candidate mapping between right +# batches: all matched, none matched, several compactions, then all matched again. +statement ok +SET datafusion.execution.batch_size = 64 + +statement ok +CREATE VIEW existence_batches AS +SELECT value / 64 AS batch_id, 1000 + value AS payload, + CASE WHEN value < 64 THEN -1 + WHEN value < 128 THEN 1000 + WHEN value < 192 THEN value - 128 ELSE 0 END AS k +FROM range(193) + +query IIIII rowsort +SELECT r.batch_id, count(*), min(r.payload), max(r.payload), sum(r.payload) +FROM (VALUES (32), (32), (48), (48), (56), (56)) l(threshold) +RIGHT SEMI JOIN existence_batches r ON l.threshold > r.k +GROUP BY r.batch_id +---- +0 64 1000 1063 66016 +2 56 1128 1183 64708 +3 1 1192 1192 1192 + +query IIIII rowsort +SELECT r.batch_id, count(*), min(r.payload), max(r.payload), sum(r.payload) +FROM (VALUES (32), (32), (48), (48), (56), (56)) l(threshold) +RIGHT ANTI JOIN existence_batches r ON l.threshold > r.k +GROUP BY r.batch_id +---- +1 64 1064 1127 70112 +2 8 1184 1191 9500 + +# Prove these are four separate right input batches, including the final short one. +query TT +EXPLAIN ANALYZE SELECT count(*) +FROM (VALUES (32), (32), (48), (48), (56), (56)) l(threshold) +RIGHT SEMI JOIN existence_batches r ON l.threshold > r.k +---- +Plan with Metrics +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)], metrics=[] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))], metrics=[] +03)----NestedLoopJoinExec: join_type=RightSemi, filter=threshold@0 > k@1, projection=[], metrics=[output_rows=121, spill_count=0, input_batches=4, input_rows=193, ] +04)------ProjectionExec: expr=[column1@0 as threshold], metrics=[] +05)--------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] +06)------ProjectionExec: expr=[CASE WHEN value@0 < 64 THEN -1 WHEN value@0 < 128 THEN 1000 WHEN value@0 < 192 THEN value@0 - 128 ELSE 0 END as k], metrics=[] +07)--------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=193, batch_size=64], metrics=[] + +# No right columns in the filter: the projected candidate batch has zero columns. +query I +SELECT count(*) FROM existence_left l RIGHT SEMI JOIN existence_right r ON TRUE +---- +129 + +# Restore the runner defaults. +statement ok +RESET datafusion.execution.batch_size + +statement ok +SET datafusion.execution.target_partitions = 4 + +statement ok +RESET datafusion.optimizer.join_reordering diff --git a/datafusion/sqllogictest/test_files/nested_loop_join_spill.slt b/datafusion/sqllogictest/test_files/nested_loop_join_spill.slt index 6eaf2b38a5bcf..0bad193823406 100644 --- a/datafusion/sqllogictest/test_files/nested_loop_join_spill.slt +++ b/datafusion/sqllogictest/test_files/nested_loop_join_spill.slt @@ -239,9 +239,133 @@ WHERE NOT EXISTS ( ---- 4950 +# --- RIGHT SEMI / RIGHT ANTI candidate pruning across spilled left chunks --- +statement ok +SET datafusion.execution.target_partitions = 1 + +statement ok +SET datafusion.optimizer.join_reordering = false + +statement ok +CREATE VIEW existence_spill_right AS +SELECT 1000 + value AS payload, + CASE WHEN (value * 2) % 129 = 128 THEN NULL ELSE (value * 2) % 129 END AS k, + 'payload-' || value AS label +FROM range(129) + +# Key 127 matches only in a later left chunk, after earlier compactions. +statement ok +CREATE VIEW existence_spill_left AS +SELECT CASE WHEN value = 4097 THEN 130 + WHEN value % 65 = 0 THEN 65 + WHEN value % 65 = 1 THEN 97 ELSE 114 END AS threshold +FROM range(5000) + +# Force multiple left chunks and multiple right batches. +statement ok +SET datafusion.runtime.memory_limit = '16K' + +statement ok +SET datafusion.execution.batch_size = 65 + +query III +SELECT count(*), sum(r.payload), max(r.k) +FROM existence_spill_left l +RIGHT SEMI JOIN existence_spill_right r ON l.threshold > r.k +---- +128 136192 127 + +query IIT +SELECT r.payload, r.k, r.label +FROM existence_spill_left l +RIGHT ANTI JOIN existence_spill_right r ON l.threshold > r.k +---- +1064 NULL payload-64 + +query TT +EXPLAIN ANALYZE SELECT count(*) +FROM existence_spill_left l +RIGHT SEMI JOIN existence_spill_right r ON l.threshold > r.k +---- +Plan with Metrics +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)], metrics=[] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))], metrics=[] +03)----NestedLoopJoinExec: join_type=RightSemi, filter=threshold@0 > k@1, projection=[], metrics=[output_rows=128, spill_count=2, input_batches=8, input_rows=516, ] +04)------ProjectionExec: expr=[CASE WHEN value@0 = 4097 THEN 130 WHEN value@0 % 65 = 0 THEN 65 WHEN value@0 % 65 = 1 THEN 97 ELSE 114 END as threshold], metrics=[] +05)--------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=5000, batch_size=65], metrics=[] +06)------ProjectionExec: expr=[CASE WHEN __common_expr_1@0 = 128 THEN NULL ELSE __common_expr_1@0 END as k], metrics=[] +07)--------ProjectionExec: expr=[value@0 * 2 % 129 as __common_expr_1], metrics=[] +08)----------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=129, batch_size=65], metrics=[] + +query TT +EXPLAIN ANALYZE SELECT count(*) +FROM existence_spill_left l +RIGHT ANTI JOIN existence_spill_right r ON l.threshold > r.k +---- +Plan with Metrics +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)], metrics=[] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))], metrics=[] +03)----NestedLoopJoinExec: join_type=RightAnti, filter=threshold@0 > k@1, projection=[], metrics=[output_rows=1, spill_count=2, input_batches=8, input_rows=516, ] +04)------ProjectionExec: expr=[CASE WHEN value@0 = 4097 THEN 130 WHEN value@0 % 65 = 0 THEN 65 WHEN value@0 % 65 = 1 THEN 97 ELSE 114 END as threshold], metrics=[] +05)--------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=5000, batch_size=65], metrics=[] +06)------ProjectionExec: expr=[CASE WHEN __common_expr_1@0 = 128 THEN NULL ELSE __common_expr_1@0 END as k], metrics=[] +07)--------ProjectionExec: expr=[value@0 * 2 % 129 as __common_expr_1], metrics=[] +08)----------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=129, batch_size=65], metrics=[] + +# Every row in both right batches matches the first left row. Later left chunks +# contain only zero thresholds, so none of those rows match again. Early stopping +# in the first chunk must preserve the global right bitmap through every replay. +statement ok +CREATE VIEW existence_spill_early_left AS +SELECT CASE WHEN value = 0 THEN 129 ELSE 0 END AS threshold +FROM range(5000) + +query IIII +SELECT count(*), sum(r.value), min(r.value), max(r.value) +FROM existence_spill_early_left l +RIGHT SEMI JOIN range(129) r ON l.threshold > r.value +---- +129 8256 0 128 + +# Both joins must spill and replay the two right batches across four left chunks. +query TT +EXPLAIN ANALYZE SELECT count(*) +FROM existence_spill_early_left l +RIGHT SEMI JOIN range(129) r ON l.threshold > r.value +---- +Plan with Metrics +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)], metrics=[] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))], metrics=[] +03)----NestedLoopJoinExec: join_type=RightSemi, filter=threshold@0 > value@1, projection=[], metrics=[output_rows=129, spill_count=2, input_batches=8, input_rows=516, ] +04)------ProjectionExec: expr=[CASE WHEN value@0 = 0 THEN 129 ELSE 0 END as threshold], metrics=[] +05)--------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=5000, batch_size=65], metrics=[] +06)------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=129, batch_size=65], metrics=[] + +query TT +EXPLAIN ANALYZE SELECT count(*) +FROM existence_spill_early_left l +RIGHT ANTI JOIN range(129) r ON l.threshold > r.value +---- +Plan with Metrics +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)], metrics=[] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))], metrics=[] +03)----NestedLoopJoinExec: join_type=RightAnti, filter=threshold@0 > value@1, projection=[], metrics=[output_rows=0, spill_count=2, input_batches=8, input_rows=516, ] +04)------ProjectionExec: expr=[CASE WHEN value@0 = 0 THEN 129 ELSE 0 END as threshold], metrics=[] +05)--------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=5000, batch_size=65], metrics=[] +06)------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=129, batch_size=65], metrics=[] + # Restore settings to slt runner defaults statement ok RESET datafusion.runtime.memory_limit statement ok RESET datafusion.catalog.create_default_catalog_and_schema + +statement ok +RESET datafusion.execution.batch_size + +statement ok +SET datafusion.execution.target_partitions = 4 + +statement ok +RESET datafusion.optimizer.join_reordering