feat: split Parquet files into row-group-sized morsels

[adriangb]

Which issue does this PR close?

• Closes #.

Rationale for this change

Follow-up to PR apache#21351 (Dynamic work scheduling in FileStream), which explicitly called out "Splitting files into smaller units (e.g. across row groups)" as a deferred next step. Today each Parquet file becomes exactly one morsel — a single ParquetPushDecoder over the entire pruned ParquetAccessPlan. That coarse granularity:

• keeps the follow-on "steal row-group work across sibling FileStreams" effort blocked — there's no row-group-sized unit for the SharedWorkSource to hand out;
• prevents pipelining row-group decode with downstream operator work inside a single partition;
• makes LIMIT and dynamic-filter early-stop wait for whole-file granularity.

The Morselizer / MorselPlanner abstraction already supports returning Vec<Box<dyn Morsel>> from a single MorselPlan, and ScanState already drains multi-morsel plans FIFO. This PR takes advantage of that: RowGroupsPrunedParquetOpen::build_stream now emits N streams, one per row-group chunk.

What changes are included in this PR?

• ParquetAccessPlan::split_into_chunks (new, pub(crate)). Packs consecutive surviving row groups into chunks bounded by a row budget (default 100k rows) and a compressed-byte budget (default 64 MiB). A single oversized row group still becomes its own chunk; no sub-row-group splitting is introduced in this PR. Skip entries are carried by the currently open chunk without forcing a boundary.
• RowGroupsPrunedParquetOpen::build_stream now returns Vec<BoxStream> instead of a single stream. Per chunk: prepare the access plan slice, build a fresh ParquetPushDecoder, mint a fresh AsyncFileReader via ParquetFileReaderFactory::create_reader (the first chunk reuses the reader that loaded metadata / page index / bloom filters so its warm cache state is preserved). Row filter is rebuilt per chunk because RowFilter is not Clone.
• ParquetOpenState::Ready now holds Vec<BoxStream>. ParquetMorselPlanner::plan maps each stream into a ParquetStreamMorsel. Empty Ready (file fully pruned) terminates the planner via Ok(None) instead of emitting an empty morsel plan.
• EarlyStoppingStream attaches to the first chunk only. FilePruner is !Clone and stateful; keeping it on chunk 0 preserves whole-file early-stop on dynamic-filter narrowing.
• Row-group reversal for sort pushdown is applied per-chunk via the existing PreparedAccessPlan::reverse, and the chunks Vec is reversed so the first emitted morsel corresponds to what was originally the file's last row groups.
• Chunk budgets are fields on ParquetMorselizer, defaulting to the new module-level constants DEFAULT_MORSEL_MAX_ROWS / DEFAULT_MORSEL_MAX_COMPRESSED_BYTES. Wiring through ParquetSource for user configuration is deliberately deferred to a follow-up — this PR keeps the public surface unchanged.

Are these changes tested?

Yes.

• Unit tests on ParquetAccessPlan::split_into_chunks cover: empty plan, all-skip, one-chunk-per-row-group when budget is tight, packing when budget allows, oversized single row group, byte-bounded splits, Skip preserved inside a chunk, Skip between chunks, Selection preserved verbatim in its chunk.
• Integration tests in opener.rs:
  • test_row_group_split_produces_multiple_morsels — 3 row groups × 3 rows with a row budget of 3 → 3 morsels; concatenated output matches the single-morsel reference.
  • test_row_group_split_packs_within_budget — budget of 6 rows packs 2 row groups into the first morsel, leaving the third in its own.
  • test_row_group_split_honors_user_skip — a user-supplied ParquetAccessPlan with Skip in the middle round-trips correctly.
  • test_row_group_split_with_reverse — reverse_row_groups=true + split budget emits morsels with the originally-last row group first.
• All 111 existing datasource-parquet library tests still pass (including the full test_reverse_scan_* suite). datafusion core lib (403) and core integration (931) also pass unchanged.

Are there any user-facing changes?

No observable semantic change — scans still produce the same rows in the same order. Runtime behavior changes: a multi-row-group Parquet scan now produces multiple morsels per file, each with its own ParquetPushDecoder + AsyncFileReader. For very small row groups this introduces per-morsel setup overhead that's amortized by the packing budget; for queries that hit LIMIT or dynamic-filter prunes mid-file, it generally lets the scan stop sooner.

🤖 Generated with Claude Code

[adriangb]

run benchmarks