feat: configurable Parquet I/O that preserves page pruning

[peterxcli]

Which issue does this PR close?

Related to apache#24393. This fork draft provides the I/O policy and an optional prefetch path; upstream submission and the default-policy decision remain open.

Rationale for this change

Parquet filter pushdown can add dependent reads for predicate and output columns. Reading the required pages together preserves filter pushdown without requiring that progressive I/O pattern. Optional next-row-group prefetch overlaps reads with decoding.

What changes are included in this PR?

datafusion.execution.parquet.progressive_io=false fetches required predicate and output pages together after statistics and page-index pruning. The default remains true, including when decoding older serialized plans. with_row_group_prefetch(bytes, memory_pool) enables at most one future row group per stream, with budget checks, cancellation, and demand-read fallback.

Whole-query profiling found that a spawned prefetch could remain in the Tokio worker's local queue while its consumer continued decoding. The decoder now yields after scheduling the read. On chia-ping-aws1, single-partition local-file I/O CPU overlap rose from 0.03% to 99.6%, and all-worker parked time fell from 426 ms to 1.9 ms. Untraced wide scans used 9–16% less time with one partition; short indexed queries remain mixed.

The full partitioned ClickBench run on chia-ping-aws1 covers all 100 files (99,997,497 rows) and all 43 queries. Before/after scheduling-fix timings improve by 0.06% and 2.04% in the two passes; 11 queries improve in both, 7 slow in both, and 25 change direction. Separate profiles show 5.94% less summed worker-blocked time. The full report and 43 query charts include all 1,032 executions, spill and VM-steal caveats, and result differences from unordered or tied LIMIT queries. This compares the previous patch with the scheduling fix; both sides already use upfront I/O and prefetch.

The Linux profiling report includes before/after CPU timelines, 384 checked executions, both timing passes, reproduction commands, and VM CPU-steal measurements. The Mac profiling report provides an independent run; the earlier benchmark compares the original patch with upstream main. These are warm local-file measurements; remote storage and Spark remain unmeasured.

I/O policy and prefetch flow

For SELECT name FROM t WHERE age > 30:

                 +--------------------------------------+
                 | Footer statistics and page indexes   |
                 | prune row groups and select pages    |
                 +-------------------+------------------+
                                     |
                         first demand for row group N
                                     |
                +--------------------+--------------------+
                |                                         |
                v                                         v
  +---------------------------+           +---------------------------+
  | progressive_io = true     |           | progressive_io = false    |
  | Read required age pages   |           | Plan age + name ranges    |
  +-------------+-------------+           | for selected pages        |
                |                         | Include dictionary pages  |
                v                         +-------------+-------------+
  +---------------------------+                         |
  | Evaluate age > 30         |                         v
  | Keep matching row numbers |           +---------------------------+
  +-------------+-------------+           | Merge overlapping/adjacent|
                |                         | ranges; keep pruned gaps  |
                v                         +-------------+-------------+
  +---------------------------+                         |
  | Read name pages needed    |                         v
  | for those matching rows   |           +---------------------------+
  +-------------+-------------+           | Fetch ranges together     |
                |                         | Buffer compressed bytes   |
                |                         +-------------+-------------+
                |                                       |
                |                                       v
                |                         +---------------------------+
                |                         | Evaluate age > 30         |
                |                         | Keep matching row numbers |
                |                         +-------------+-------------+
                |                                       |
                +--------------------+------------------+
                                     v
                 +--------------------------------------+
                 | Decode/project name for matching rows|
                 | Emit Arrow batches                   |
                 +--------------------------------------+

Prefetch is a separate option:

time --------------------------------------------------------------->

scan       fetch N | decode batches from N | use buffered N+1 | decode
                   |                      ^
prefetch           +-- reserve budget ----|
                       read selected      |
                       pages from N+1 ----+

                   At most one future row group is in flight.
                   Pruned group -> cancel read, discard its bytes.
                   Budget unavailable -> continue with demand reads.

What is the testing strategy for this PR?

The new regression reproduces a consumer keeping a runtime worker busy while a read is queued; it fails before the fix and passes afterward. This follow-up passes 259 Parquet library tests, the SQL page-pruning regression, formatting, workspace Clippy, and dev/rust_lint.sh. All 384 synthetic benchmark executions return the independently expected row count and checksum, with identical paired requested-byte and prefetch counters. In full ClickBench, 35 queries return matching result multisets; eight have unordered or tied LIMIT results, documented separately in the report.

Existing coverage includes page pruning, predicate-only columns, dictionaries, nested output, dynamic-filter cancellation, memory pressure, speculative I/O failures, and serialization defaults.

Are there any user-facing changes?

Configure the I/O policy before registering the table:

SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.execution.parquet.progressive_io = false;

