fix: Disable join dynamic filters for null-equal joins (#22965)

## Which issue does this PR close?

- Closes ##22964

## Rationale for this change

We presently allow dynamic filter pushdown to be applied to null-equal
hash joins. This might result in pushing a predicate down into the
probe-side plan, where the predicate will not be evaluated with the
null-equal semantics that are required.

Longer-term, we might consider supporting this case with the correct
semantics (e.g., generate a predicate with `OR IS NULL ...`), but for
now disabling pushdown for null-equal joins seems much more practical.

## What changes are included in this PR?

* Disable hash join dynamic filter pushdown for null-equal joins
* Add SLT test with end-to-end repro
* Add unit test

## Are these changes tested?

Yes.

## Are there any user-facing changes?

No.

Author: neilconway

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

@@ -844,6 +844,14 @@ 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
@@ -6417,6 +6425,35 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_dynamic_filter_pushdown_rejects_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]));
+
+        let mut session_config = SessionConfig::default();
+        session_config
+            .options_mut()
+            .optimizer
+            .enable_join_dynamic_filter_pushdown = true;
+
+        let join = HashJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            &JoinType::RightSemi,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNull,
+            false,
+        )?;
+
+        assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options()));
+
+        Ok(())
+    }
+
     #[test]
     fn test_with_dynamic_filter_rejects_invalid_columns() -> Result<()> {
         let (_, _, on) = build_schema_and_on()?;

datafusion/sqllogictest/test_files/push_down_filter_parquet.slt

@@ -1023,6 +1023,48 @@ statement ok
 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.
+########
+
+statement ok
+COPY (SELECT * FROM (VALUES (11), (22), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nej_probe.parquet' STORED AS PARQUET;
+
+statement ok
+COPY (SELECT * FROM (VALUES (11), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nej_build.parquet' STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE nej_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nej_probe.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE nej_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nej_build.parquet';
+
+# The probe-side NULL key must survive to match the build-side NULL
+query II rowsort
+SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id IS NOT DISTINCT FROM nej_probe.id
+----
+11 11
+NULL NULL
+
+# No DynamicFilter predicate may appear on the probe side of a null-equal join
+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
+
+statement ok
+drop table nej_build;
+
+statement ok
+drop table nej_probe;
+
+
 # Config reset
 statement ok
 RESET datafusion.explain.physical_plan_only;