perf: prune ListingTable files before grouping - #26441

Heads up: this is an AI-assisted review pass. I checked each point below locally against bebaf2e, and against current main with your catalog-listing changes applied.

Thanks for working on this, @rgbuilds. To answer your question: yes, pruning before split_files is what I had in mind in #26130. On the issue's example the scan now gets [[f2.parquet], [f3.parquet]] and Rows=Inexact(20), instead of 8 files in 4 groups and Rows=Inexact(80). Reusing the statistics that were already collected, keeping files with unknown bounds, and leaving declared output_partitioning alone all match the issue. For Parquet, the planner only drops files that the opener's FilePruner would already skip at open time using the same statistics. That keeps the risk to single-scan results low.

One request before I approve: keep partition-value groups intact

When datafusion.optimizer.preserve_file_partitions is enabled, list_files_for_regular_scan prunes before group_by_partition_values. If the predicate removes every file of one Hive value on one side of a join, that side's groups stop lining up with the other side's.

Two Hive tables, each with values A–D and one file per value. Only dim's d_dkey=B file lacks service = 'log', and only fact's f_dkey=B file has value = 20:

SET datafusion.execution.target_partitions = 3;
SET datafusion.optimizer.preserve_file_partitions = 1;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;

-- fact/f_dkey={A,B,C,D}/data.parquet hold value 10, 20, 30, 40
-- dim/d_dkey={A,B,C,D}/data.parquet hold service 'log', 'trace', 'log', 'log'
CREATE EXTERNAL TABLE fact (value BIGINT) STORED AS PARQUET PARTITIONED BY (f_dkey STRING) LOCATION '.../fact/';
CREATE EXTERNAL TABLE dim (service STRING) STORED AS PARQUET PARTITIONED BY (d_dkey STRING) LOCATION '.../dim/';

-- Q1 prunes dim's B file; Q2 prunes fact's B file
SELECT f.f_dkey, f.value FROM fact f JOIN dim d ON f.f_dkey = d.d_dkey WHERE d.service = 'log' ORDER BY f.f_dkey;
SELECT f.f_dkey, f.value FROM fact f JOIN dim d ON f.f_dkey = d.d_dkey WHERE f.value <> 20 ORDER BY f.f_dkey;

Both queries should return A, C and D.

• On this PR's base (5f21582), both return only A. The pruned side is grouped as [A], [C], [D] while the other side stays [A, D], [B], [C]. Both sides still declare Hash([key], 3), so the partitioned join runs without repartitioning. This is the mismatch from #23436, now reachable through a data predicate.
• After rebasing onto main, the results are correct but the plan gets an extra exchange. #25279 (merged 2026-10-22) declares these scans as Range with split points. Pruning changes the pruned side's split points, so the planner adds RepartitionExec: partitioning=Range([f_dkey@1 ASC], [(B), (C)], 3) on that side. In Q2 that means repartitioning the fact table. main needs no exchange there: it keeps the B file in its group, and the opener skips it without reading data. In Ballista that is an extra shuffle stage.

I think the value-grouped path should follow the same rule you already apply to declared layouts. Build the groups and split points from all listed files, then prune inside each group and keep the empty groups. I tried that on main with your patch applied (diff below). Both queries return A, C and D, there is no RepartitionExec, and the pruned side becomes [[A], [], [C, D]]. The full SLT suite gives the same results as your patch.

@@ -1072,7 +1072,6 @@ impl ListingTable {
         let (file_group, inexact_stats) = self
             .collect_files_for_scan(ctx, &store, filters, limit)
             .await?;
-        let file_group = self.prune_files_by_statistics(ctx, data_filters, file_group);
 
         // Threshold: 0 = disabled, N > 0 = enabled when distinct_keys >= N
         //
@@ -1086,17 +1085,26 @@ impl ListingTable {
                 let (grouped, split_points) = file_group
                     .group_by_partition_values_with_split_points(file_group_count);
                 if grouped.len() >= threshold {
+                    // Prune inside each value group so the split points (and the
+                    // declared Range partitioning) do not change.
+                    let grouped = grouped
+                        .into_iter()
+                        .map(|g| self.prune_files_by_statistics(ctx, data_filters, g))
+                        .collect();
                     (grouped, true, split_points)
                 } else {
                     let all_files: Vec<_> =
                         grouped.into_iter().flat_map(|g| g.into_inner()).collect();
-                    (
-                        FileGroup::new(all_files).split_files(file_group_count),
-                        false,
-                        vec![],
-                    )
+                    let file_group = self.prune_files_by_statistics(
+                        ctx,
+                        data_filters,
+                        FileGroup::new(all_files),
+                    );
+                    (file_group.split_files(file_group_count), false, vec![])
                 }
             } else {
+                let file_group =
+                    self.prune_files_by_statistics(ctx, data_filters, file_group);
                 (file_group.split_files(file_group_count), false, vec![])
             };

The prototype rebuilds the PruningPredicate for every group, so a real version should build it once and reuse it. A test in preserve_file_partitioning.slt would lock this in: a partitioned join plus a data filter that prunes one value on one side, checking both the rows and that no RepartitionExec is added.

Rebase

The PR now conflicts with main in repartition_subset_satisfaction.slt, and #25279 rewrote the grouping code in list_files_for_regular_scan that the request above touches. With your catalog-listing changes applied to current main, the full SLT suite fails only on EXPLAIN blocks in the 9 files you already update. No query result changes.

Smaller things

• prunes_before_grouping_and_aggregates_survivor_statistics still has three debug println!s ("AFTER raw scan", ...).
• file_statistics_pruning.rs and the new unit tests in table.rs have no blank lines between functions.
• In clickbench.slt, 8 of the 9 updated EXPLAIN blocks now plan FilterExec over EmptyExec, and the COUNT(*) ... WHERE "AdvEngineID" <> 0 block becomes 0 as count(*) over PlaceholderRowExec. Those blocks no longer show the scan and aggregate shape these queries get on the real data. If others want to keep that coverage, running those blocks with collect_statistics = false would do it.

Follow-ups (non-blocking)

• Files dropped at planning no longer show up in files_ranges_pruned_statistics. On the issue's example, EXPLAIN ANALYZE goes from 8 total → 2 matched to 2 total → 2 matched, and nothing records the 6 skipped files. A count on the DataSourceExec line, or a planning metric, would keep that visible.
• prune_collected_files builds one PrunableStatistics per file. That is needed because PrunableStatistics fills non-exact bounds with an untyped ScalarValue::Null, so iter_to_array fails for the whole column once files are mixed. The cost is small. With 10k files and a predicate that prunes nothing, planning goes from 0.276 s to 0.296 s (median of 10 runs, three rounds), about 2 µs per file. Typed nulls taken from the schema field in PrunableStatistics would allow a single prune call over all files, and would fix the same problem for other callers that pass more than one container.

What I checked

• With 10k files of which 10 can match, planning is no slower (0.34–0.42 s vs 0.37–0.54 s on a loaded machine), and the query drops from about 0.45 s to 0.30 s.
• A volatile conjunct is skipped while the rest still prunes. An OR with a volatile branch turns pruning off. An uncorrelated scalar-subquery conjunct doesn't block it.
• The full SLT suite passes on main + this PR apart from the EXPLAIN updates above. CI is green on bebaf2e.

@rgbuilds thanks for this! Approving. I checked that planning-time pruning gives the same results as the Parquet opener on schema-evolution, missing-column and partition-column predicates. The three inline comments are optional.