feat: report statistics of materialized CTE scans - #26271

Which issue does this PR close?

• Closes Materialized CTE scans report unknown statistics, ordering and partitioning #26230.

Rationale for this change

A reference to a MATERIALIZED CTE reported unknown statistics. A query could then get a worse plan than the same query with the CTE inlined.

SET datafusion.sql_parser.dialect = 'PostgreSQL';
CREATE TABLE small AS SELECT value AS k FROM generate_series(1, 5);
CREATE TABLE huge  AS SELECT value AS k FROM generate_series(1, 100000);

EXPLAIN WITH c AS MATERIALIZED (SELECT k FROM small)
SELECT count(*) FROM huge JOIN c ON huge.k = c.k;

Before, the hash table is built on the 100000 rows of huge:

HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)], projection=[]
  DataSourceExec: partitions=1, partition_sizes=[13]
  MaterializedCteScanExec: name=c

After, it is built on the 5 rows of the CTE:

HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)], projection=[]
  MaterializedCteScanExec: name=c
  DataSourceExec: partitions=1, partition_sizes=[13]

What changes are included in this PR?

The scan is a leaf and the body is a child of MaterializedCteExec, so the planner copies the statistics to the scan in DefaultPhysicalPlanner::optimize_physical_plan, before the optimizer runs. The optimizer runs one time, as before. The EXPLAIN paths use the same function.

Each scan gets the statistics of the body, computed with StatisticsContext. A body that reads another CTE gets the statistics of that CTE first. A rewrite of the body does not change its rows, so the statistics stay correct. The statistics of one scan partition are known only when the scan has one partition. The statistics API is not changed.

Other plan changes that follow from the statistics: an exact row count of the body can answer count(*), and the body then does not run. One pinned plan in cte_materialized.slt has a new join order.

The ordering and the hash partitioning of the scans are a follow-up: #26424.

The scan gets the complete statistics of the body, not only the row count: column min, max, null count, distinct count and byte size. Thus the outer query uses the column statistics of the CTE. For example, a filter above the scan derives its bounds from the column bounds of the body, the same as in the inlined plan. The statistics of one partition are given only when the scan has one partition. Otherwise they are unknown, because scan partitions do not map to body partitions.

Comparison with PostgreSQL and DuckDB. This is what PostgreSQL and DuckDB do. PostgreSQL gives a CTE scan the row estimate of the CTE plan (allpaths.c). DuckDB copies the statistics of the body to each reference (propagate_cte.cpp).

Are these changes tested?

Yes. cte_materialized.slt: the join-order plan, the statistics in EXPLAIN, count(*) from exact statistics, a chain of two CTEs, a scalar subquery and a view. A unit test for the statistics of a scan.

Are there any user-facing changes?

Plans of queries with MATERIALIZED CTEs can change. New public functions: MaterializedCteScanExec::with_body_statistics and MaterializedCteExec::body.