apache / apache/iceberg

Parquet: Page Index pruning in the custom Parquet reader

Open
#17,596 0 comments 1 reaction 0 assignees View on GitHub
improvement
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 11h
Merged PRs (30d)
132

Description

### Feature Request / Improvement

## Motivation

Iceberg's custom row-based Parquet reader currently performs pruning at row-group granularity but does not use Parquet ColumnIndex/OffsetIndex metadata to skip individual pages within retained row groups.

This can result in substantially more I/O when a selective predicate matches only a small portion of a large row group, particularly when data is physically clustered by the predicate column.

## Current POC

I built a POC against current `main` using parquet-java 1.17.1.

The POC integrates Parquet Page Index pruning into Iceberg's custom row-based reader using:

```text
Iceberg Expression
-> Parquet predicate
-> ColumnIndex filtering
-> readFilteredRowGroup(rowGroupIndex)
-> filtered PageReadStore
-> Iceberg ParquetValueReader
```

This remains conservative page-level pruning. It does not replace Iceberg's residual predicate evaluation.

## Correctness findings

The POC currently verifies the following:

- ColumnIndex and OffsetIndex generated by Iceberg writers can be consumed by parquet-java for page-level pruning.
- A narrow predicate can produce non-contiguous candidate row ranges.
- `PageReadStore.getRowIndexes()` exposes row-group-relative row indexes.
- `PageReadStore.getRowIndexOffset()` provides the file-level row-group start.
- Physical positions can therefore be reconstructed as:

```text
file position = rowIndexOffset + relative row index
```

- Iceberg `_pos` remains the original physical file position after page pruning.
- fallback `_row_id` values derived from `first_row_id + _pos` remain correct.
- multiple row groups are handled correctly.
- row groups for which the ColumnIndex eliminates every page are skipped correctly.
- files without a usable ColumnIndex conservatively fall back to reading the retained row group.

These position semantics are important because candidate pages cannot be renumbered without breaking metadata-column correctness and row-level delete semantics.

## File schema / filter conversion

One issue exposed by the POC is that `ParquetFilters.convert` requires file-specific physical aliases.

A plain logical Iceberg read schema is insufficient because the Parquet predicate needs to resolve the current Iceberg field ID to the physical path used in the particular Parquet file.

The current POC therefore:

```text
opens the Parquet file
-> discovers the physical schema and aliases
-> constructs the Parquet predicate
-> reopens the reader with the final ParquetReadOptions
```

This extra open is acceptable for demonstrating feasibility but may not intended as the final reader architecture.

This is also closely related to the schema/alias concerns discussed in #1566.

## Initial I/O measurements

I also ran a small local POC microbenchmark using:

- 500,000 rows
- one row group
- approximately 1,000 rows per page
- uncompressed Parquet
- projection: `id`, `payload`
- predicate: `id >= 250000 AND id < 250100`

The current double-open overhead is included in the byte measurements.

| Dataset | Page Index | Candidate Rows | Bytes Read | Median Time |
| --- | ---: | ---: | ---: | ---: |
| Sorted | OFF | 500,000 | 131.054 MiB | 67.79 ms |
| Sorted | ON | 1,000 | 0.288 MiB | 8.16 ms |
| Random | OFF | 500,000 | 131.054 MiB | 66.56 ms |
| Random | ON | 500,000 | 131.079 MiB | 69.72 ms |
| Sorted / no index | OFF | 500,000 | 131.054 MiB | 66.02 ms |
| Sorted / no index | ON | 500,000 | 131.070 MiB | 69.15 ms |

This is not a formal JMH benchmark. The timing numbers are directional; the byte-count results are the main signal.

For sorted data, page pruning reduced candidate rows by 99.8% and measured bytes read by approximately 99.78%.

For randomly distributed data, the ColumnIndex could not eliminate pages.

A file without a predicate-column ColumnIndex also fell back conservatively.

## Proposed initial scope

I suggest keeping the first implementation intentionally narrow:

- custom row-based Parquet reader only;
- Page Index used only as conservative I/O pruning;
- residual predicate evaluation remains unchanged;
- simple supported Parquet predicates first;
- missing/unsupported Page Index information falls back conservatively;
- no public Spark/Flink/table configuration in the first implementation.

Vectorized-reader support, broader predicate coverage, and configuration/default policy can be handled separately once the row-based path is well defined.

### Query engine

None

### Willingness to contribute

- [x] I can contribute this improvement/feature independently
- [x] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time

Contributor guide

Open the contributing guide

Research direction

Start with the custom row-based Parquet reader and the POC flow through readFilteredRowGroup, PageReadStore, and ParquetFilters.convert. Review ParquetReadOptions and the schema/alias concerns in #1566 before deciding how the physical schema is obtained. Done means conservative page pruning for supported predicates, correct position metadata, and fallback behavior when indexes are missing or unusable.

Written by the indexing model from the issue text.

Assessment

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.