bharath-techie opened a new issue, #24288:
URL: https://github.com/apache/datafusion/issues/24288
### Is your feature request related to a problem or challenge?
DataFusion can skip loading the Parquet page index when row-group statistics
show that page pruning cannot help. However, when page pruning is useful, it
loads and decodes the complete column-index and offset-index for all the column
of the row group.
For wide files, this cost is disproportionate to the query. A query
filtering on 2 columns and projecting 5 out of 400 generally needs:
- Column indexes for predicate columns, for page pruning.
- Offset indexes for physical columns involved in pruning, row-selection
application, and decoding.
Indexes for the remaining columns are fetched, decoded, and retained
without benefiting the query.
We measured this while implementing scoped page-index loading downstream
in OpenSearch:
https://github.com/opensearch-project/OpenSearch/pull/22254
On a wide, one-billion-row
[textbench](http://github.com/ClickHouse/TextBench/) dataset:
| Metadata | Memory |
|---|---:|
| Complete `ParquetMetaData` with page indexes | ~1750 MB |
| Footer metadata | ~157 MB |
| Required offset indexes | ~158 MB |
| Required column indexes | ~20-30 MB |
The required metadata was approximately 335–345 MB instead of 1750 MB. Exact
values depend on the schema and workload.
So, raising this feature request to see if the wider community can benefit
from these changes.
### Describe the solution you'd like
## Current behavior
The Parquet opener initially requests metadata with
`PageIndexPolicy::Skip` and prunes row groups using footer statistics.
`should_load_page_index` then checks whether:
- A page-pruning predicate exists.
- At least one surviving row group is not fully matched.
- At least one predicate column has column-index and offset-index
locations.
This already avoids many unnecessary page-index loads.
When indexes are required, `load_page_index` runs `ParquetMetaDataReader`
with `PageIndexPolicy::Optional`. Arrow then fetches and decodes complete
page-index matrices. DataFusion cannot request particular columns or row
groups through this API.
`FileMetadataCache` stores one `Arc<ParquetMetaData>` per file. An entry
is treated as either:
- Footer-only, when column_index() or offset_index() is None.
- Fully indexed, when both are Some.
The deferred load currently bypasses this cache, as described in #23978.
Caching the complete index would avoid repeated I/O but could substantially
increase resident memory for wide files.
## Proposed behavior
After footer-based row-group pruning, derive the page-index entries
required by the remaining scan:
- Load column indexes only for Parquet leaf columns referenced by
page-pruning predicates.
- Load offset indexes only for physical columns needed by page pruning,
row-selection application, pushed-down filtering, or decoding.
- Do not decode indexes for pruned row groups.
- Do not decode indexes for fully matched row groups unless another row
selection or read requirement needs them.
The opener has the required information at this point:
- Surviving and fully matched row groups.
- Page-pruning predicate columns.
- Decoder projection.
- Columns decoded by pushed-down row filters.
This also allows row-group scoping, which is difficult to implement safely
in an external reader factory because DataFusion determines the final access
plan after obtaining metadata.
If an entry is absent or optional decoding fails, DataFusion should
preserve its conservative behavior and scan the affected data without page
pruning.
## Separate fetch from decode
Selecting entries for decoding should not require issuing one object-store
request per entry.
The implementation should separate:
1. The logical selection of row-group and column indexes.
2. The physical byte ranges fetched to satisfy that selection.
The appropriate strategy depends on storage characteristics. Local files
can benefit from narrow reads, while remote object stores generally prefer
fewer and larger requests.
For example, the OpenSearch implementation uses different strategies by
storage type:
- For local storage, it fetches ranges covering only selected column
chunks.
- For remote storage backed by Foyer, shard warmup stores the complete
column-index and offset-index regions under exact range keys. Query-time
loading requests those same complete regions, producing cache hits without
remote I/O.
But decodes are done only for selected entries.
This preserves the main CPU and memory benefits even when reducing fetched
bytes is not worthwhile.
A DataFusion implementation could initially use a simple per-reader policy:
```
enum PageIndexFetchPolicy {
Exact,
Coalesce {
max_gap: usize,
target_size: usize,
},
WholeIndexRegion,
}
```
The exact API is open for discussion. The important requirement is that
the logical decode selection remains independent from range coalescing.
Longer term, the object-store or reader implementation could provide its
preferred policy.
## Scoped caching
- The current metadata cache can be extended to contain decoded page-index
entries at a finer granularity than the complete file index.
- A cache entry needs to identify at least:
- File identity, including freshness information.
- Index type: column index or offset index.
- Parquet physical leaf-column index.
- Row-group index, unless entries are deliberately grouped across row
groups.
## Arrow support
Current `arrow-rs` main has no public Parquet API for decoding a selected
set of page-index entries.
`ParquetMetaDataReader` and `ParquetMetaDataPushDecoder` support separate
policies for column and offset indexes, but each applies to the whole file.
Their parsers produce complete dense matrices.
Relevant arrow-rs issues include:
- apache/arrow-rs#9609: decode page indexes only for selected columns.
- apache/arrow-rs#8643: metadata projection and progressive loading.
- apache/arrow-rs#8818: optional or sparse page-index entries.
A clean integration likely requires changes in `arrow-rs` as well to
support scoped page-index decoding.
### Possible implementation stages
1. Add an `arrow-rs` API that accepts independent column-index and
offset-index selections, optionally restricted to selected row groups.
2. Initially fetch the existing covering page-index range, but decode only
selected entries. This provides the main CPU and memory benefit without
introducing additional object-store requests.
3. In the DataFusion opener, derive the required columns and row groups
after footer-statistics pruning.
4. Pass the scoped indexes to page pruning and Parquet decoding, with
conservative fallback when a required index is unavailable.
5. Add bounded caching for decoded page-index entries. This could extend
the existing metadata cache or use a separate cache associated with the cached
footer metadata.
6. Add storage-aware range fetching so readers can choose between exact
ranges, coalesced ranges, and the complete index region.
7. Add metrics and benchmarks
I am happy to contribute / help on the DataFusion implementation.
I might be missing internal details that might make some of this tricky and
I'd love to hear feedback from community.
### Describe alternatives you've considered
### Additional context
### Relevant issues
- #16200 describes the general goal of reading page metadata only for
required row groups and columns. This proposal adds independent column-index
and offset-index selections, storage-aware fetching, incremental caching, and
downstream implementation evidence.
### Downstream custom implementation
For reference, In the downstream , OpenSearch implemented scoped loading
without changing DataFusion, we implemented workaround as follows :
1. A custom `ParquetFileReaderFactory` returns footer metadata with
selected page indexes attached.
2. A physical optimizer rule replaces the reader factory installed by
ParquetFormat so the custom reader receives predicate and projection columns.
3. Because `ParquetMetaData` expects dense [row_group][column] matrices,
unrequested column-index cells are populated with ColumnIndexMetaData::NONE.
4. Offset indexes have no equivalent missing-cell representation,
requiring synthetic entries for unrequested cells - This is a hacky workaround.
5. The listing-table path cannot safely scope by row group because
DataFusion determines the final row-group access plan after invoking the reader
factory.
6. Selective decoding uses the deprecated `read_columns_indexes` and
`read_offset_indexes` APIs available in the older Parquet 58.3 version.
Those subset decoder APIs were removed from arrow-rs in
apache/arrow-rs#10035, so this workaround has no upgrade path to current Arrow.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
