feat: TopK stats init + cumulative RG pruning for pure-TopK parquet scans (no-WHERE)

[zhuqi-lucas]

Which issue does this PR close?

Carved out of #21580 — after #21956 merged the file/RG reorder + reverse machinery, this PR ships the two remaining pieces of the statistics-driven TopK pipeline, restricted to the simplest scope (no WHERE).

Rationale for this change

For ORDER BY <col> LIMIT N queries on parquet:

1. TopK's dynamic filter starts as lit(true) — the first row groups read are never pruned by it.
2. After file/RG reorder normalises data to ASC-by-min (and optional reverse for DESC), the first few row groups already contain the rows that will end up in the heap. The rest can be pruned upfront from parquet min/max statistics if the threshold is known before scanning.

This PR initialises the threshold from per-RG stats before the PruningPredicate is built, then accumulates row counts from the front of the reordered RG list and truncates once $\geq K$. The net effect on a sorted, non-overlapping dataset is that only one row group is decoded for a LIMIT N query — observed 17×–60× speedups in #21580's benchmark.

Design — using existing sort-pushdown plumbing

Both optimisations need to know K (the LIMIT) and the sort direction/column. Rather than touching DynamicFilterPhysicalExpr, this PR follows the pattern established by #21956:

• Sort direction + column — read from ParquetSource::sort_order_for_reorder (already set by try_pushdown_sort and merged).
• K — new topk_fetch: Option<usize> field on ParquetSource, plumbed by PushdownSort rule via a new FileSource::with_topk_fetch_hint(fetch) trait method (default None, only overridden on ParquetSource).

DynamicFilterPhysicalExpr is not modified.

What changes are included in this PR?

Six small focused commits.

1. feat: add topk_fetch field on ParquetSource

New optional field next to sort_order_for_reorder / reverse_row_groups. Not surfaced in EXPLAIN (the SortExec.fetch above is the same value).

2. feat: add FileSource::with_topk_fetch_hint trait method

Adds the trait method with a None default; ParquetSource overrides it to clone-and-set topk_fetch. Other file source impls don't need to change.

3. feat: plumb SortExec.fetch via with_topk_fetch_hint in PushdownSort

Mirrors try_pushdown_sort's plumbing chain — with_topk_fetch_hint exposed on ExecutionPlan, DataSource, DataSourceExec, FileScanConfig, FileSource. Both Inexact branches in PushdownSort (the SPM+Sort pattern and the standalone Sort) now call inner.with_topk_fetch_hint(fetch) when the SortExec has a fetch.

4. feat: add PreparedAccessPlan::truncate_row_groups

Small pub(crate) helper that keeps the first count entries of row_group_indexes. Bails unchanged if row_selection is set (page-level state would be hard to remap). Three unit tests.

5. feat: TopK stats init + cumulative RG pruning in opener

Two opener-side optimisations, both dual-gated on predicate_is_pure_dynamic_filter(pred) AND sort_order_for_reorder.is_some() AND topk_fetch.is_some().

Stats init (try_init_topk_threshold, runs in prepare_filters before PruningPredicate build):

• DESC LIMIT K: T_min = max(min(col)) across RGs with num_rows >= K → filter becomes col >= T_min.
• ASC LIMIT K: T_max = min(max(col)) across RGs with num_rows >= K → filter becomes col <= T_max.
• >= / <= (not > / <) — T from stats is a safe bound, not strict; the real TopK will tighten it on its first heap update.
• Sort direction + column come from sort_order_for_reorder, not from the dynamic filter.
• Skips silently on: no qualifying RGs, sort expression not a simple column (unwraps single-child wrappers like CastExpr), column outside file schema, stats unavailable, type cast failure.

Cumulative RG prune (inside prepare_access_plan, after reorder_by_statistics and reverse):

• Accumulate num_rows from the front of row_group_indexes until the running total reaches K, then truncate_row_groups.

Helpers: find_dynamic_filter, find_column_in_expr (unwraps CastExpr etc.), compute_best_threshold, predicate_is_pure_dynamic_filter.

6. test: add Test I SLT

Six sub-tests in sort_pushdown.slt:

• I.1: DESC LIMIT — EXPLAIN shows DynamicFilter [ empty ] + sort_order_for_reorder=[id@0 DESC] + reverse_row_groups=true
• I.2: DESC LIMIT result correctness
• I.3: ASC LIMIT in file order — Exact path, SortExec eliminated, LIMIT becomes a static fetch
• I.4: DESC LIMIT with WHERE — both optimisations skip, result still correct via the regular dynamic-filter path
• I.5: Larger LIMIT spanning multiple RGs
• I.6: LIMIT > total rows — returns all rows

Scope

Intentionally narrow:

• No WHERE only. Cumulative prune with WHERE needs filter selectivity reasoning to avoid under-returning rows; out of scope for this PR.
• Sort pushdown path only. Non-sort-pushdown TopK on parquet is a future PR on top of this one (feat: statistics-driven TopK optimization for parquet (file reorder + RG reorder + threshold init + cumulative prune) #21580 went on to do this via DynamicFilterPhysicalExpr.sort_options, but it's a separable concern).
• Single-column leading sort (matches the existing sort_order_for_reorder plain-Column constraint).

Are these changes tested?

• 125 datafusion-datasource-parquet lib tests pass (including 3 new tests for truncate_row_groups).
• sort_pushdown SLT passes (existing tests unchanged, new Test I added).
• cargo clippy --all-targets -- -D warnings clean on touched crates.

Are there any user-facing changes?

No. Transparent optimisation — same query results, faster TopK on sorted parquet when no WHERE clause is present.