[python] Read Parquet row windows with OffsetIndex

[XiaoHongbo-Hope]

What

Add Parquet OffsetIndex reads for one contiguous row-ID window per row group. PyPaimon fetches the selected data pages and required dictionary pages, then decodes them with PyArrow. This supports scalar and nested STRUCT/ARRAY/MAP fields, including nested projections. OffsetIndex and selected PageHeaders use bounded typed decoders, footer materialization is bounded, and adjacent index ranges are read together.

Disable it with the table option:

table = table.copy({"parquet.filter.columnindex.enabled": "false"})

The option is enabled by default, matching Java. Disjoint windows, VARIANT, missing indexes, decoded row-group cache reads, and unsupported or unprofitable selections use the existing reader. This does not change the table format or add ColumnIndex predicate filtering.

Trade-off

Reading fewer bytes can require more range requests, so lower latency is not guaranteed. A read-only OSS A/B on an isolated table confirmed identical results and this byte/request trade-off; it is not an end-to-end throughput claim.

Validation

Tests passed on PyArrow 16, 19 and 24, including nested fields, projections, bounded metadata decoding, fallback and error paths. Java parquet-mr nested fixtures matched full reads; an older PyArrow fixture exercised the fallback.

[JingsongLi]

[P2] Preserve the existing default for the shared Parquet option

parquet.filter.columnindex.enabled is an existing parquet-mr option whose default is true, and Paimon forwards parquet.* table options to Java ParquetReadOptions. Defining the same key with a default of false makes its effective behavior differ between Java and PyPaimon when the option is absent. Please change this default to true. If the new PyPaimon OffsetIndex path must remain opt-in, it should use a separate PyPaimon-specific option instead.

[JingsongLi]

[P1] Bound decoded OffsetIndex state, not only the encoded buffer

_MAX_INDEX_BYTES caps serialized bytes, but this generic decoder materializes the complete collection as Python lists, tuples, and dictionaries before the OffsetIndex/PageLocation fields are validated. For example, an encoded list of empty structs passes the remaining-byte check and can expand an 8 MiB index into hundreds of MiB before the later row-boundary validation rejects it. The parquet-mr path uses generated PageLocation decoding, which validates each required field while decoding instead of first building a generic metadata tree. Please at least align with that behavior: decode OffsetIndex into typed fields, validate each PageLocation before retaining it, and enforce an explicit page-location/object budget before constructing the collection.

[JingsongLi]

[P2] Batch OffsetIndex reads before deciding to fall back

Each physical leaf performs a separate read_at here, while selected_bytes plus index_bytes is checked only after every index has been fetched. With 200 scalar columns and a one-page row group, selecting rows 0 through 1 adds about 200 index range reads and then falls back to the full-row-group reader because no page can be skipped. parquet-mr retains decoded indexes in a row-group ColumnIndexStore and coalesces or vector-reads selected data ranges; its source also explicitly marks batching consecutive OffsetIndexes as a TODO. Please keep this default-on Python path at least no worse than that model by planning and coalescing adjacent index ranges from the footer, reusing decoded indexes, and/or rejecting the optimization from footer metadata before issuing per-column reads.

[JingsongLi]

[P1] Charge struct fields against the OffsetIndex object budget

The typed decoder budgets collection elements, but every decoded struct retains an unbudgeted seen set. Unknown inline-boolean fields consume no remaining_items budget, so a syntactically valid compact payload can expand far beyond the serialized cap before being rejected. I reproduced a 1,048,585-byte OffsetIndex that decoded successfully while peaking at 100,513,560 bytes; the current 8 MiB byte limit can therefore still create worker-threatening allocation on corrupt or forward-extended metadata.

Please count every struct field against a global decoder budget, or avoid retaining unknown IDs, and raise _PageIndexBudgetExceeded so read_row_group falls back safely. A regression should bound allocation for many unknown fields, not only collection size.

[JingsongLi]

[P2] Avoid materializing every unselected row group footer tree

create serializes and generic-decodes the complete FileMetaData, then retains the whole Python object tree even when row_groups selects only one group. On a valid one-column file with 5,000 row groups, create(..., row_groups=[0]) turned a 574,774-byte footer into 25,377,527 retained bytes, peaked at 26,528,093 bytes, and took 0.742 seconds locally. This is paid on the default-enabled row-range path and scales with irrelevant groups.

Please stream or retain only schema plus selected row groups, cache a bounded decoded representation, or reject oversized or over-fragmented metadata before generic materialization and fall back to the ordinary reader.

[JingsongLi]

OffsetIndex has been switched to a typed/bounded decoder, but the PageHeader of the data page is still decoded using the generic _Compact.value(12), which fully expands unknown collections or structs into Python lists or dictionaries. _MAX_PAGE_BYTES limits only the encoded data size (32 MiB), not the size of the decoded objects.