Fix Subtraction overflow in max_distinct_count when hash join has a pushed-down limit

[KARTIK64-rgb]

Which issue does this PR close?

• Closes Subtraction overflow in max_distinct_count when hash join has a pushed-down limit #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:

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

Before #20228 this subtraction was always safe because num_rows was never smaller
than null_count. But #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.

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

[jonathanc-n]

Sorry didn't catch this would happen in my PR. Thanks for the fix!

[jonathanc-n]

cc @gabotechs

[jonathanc-n]

I created a test for this for sqllogictests @KARTIK64-rgb, can add:

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;

i verified it reproduces the bug

[gabotechs]

Looks good, thanks @KARTIK64-rgb for quick fix and @jonathanc-n for the review!

As soon as the CI issues are addressed, this is good to merge.

[gabotechs]

👍 nice, even if this is a good safeguard, the fact that this can even happen makes me think that there is some further work to be done in the stats propagation mechanism.

Ideally, this would not even be possible by construction, but that's a topic for another PR.

[KARTIK64-rgb]

Welcome @gabotechs @jonathanc-n !

[alamb]

@KARTIK64-rgb can you help fix the CI tests on this PR?

[alamb]

I pushed a commit to try and resolve the build error

[gabotechs]

Should be good now, in it goes. Thanks @KARTIK64-rgb for the PR, @jonathanc-n for the review and @alamb for fixing the build error.

[alamb]

woohoo -- progress every day

[KARTIK64-rgb]

Thanks @alamb for solving the ci-issues.
thanks @jonathanc-n @gabotechs for guiding me .