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/10/05 13:12:18 UTC
(datafusion) 01/01: fix: allow NULL-padded rows in sort-merge join filter evaluation (#26027)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26027-5741773891f25843bc4fc2fbaf31f6d944fe5459
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 8248a579695b92b7ca94709eedd63a71689513ce
Author: Yu-Chuan Hung <86...@users.noreply.github.com>
AuthorDate: Mon Oct 5 13:11:53 2026 +0000
fix: allow NULL-padded rows in sort-merge join filter evaluation (#26027)
## Which issue does this PR close?
- Closes #26001.
## Rationale for this change
With sort-merge join (`datafusion.optimizer.prefer_hash_join = false`),
a `LEFT`, `RIGHT` or `FULL` join with an extra join filter fails with
`Arrow error: Invalid argument error: Column '...' is declared as
non-nullable but contains null values` when the NULL-padded side has
`NOT NULL` columns referenced by the filter. Hash join returns the
correct result for the same queries.
## What changes are included in this PR?
**Root cause.** As @jayzhan211 pointed out in the issue, when a streamed
row has no match but the buffered side still holds a batch,
`null_join_streamed_row` appends it with `Some(scanning_batch_idx)`, so
it is materialized in the same chunk as matched rows. Its NULL-padded
buffered-side columns then reach join filter evaluation, but the
filter's intermediate schema declares them non-nullable, so
`RecordBatch::try_new` fails.
**Fix.** Following the fix suggested in the issue,
`MaterializingSortMergeJoinStream::try_new` builds the filter schema
once with buffered-side fields marked nullable (`JoinSide::Right` for
`Left`/`Full`, `JoinSide::Left` for `Right`), and filter evaluation uses
it. It is only built when `deferred_filtering` is set (outer join with a
filter); inner joins keep the original schema.
**Why this is safe.** The filter result for a NULL-padded row does not
affect the output. Such a row is the only entry for its streamed row, so
`get_corrected_filter_mask` keeps it as-is if the filter returns true
and null-joins it if false; either way the same NULL-padded row is
emitted. The `Full` join filter-status tracking already skips NULL
buffered indices, so this path already expects such rows.
**Alternative considered.** Keep NULL-padded rows out of the filter by
appending them with `buffered_batch_idx = None`. Done naively, this
reorders output: `freeze_streamed` pushes `None` chunks immediately but
pushes matched chunks together at the end, so sort order would be
silently broken. Flushing matched chunks before each `None` chunk would
preserve order, but splits the single batched filter evaluation in
`freeze_streamed_matched`, which matters most when keys are near-unique.
Making the schema nullable avoids both.
## What is the testing strategy for this PR?
Added sqllogictest cases to `sort_merge_join.slt` with `NOT NULL`
columns and `target_partitions = 2` (needed to reproduce; the queries
pass with 1 partition):
- `LEFT`, `FULL` and `RIGHT` joins from the issue, with expected results
matching hash join.
- A `LEFT` join whose filter divides by a buffered-side column (`a.v /
b.w < 1`), to check that the zero-valued slots under NULL-padded rows do
not cause a divide-by-zero error.
All four queries fail on `main` with the Arrow error above and pass with
this change. The full extended test suite also passes locally.
## Are there any user-facing changes?
No. Bug fix only; no public API change.
---
.../joins/sort_merge_join/materializing_stream.rs | 38 ++++++++++++--
.../sqllogictest/test_files/sort_merge_join.slt | 58 ++++++++++++++++++++++
2 files changed, 92 insertions(+), 4 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 19f9e49973..6e5193f535 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
@@ -45,11 +45,11 @@ use arrow::compute::{
self, BatchCoalescer, SortOptions, concat_batches, filter_record_batch, interleave,
take_arrays,
};
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{Schema, SchemaRef};
use datafusion_common::cast::as_uint64_array;
use datafusion_common::instant::Instant;
use datafusion_common::{
- DataFusionError, JoinType, NullEquality, Result, exec_err, internal_err,
+ DataFusionError, JoinSide, JoinType, NullEquality, Result, exec_err, internal_err,
};
use datafusion_execution::memory_pool::MemoryReservation;
use datafusion_execution::runtime_env::RuntimeEnv;
@@ -295,6 +295,10 @@ pub(super) struct MaterializingSortMergeJoinStream {
/// Cached `needs_deferred_filtering(filter, join_type)` — both inputs
/// are fixed at construction time.
pub deferred_filtering: bool,
+ /// The join filter's schema with buffered-side fields marked nullable,
+ /// since NULL-padded rows reach filter evaluation. `None` unless
+ /// `deferred_filtering`.
+ pub filter_schema: Option<SchemaRef>,
/// Target output batch size
pub batch_size: usize,
@@ -551,6 +555,30 @@ impl MaterializingSortMergeJoinStream {
semi/anti/mark joins use BitwiseSortMergeJoinStream"
);
let join_time = join_metrics.join_time();
+ let buffered_side = if join_type == JoinType::Right {
+ JoinSide::Left
+ } else {
+ JoinSide::Right
+ };
+ let deferred_filtering = needs_deferred_filtering(filter.as_ref(), join_type);
+ let filter_schema = if deferred_filtering && let Some(f) = &filter {
+ let mut fields = Vec::with_capacity(f.schema().fields().len());
+ for (column_index, field) in
+ f.column_indices().iter().zip(f.schema().fields().iter())
+ {
+ if column_index.side == buffered_side {
+ fields.push(Arc::new(field.as_ref().clone().with_nullable(true)));
+ } else {
+ fields.push(Arc::clone(field));
+ }
+ }
+ Some(Arc::new(Schema::new_with_metadata(
+ fields,
+ f.schema().metadata().clone(),
+ )))
+ } else {
+ None
+ };
let mut this = Self {
sort_options,
null_equality,
@@ -566,7 +594,8 @@ impl MaterializingSortMergeJoinStream {
buffered_exhausted: false,
on_streamed,
on_buffered,
- deferred_filtering: needs_deferred_filtering(filter.as_ref(), join_type),
+ deferred_filtering,
+ filter_schema,
filter,
joined_record_batches: JoinedRecordBatches {
joined_batches: new_output_coalescer(Arc::clone(&schema), batch_size),
@@ -1601,8 +1630,9 @@ impl MaterializingSortMergeJoinStream {
if !filter_columns.is_empty() {
if let Some(f) = &self.filter {
+ let schema = self.filter_schema.as_ref().unwrap_or_else(|| f.schema());
let filter_batch =
- RecordBatch::try_new(Arc::clone(f.schema()), filter_columns)?;
+ RecordBatch::try_new(Arc::clone(schema), filter_columns)?;
let filter_result = f
.expression()
.evaluate(&filter_batch)?
diff --git a/datafusion/sqllogictest/test_files/sort_merge_join.slt b/datafusion/sqllogictest/test_files/sort_merge_join.slt
index 5ae8117290..5c563c8e81 100644
--- a/datafusion/sqllogictest/test_files/sort_merge_join.slt
+++ b/datafusion/sqllogictest/test_files/sort_merge_join.slt
@@ -971,6 +971,64 @@ WHERE t1_sorted.data < 0
----
100
+# Regression test for #26001: outer SMJ with a join filter must not fail
+# when the NULL-padded side has NOT NULL columns. target_partitions = 2
+# is needed to reproduce; with 1 partition the bug does not trigger.
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+statement ok
+CREATE TABLE a (k INT NOT NULL, v INT NOT NULL) AS VALUES (1, 10), (2, 20), (3, 30);
+
+statement ok
+CREATE TABLE b (k INT NOT NULL, w INT NOT NULL) AS VALUES (1, 100), (3, 300);
+
+statement ok
+CREATE TABLE c (k INT NOT NULL, w INT NOT NULL) AS VALUES (1, 100), (2, 200), (3, 300);
+
+query IIII
+SELECT * FROM a LEFT JOIN b ON a.k = b.k AND a.v < b.w ORDER BY a.k
+----
+1 10 1 100
+2 20 NULL NULL
+3 30 3 300
+
+query IIII
+SELECT * FROM a FULL JOIN b ON a.k = b.k AND a.v < b.w ORDER BY a.k
+----
+1 10 1 100
+2 20 NULL NULL
+3 30 3 300
+
+query IIII
+SELECT * FROM b RIGHT JOIN c ON b.k = c.k AND b.w <= c.w ORDER BY c.k
+----
+1 100 1 100
+NULL NULL 2 200
+3 300 3 300
+
+# NULL-padded rows reach the filter; their zero-valued slots must not
+# cause divide-by-zero errors
+query IIII
+SELECT * FROM a LEFT JOIN b ON a.k = b.k AND a.v / b.w < 1 ORDER BY a.k
+----
+1 10 1 100
+2 20 NULL NULL
+3 30 3 300
+
+statement ok
+DROP TABLE a;
+
+statement ok
+DROP TABLE b;
+
+statement ok
+DROP TABLE c;
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
# Config reset
statement ok
RESET datafusion.optimizer.prefer_hash_join;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org