From 7c1807538d2058ce35ec7da505812b60f9f243c8 Mon Sep 17 00:00:00 2001 From: mrhard9090 Date: Sat, 3 Oct 2026 21:40:50 +0300 Subject: [PATCH] fix: evaluate sort-merge join filters that reference no columns Co-Authored-By: Claude Sonnet 5.5 --- .../sort_merge_join/materializing_stream.rs | 122 +++++++++--------- .../test_files/sort_merge_join.slt | 38 ++++++ 2 files changed, 101 insertions(+), 59 deletions(-) diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs index 19f9e49973bf8..3ce41b0d95be9 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs @@ -46,6 +46,7 @@ use arrow::compute::{ take_arrays, }; use arrow::datatypes::SchemaRef; +use arrow::record_batch::RecordBatchOptions; use datafusion_common::cast::as_uint64_array; use datafusion_common::instant::Instant; use datafusion_common::{ @@ -1599,72 +1600,75 @@ impl MaterializingSortMergeJoinStream { }; let output_batch = RecordBatch::try_new(Arc::clone(&self.schema), columns)?; - if !filter_columns.is_empty() { - if let Some(f) = &self.filter { - let filter_batch = - RecordBatch::try_new(Arc::clone(f.schema()), filter_columns)?; - let filter_result = f - .expression() - .evaluate(&filter_batch)? - .into_array(filter_batch.num_rows())?; - - let filter_result_mask = - datafusion_common::cast::as_boolean_array(&filter_result)?; - - // Convert NULL filter results to false — NULL means "not satisfied" - // per SQL semantics, same as Left/Right outer joins. - let mask = if filter_result_mask.null_count() > 0 { - compute::prep_null_mask_filter(filter_result_mask) - } else { - filter_result_mask.clone() - }; + // A filter that reads no columns (e.g. a bound parameter or a volatile + // function) still has to be evaluated, so key on the filter itself. + if let Some(f) = &self.filter { + let filter_batch = RecordBatch::try_new_with_options( + Arc::clone(f.schema()), + filter_columns, + &RecordBatchOptions::new().with_row_count(Some(total_matched_rows)), + )?; + let filter_result = f + .expression() + .evaluate(&filter_batch)? + .into_array(filter_batch.num_rows())?; + + let filter_result_mask = + datafusion_common::cast::as_boolean_array(&filter_result)?; + + // Convert NULL filter results to false — NULL means "not satisfied" + // per SQL semantics, same as Left/Right outer joins. + let mask = if filter_result_mask.null_count() > 0 { + compute::prep_null_mask_filter(filter_result_mask) + } else { + filter_result_mask.clone() + }; - if self.deferred_filtering { - self.joined_record_batches.push_batch_with_filter_metadata( - output_batch, - &combined_left_indices, - &mask, - self.streamed_batch_counter, - self.join_type, - ); - } else { - let filtered_batch = filter_record_batch(&output_batch, &mask)?; - self.joined_record_batches - .push_batch_without_metadata(filtered_batch); - } + if self.deferred_filtering { + self.joined_record_batches.push_batch_with_filter_metadata( + output_batch, + &combined_left_indices, + &mask, + self.streamed_batch_counter, + self.join_type, + ); + } else { + let filtered_batch = filter_record_batch(&output_batch, &mask)?; + self.joined_record_batches + .push_batch_without_metadata(filtered_batch); + } - // Track which buffered rows had all filter matches fail, - // so full join can emit them as null-joined later. - if self.join_type == JoinType::Full { - let mut offset = 0usize; - for (batch_idx, _left, right) in matched_chunks { - let chunk_len = right.len(); - let buffered_batch = &mut self.buffered_data.batches[*batch_idx]; - - for i in 0..chunk_len { - if right.is_null(i) { - continue; + // Track which buffered rows had all filter matches fail, + // so full join can emit them as null-joined later. + if self.join_type == JoinType::Full { + let mut offset = 0usize; + for (batch_idx, _left, right) in matched_chunks { + let chunk_len = right.len(); + let buffered_batch = &mut self.buffered_data.batches[*batch_idx]; + + for i in 0..chunk_len { + if right.is_null(i) { + continue; + } + let idx = right.value(i) as usize; + match buffered_batch.join_filter_status[idx] { + FilterState::SomePassed => {} + _ if mask.value(offset + i) => { + buffered_batch.join_filter_status[idx] = + FilterState::SomePassed; } - let idx = right.value(i) as usize; - match buffered_batch.join_filter_status[idx] { - FilterState::SomePassed => {} - _ if mask.value(offset + i) => { - buffered_batch.join_filter_status[idx] = - FilterState::SomePassed; - } - _ => { - buffered_batch.join_filter_status[idx] = - FilterState::AllFailed; - } + _ => { + buffered_batch.join_filter_status[idx] = + FilterState::AllFailed; } } - offset += chunk_len; } - debug_assert_eq!( - offset, total_matched_rows, - "offset must advance through every chunk exactly once" - ); + offset += chunk_len; } + debug_assert_eq!( + offset, total_matched_rows, + "offset must advance through every chunk exactly once" + ); } } else { self.joined_record_batches diff --git a/datafusion/sqllogictest/test_files/sort_merge_join.slt b/datafusion/sqllogictest/test_files/sort_merge_join.slt index 5ae8117290e40..aa4871b93e1b7 100644 --- a/datafusion/sqllogictest/test_files/sort_merge_join.slt +++ b/datafusion/sqllogictest/test_files/sort_merge_join.slt @@ -971,6 +971,44 @@ WHERE t1_sorted.data < 0 ---- 100 +# Join filter that reads no columns: it must still be evaluated for every +# candidate pair (hash join already does), so FULL JOIN null-pads both sides. +statement ok +CREATE TABLE smj_nocol_a AS SELECT * FROM (VALUES (1, 10), (2, 20), (3, 30)) t(k, v); + +statement ok +CREATE TABLE smj_nocol_b AS SELECT * FROM (VALUES (1, 100), (2, 200), (4, 400)) t(k, w); + +statement ok +set datafusion.execution.target_partitions = 2; + +query IIII +SELECT * FROM smj_nocol_a a FULL JOIN smj_nocol_b b ON a.k = b.k AND random() > 2 ORDER BY a.k, b.k; +---- +1 10 NULL NULL +2 20 NULL NULL +3 30 NULL NULL +NULL NULL 1 100 +NULL NULL 2 200 +NULL NULL 4 400 + +query IIII +SELECT * FROM smj_nocol_a a FULL JOIN smj_nocol_b b ON a.k = b.k AND random() < 2 ORDER BY a.k, b.k; +---- +1 10 1 100 +2 20 2 200 +3 30 NULL NULL +NULL NULL 4 400 + +statement ok +set datafusion.execution.target_partitions = 4; + +statement ok +DROP TABLE smj_nocol_a; + +statement ok +DROP TABLE smj_nocol_b; + # Config reset statement ok RESET datafusion.optimizer.prefer_hash_join;