Fix Subtraction overflow in max_distinct_count when hash join has a pushed-down limit (apache#20799)

## Which issue does this PR close?

- Closes apache#20779.

## Rationale for this change

In `max_distinct_count` (inside
`datafusion/physical-plan/src/joins/utils.rs`), the
`Precision::Exact` branch computes the number of non-null rows by doing:

```rust
let count = count - stats.null_count.get_value().unwrap_or(&0);
```

Before apache#20228 this subtraction was always safe because `num_rows` was
never smaller
than `null_count`. But apache#20228 added `fetch` (limit push-down) support to
`HashJoinExec`, and when a limit is applied, `partition_statistics()`
caps
`num_rows` to `Exact(fetch_value)` without also capping the per-column
`null_count`. This means `null_count` can legally exceed `num_rows`,
causing a
panic with *"attempt to subtract with overflow"*.

## What changes are included in this PR?

- **Bug fix** in `max_distinct_count` (`utils.rs` ~line 725): replaced
the bare
subtraction with a saturating subtraction so that when `null_count`
exceeds
  `num_rows` the result is clamped to `0` instead of panicking.

  ```rust
  // Before
  let count = count - stats.null_count.get_value().unwrap_or(&0);

  // After
let count =
count.saturating_sub(*stats.null_count.get_value().unwrap_or(&0));
  ```

- **Regression test** added at the bottom of the `mod tests` block in
the same
file. The test deliberately constructs a scenario where `null_count (5)
>
num_rows (2)` and asserts that `max_distinct_count` returns `Exact(0)`
without
  panicking.

## Are these changes tested?

Yes. A new unit test
`test_max_distinct_count_no_overflow_when_null_count_exceeds_num_rows`
is added
directly in `datafusion/physical-plan/src/joins/utils.rs`. It covers the
exact
edge-case from the bug report (null_count > num_rows after a fetch/limit
push-down) and would have caught the panic before the fix.

## Are there any user-facing changes?

No user-facing or API changes. This is a purely internal arithmetic fix
in the
statistics estimation logic. Queries that previously panicked when a
limit was
pushed down into a `HashJoinExec` will now complete successfully.

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: KARTIK64-rgb

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

@@ -722,11 +722,12 @@ fn max_distinct_count(
                     }
                 }
                 Precision::Exact(count) => {
-                    let count = count - stats.null_count.get_value().unwrap_or(&0);
+                    let null_count = *stats.null_count.get_value().unwrap_or(&0);
+                    let non_null_count = count.checked_sub(null_count).unwrap_or(0);
                     if stats.null_count.is_exact().unwrap_or(false) {
-                        Precision::Exact(count)
+                        Precision::Exact(non_null_count)
                     } else {
-                        Precision::Inexact(count)
+                        Precision::Inexact(non_null_count)
                     }
                 }
             };
@@ -2939,4 +2940,19 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_max_distinct_count_no_overflow_when_null_count_exceeds_num_rows() {
+        let num_rows = Exact(2);
+        let stats = ColumnStatistics {
+            distinct_count: Absent,
+            null_count: Exact(5),
+            min_value: Absent,
+            max_value: Absent,
+            sum_value: Absent,
+            byte_size: Absent,
+        };
+        let result = max_distinct_count(&num_rows, &stats);
+        assert_eq!(result, Exact(0));
+    }
 }

datafusion/sqllogictest/test_files/joins.slt

@@ -5404,3 +5404,26 @@ DROP TABLE t1;
 
 statement count 0
 DROP TABLE t2;
+
+statement ok
+CREATE TABLE t1(a INT, b INT) AS VALUES 
+  (NULL, 1), (NULL, 2), (NULL, 3), (NULL, 4), (NULL, 5);
+
+statement ok
+CREATE TABLE t2(c INT) AS VALUES (1), (2);
+
+# This query panicked before the fix: the ORDER BY forces a SortExec,
+# the LIMIT gets pushed into SortExec.fetch, and the HashJoinExec
+# calls partition_statistics() on the SortExec child during execution.
+query II
+SELECT sub.a, sub.b FROM (
+  SELECT * FROM t1 ORDER BY b LIMIT 1
+) sub 
+JOIN t2 ON sub.a = t2.c;
+----
+
+statement ok
+DROP TABLE t1;
+
+statement ok
+DROP TABLE t2;