Skip to content

Commit 9c02dad

Browse files
adriangbclaude
andcommitted
fix: do not install a Parquet row selection when the page index prunes nothing
Page-index pruning installed a row selection for each row group that it examined, also when the selection skipped no rows. The opener treats any row selection as live, so a select-all selection disabled runtime (dynamic filter) row-group pruning and statistics-based row-group reordering for the whole file. Keep the row group as a full scan when the page index skips no rows. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 696eaf5 commit 9c02dad

2 files changed

Lines changed: 155 additions & 1 deletion

File tree

‎datafusion/datasource-parquet/src/page_filter.rs‎

Lines changed: 68 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -373,7 +373,12 @@ impl PagePruningAccessPlanFilter {
373373
);
374374
total_skip += rows_skipped;
375375
total_select += rows_selected;
376-
access_plan.scan_selection(row_group_index, overall_selection)
376+
// A selection that skips no rows is equal to a full scan.
377+
// Do not install it: a live selection disables runtime
378+
// row-group pruning and row-group reordering.
379+
if rows_skipped > 0 {
380+
access_plan.scan_selection(row_group_index, overall_selection)
381+
}
377382
} else {
378383
// Selection skips all rows, so skip the entire row group
379384
let rows_skipped = groups[row_group_index].num_rows() as usize;
@@ -980,6 +985,68 @@ mod tests {
980985
}
981986
}
982987

