perf: move stats init before RG pruning so first file also benefits

Move try_init_topk_threshold() from build_stream() to
prune_row_groups(), before prune_by_statistics(). This way:

- File 1: stats init sets threshold from ALL its RG statistics,
  then prune_by_statistics uses it to prune file 1's own RGs.
  Only the best RG(s) are read, rest skipped with zero I/O.

- File 2+: dynamic filter already has tight threshold from file 1,
  most RGs pruned immediately.

This effectively achieves dynamic RG pruning without needing morsel-
level scheduling — the threshold is computed from statistics (no data
read), then used to prune RGs in the same file.

Author: zhuqi-lucas

datafusion/datasource-parquet/src/opener.rs

@@ -905,6 +905,22 @@ impl FiltersPreparedParquetOpen {
             row_groups.prune_by_range(rg_metadata, range);
         }
 
+        // Initialize TopK dynamic filter threshold from row group statistics
+        // BEFORE row group pruning, so that this file's own RGs can be pruned
+        // by the threshold. For the first file, this sets the initial threshold;
+        // for subsequent files, the threshold is already set by earlier files.
+        if let (Some(predicate), Some(limit)) = (&prepared.predicate, prepared.limit)
+            && let Err(e) = try_init_topk_threshold(
+                predicate,
+                limit,
+                rg_metadata,
+                &prepared.physical_file_schema,
+                loaded.reader_metadata.parquet_schema(),
+            )
+        {
+            debug!("Skipping TopK threshold initialization from statistics: {e}");
+        }
+
         // If there is a predicate that can be evaluated against the metadata
         if let Some(predicate) = self.pruning_predicate.as_ref().map(|p| p.as_ref()) {
             if prepared.enable_row_group_stats_pruning {
@@ -1087,23 +1103,6 @@ impl RowGroupsPrunedParquetOpen {
         let file_metadata = Arc::clone(reader_metadata.metadata());
         let rg_metadata = file_metadata.row_groups();
 
-        // Initialize TopK dynamic filter from row group statistics.
-        // This sets an initial threshold before any data is read, so that
-        // subsequent row filtering can benefit immediately.
-        // Sort direction is read from the DynamicFilterPhysicalExpr's
-        // sort_options (set by SortExec for TopK queries).
-        if let (Some(predicate), Some(limit)) = (&prepared.predicate, prepared.limit)
-            && let Err(e) = try_init_topk_threshold(
-                predicate,
-                limit,
-                rg_metadata,
-                &prepared.physical_file_schema,
-                reader_metadata.parquet_schema(),
-            )
-        {
-            debug!("Skipping TopK threshold initialization from statistics: {e}");
-        }
-
         // Filter pushdown: evaluate predicates during scan
         let row_filter = if let Some(predicate) = prepared
             .pushdown_filters