[jayzhan211]

Which issue does this PR close?

• Closes Scans of tables with a declared ordering use count-based file groups, so uneven files make them up to 3x slower #26457.

Rationale for this change

A scan with a declared ordering keeps the count-based file groups from split_files, while an unordered scan is cut into size-balanced byte ranges. With uneven files, one partition of an ordered scan gets most of the data and every query over the table waits for it (see the issue: 233 ms ordered vs 93 ms unordered for the same full scan).

What changes are included in this PR?

FileScanConfig::repartitioned: when the scan has an output ordering, first ask the file source for the split it would use without an ordering (the size-balanced split, which keeps files in group order and only moves the group boundaries). Use it only if validated_output_ordering() is unchanged, that is, statistics prove every new group is still sorted. Otherwise fall back to the current order-preserving split.

• Groups from split_groups_by_statistics interleave value ranges, so the re-cut fails validation and they are kept as they are.
• Files without statistics cannot be proven sorted in a multi-file group, so they also keep today's behavior.
• File sources that cannot split files (compressed files, Arrow IPC stream) return None and keep today's behavior.

What is the testing strategy for this PR?

• repartition_balances_ordered_groups: groups [[8000 B, 8000 B], [1000 B, 1000 B]] become four 4500 B groups and keep the ordering.
• repartition_keeps_ordered_groups_when_rebalancing_breaks_order: overlapping files across the old boundary keep the old groups. Fails if the validation check is removed.
• repartition_falls_back_to_ordered_split_when_rebalancing_breaks_order: [[f1, f2], [f3]] with f3 overlapping both; the balanced split is rejected and the ordered split keeps [f1, f2] and splits f3.
• repartition_keeps_ordered_groups_when_rebalancing_misplaces_nulls: the balanced split would read a file with NULLs after another file; it is rejected.
• repartition_scan.slt: a new ordered_files table whose files are in value order gets size-balanced groups, one of them spanning both files, and returns the rows in order. The existing ordered plan is unchanged, because its statistics cannot prove the files are sorted.
• parquet_sorted_statistics.slt: two files with NULLs for NULLS FIRST and NULLS LAST, read in one partition. When the NULLs would be in the wrong place the plan keeps a sort, otherwise the ordering is used; both read the rows back.
• parallelization_multiple_files snapshot updated: at 8 partitions each group is still one range of one file, only the group order changes; the 3-partition case keeps its old groups because the balanced split would put two files without statistics in one group.

Benchmarks (release-nonlto, median of 5 interleaved runs, ms). 16 sorted files (4 x 8M rows, 12 x 0.5M), WITH ORDER (a ASC), target_partitions = 8:

Query	pushdown_filters	main	this PR
SELECT sum(a + b) FROM t WHERE b > 0.5	off / on	228 / 304	78 / 105
SELECT * FROM t WHERE b > 0.999999	off / on	31 / 29	12 / 11
SELECT a, count(*) FROM t WHERE b > 0.99 GROUP BY a ORDER BY a LIMIT 5	off / on	103 / 102	74 / 74
SELECT * FROM t WHERE b > 0.999999 LIMIT 10	off / on	11 / 3	8 / 4
SELECT * FROM t WHERE b > 0.999 ORDER BY a LIMIT 10	off / on	4 / 4	4 / 5
SELECT * FROM t ORDER BY a LIMIT 10	off / on	3 / 3	3 / 3

TPC-H SF10 lineitem (12 even files) declared WITH ORDER (l_orderkey), 5 queries x pushdown on/off, two interleaved rounds: all within noise (for example ORDER BY l_orderkey LIMIT 10: 19-20 vs 18 ms; WHERE l_shipdate > '1998-11-25' ORDER BY l_orderkey LIMIT 10: 35-37 vs 35-36 ms).

Worst case: when the old groups were already balanced, the new split moves the boundaries and can leave a small byte range of a file in a group (the new repartition_scan.slt plan has 1.parquet:516..521), as unordered scans already do.

Are there any user-facing changes?

EXPLAIN output for ordered file scans shows byte-range file groups more often. No API changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 95.83333% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.90%. Comparing base (290ad15) to head (d8d0eb3).

