From 64cd4b82f50478396e12964b86d05c7f8c2e33e1 Mon Sep 17 00:00:00 2001 From: Moe Date: Mon, 22 Jun 2026 20:07:08 -0700 Subject: [PATCH 1/2] Re-enabled null-equal join dynamic filters via an IS NULL predicate. build-side predicate prunes a probe-side NULL that can null-match a build-side NULL. Push the filter with `OR key IS NULL` over the nullable probe keys instead, the way #23104 does for null-aware anti joins. A NOT NULL key never widens the filter, so an all-NOT-NULL join keeps full selectivity. --- .../physical-plan/src/joins/hash_join/exec.rs | 15 +-- .../src/joins/hash_join/shared_bounds.rs | 117 ++++++++++++++---- .../test_files/push_down_filter_parquet.slt | 39 +++++- 3 files changed, 134 insertions(+), 37 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 9d9c867c2724b..1f5c9eed0e45e 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -863,14 +863,6 @@ impl HashJoinExec { return false; } - // Bounds and membership filters derived from the build side do not - // account for null-equal matching: a probe-side NULL key evaluates - // such predicates to NULL and would be pruned, even though it can - // match a build-side NULL when nulls compare equal. - if self.null_equality == NullEquality::NullEqualsNull { - return false; - } - // A null-aware anti join emits a build-side NULL only when the probe // is truly empty. The pushed filter can empty the probe by pruning // every row, which would surface that NULL wrongly. A NOT NULL build @@ -1411,6 +1403,7 @@ impl ExecutionPlan for HashJoinExec { filter, on_right, repartition_random_state, + self.null_equality, self.null_aware, )) }))) @@ -6933,7 +6926,7 @@ mod tests { } #[test] - fn test_dynamic_filter_pushdown_rejects_null_equal_join() -> Result<()> { + fn test_dynamic_filter_pushdown_allowed_for_null_equal_join() -> Result<()> { let (_, _, on) = build_schema_and_on()?; let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1])); let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1])); @@ -6956,7 +6949,9 @@ mod tests { false, )?; - assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options())); + // Null-equal joins keep dynamic filter pushdown: the pushed predicate carries an + // `IS NULL` disjunct so a probe-side NULL still reaches the join. + assert!(join.allow_join_dynamic_filter_pushdown(session_config.options())); Ok(()) } diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 1fa06b5c6ca23..54930799f36b9 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -33,7 +33,9 @@ use crate::joins::hash_join::partitioned_hash_eval::{ use arrow::array::ArrayRef; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::config::ConfigOptions; -use datafusion_common::{DataFusionError, Result, ScalarValue, SharedResult}; +use datafusion_common::{ + DataFusionError, NullEquality, Result, ScalarValue, SharedResult, +}; use datafusion_expr::Operator; use datafusion_functions::core::r#struct as struct_func; use datafusion_physical_expr::expressions::{ @@ -255,6 +257,9 @@ pub(crate) struct SharedBuildAccumulator { repartition_random_state: SeededRandomState, /// Schema of the probe (right) side for evaluating filter expressions probe_schema: Arc, + /// Null equality of the join. Under `NullEqualsNull` a probe-side NULL can match a + /// build-side NULL, so the pushed filter must keep NULL rows here too. + null_equality: NullEquality, /// Null-aware anti join (`NOT IN`). A probe-side NULL must reach the join so its /// three-valued logic can collapse the result, so the pushed filter keeps NULL rows. null_aware: bool, @@ -354,6 +359,7 @@ impl SharedBuildAccumulator { /// We cannot build a partial filter from some partitions - it would incorrectly eliminate /// valid join results. We must wait until we have complete information from ALL /// relevant partitions before updating the dynamic filter. + #[expect(clippy::too_many_arguments)] pub(crate) fn new_from_partition_mode( partition_mode: PartitionMode, left_child: &dyn ExecutionPlan, @@ -361,6 +367,7 @@ impl SharedBuildAccumulator { dynamic_filter: Arc, on_right: Vec, repartition_random_state: SeededRandomState, + null_equality: NullEquality, null_aware: bool, ) -> Self { // Troubleshooting: If partition counts are incorrect, verify this logic matches @@ -408,6 +415,7 @@ impl SharedBuildAccumulator { on_right, repartition_random_state, probe_schema: right_child.schema(), + null_equality, null_aware, } } @@ -585,7 +593,7 @@ impl SharedBuildAccumulator { combine_membership_and_bounds(membership_expr, bounds_expr) { self.dynamic_filter - .update(self.null_aware_filter(filter_expr))?; + .update(self.preserve_probe_nulls(filter_expr))?; } } PartitionStatus::Pending => { @@ -692,38 +700,48 @@ impl SharedBuildAccumulator { }; self.dynamic_filter - .update(self.null_aware_filter(filter_expr))?; + .update(self.preserve_probe_nulls(filter_expr))?; } } Ok(()) } - /// Wraps a pushdown filter so a null-aware anti join keeps its probe-side NULL rows. + /// Keeps probe rows with a NULL key when the join semantics need them. /// - /// The build-side predicate drops probe rows whose key is NULL, but `NOT IN` three-valued - /// logic needs that NULL to reach the join. OR-ing `probe_key IS NULL` preserves the dynamic - /// filter's selectivity for non-NULL rows while letting the NULL through. - fn null_aware_filter( + /// The build-side predicate drops probe rows whose key is NULL. A null-aware anti join + /// (`NOT IN`) needs that NULL to reach the join so three-valued logic can collapse the + /// result, and a null-equal join needs it to match a build-side NULL. OR-ing `key IS NULL` + /// keeps those rows while preserving the filter's selectivity for the rest; the join refines + /// whatever the widened filter lets through. + fn preserve_probe_nulls( &self, filter_expr: Arc, ) -> Arc { - if !self.null_aware { + if self.null_equality != NullEquality::NullEqualsNull && !self.null_aware { return filter_expr; } - debug_assert_eq!( - self.on_right.len(), - 1, - "null_aware anti join must have exactly one probe key" - ); - let probe_key_is_null: Arc = - Arc::new(IsNullExpr::new(Arc::clone(&self.on_right[0]))); + // Only a key that can actually be NULL needs the disjunct; a NOT NULL key never widens. + // Null-aware joins are single-key; null-equal joins can be multi-key, so OR every nullable + // key. If every key is NOT NULL the filter is left untouched, at full selectivity. + let any_key_is_null = self + .on_right + .iter() + .filter(|key| key.nullable(&self.probe_schema).unwrap_or(true)) + .map(|key| { + Arc::new(IsNullExpr::new(Arc::clone(key))) as Arc + }) + .reduce(|acc, is_null| { + Arc::new(BinaryExpr::new(acc, Operator::Or, is_null)) + as Arc + }); // Cheap null check first short-circuits before the costlier dynamic filter. - Arc::new(BinaryExpr::new( - probe_key_is_null, - Operator::Or, - filter_expr, - )) + match any_key_is_null { + Some(any_key_is_null) => { + Arc::new(BinaryExpr::new(any_key_is_null, Operator::Or, filter_expr)) + } + None => filter_expr, + } } } @@ -756,6 +774,7 @@ pub(super) fn make_partitioned_accumulator_for_test( on_right: vec![], repartition_random_state: SeededRandomState::with_seed(1), probe_schema, + null_equality: NullEquality::NullEqualsNothing, null_aware: false, } } @@ -813,6 +832,7 @@ mod tests { on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), + null_equality: NullEquality::NullEqualsNothing, null_aware: false, } } @@ -1073,4 +1093,59 @@ mod tests { assert!(matches!(partitions[0], PartitionStatus::CanceledUnknown)); assert_eq!(completed, 1); } + + fn null_equal_accumulator( + probe_schema: Arc, + on_right: Vec, + ) -> SharedBuildAccumulator { + SharedBuildAccumulator { + inner: Mutex::new(AccumulatorState { + data: AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; 1], + completed_partitions: 0, + }, + completion: CompletionState::Pending, + }), + completion_notify: Notify::new(), + dynamic_filter: Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true))), + on_right, + repartition_random_state: SeededRandomState::with_seed(1), + probe_schema, + null_equality: NullEquality::NullEqualsNull, + null_aware: false, + } + } + + #[test] + fn preserve_probe_nulls_only_widens_nullable_keys() { + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("k_nullable", DataType::Int32, true), + Field::new("k_not_null", DataType::Int32, false), + ])); + let on_right: Vec = vec![ + Arc::new(Column::new("k_nullable", 0)), + Arc::new(Column::new("k_not_null", 1)), + ]; + let acc = null_equal_accumulator(probe_schema, on_right); + + // Only the nullable key earns an IS NULL disjunct; the NOT NULL key is left out. + let widened = acc.preserve_probe_nulls(lit(true)); + assert_eq!(format!("{widened}").matches("IS NULL").count(), 1); + } + + #[test] + fn preserve_probe_nulls_leaves_all_not_null_keys_untouched() { + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ])); + let on_right: Vec = + vec![Arc::new(Column::new("a", 0)), Arc::new(Column::new("b", 1))]; + let acc = null_equal_accumulator(probe_schema, on_right); + + // Every key is NOT NULL, so there is nothing to OR in and the filter is returned as-is. + let filter = lit(true); + let result = acc.preserve_probe_nulls(Arc::clone(&filter)); + assert_eq!(format!("{result}"), format!("{filter}")); + } } diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index f1e787441d5e1..a857df18f0a08 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -1025,10 +1025,10 @@ drop table int_probe; ######## -# Dynamic filters must not be created for null-equal joins (IS NOT DISTINCT -# FROM, INTERSECT): min/max bounds and membership filters derived from the -# build side evaluate to NULL for probe-side NULL keys and would prune rows -# that can null-match a build-side NULL. +# Null-equal joins (IS NOT DISTINCT FROM, INTERSECT) keep dynamic filter pushdown. +# Min/max bounds and membership filters derived from the build side evaluate to NULL +# for a probe-side NULL key, so the pushed predicate carries an `IS NULL` disjunct that +# lets the probe NULL reach the join and null-match a build-side NULL. ######## statement ok @@ -1050,14 +1050,14 @@ SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id 11 11 NULL NULL -# No DynamicFilter predicate may appear on the probe side of a null-equal join +# The probe side now carries a DynamicFilter for a null-equal join (widened with IS NULL at runtime) query TT EXPLAIN SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id IS NOT DISTINCT FROM nej_probe.id ---- physical_plan 01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], NullsEqual: true 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_build.parquet]]}, projection=[id], file_type=parquet -03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_probe.parquet]]}, projection=[id], file_type=parquet +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_probe.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible statement ok drop table nej_build; @@ -1066,6 +1066,33 @@ statement ok drop table nej_probe; +# Multi-key null-equal join: the IS NULL disjunct covers every nullable key, so a probe row with a +# NULL in either key still reaches the join and null-matches the build side. +statement ok +COPY (SELECT * FROM (VALUES (1, 10), (2, NULL), (NULL, 30)) v(a, b)) TO 'test_files/scratch/push_down_filter_parquet/mnej_probe.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (1, 10), (2, NULL)) v(a, b)) TO 'test_files/scratch/push_down_filter_parquet/mnej_build.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE mnej_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/mnej_probe.parquet'; + +statement ok +CREATE EXTERNAL TABLE mnej_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/mnej_build.parquet'; + +query IIII rowsort +SELECT mnej_build.a, mnej_build.b, mnej_probe.a, mnej_probe.b FROM mnej_build JOIN mnej_probe ON (mnej_build.a IS NOT DISTINCT FROM mnej_probe.a) AND (mnej_build.b IS NOT DISTINCT FROM mnej_probe.b) +---- +1 10 1 10 +2 NULL 2 NULL + +statement ok +drop table mnej_build; + +statement ok +drop table mnej_probe; + + ######## # Regression test for build-NULL + emptied-probe interaction in null-aware LeftAnti joins. # From 8ecd68fa8d3e17d179ac0b6a6b02d6a988655b37 Mon Sep 17 00:00:00 2001 From: Moe Date: Wed, 24 Jun 2026 13:44:54 -0700 Subject: [PATCH 2/2] Documented the conservative nullability fallback. The `unwrap_or(true)` widening on an unresolved nullability check wasn't obvious. An extra NULL row is safe; dropping a needed one isn't. --- datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 54930799f36b9..51d9cdc6524b5 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -727,6 +727,7 @@ impl SharedBuildAccumulator { let any_key_is_null = self .on_right .iter() + // Widen on unresolved nullability: an extra NULL row is safe, a dropped one isn't. .filter(|key| key.nullable(&self.probe_schema).unwrap_or(true)) .map(|key| { Arc::new(IsNullExpr::new(Arc::clone(key))) as Arc