fix: re-enable null-equal join dynamic filters with an IS NULL predicate

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(())
     }

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<Schema>,
+    /// 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,13 +359,15 @@ 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,
         right_child: &dyn ExecutionPlan,
         dynamic_filter: Arc<DynamicFilterPhysicalExpr>,
         on_right: Vec<PhysicalExprRef>,
         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,51 @@ 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<dyn PhysicalExpr>,
-    ) -> Arc<dyn PhysicalExpr> {
-        if !self.null_aware {
-            return filter_expr;
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        if self.null_equality != NullEquality::NullEqualsNull && !self.null_aware {
+            return Ok(filter_expr);
+        }
+        // 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 mut any_key_is_null: Option<Arc<dyn PhysicalExpr>> = None;
+        for key in &self.on_right {
+            // `nullable` fails only when a key is out of sync with the probe schema. That is
+            // a construction bug, so surface it instead of widening around it.
+            if !key.nullable(&self.probe_schema)? {
+                continue;
+            }
+            let is_null =
+                Arc::new(IsNullExpr::new(Arc::clone(key))) as Arc<dyn PhysicalExpr>;
+            any_key_is_null = Some(match any_key_is_null {
+                Some(acc) => Arc::new(BinaryExpr::new(acc, Operator::Or, is_null)) as _,
+                None => is_null,
+            });
         }
-        debug_assert_eq!(
-            self.on_right.len(),
-            1,
-            "null_aware anti join must have exactly one probe key"
-        );
-        let probe_key_is_null: Arc<dyn PhysicalExpr> =
-            Arc::new(IsNullExpr::new(Arc::clone(&self.on_right[0])));
         // Cheap null check first short-circuits before the costlier dynamic filter.
-        Arc::new(BinaryExpr::new(
-            probe_key_is_null,
-            Operator::Or,
-            filter_expr,
-        ))
+        Ok(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 +777,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 +835,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 +1096,71 @@ mod tests {
         assert!(matches!(partitions[0], PartitionStatus::CanceledUnknown));
         assert_eq!(completed, 1);
     }
+
+    fn null_equal_accumulator(
+        probe_schema: Arc<Schema>,
+        on_right: Vec<PhysicalExprRef>,
+    ) -> 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<PhysicalExprRef> = 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)).unwrap();
+        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<PhysicalExprRef> =
+            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)).unwrap();
+        assert_eq!(format!("{result}"), format!("{filter}"));
+    }
+
+    #[test]
+    fn preserve_probe_nulls_rejects_out_of_sync_key() {
+        let probe_schema =
+            Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+        // The key's column index points past the probe schema: a construction bug that
+        // must surface as an error, not get widened around.
+        let on_right: Vec<PhysicalExprRef> = vec![Arc::new(Column::new("b", 1))];
+        let acc = null_equal_accumulator(probe_schema, on_right);
+
+        assert!(acc.preserve_probe_nulls(lit(true)).is_err());
+    }
 }

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.
 #