Files with missing lines	Patch %	Lines
datafusion/datasource/src/file_scan_config/mod.rs	95.83%	0 Missing and 3 partials ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #26458   +/-   ##
=======================================
  Coverage   82.89%   82.90%           
=======================================
  Files        1154     1154           
  Lines      457656   457728   +72     
  Branches   457656   457728   +72     
=======================================
+ Hits       379391   379464   +73     
+ Misses      55261    55259    -2     
- Partials    23004    23005    +1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[kosiew]

@jayzhan211,

Thanks for working on this. The size-balanced repartitioning approach addresses the skew in ordered scans, but I found a correctness issue with nullable sort keys that needs to be fixed before merging. I've also left two suggestions to improve the test coverage and documentation.

[kosiew]

The ordering check can accept an invalid merge when sort keys contain NULLs, because the min/max validation ignores null counts. For example, with ASC NULLS LAST, combining [1, NULL] and [2, 3] can place NULL before non-null values while still passing validation. Please add a null-aware check that falls back when ordering cannot be proven, including unknown null counts, without rejecting valid singleton groups. Also add regression tests for both NULLS FIRST and NULLS LAST, including a SQL test that reads rows from both files and verifies the results.

[jayzhan211]

This is already reachable on main without this PR: with [1, NULL] and [2, 3] in two files declared WITH ORDER (a ASC NULLS LAST) and target_partitions = 1, both files are read in one partition, the sort is removed, and the query returns 1, NULL, 2, 3. #26280 (for #26278) fixes it in MinMaxStatistics, which is the check this PR relies on: it puts NULLs into the file bounds when a file may contain them, treats unknown null counts as possibly NULL, and leaves single-file groups alone. So I'd rather build on that than add a second check here.

I checked this PR on top of #26280. Once it merges I'll rebase and add a unit test where the balanced split would put a file with NULLs after another file under NULLS FIRST (it is rejected and the ordered split is used), plus sqllogictest cases with two files for NULLS FIRST and NULLS LAST that read the rows back, with the NULLs at the allowed end (ordering kept) and at the wrong end (sort added).

[kosiew]

The nearby comment says EXPLAIN should never contain multi-file groups, but the updated plan intentionally includes one. Could you update it to explain that groups may span files as long as their concatenation preserves the declared ordering?

[jayzhan211]

While looking into this I found that the updated plan here was itself wrong. The table declares int over files that store bigint, and in that case the file statistics come back with NULL min/max values and no null count, so the check accepted [1.parquet (100, 200), 2.parquet (1..5)] as one sorted group. The results only looked right because 2.parquet:0..6 reads no rows. On top of #26280 that split is rejected and this plan goes back to what it is on main, so the comment is accurate again. I'll add a separate case whose files are in value order, which shows a group spanning two files and explains when that is allowed.

[kosiew]

The current rejection test only covers a case where both repartitioning strategies return None, so it doesn't verify that the ordered fallback can succeed. Could you add a mixed-group test using [[f1(8000, 0..9), f2(8000, 10..19)], [f3(1000, 5..15)]] with four target partitions, and verify that the balanced candidate is rejected while the ordered fallback returns Some with the first group intact, f3 split, and ordering preserved?

[jayzhan211]

Added repartition_falls_back_to_ordered_split_when_rebalancing_breaks_order with this layout and 4 target partitions. The balanced split is rejected, and the ordered split keeps [f1, f2] as the first group, splits f3 into the other three groups, and keeps the ordering.

[kosiew]

@jayzhan211,

Thanks for the follow-up. The size-balanced repartitioning approach is appropriate, and the additional tests cover the ordering correctness concerns raised in the previous review.

The architectural boundary is sound: ordered scans can adopt balanced file groups only when their validated output ordering is preserved. Otherwise, the existing ordering-aware splitter remains responsible for safe repartitioning. This improves workload distribution without weakening the scan's ordering contract.

The previous correctness blocker is resolved by the shared statistics fix in the rebased branch. The NULL-handling, statistics, and fallback regression coverage addresses the remaining concerns. I have no further blocking findings. Approved.

[adriangb]

Thanks folks!