988+
/// Page pruning that skips no rows must keep the row group as a full
989+
/// scan. A select-all selection disables runtime row-group pruning.
990+
#[test]
991+
fn page_pruning_that_skips_no_rows_installs_no_selection() {
992+
let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
993+
let props = WriterProperties::builder()
994+
.set_data_page_row_count_limit(2)
995+
.set_write_batch_size(2)
996+
.build();
997+
let mut writer =
998+
ArrowWriter::try_new(Vec::new(), Arc::clone(&schema), Some(props)).unwrap();
999+
// RG 0: every page holds a row with `a > 5`, so no page is pruned.
1000+
// RG 1: the middle page holds no row with `a > 5`, so it is pruned.
1001+
for values in [vec![0, 10, 0, 10, 0, 10], vec![0, 10, 0, 0, 0, 10]] {
1002+
let batch = RecordBatch::try_new(
1003+
Arc::clone(&schema),
1004+
vec![Arc::new(Int32Array::from(values))],
1005+
)
1006+
.unwrap();
1007+
writer.write(&batch).unwrap();
1008+
writer.flush().unwrap();
1009+
}
1010+
let options =
1011+
ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Required);
1012+
let reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
1013+
Bytes::from(writer.into_inner().unwrap()),
1014+
options,
1015+
)
1016+
.unwrap();
1017+
let expr: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(
1018+
Arc::new(Column::new("a", 0)),
1019+
Operator::Gt,
1020+
lit(5i32),
1021+
));
1022+
let filter = PagePruningAccessPlanFilter::new(&expr, Arc::clone(&schema));
1023+
let metrics = ParquetFileMetrics::new(0, "test", &ExecutionPlanMetricsSet::new());
1024+
let result = filter.prune_plan_with_page_index_and_metrics(
1025+
ParquetAccessPlan::new_all(2),
1026+
&schema,
1027+
reader.parquet_schema(),
1028+
reader.metadata(),
1029+
&metrics,
1030+
None,
1031+
);
1032+
assert_eq!(
1033+
result.access_plan.inner(),
1034+
&[
1035+
RowGroupAccess::Scan,
1036+
RowGroupAccess::Selection(
1037+
vec![
1038+
RowSelector::select(2),
1039+
RowSelector::skip(2),
1040+
RowSelector::select(2),
1041+
]
1042+
.into()
1043+
),
1044+
]
1045+
);
1046+
assert_eq!(metrics.page_index_rows_pruned.pruned(), 2);
1047+
assert_eq!(metrics.page_index_rows_pruned.matched(), 10);
1048+
}
1049+
9831050
#[test]
9841051
fn limit_plan_preserves_existing_selections() {
9851052
let groups = vec![row_group_metadata(6), row_group_metadata(6)];

‎datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -329,3 +329,90 @@ set datafusion.execution.target_partitions = 4;
329329

330330
statement ok
331331
RESET datafusion.execution.parquet.pushdown_filters;
332+
333+
# Page-index pruning that skips no rows must not disable runtime row-group
334+
# pruning. `w = 1` cannot be proven by row-group statistics (each row group
335+
# holds both `w = 0` and `w = 1`), so page-index pruning runs, but it prunes
336+
# no page: it selects all rows of each row group. After the first row group,
337+
# the TopK dynamic filter `v < 1` proves that row groups 1..=4 cannot hold the
338+
# top-1 row, so the runtime pruner must skip them with the page index enabled
339+
# and with it disabled. The rows of each row group are in descending order, so
340+
# the file has no output ordering and the TopK cannot stop early. Filter
341+
# pushdown removes the `FilterExec`, which would otherwise buffer all rows
342+
# before the TopK sees them.
343+
# RG 0: v=2,1,0 RG 1: v=5,4,3 RG 2: v=8,7,6 RG 3: v=11,10,9 RG 4: v=14,13,12
344+
# w = v % 2
345+
statement ok
346+
set datafusion.execution.target_partitions = 1;
347+
348+
statement ok
349+
set datafusion.execution.parquet.pushdown_filters = true;
350+
351+
statement ok
352+
set datafusion.explain.analyze_level = summary;
353+
354+
statement ok
355+
COPY (
356+
SELECT v, v % 2 AS w FROM (
357+
SELECT CAST(3 * (i / 3) + 2 - i % 3 AS BIGINT) AS v FROM generate_series(0, 14) AS t(i)
358+
)
359+
)
360+
TO 'test_files/scratch/dynamic_row_group_pruning/select_all.parquet'
361+
STORED AS PARQUET
362+
OPTIONS ('format.max_row_group_size' '3');
363+
364+
statement ok
365+
CREATE EXTERNAL TABLE select_all (v BIGINT NOT NULL, w BIGINT NOT NULL)
366+
STORED AS PARQUET
367+
LOCATION 'test_files/scratch/dynamic_row_group_pruning/select_all.parquet';
368+
369+
query I
370+
SELECT v FROM select_all WHERE w = 1 ORDER BY v LIMIT 1;
371+
----
372+
1
373+
374+
# Page index enabled (the default): `page_index_pages_pruned` shows that no
375+
# page is pruned, and `row_groups_pruned_dynamic_filter=4` shows that runtime
376+
# row-group pruning still happens.
377+
query TT
378+
explain analyze SELECT v FROM select_all WHERE w = 1 ORDER BY v LIMIT 1;
379+
----
380+
Plan with Metrics
381+
01)SortExec: TopK(fetch=1), expr=[v@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[v@0 < 1], metrics=[output_rows=1, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
382+
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/select_all.parquet]]}, projection=[v], file_type=parquet, predicate=w@1 = 1 AND DynamicFilter [ v@0 < 1 ], sort_order_for_reorder=[v@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=w_null_count@2 != row_count@3 AND w_min@0 <= 1 AND 1 <= w_max@1 AND v_null_count@5 != row_count@3 AND v_min@4 < 1, required_guarantees=[w in (1)], metrics=[output_rows=1, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=5 total → 5 matched, row_groups_pruned_bloom_filter=5 total → 5 matched, page_index_pages_pruned=5 total → 5 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=<slt:ignore>, bytes_scanned=<slt:ignore>, row_groups_pruned_dynamic_filter=4, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
383+
384+
statement ok
385+
drop table select_all;
386+
387+
# Page index disabled: the same runtime row-group pruning. Create the table
388+
# again after the option change, because the table keeps the parquet options
389+
# of its creation time.
390+
statement ok
391+
set datafusion.execution.parquet.enable_page_index = false;
392+
393+
statement ok
394+
CREATE EXTERNAL TABLE select_all (v BIGINT NOT NULL, w BIGINT NOT NULL)
395+
STORED AS PARQUET
396+
LOCATION 'test_files/scratch/dynamic_row_group_pruning/select_all.parquet';
397+
398+
query TT
399+
explain analyze SELECT v FROM select_all WHERE w = 1 ORDER BY v LIMIT 1;
400+
----
401+
Plan with Metrics
402+
01)SortExec: TopK(fetch=1), expr=[v@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[v@0 < 1], metrics=[output_rows=1, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
403+
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/select_all.parquet]]}, projection=[v], file_type=parquet, predicate=w@1 = 1 AND DynamicFilter [ v@0 < 1 ], sort_order_for_reorder=[v@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=w_null_count@2 != row_count@3 AND w_min@0 <= 1 AND 1 <= w_max@1 AND v_null_count@5 != row_count@3 AND v_min@4 < 1, required_guarantees=[w in (1)], metrics=[output_rows=1, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=5 total → 5 matched, row_groups_pruned_bloom_filter=5 total → 5 matched, page_index_pages_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=<slt:ignore>, bytes_scanned=<slt:ignore>, row_groups_pruned_dynamic_filter=4, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
404+
405+
statement ok
406+
drop table select_all;
407+
408+
statement ok
409+
RESET datafusion.execution.parquet.enable_page_index;
410+
411+
statement ok
412+
set datafusion.execution.target_partitions = 4;
413+
414+
statement ok
415+
RESET datafusion.explain.analyze_level;
416+
417+
statement ok
418+
RESET datafusion.execution.parquet.pushdown_filters;

0 commit comments

Comments
 (0)