apache / apache/arrow-rs

Allow forward skipping and bounded reads on a retained ParquetRecordBatchReader

Open
#10,655 9 comments 1 reaction 0 assignees View on GitHub
enhancement
Dominant language
Rust
Stars
3.6k
Forks
1.3k
Avg merge
2d 18h
Merged PRs (30d)
169

Description

### Is your feature request related to a problem or challenge?

ParquetRecordBatchReader has its batch size and optional RowSelection **fixed** when the reader is built.

This works for full scans and when all required rows are known in advance. It does not work well when row positions are discovered incrementally, for example from a secondary index or another operator producing monotonically increasing row numbers during the scan.

In that case, ideally , a consumer needs to:

1. Skip forward from the current reader position.
2. Read up to a requested number of rows.
3. Keep the existing decoder state for the next request.

We do have ways to do this across row groups , but we don't have a way to skip ahead within a row group.

Also skip forward's use case is easy to understand, but having adaptive batch size can help in consumers adaptively adjusting the batch size as they read through the parquet file. If we are iterating a secondary index and it has very sparse matches [ 0, 100k, 160k ] etc , then we can read very less number of rows by having a small batch size.

### Describe the solution you'd like

The internal ArrayReader supports these operations through skip_records, read_records, and consume_batch, but they are not exposed through ParquetRecordBatchReader.

## Proposed API

```
impl ParquetRecordBatchReader {

/// Skip up to `num_rows` rows in reader order from the current position.
///
/// Returns the number of rows actually skipped, which may be less than
/// `num_rows` at the end of the reader.
pub fn skip_rows(&mut self, num_rows: usize) -> Result;

/// Read up to `max_rows` output rows from the current position.
///
/// `max_rows` applies only to this call. A later call to `Iterator::next`
/// continues using the batch size configured on the builder.
pub fn read_next_batch(
&mut self,
max_rows: usize,
) -> Result>;

/// ..... existing functionality
}
```

### API usage

- `skip_rows` delegates to `ArrayReader::skip_records` . When an offset index is loaded, complete pages in the skipped range can be skipped without being fetched or decoded. Without an offset index, it retains
the existing `skip_records fallback` behavior.

- Row counts refer to top-level records, not leaf values. Skipping two rows of a list column skips two lists.
- `skip_rows(0)` returns 0.
- `read_next_batch(0)` returns an error.
- `skip_rows` initially returns an error for readers built with a RowSelection, because physical-row versus selected-row skipping is ambiguous.
- `read_next_batch` can use the existing selection cursor. In that case, `max_rows` counts selected output rows.
- Both methods advance the same reader state used by `Iterator::next.`

## Row numbers

The virtual RowNumber column gives consumers file row identifiers. This proposal deliberately does not add an absolute seek_row(row_number) API.

A reader may cover a subset or reordered list of row groups, making absolute positioning ambiguous. Relative skipping has well-defined semantics for every reader configuration. Callers using row numbers can calculate the forward delta when appropriate.

## Scope

This proposal covers the synchronous reader only. An equivalent API for ParquetRecordBatchStream can be considered separately once the API shape is settled.

## Proof of concept

An earlier proof of concept with tests for primitive columns, repeated columns, sized reads, and RowSelection rejection is available here:

https://github.com/bharath-techie/arrow-rs/commit/38b2f7a6a7f58d0dde9f932be1497444f7646a4c

The proof of concept implements simpler semantics: both methods reject readers with a RowSelection, and the batch size passed to read_next_batch persists for later calls. It also predates the current main branch and must be rebased. The API described above is the intended behavior.

Happy to submit a PR if there is agreement on the API direction.

### Describe alternatives you've considered

## Why can't we use RowSelection for this use case?

RowSelection should remain the preferred API when the complete set of required rows is known before constructing the reader.

The problem here is that a new selection cannot be applied to an existing reader.

```
RowSelection::from(vec![
RowSelector::skip(target),
RowSelector::select(count),
])
```

Rebuilding the reader for every request as given above is correct and can avoid reading unrelated pages when an offset index is loaded. But, it recreates the array readers, loses the current decoder position, must re-establish the target from the beginning of the reader, and may repeatedly decode the same partial page.

### Additional context

## Related

- #9968 — Reconfigure projection and filters at row-group boundaries
- #5343 — Random-access reads at row-group granularity
- #7299 — File row-number support in the Parquet reader

Contributor guide

Open the contributing guide

Research direction

Start by reviewing ParquetRecordBatchReader and the internal ArrayReader operations named in the proposal, then compare the proof of concept in commit 38b2f7a6a7f58d0dde9f932be1497444f7646a4c with the current main branch. Check the existing reader behavior and related issues #9968, #5343, and #7299 before settling the API semantics. Done means implementing and testing forward skipping and bounded reads, including zero values, end-of-reader behavior, nested columns, decoder state, and RowSelection handling.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.