Upfront reads can fetch pages that later row filtering would skip. Prefetch stays disabled by default and is not serialized; the ClickBench runner accepts --prefetch-bytes. The Comet POC remains pinned to its earlier DataFusion implementation.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion v55.0.0 (current)
       Built [  51.304s] (current)
     Parsing datafusion v55.0.0 (current)
      Parsed [   0.030s] (current)
    Building datafusion v55.0.0 (baseline)
       Built [  46.327s] (baseline)
     Parsing datafusion v55.0.0 (baseline)
      Parsed [   0.030s] (baseline)
    Checking datafusion v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.737s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 100.200s] datafusion
    Building datafusion-common v55.0.0 (current)
       Built [  26.896s] (current)
     Parsing datafusion-common v55.0.0 (current)
      Parsed [   0.052s] (current)
    Building datafusion-common v55.0.0 (baseline)
       Built [  27.053s] (baseline)
     Parsing datafusion-common v55.0.0 (baseline)
      Parsed [   0.054s] (baseline)
    Checking datafusion-common v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.754s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ParquetOptions.progressive_io in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:1301

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  56.102s] datafusion-common
    Building datafusion-datasource-parquet v55.0.0 (current)
       Built [  40.416s] (current)
     Parsing datafusion-datasource-parquet v55.0.0 (current)
      Parsed [   0.029s] (current)
    Building datafusion-datasource-parquet v55.0.0 (baseline)
       Built [  40.242s] (baseline)
     Parsing datafusion-datasource-parquet v55.0.0 (baseline)
      Parsed [   0.029s] (baseline)
    Checking datafusion-datasource-parquet v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.153s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  81.940s] datafusion-datasource-parquet
    Building datafusion-proto v55.0.0 (current)
       Built [  43.886s] (current)
     Parsing datafusion-proto v55.0.0 (current)
      Parsed [   0.015s] (current)
    Building datafusion-proto v55.0.0 (baseline)
       Built [  44.621s] (baseline)
     Parsing datafusion-proto v55.0.0 (baseline)
      Parsed [   0.016s] (baseline)
    Checking datafusion-proto v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.116s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  89.815s] datafusion-proto
    Building datafusion-proto-common v55.0.0 (current)
       Built [  18.008s] (current)
     Parsing datafusion-proto-common v55.0.0 (current)
      Parsed [   0.041s] (current)
    Building datafusion-proto-common v55.0.0 (baseline)
       Built [  17.842s] (baseline)
     Parsing datafusion-proto-common v55.0.0 (baseline)
      Parsed [   0.041s] (baseline)
    Checking datafusion-proto-common v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   1.188s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ParquetOptions.progressive_io in /home/runner/work/datafusion/datafusion/datafusion/proto-common/src/generated/prost.rs:822
  field ParquetOptions.progressive_io in /home/runner/work/datafusion/datafusion/datafusion/proto-common/src/generated/prost.rs:822
  field ParquetOptions.progressive_io in /home/runner/work/datafusion/datafusion/datafusion/proto-common/src/generated/prost.rs:822

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  38.434s] datafusion-proto-common
    Building datafusion-proto-models v55.0.0 (current)
       Built [  20.312s] (current)
     Parsing datafusion-proto-models v55.0.0 (current)
      Parsed [   0.110s] (current)
    Building datafusion-proto-models v55.0.0 (baseline)
       Built [  20.457s] (baseline)
     Parsing datafusion-proto-models v55.0.0 (baseline)
      Parsed [   0.112s] (baseline)
    Checking datafusion-proto-models v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   1.981s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ParquetOptions.progressive_io in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/datafusion_proto_common.rs:822
  field ParquetOptions.progressive_io in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/datafusion_proto_common.rs:822

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  44.033s] datafusion-proto-models
    Building datafusion-pruning v55.0.0 (current)
       Built [  33.128s] (current)
     Parsing datafusion-pruning v55.0.0 (current)
      Parsed [   0.013s] (current)
    Building datafusion-pruning v55.0.0 (baseline)
       Built [  33.355s] (baseline)
     Parsing datafusion-pruning v55.0.0 (baseline)
      Parsed [   0.014s] (baseline)
    Checking datafusion-pruning v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.078s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  67.604s] datafusion-pruning
    Building datafusion-sqllogictest v55.0.0 (current)
       Built [  83.166s] (current)
     Parsing datafusion-sqllogictest v55.0.0 (current)
      Parsed [   0.018s] (current)
    Building datafusion-sqllogictest v55.0.0 (baseline)
       Built [  83.866s] (baseline)
     Parsing datafusion-sqllogictest v55.0.0 (baseline)
      Parsed [   0.021s] (baseline)
    Checking datafusion-sqllogictest v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.096s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 169.632s] datafusion-sqllogictest