fix: Scale semi/anti-join column stats by estimated row count

datafusion/core/tests/physical_optimizer/partition_statistics.rs

@@ -661,39 +661,36 @@ mod test {
         let full_statistics = nested_loop_join.partition_statistics(None)?;
         // With empty join columns, estimate_join_statistics returns Inexact row count
         // based on the outer side (right side for RightSemi)
-        let mut expected_full_statistics = create_partition_statistics(
+        let expected_full_statistics = create_partition_statistics(
             4,
             32,
             1,
             4,
             Some((DATE_2025_03_01, DATE_2025_03_04)),
-        );
-        expected_full_statistics.num_rows = Precision::Inexact(4);
-        expected_full_statistics.total_byte_size = Precision::Absent;
+        )
+        .to_inexact();
         assert_eq!(*full_statistics, expected_full_statistics);
 
         // Test partition_statistics(Some(idx)) - returns partition-specific statistics
         // Partition 1: ids [3,4], dates [2025-03-01, 2025-03-02]
-        let mut expected_statistic_partition_1 = create_partition_statistics(
+        let expected_statistic_partition_1 = create_partition_statistics(
             2,
             16,
             3,
             4,
             Some((DATE_2025_03_01, DATE_2025_03_02)),
-        );
-        expected_statistic_partition_1.num_rows = Precision::Inexact(2);
-        expected_statistic_partition_1.total_byte_size = Precision::Absent;
+        )
+        .to_inexact();
 
         // Partition 2: ids [1,2], dates [2025-03-03, 2025-03-04]
-        let mut expected_statistic_partition_2 = create_partition_statistics(
+        let expected_statistic_partition_2 = create_partition_statistics(
             2,
             16,
             1,
             2,
             Some((DATE_2025_03_03, DATE_2025_03_04)),
-        );
-        expected_statistic_partition_2.num_rows = Precision::Inexact(2);
-        expected_statistic_partition_2.total_byte_size = Precision::Absent;
+        )
+        .to_inexact();
 
         let statistics = (0..nested_loop_join.output_partitioning().partition_count())
             .map(|idx| nested_loop_join.partition_statistics(Some(idx)))

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

@@ -1448,6 +1448,7 @@ impl ExecutionPlan for HashJoinExec {
                     Arc::unwrap_or_clone(left_stats),
                     Arc::unwrap_or_clone(right_stats),
                     &self.on,
+                    self.null_equality,
                     &self.join_type,
                     &self.join_schema,
                 )?
@@ -1463,6 +1464,7 @@ impl ExecutionPlan for HashJoinExec {
                     Arc::unwrap_or_clone(left_stats),
                     Arc::unwrap_or_clone(right_stats),
                     &self.on,
+                    self.null_equality,
                     &self.join_type,
                     &self.join_schema,
                 )?
@@ -1480,6 +1482,7 @@ impl ExecutionPlan for HashJoinExec {
                     Arc::unwrap_or_clone(left_stats),
                     Arc::unwrap_or_clone(right_stats),
                     &self.on,
+                    self.null_equality,
                     &self.join_type,
                     &self.join_schema,
                 )?

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -61,8 +61,9 @@ use arrow::record_batch::RecordBatch;
 use arrow_schema::DataType;
 use datafusion_common::cast::as_boolean_array;
 use datafusion_common::{
-    JoinSide, Result, ScalarValue, Statistics, arrow_err, assert_eq_or_internal_err,
-    internal_datafusion_err, internal_err, project_schema, unwrap_or_internal_err,
+    JoinSide, NullEquality, Result, ScalarValue, Statistics, arrow_err,
+    assert_eq_or_internal_err, internal_datafusion_err, internal_err, project_schema,
+    unwrap_or_internal_err,
 };
 use datafusion_execution::TaskContext;
 use datafusion_execution::disk_manager::RefCountedTempFile;
@@ -713,6 +714,7 @@ impl ExecutionPlan for NestedLoopJoinExec {
             left_stats,
             right_stats,
             &join_columns,
+            NullEquality::NullEqualsNothing,
             &self.join_type,
             &self.join_schema,
         )?;

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

@@ -582,6 +582,7 @@ impl ExecutionPlan for SortMergeJoinExec {
             left_stats,
             right_stats,
             &self.on,
+            self.null_equality,
             &self.join_type,
             &self.schema,
         )?))