You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/04/16 15:15:53 UTC
(datafusion) branch gh-readonly-queue/main/pr-21660-9873357d8d8ad2bffea38fbff0e7ddb21abb68de created (now 2ef0217705)
This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a change to branch gh-readonly-queue/main/pr-21660-9873357d8d8ad2bffea38fbff0e7ddb21abb68de
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 2ef0217705 fix: SortMergeJoin full outer join incorrectly matches rows when filter evaluates to NULL (#21660)
This branch includes the following new commits:
new 2ef0217705 fix: SortMergeJoin full outer join incorrectly matches rows when filter evaluates to NULL (#21660)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: SortMergeJoin full outer join incorrectly matches rows when filter evaluates to NULL (#21660)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch gh-readonly-queue/main/pr-21660-9873357d8d8ad2bffea38fbff0e7ddb21abb68de
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 2ef021770598b723210f81d16c897fd98dab7771
Author: Matt Butrovich <mb...@users.noreply.github.com>
AuthorDate: Thu Apr 16 11:15:08 2026 -0400
fix: SortMergeJoin full outer join incorrectly matches rows when filter evaluates to NULL (#21660)
## Which issue does this PR close?
- Closes #.
## Rationale for this change
While enabling `SortMergeJoinExec` with filters in [DataFusion
Comet](https://github.com/apache/datafusion-comet), I hit a correctness
bug in DataFusion's `SortMergeJoinExec` for full outer joins with
nullable filter columns. The bug was originally surfaced via the
[SPARK-43113](https://issues.apache.org/jira/browse/SPARK-43113)
reproducer.
When a join filter expression evaluates to `NULL` (e.g., `l.b < (r.b +
1)` where `r.b` is `NULL`), the full outer join treats the row pair as
**matched** instead of **unmatched**. Per SQL semantics, `NULL` in a
boolean filter context is `false` (not satisfied), so the rows should be
emitted as separate unmatched rows.
The bug has been present since filtered full outer join support was
added to SMJ in #12764 / #13369. It was never caught because:
1. The join fuzz tests generate filter column data with
`Int32Array::from_iter_values()`, which never produces `NULL` values.
2. No existing unit test or sqllogictest exercised a full outer join
filter that evaluates to `NULL`.
## What changes are included in this PR?
**Root cause:** The full outer join code path had a special case that
preserved raw `NULL` values from the filter expression result
(`pre_mask`) instead of converting them to `false` via
`prep_null_mask_filter` like left/right outer joins do. This caused two
problems:
1. **Streamed (left) side:** In `get_corrected_filter_mask()`, `NULL`
entries in `filter_mask` are treated as "pass through" (for
pre-null-joined rows from `append_nulls()`). But when the filter
expression itself returns `NULL`, those entries also appear as `NULL` in
the mask — and get incorrectly treated as matched. This produced wrong
join output (matched rows instead of unmatched).
2. **Buffered (right) side:** `BooleanArray::value()` was called on
`NULL` positions in `pre_mask` to update `FilterState`. At NULL
positions, the values buffer contains a deterministic but semantically
meaningless result (computed from the default zero-storage of NULL
inputs). For some rows this value happens to be `true`, which
incorrectly marks unmatched buffered rows as `SomePassed` and silently
drops them from the output.
**Fix:** Remove the full outer join special case in
`materializing_stream.rs`. All outer join types now uniformly use the
null-corrected `mask` (where `NULL` → `false` via
`prep_null_mask_filter`) for both deferred filtering metadata and
`FilterState` tracking. Semi/anti/mark joins are unaffected — they use
`BitwiseSortMergeJoinStream` which already converts NULLs to `false`.
**Tests:**
- New unit test `join_full_null_filter_result` reproducing the
SPARK-43113 scenario with a nullable right-side column.
- Modified `make_staggered_batches_i32` in `join_fuzz.rs` to inject ~10%
`NULL` values into the filter column (`x`), so the fuzz tests exercise
`NULL` filter handling across all join types.
## Are these changes tested?
Yes.
- New unit test (`join_full_null_filter_result`) directly reproduces the
bug.
- Existing 57 SMJ unit tests all pass.
- All 41 join fuzz tests pass with the new nullable filter column data,
including `test_full_join_1k_filtered` which compares `HashJoinExec` vs
`SortMergeJoinExec` and would have caught this bug if the fuzz data had
included `NULL`s.
- Will run 100 iterations of the fuzz tests overnight to shake out any
remaining nondeterministic issues.
- Testing in Comet CI (all Spark SQL tests)
https://github.com/apache/datafusion-comet/pull/3916
## Are there any user-facing changes?
Full outer sort-merge joins with filters involving nullable columns now
produce correct results. Previously, rows where the filter evaluated to
`NULL` were incorrectly included as matched; they are now correctly
emitted as unmatched (null-joined) rows.
---
datafusion/core/tests/fuzz_cases/join_fuzz.rs | 9 +-
.../joins/sort_merge_join/materializing_stream.rs | 23 ++---
.../src/joins/sort_merge_join/tests.rs | 109 ++++++++++++++++++++-
3 files changed, 122 insertions(+), 19 deletions(-)
diff --git a/datafusion/core/tests/fuzz_cases/join_fuzz.rs b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
index f570335e92..fdb2934817 100644
--- a/datafusion/core/tests/fuzz_cases/join_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
@@ -1276,7 +1276,14 @@ fn make_staggered_batches_i32(len: usize, with_extra_column: bool) -> Vec<Record
input12.sort_unstable();
let input1 = Int32Array::from_iter_values(input12.clone().into_iter().map(|k| k.0));
let input2 = Int32Array::from_iter_values(input12.clone().into_iter().map(|k| k.1));
- let input3 = Int32Array::from_iter_values(input3);
+ let input3 = Int32Array::from_iter(input3.into_iter().map(|v| {
+ // ~10% NULLs in filter column to exercise NULL filter handling
+ if rng.random_range(0..10) == 0 {
+ None
+ } else {
+ Some(v)
+ }
+ }));
let input4 = Int32Array::from_iter_values(input4);
let mut columns = vec![
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 32ed1627fb..4840b56f55 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
@@ -1423,27 +1423,22 @@ impl MaterializingSortMergeJoinStream {
.evaluate(&filter_batch)?
.into_array(filter_batch.num_rows())?;
- let pre_mask = datafusion_common::cast::as_boolean_array(&filter_result)?;
+ let filter_result_mask =
+ datafusion_common::cast::as_boolean_array(&filter_result)?;
- let mask = if pre_mask.null_count() > 0 {
- compute::prep_null_mask_filter(pre_mask)
+ // 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 {
- pre_mask.clone()
+ filter_result_mask.clone()
};
if needs_deferred_filtering(&self.filter, self.join_type) {
- // Full join uses pre_mask (preserving nulls) for
- // get_corrected_filter_mask; other outer joins use mask.
- let mask_to_use = if self.join_type != JoinType::Full {
- &mask
- } else {
- pre_mask
- };
-
self.joined_record_batches.push_batch_with_filter_metadata(
output_batch,
&combined_left_indices,
- mask_to_use,
+ &mask,
self.streamed_batch_counter.load(Relaxed),
self.join_type,
);
@@ -1468,7 +1463,7 @@ impl MaterializingSortMergeJoinStream {
let idx = right.value(i) as usize;
match buffered_batch.join_filter_status[idx] {
FilterState::SomePassed => {}
- _ if pre_mask.value(offset + i) => {
+ _ if mask.value(offset + i) => {
buffered_batch.join_filter_status[idx] =
FilterState::SomePassed;
}
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
index 5824c2328d..5d70530528 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
@@ -55,7 +55,7 @@ use datafusion_common::{
test_util::{batches_to_sort_string, batches_to_string},
};
use datafusion_common::{
- JoinType, NullEquality, Result, assert_batches_eq, assert_contains,
+ JoinType, NullEquality, Result, ScalarValue, assert_batches_eq, assert_contains,
};
use datafusion_common_runtime::JoinSet;
use datafusion_execution::config::SessionConfig;
@@ -65,6 +65,7 @@ use datafusion_execution::runtime_env::RuntimeEnvBuilder;
use datafusion_execution::{SendableRecordBatchStream, TaskContext};
use datafusion_expr::Operator;
use datafusion_physical_expr::expressions::BinaryExpr;
+use datafusion_physical_expr::expressions::Literal;
use datafusion_physical_expr_common::physical_expr::PhysicalExprRef;
use futures::{Stream, StreamExt};
use insta::assert_snapshot;
@@ -2049,6 +2050,108 @@ async fn join_full_multiple_batches() -> Result<()> {
Ok(())
}
+/// Full outer join where the filter evaluates to NULL due to a nullable column.
+/// NULL filter results must be treated as unmatched, not matched.
+/// Reproducer for SPARK-43113.
+#[tokio::test]
+async fn join_full_null_filter_result() -> Result<()> {
+ // Left: (a, b) all non-null, sorted on a
+ let left = build_table_two_cols(
+ ("a1", &vec![1, 1, 2, 2, 3, 3]),
+ ("b1", &vec![1, 2, 1, 2, 1, 2]),
+ );
+
+ // Right: (a, b) with b nullable, sorted on a
+ let right_schema = Arc::new(Schema::new(vec![
+ Field::new("a2", DataType::Int32, false),
+ Field::new("b2", DataType::Int32, true),
+ ]));
+ let right_batch = RecordBatch::try_new(
+ Arc::clone(&right_schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(Int32Array::from(vec![None, Some(2)])),
+ ],
+ )?;
+ let right =
+ TestMemoryExec::try_new_exec(&[vec![right_batch]], right_schema, None).unwrap();
+
+ let on = vec![(
+ Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("a2", &right.schema())?) as _,
+ )];
+
+ // Filter: b1 < (b2 + 1) AND b1 < (a2 + 1)
+ // When b2 is NULL, (b2 + 1) is NULL, so b1 < NULL is NULL → unmatched.
+ let lit_1: PhysicalExprRef = Arc::new(Literal::new(ScalarValue::Int32(Some(1))));
+ let b1_lt_b2_plus_1: PhysicalExprRef = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b1", 0)),
+ Operator::Lt,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b2", 1)),
+ Operator::Plus,
+ Arc::clone(&lit_1),
+ )),
+ ));
+ let b1_lt_a2_plus_1: PhysicalExprRef = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b1", 0)),
+ Operator::Lt,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a2", 2)),
+ Operator::Plus,
+ Arc::clone(&lit_1),
+ )),
+ ));
+ let filter_expr: PhysicalExprRef = Arc::new(BinaryExpr::new(
+ b1_lt_b2_plus_1,
+ Operator::And,
+ b1_lt_a2_plus_1,
+ ));
+
+ let filter = JoinFilter::new(
+ filter_expr,
+ vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ },
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Right,
+ },
+ ],
+ Arc::new(Schema::new(vec![
+ Field::new("b1", DataType::Int32, true),
+ Field::new("b2", DataType::Int32, true),
+ Field::new("a2", DataType::Int32, true),
+ ])),
+ );
+
+ let (_, batches) = join_collect_with_filter(left, right, on, filter, Full).await?;
+
+ // r=(1,NULL): b2 is NULL → b1 < (NULL+1) is NULL → all a=1 rows unmatched
+ // r=(2,2): b1 < 3 AND b1 < 3 → both l=(2,1) and l=(2,2) match
+ // l=(3,*): no right row with a=3 → unmatched
+ assert_snapshot!(batches_to_sort_string(&batches), @r"
+ +----+----+----+----+
+ | a1 | b1 | a2 | b2 |
+ +----+----+----+----+
+ | | | 1 | |
+ | 1 | 1 | | |
+ | 1 | 2 | | |
+ | 2 | 1 | 2 | 2 |
+ | 2 | 2 | 2 | 2 |
+ | 3 | 1 | | |
+ | 3 | 2 | | |
+ +----+----+----+----+
+ ");
+ Ok(())
+}
+
#[tokio::test]
async fn overallocation_single_batch_no_spill() -> Result<()> {
let left = build_table(
@@ -3589,9 +3692,7 @@ async fn join_filtered_with_multiple_buffered_batches() -> Result<()> {
Arc::new(Column::new("val_r", 1)),
)),
Operator::Lt,
- Arc::new(datafusion_physical_expr::expressions::Literal::new(
- datafusion_common::ScalarValue::Int32(Some(350)),
- )),
+ Arc::new(Literal::new(ScalarValue::Int32(Some(350)))),
)),
vec![
ColumnIndex {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org