sunchao opened a new pull request, #11165:
URL: https://github.com/apache/arrow-rs/pull/11165
## Why are the changes needed?
A caller-driven Parquet reader needs to know **which bytes the decoder will
ask for next** before it can schedule a useful optional read. Knowing only the
next row-group index is not enough: projection, row selection, offset/limit,
and existing input buffers can all change that request. Reimplementing this
logic in an I/O scheduler risks drifting from the decoder. Advancing the
decoder just to discover a future request also changes ordered execution.
For example, consider a file with many columns where a scan projects only
column `a` and selects a small interval in the next row group. A useful preview
follows the existing planner's column/page decisions; it does not simply
request the whole row group. When the scan reaches its limit, later row groups
should disappear from the preview as well.
There is a related ownership gap during scan preparation. A custom
`AsyncFileReader` may retain metadata, cache entries, or a file-version token.
After moving it into `ParquetRecordBatchStreamBuilder` to read metadata or
Bloom filters, a caller that wants to drive a push decoder needs that **same
reader** back. Constructing a replacement can discard the state established
during preparation.
## What changes were proposed in this PR?
This adds two small, opt-in APIs for that workflow. Neither API starts reads
or introduces a scheduler.
`ParquetPushDecoder::preview_row_group_ranges(max_row_groups)` returns an
advisory snapshot containing the selected row-group indices and the ranges that
the demand planner would request against the current input buffers. It walks a
clone of the remaining row-group plan, applies the existing selection and
offset/limit accounting, and shares the projected-range construction with
ordinary demand. The decoder itself does not advance.
```mermaid
flowchart LR
A["Decoder at a row-group boundary"] --> B["Preview one or two groups"]
B --> C["Caller may schedule optional I/O"]
A --> D["Normal ordered decoding: NeedsData"]
C --> E["Reconcile with actual demand"]
D --> E
```
For a concrete buffering example, if the planner requests `[100, 200)` and
one input buffer already covers that whole range, the preview omits it. If only
`[100, 150)` is buffered, the preview still reports `[100, 200)`, exactly like
`NeedsData`; it does not invent a partial-range subtraction policy. Likewise,
the two-group snapshot does not predict how reading the first group will
consume buffers. Callers must reconcile a preview with later ordered demand
before using speculative results.
The initial API accepts a depth of one or two. It returns `None` while a row
group is active or row predicates are present, and an empty preview when no
selected work remains. It never evaluates a predicate, decodes a batch, or
reads data. Errors leave the decoder unchanged and should be deferred to normal
demand by speculative callers. Previewing does clone the remaining plan and
selections, so it is not a constant-time operation.
`ParquetRecordBatchStreamBuilder::into_inner()` consumes the builder and
returns its original reader without I/O. The caller can retain metadata
separately, recover the reader after metadata/Bloom-filter preparation, and
move it into its own I/O loop. Builder settings are discarded. The test checks
both metadata-only preparation and a real Bloom-filter read, including
reader-state identity and the resulting decoded batch.
## How was this PR tested?
Using Rust 1.98.1 with the `async` feature:
```text
cargo test --locked -p parquet --features async --lib arrow::push_decoder::
# 67 passed, including the eight new preview tests
cargo test --locked -p parquet --features async --lib
test_builder_into_inner_preserves_reader_after_bloom_filter
# 1 passed
cargo fmt --all -- --check
git diff --check
# both passed
```
The preview tests compare actual `NeedsData` requests and decoded batches,
not just expected row-group indices. They cover projection, global and
per-row-group selections (including bitmap/RLE, reversed order, empty
selections, and duplicates), offset/limit, full and partial buffered ranges,
unsupported predicate/active states, and a preview error that must not consume
an earlier valid group.
**Local dependency qualification:** the configured registry did not provide
the upstream lockfile's `opendal 0.59.2`. These focused tests used a temporary
local resolution with `opendal`/`opendal-core 0.59.1`, `asyncband 0.6.7`, and
its resolved transitive graph. No manifest changes were made, and the exact
upstream `Cargo.lock` was restored before committing. The PR contains no
dependency changes; CI coverage with the committed lockfile is still pending.
No full local suite was run. This PR makes no standalone query-time,
throughput, or memory-saving claim; scheduling policy and performance
measurement remain the caller's responsibility.
## Which issue does this PR close?
Closes #11164.
## Are there any user-facing changes?
Additive public APIs: `RowGroupRangePreview`,
`ParquetPushDecoder::preview_row_group_ranges`, and
`ParquetRecordBatchStreamBuilder::into_inner`. Existing decoding and I/O
behavior are unchanged when the APIs are not used. Their boundaries,
invalidation rules, and ownership behavior are documented in the API comments.
No new dependency or configuration default is introduced.
## AI assistance
OpenAI Codex assisted with porting the implementation, writing focused
tests, reviewing the change, and drafting this description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
