feat: configurable Parquet I/O that preserves page pruning

[peterxcli]

Which issue does this PR close?

It narrows the I/O-policy work in apache#24393 and does not close an upstream issue. The broader filter-pushdown work in apache#20324 and apache#3463 remains open.

Rationale for this change

Parquet filter pushdown can introduce dependent reads: fetch predicate columns, evaluate the filter, then fetch output columns. progressive_io=false fetches the required predicate and output pages together, trading opportunities to avoid reads after row filtering for fewer dependent reader calls. Statistics and page-index pruning still run first.

What changes are included in this PR?

datafusion.execution.parquet.progressive_io is a session/table option, also set by ParquetSource::with_progressive_io. Its default is true, including for older serialized plans. For SELECT name FROM t WHERE age > 30, the two paths are:

                 +--------------------------------------+
                 | 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                   |
                 +--------------------------------------+

When progressive_io=false, the reader fetches the selected filter and output pages for the current row group together, before evaluating the row filter. It submits these ranges through get_byte_ranges when the decoder first requests data from that group. Without a usable page selection and offset index, it reads whole column chunks. The request also includes any ranges the decoder needs for its predicate cache; the decoder can request more data later. If every row is known to match, the reader skips row filtering and omits columns used only by the filter. A runtime filter that can still change cannot establish that guarantee.

with_row_group_prefetch(bytes, memory_pool) is a separate execution option, disabled by default and not serialized. Setting progressive_io=false alone does not start background I/O:

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.

Prefetch uses the same page ranges, preserves scan ordering, and retries speculative I/O errors on demand. It follows upstream's restriction that runtime row-group pruning is disabled while a page selection is active. Its memory budget covers additional compressed reservations; current-reader buffers, decoded buffers, and process memory are outside that metric.

The benchmark report compares implementation 801cb0e52 with its upstream base 408696966, using the same dependency lockfile, Parquet 59.3.0, input files, and two passes with revision order reversed.

On the 2 GiB synthetic files, reading filter and output columns together reduced reader API calls from 512 to 256 without indexes and from 513 to 257 with indexes, with identical requested bytes. With page indexes and matching rows grouped into pages, returning 7 columns used 9–14% less elapsed time than upstream reading filter columns first; returning 1 column used 15–21% less. These comparisons have prefetch disabled. Other workloads varied, including a reversal between passes for scattered matching rows and 7 output columns without indexes. The patch reading filter columns first was 3.4%/9.7% slower than upstream using that same mode on scattered matching rows, 7 output columns, and page indexes. This control warrants profiling.

The ClickBench runner now accepts --prefetch-bytes. All three reading modes were measured with prefetch enabled and disabled, using a 64 MiB budget so the projected row groups fit. Prefetch completed in every enabled execution. Its timing effect was mixed: Q22 used 3–6% less time with prefetch when filtering after decoding, and Q11 used 7–12% less when reading filter and output columns together. Other cases changed direction between passes. Both modes that filter during decoding remained slower than upstream filtering after decoding for all six queries in both passes, even with prefetch enabled.

The charts spell out each reading mode, prefetch setting, measurement pass, and upstream reference. These warm local-file samples do not establish full-dataset, remote-storage, or Spark performance; ClickBench uses a one-million-row file containing two row groups.

What is the testing strategy for this PR?

• Validation after the rebase covered 11,322 passing Rust tests (9 ignored) and all 513 SQLLogicTest files. Twelve metric expectations in three SQLLogicTest files were updated for the runtime-filter guard and those files were rerun; the remaining workspace test binaries also passed.
• Coverage includes 48 SQL policy/data combinations for page pruning, predicate-only columns, dictionaries, nested output and reversed row groups; exact-byte checks for fully matched groups; dynamic-filter cancellation; budget exhaustion; speculative-read failures; and serialization round trips with the older-plan default.
• cargo fmt --all, cargo clippy --all-targets --all-features -- -D warnings, dev/rust_lint.sh, and documentation formatting passed.
• The comparison passed 256 smoke scans and 512 full-size scans with independent row-count/checksum assertions. Input sizes, results, and requested bytes matched across revisions and policies in all 512 full-size scans. All 384 ClickBench executions returned ten rows; all 144 executions with prefetch enabled recorded one successful prefetch. Their result values were not independently compared.

Are there any user-facing changes?

