jayzhan211 commented on code in PR #25460:
URL: https://github.com/apache/datafusion/pull/25460#discussion_r4052344058
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -1291,35 +1294,86 @@ impl RowGroupsPrunedParquetOpen {
/// Returns true if the reader would benefit from a page index load, given
/// the current pruning predicate and row group access plan.
///
- /// The page index is used for data page pruning, and it is only useful
- /// when:
+ /// Offset indexes also allow an existing row selection to skip data pages,
+ /// even without a predicate or when row-group statistics fully match it.
+ /// Otherwise, the page index is useful for predicate-based pruning when:
///
/// 1. There is at least one row group that may have filtered rows
/// (if it is fully matched we know no rows will be filtered)
///
/// 2. There is a page index for at least one predicate column (some
/// parquet writers do not write the page index).
- fn should_load_page_index(&self) -> bool {
+ fn should_load_page_index(&self) -> Result<bool> {
+ if !self.prepared.loaded.prepared.enable_page_index {
+ return Ok(false);
+ }
+ let row_groups = &self.row_groups;
+ let parquet_metadata = self.prepared.loaded.reader_metadata.metadata();
+ // External row selections need offset indexes to skip pages without
+ // decoding them. They do not require column statistics or a predicate.
+ let mut selected_row_groups = row_groups
+ .row_group_indexes()
+ .filter(|&idx| {
+ let RowGroupAccess::Selection(selection) =
+ &row_groups.access_plan().inner()[idx]
+ else {
+ return false;
+ };
+ // Runs alternate between selected and skipped rows, so two
runs
+ // suffice. Stream bitmap runs without materializing all
selectors.
+ match selection.as_mask() {
+ Some(mask) => MaskRunIter::new(mask).nth(1).is_some(),
+ None => selection.iter().nth(1).is_some(),
+ }
+ })
+ .peekable();
+ if selected_row_groups.peek().is_some() {
+ let prepared = &self.prepared.loaded.prepared;
+ // Resolve the same file-column projection as the decoder,
excluding
+ // virtual columns and respecting nested field projections.
+ let projection = match prepared.virtual_state.as_deref() {
+ None => prepared.projection.clone(),
+ Some(state) =>
prepared.projection.clone().try_map_exprs(|expr| {
+ replace_columns_with_literals(expr,
state.null_replacements())
+ })?,
+ };
+ let read_plan = build_projection_read_plan(
+ projection.expr_iter(),
+ &prepared.physical_file_schema,
+ parquet_metadata.file_metadata().schema_descr(),
+ );
+ if selected_row_groups.any(|idx| {
Review Comment:
`any(leaf_included && offset_index_offset().is_some())` doesn't match what
the load can deliver: with `PageIndexPolicy::Optional`, `parse_offset_index`
sets both indexes to `None` if *any* column chunk in *any* row group has no
offset index (parquet 59.3 `file/metadata/parser.rs:309-321`). On a mixed file
this returns `true`, pays the fetch, and gets nothing —
`should_load_page_index_with_row_selection_checks_projection` asserts `true` on
exactly that shape (`[("a", false), ("b", true)]`).
Fix: require all chunks, keep the mask only for "reads any leaf":
```rs
let all_have_offset_index = parquet_metadata
.row_groups()
.iter()
.flat_map(|rg| rg.columns())
.all(|column| column.offset_index_offset().is_some());
let reads_any_leaf =
(0..parquet_metadata.file_metadata().schema_descr().num_columns())
.any(|leaf_idx| read_plan.projection_mask.leaf_included(leaf_idx));
if all_have_offset_index && reads_any_leaf {
return Ok(true);
}
```
and flip the mixed-file expectations in the test to `false`.
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -1291,35 +1294,86 @@ impl RowGroupsPrunedParquetOpen {
/// Returns true if the reader would benefit from a page index load, given
/// the current pruning predicate and row group access plan.
///
- /// The page index is used for data page pruning, and it is only useful
- /// when:
+ /// Offset indexes also allow an existing row selection to skip data pages,
+ /// even without a predicate or when row-group statistics fully match it.
+ /// Otherwise, the page index is useful for predicate-based pruning when:
///
/// 1. There is at least one row group that may have filtered rows
/// (if it is fully matched we know no rows will be filtered)
///
/// 2. There is a page index for at least one predicate column (some
/// parquet writers do not write the page index).
- fn should_load_page_index(&self) -> bool {
+ fn should_load_page_index(&self) -> Result<bool> {
+ if !self.prepared.loaded.prepared.enable_page_index {
+ return Ok(false);
+ }
+ let row_groups = &self.row_groups;
+ let parquet_metadata = self.prepared.loaded.reader_metadata.metadata();
+ // External row selections need offset indexes to skip pages without
+ // decoding them. They do not require column statistics or a predicate.
+ let mut selected_row_groups = row_groups
+ .row_group_indexes()
+ .filter(|&idx| {
+ let RowGroupAccess::Selection(selection) =
+ &row_groups.access_plan().inner()[idx]
+ else {
+ return false;
+ };
+ // Runs alternate between selected and skipped rows, so two
runs
+ // suffice. Stream bitmap runs without materializing all
selectors.
+ match selection.as_mask() {
+ Some(mask) => MaskRunIter::new(mask).nth(1).is_some(),
+ None => selection.iter().nth(1).is_some(),
+ }
+ })
+ .peekable();
+ if selected_row_groups.peek().is_some() {
+ let prepared = &self.prepared.loaded.prepared;
+ // Resolve the same file-column projection as the decoder,
excluding
+ // virtual columns and respecting nested field projections.
+ let projection = match prepared.virtual_state.as_deref() {
+ None => prepared.projection.clone(),
+ Some(state) =>
prepared.projection.clone().try_map_exprs(|expr| {
+ replace_columns_with_literals(expr,
state.null_replacements())
+ })?,
+ };
+ let read_plan = build_projection_read_plan(
Review Comment:
Trigger only looks at `prepared.projection`; with `pushdown_filters` the
row-filter columns are decoded under the same external selection but never
trigger the load. Repro: e2e test with `.with_projection_indices(&[])`,
`.with_pushdown_filters(true)`, predicate `a >= 9950` → `bytes_scanned`
`[42700, 42700]` (disabled/enabled); with `a` projected it drops. Fine as a
follow-up: "Parquet: consider row-filter columns when deciding to load offset
index for external selections".
Fix:
```diff
+ let pushdown_predicate = prepared
+ .pushdown_filters
+ .then_some(prepared.predicate.as_ref())
+ .flatten();
let read_plan = build_projection_read_plan(
- projection.expr_iter(),
+ projection.expr_iter().chain(pushdown_predicate.cloned()),
&prepared.physical_file_schema,
parquet_metadata.file_metadata().schema_descr(),
);
```
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
Review Comment:
Selection path needs only the offset index, but `load_page_index` fetches +
decodes the column index too, and the trigger fires for any partial selection:
`skip(1)/select(9_999)` on the PR's e2e file → `bytes_scanned` 42700 → 43800
plus an extra request, no pages skipped. Skip the column index when there is no
page predicate; selectivity threshold is fine as a follow-up: "Parquet: load
only offset index (and only when selective) for external row selections".
```diff
- .with_page_index_policy(PageIndexPolicy::Optional),
+ .with_offset_index_policy(PageIndexPolicy::Optional)
+ .with_column_index_policy(
+ if self.prepared.page_pruning_predicate.is_some() {
+ PageIndexPolicy::Optional
+ } else {
+ PageIndexPolicy::Skip
+ },
+ ),
```
The free fn `load_page_index` hardcodes `.with_page_index_policy(Optional)`
on the `ParquetMetaDataReader` and early-returns on `missing_column_index ||
missing_offset_index`; both need to follow the passed policy, otherwise
offset-only metadata from a caching reader is re-fetched on every open.
--
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]
