Re-enabled null-equal join dynamic filters via an IS NULL predicate.

apache#22965 disabled dynamic filter pushdown for null-equal joins because the
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 apache#23104 does for null-aware anti joins. A NOT NULL key never
widens the filter, so an all-NOT-NULL join keeps full selectivity.

Author: mdashti

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -862,14 +862,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;
-        }
-
         // `preserve_file_partitions` can report Hash partitioning for Hive-style
         // file groups, but those partitions are not actually hash-distributed.
         // Partitioned dynamic filters rely on hash routing, so disable them in
@@ -1363,6 +1355,7 @@ impl ExecutionPlan for HashJoinExec {
                             filter,
                             on_right,
                             repartition_random_state,
+                            self.null_equality,
                             self.null_aware,
                         ))
                     })))
@@ -6629,7 +6622,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]));
@@ -6652,7 +6645,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,
@@ -361,6 +366,7 @@ impl SharedBuildAccumulator {
         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 +414,7 @@ impl SharedBuildAccumulator {
             on_right,
             repartition_random_state,
             probe_schema: right_child.schema(),
+            null_equality,
             null_aware,
         }
     }
@@ -585,7 +592,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,33 +699,47 @@ 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 {
+        if self.null_equality != NullEquality::NullEqualsNull && !self.null_aware {
             return filter_expr;
         }
-        // A null-aware anti join is validated to a single probe key.
-        let probe_key_is_null: Arc<dyn PhysicalExpr> =
-            Arc::new(IsNullExpr::new(Arc::clone(&self.on_right[0])));
-        Arc::new(BinaryExpr::new(
-            filter_expr,
-            Operator::Or,
-            probe_key_is_null,
-        ))
+        // 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<dyn PhysicalExpr>
+            })
+            .reduce(|acc, is_null| {
+                Arc::new(BinaryExpr::new(acc, Operator::Or, is_null))
+                    as Arc<dyn PhysicalExpr>
+            });
+        match any_key_is_null {
+            Some(any_key_is_null) => {
+                Arc::new(BinaryExpr::new(filter_expr, Operator::Or, any_key_is_null))
+            }
+            None => filter_expr,
+        }
     }
 }
 
@@ -751,6 +772,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,
     }
 }
@@ -771,6 +793,7 @@ pub(super) fn completed_partitions_for_test(acc: &SharedBuildAccumulator) -> usi
 #[cfg(test)]
 mod tests {
     use super::*;
+    use datafusion_physical_expr::expressions::Column;
 
     fn partitioned_state(acc: &SharedBuildAccumulator) -> (Vec<PartitionStatus>, usize) {
         let guard = acc.inner.lock();
@@ -840,4 +863,59 @@ 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));
+        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));
+        assert_eq!(format!("{result}"), format!("{filter}"));
+    }
 }

datafusion/sqllogictest/test_files/push_down_filter_parquet.slt

@@ -1024,10 +1024,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
@@ -1049,14 +1049,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;
@@ -1065,6 +1065,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;
+
+
 # Config reset
 statement ok
 RESET datafusion.explain.physical_plan_only;