The I/O policy can be configured through SQL before registering the Parquet table:

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

Reading filter and output columns together can fetch pages that row filtering would otherwise skip. Filter pushdown and prefetch retain their defaults. The ClickBench runner enables prefetch with --prefetch-bytes 67108864; its default of zero disables prefetch. The separate Comet POC remains pinned to its earlier implementation commit 8411b35ba; it has not been rebuilt against this rebase.

[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 [  76.287s] (current)
     Parsing datafusion v55.0.0 (current)
      Parsed [   0.042s] (current)
    Building datafusion v55.0.0 (baseline)
       Built [  73.960s] (baseline)
     Parsing datafusion v55.0.0 (baseline)
      Parsed [   0.044s] (baseline)
    Checking datafusion v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.640s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 152.763s] datafusion
    Building datafusion-common v55.0.0 (current)
       Built [  43.129s] (current)
     Parsing datafusion-common v55.0.0 (current)
      Parsed [   0.073s] (current)
    Building datafusion-common v55.0.0 (baseline)
       Built [  42.209s] (baseline)
     Parsing datafusion-common v55.0.0 (baseline)
      Parsed [   0.073s] (baseline)
    Checking datafusion-common v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.670s] 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 [  87.439s] datafusion-common
    Building datafusion-datasource-parquet v55.0.0 (current)
       Built [  61.331s] (current)
     Parsing datafusion-datasource-parquet v55.0.0 (current)
      Parsed [   0.044s] (current)
    Building datafusion-datasource-parquet v55.0.0 (baseline)
       Built [  61.747s] (baseline)
     Parsing datafusion-datasource-parquet v55.0.0 (baseline)
      Parsed [   0.042s] (baseline)
    Checking datafusion-datasource-parquet v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.154s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 124.627s] datafusion-datasource-parquet
    Building datafusion-proto v55.0.0 (current)
       Built [  66.553s] (current)
     Parsing datafusion-proto v55.0.0 (current)
      Parsed [   0.020s] (current)
    Building datafusion-proto v55.0.0 (baseline)
       Built [  66.197s] (baseline)
     Parsing datafusion-proto v55.0.0 (baseline)
      Parsed [   0.023s] (baseline)
    Checking datafusion-proto v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.121s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 134.330s] datafusion-proto
    Building datafusion-proto-common v55.0.0 (current)
       Built [  26.783s] (current)
     Parsing datafusion-proto-common v55.0.0 (current)
      Parsed [   0.055s] (current)
    Building datafusion-proto-common v55.0.0 (baseline)
       Built [  26.703s] (baseline)
     Parsing datafusion-proto-common v55.0.0 (baseline)
      Parsed [   0.058s] (baseline)
    Checking datafusion-proto-common v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   1.213s] 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 [  55.842s] datafusion-proto-common
    Building datafusion-proto-models v55.0.0 (current)
       Built [  31.193s] (current)
     Parsing datafusion-proto-models v55.0.0 (current)
      Parsed [   0.147s] (current)
    Building datafusion-proto-models v55.0.0 (baseline)
       Built [  30.400s] (baseline)
     Parsing datafusion-proto-models v55.0.0 (baseline)
      Parsed [   0.147s] (baseline)
    Checking datafusion-proto-models v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   1.741s] 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 [  64.849s] datafusion-proto-models
    Building datafusion-pruning v55.0.0 (current)
       Built [  48.083s] (current)
     Parsing datafusion-pruning v55.0.0 (current)
      Parsed [   0.019s] (current)
    Building datafusion-pruning v55.0.0 (baseline)
       Built [  48.194s] (baseline)
     Parsing datafusion-pruning v55.0.0 (baseline)
      Parsed [   0.020s] (baseline)
    Checking datafusion-pruning v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.082s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  97.603s] datafusion-pruning
    Building datafusion-sqllogictest v55.0.0 (current)
       Built [ 119.933s] (current)
     Parsing datafusion-sqllogictest v55.0.0 (current)
      Parsed [   0.032s] (current)
    Building datafusion-sqllogictest v55.0.0 (baseline)
       Built [ 121.927s] (baseline)
     Parsing datafusion-sqllogictest v55.0.0 (baseline)
      Parsed [   0.029s] (baseline)
    Checking datafusion-sqllogictest v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.105s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 245.490s] datafusion-sqllogictest