fix: preserve Inexact precision in Statistics (#22146)

## Which issue does this PR close?

No issue reported

## Rationale for this change

`Statistics::with_fetch` unconditionally returned `Precision::Exact(0)`
when the input had `nr <= skip`, even when the input was `Inexact(nr)` —
promoting an estimated upper bound into an exact zero. The exactness
flag then misleads downstream consumers (notably `AggregateStatistics`
via `Count::value_from_stats`) into trusting a derived "0" and folding
the count subtree to a literal.

Concrete user-visible symptom reported on TPC-H Q22:

```rust
let df = ctx.sql(q22_sql).await?;
df.clone().show().await?;   // prints 7 rows (correct)
df.count().await?;          // returns 0 (wrong)
```

`EXPLAIN` for the count plan shows the outer count aggregate collapsed
to `ProjectionExec([lit(0)]) -> PlaceholderRowExec`.

After PR #21240 left uncorrelated scalar subqueries in the filter rather
than rewriting them to joins, `FilterExec` can't use interval analysis
on `ScalarSubqueryExpr`, falls back to the 20% default selectivity, and
produces a small `Inexact` row estimate. A `LeftAnti` join whose
estimated semi-overlap covers the outer estimate then yields
`Inexact(0)`. That zero propagates through grouped aggregates whose
`estimate_num_rows` returns the child stats unchanged when `value == 0`.
The pre-existing `with_fetch` bug on a downstream `SortExec` finally
promotes it to `Exact(0)`, which `AggregateStatistics` trusts.

The root cause is the precision promotion in `with_fetch`. The PR fixes
that; the surrounding plan-shape changes after #21240 just made it
reachable.

## What changes are included in this PR?

- `Statistics::with_fetch`: when `nr <= skip`, preserve the exactness of
the input via `check_num_rows(Some(0),
self.num_rows.is_exact().unwrap())` instead of always returning
`Exact(0)`.
- `datafusion-common`: new unit test
`test_with_fetch_skip_all_rows_inexact` pinning the new behaviour.
- `datafusion-physical-plan`: update the existing
`test_row_number_statistics_for_global_limit` expectation that encoded
the old (incorrect) promotion to expect `Inexact(0)` now.
- `datafusion/sqllogictest/test_files/subquery.slt`: SLT regression test
reproducing the user-visible `count(*)` symptom over a query that
contains a scalar subquery, `not exists`, and a group-by on a derived
column, backed by parquet sources so the data sources report Exact
statistics.

## Are these changes tested?

Yes:
- Unit test in `datafusion-common` for the precision-preserving
behaviour.
- Updated unit test in `datafusion-physical-plan/limit.rs`.
- SLT regression test in `subquery.slt` that fails without the fix
(`count(*)` returns `0` instead of `2`) and passes with it.

## Are there any user-facing changes?

Only the bug fix. No public API changes.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: timsaucer

datafusion/common/src/stats.rs

@@ -553,10 +553,12 @@ impl Statistics {
                 num_rows: Precision::Inexact(nr),
                 ..
             } => {
-                // Here, the inexact case gives us an upper bound on the number of rows.
+                // Here, the inexact case gives us an estimate of the number of rows.
                 if nr <= skip {
-                    // All input data will be skipped:
-                    Precision::Exact(0)
+                    // All input data will be skipped. Preserve the exactness of
+                    // the input estimate: if the input was inexact, the
+                    // resulting zero is also inexact.
+                    check_num_rows(Some(0), self.num_rows.is_exact().unwrap())
                 } else if nr <= fetch_val && skip == 0 {
                     // If the input does not reach the `fetch` globally, and `skip`
                     // is zero (meaning the input and output are identical), return
@@ -2336,6 +2338,22 @@ mod tests {
         assert_eq!(result.total_byte_size, Precision::Inexact(0));
     }
 
+    #[test]
+    fn test_with_fetch_skip_all_rows_inexact() {
+        // When the input num_rows is Inexact (an upper-bound estimate), an
+        // `nr <= skip` outcome must remain Inexact: the estimate could be
+        // wrong, so we cannot promote 0 to Exact.
+        let original_stats = Statistics {
+            num_rows: Precision::Inexact(0),
+            total_byte_size: Precision::Inexact(0),
+            column_statistics: vec![col_stats_i64(10)],
+        };
+
+        let result = original_stats.clone().with_fetch(None, 0, 1).unwrap();
+
+        assert_eq!(result.num_rows, Precision::Inexact(0));
+    }
+
     #[test]
     fn test_with_fetch_no_limit() {
         // Test when fetch is None and skip is 0 (no limit applied)

datafusion/physical-plan/src/limit.rs

@@ -791,9 +791,12 @@ mod tests {
             row_number_inexact_statistics_for_global_limit(5, Some(10)).await?;
         assert_eq!(row_count, Precision::Inexact(10));
 
+        // Input was Inexact, so an `nr <= skip` outcome must remain Inexact:
+        // the inexact estimate could be wrong, so we cannot promote 0 to
+        // Exact.
         let row_count =
             row_number_inexact_statistics_for_global_limit(400, Some(10)).await?;
-        assert_eq!(row_count, Precision::Exact(0));
+        assert_eq!(row_count, Precision::Inexact(0));
 
         let row_count =
             row_number_inexact_statistics_for_global_limit(398, Some(10)).await?;

datafusion/sqllogictest/test_files/subquery.slt

@@ -2187,3 +2187,87 @@ ORDER BY column1;
 
 statement ok
 DROP TABLE subquery_partitioned;
+
+# Regression test: a `count(*)` aggregate wrapping a query whose filter
+# contains an uncorrelated scalar subquery used to be folded to a literal
+# `0` by the `AggregateStatistics` physical-optimizer rule.
+#
+# The chain that triggered it requires Exact source statistics (parquet),
+# a filter the interval analyzer cannot reason about (the scalar subquery)
+# so a small default-selectivity Inexact upper-bound flows out, and a
+# LeftAnti join whose semi-join estimate matches the outer estimate.
+# `Statistics::with_fetch` then incorrectly promoted the resulting
+# `Inexact(0)` to `Exact(0)`, which the count statistics fast-path
+# trusted.
+query I
+COPY (SELECT column1 AS c_custkey,
+             column2 AS c_phone,
+             arrow_cast(column3, 'Decimal128(15, 2)') AS c_acctbal
+      FROM (VALUES (1::BIGINT, '13-a', 10.0),
+                   (2::BIGINT, '17-b', 20.0),
+                   (3::BIGINT, '18-c', 30.0),
+                   (4::BIGINT, '23-d',  5.0),
+                   (5::BIGINT, '29-e', 40.0)))
+TO 'test_files/scratch/subquery/count_scalar_sq/customer.parquet';
+----
+5
+
+query I
+COPY (SELECT column1 AS o_custkey FROM (VALUES (1::BIGINT), (4::BIGINT)))
+TO 'test_files/scratch/subquery/count_scalar_sq/orders.parquet';
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE sq_count_customer
+STORED AS PARQUET
+LOCATION 'test_files/scratch/subquery/count_scalar_sq/customer.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE sq_count_orders
+STORED AS PARQUET
+LOCATION 'test_files/scratch/subquery/count_scalar_sq/orders.parquet';
+
+# Inner query result: 2 distinct cntrycodes survive the filters/anti-join.
+query TIR
+select cntrycode, count(*) as numcust, sum(c_acctbal) as totacctbal
+from (
+    select substring(c_phone from 1 for 2) as cntrycode, c_acctbal
+    from sq_count_customer
+    where c_acctbal > (
+        select avg(c_acctbal) from sq_count_customer where c_acctbal > 0.00
+    )
+      and not exists (
+        select * from sq_count_orders where o_custkey = c_custkey
+      )
+) as custsale
+group by cntrycode
+order by cntrycode;
+----
+18 1 30
+29 1 40
+
+# `count(*)` over the same query must agree with the row count above.
+query I
+select count(*) from (
+    select cntrycode, count(*) as numcust, sum(c_acctbal) as totacctbal
+    from (
+        select substring(c_phone from 1 for 2) as cntrycode, c_acctbal
+        from sq_count_customer
+        where c_acctbal > (
+            select avg(c_acctbal) from sq_count_customer where c_acctbal > 0.00
+        )
+          and not exists (
+            select * from sq_count_orders where o_custkey = c_custkey
+          )
+    ) as custsale
+    group by cntrycode
+) as q;
+----
+2
+
+statement ok
+DROP TABLE sq_count_customer;
+
+statement ok
+DROP TABLE sq_count_orders;