apache / apache/arrow-rs

parquet: Support reverse page iteration for reverse streaming with filter pushdown

Open
#9,934 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Rust
Stars
3.6k
Forks
1.3k
Avg merge
2d 14h
Merged PRs (30d)
167

Description

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

> **Note**: This is a refined / scoped follow-up to #3922. To avoid the
> historical objections in that thread, this issue **explicitly does NOT
> request reverse-decoding of rows within a page** (which is indeed
> impractical due to RLE / dictionary / delta / bit-packing being forward
> streams). Instead, it requests **page-level reverse iteration**:
> traversing pages in reverse order while still decoding each page
> forward. See "Scope" below.

DataFusion has merged row-group-level reverse scan support
(apache/datafusion#18817) to optimize queries like
`ORDER BY sorted_col DESC LIMIT N` on ascending-sorted Parquet files.
The current implementation works at **row-group granularity** (~128MB),
which has two limitations:

1. **High first-batch latency**: must decode an entire row group
(~1M rows) before emitting the first reversed batch.
2. **High peak memory**: must buffer one full row group to reverse it.

For the **no-filter** case, `RowSelection::skip(rg_size - N).select(N)`
is sufficient, as already discussed in #3922. **However, this does not
work when there is a filter predicate.** Consider:

```sql
SELECT * FROM events
WHERE user_id = 42
ORDER BY ts DESC
LIMIT 10
```

We do not know which rows match `user_id = 42`, so we cannot
pre-compute a `RowSelection` for "the last 10 matches". The only
correct approach is **streaming pages in reverse order, decoding +
filtering each page, until 10 matches are accumulated**.

This is the killer use case for page-level reverse iteration.

## Scope (what this issue is and is NOT)

**This issue requests** (achievable today via `OffsetIndex`):

- Iterate `OffsetIndex.page_locations` **in reverse order**
- For each page, **decode forward as usual** (using existing
`SerializedPageReader` machinery)
- Reverse the resulting `RecordBatch` (or decoded values) at the page
boundary using existing kernels (`take` etc.)
- Apply `RowFilter` on each reversed page; emit and accumulate against
`LIMIT`

**This issue does NOT request**:

- Reverse-decoding rows *within* a single page (impossible due to
forward-only encodings — agreed with the analysis in #3922)
- Reordering row groups (already handled at the DataFusion layer)
- Any change to the on-disk Parquet format

## Why this is feasible

Page-level reverse iteration only needs random access to pages, which
the **`OffsetIndex` already provides** for any Parquet file written with
page indexes (the default for most modern writers, including arrow-rs).

`OffsetIndexMetaData::page_locations()` exposes
`Vec`,
which is sufficient to seek directly to any page in any order.

The dictionary page is at the start of the column chunk and must be
read once before any data page — this is a known one-time cost, not a
blocker.

## Describe the solution you'd like

Two possible API surfaces (open to discussion):

### Option A: High-level builder flag

```rust
let reader = ParquetRecordBatchReaderBuilder::try_new(file)?
.with_row_filter(filter)
.with_limit(10)
.with_reverse_pages(true) // <-- new
.build()?;
```

The builder reverse-traverses pages within each row group, applies
the filter, and emits `RecordBatch`es with rows already in reverse
order. `with_limit` causes early termination once N matches are
collected.

### Option B: Low-level `PageReader` variant

```rust
pub struct ReverseSerializedPageReader { ... }

impl ReverseSerializedPageReader {
pub fn new(
reader: Arc,
meta: &ColumnChunkMetaData,
offset_index: &OffsetIndexMetaData,
) -> Result;
}

impl PageReader for ReverseSerializedPageReader { ... }
```

Option B is more flexible but pushes the orchestration burden onto
callers (DataFusion, etc.). My current preference is to start with B as
the foundation and layer A on top.

## Performance expectation

Compared to the current row-group-level reverse in DataFusion, on
`SELECT * FROM t WHERE rare_predicate ORDER BY ts DESC LIMIT 10`:

| Metric | Row-group reverse | Page reverse |
|---|---|---|
| First-batch latency | decode ~1M rows | decode ~1 page (~10K rows) |
| Peak buffer memory | ~128MB (1 row group) | ~1MB (1 page) |
| Wasted decode work (filter selectivity 1%) | ~1M rows for 10 matches | scan from tail until 10 matches |

Concrete benchmarks will follow once a POC is in place.

## Describe alternatives I've considered

1. **`RowSelection::skip(N).select(M)`** — only works without filter.
Does not solve the filter+reverse case.
2. **Row-group-level reverse** — already done at DataFusion layer
(apache/datafusion#18817). Works but has the latency / memory
limitations described above.
3. **Read entire file, sort in memory** — current default. Defeats the
purpose of having sorted Parquet files.

## Additional context

- Related: #3922 (broader reverse-stream request)
- DataFusion umbrella: apache/datafusion#17172
- DataFusion follow-up: apache/datafusion#19486
- DataFusion row-group reverse PR : apache/datafusion#18817

I'm happy to drive the implementation if there is consensus on
direction.

Contributor guide

Open the contributing guide

Research direction

Start by tracing OffsetIndexMetaData::page_locations(), SerializedPageReader, and the PageReader interface to understand how pages are accessed and dictionary pages are handled. Then review ParquetRecordBatchReaderBuilder and RowFilter to determine where reverse page traversal and page-boundary reversal would fit. Done means a scoped API and implementation that iterates indexed pages in reverse, decodes each page forward, applies filtering, and supports early termination for a limit.

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
Quiet
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.