Skip to content

Commit 102a162

Browse files
authored
fix: report Parquet Bloom filter pruning metrics accurately (#25822)
## Which issue does this PR close? - Closes #18355. ## Rationale for this change `row_groups_pruned_bloom_filter` currently reports retained row groups as Bloom filter matches even when Bloom pruning was not evaluated—for example, when Bloom filters are disabled or unavailable, no relevant Bloom statistics were loaded, or predicate evaluation failed. This makes `EXPLAIN ANALYZE` suggest that Bloom filters evaluated and matched row groups when they did not participate in the pruning decision. It also displays an idle Bloom pruning metric when all its counters are zero. A retained row group is not necessarily a Bloom filter match. The metric should describe actual Bloom pruning outcomes. ## What changes are included in this PR? - Record a Bloom filter match only when the Bloom pruning predicate is successfully evaluated and determines that the row group may match. - Leave Bloom pruning counters unchanged when Bloom pruning is unavailable or cannot be evaluated. - Continue retaining row groups conservatively when Bloom predicate evaluation fails, while recording the failure in `predicate_evaluation_errors`. - Omit `row_groups_pruned_bloom_filter` from physical-plan displays when all of its pruning counters are zero. - Preserve the registered metric and direct metric lookup; only its accounting and displayed output change. - Update affected Rust assertions and SQL logic test snapshots. The row-group access decisions and query results are unchanged. ## What is the testing strategy for this PR? Added `bloom_filter_pruning_error_retains_row_group_without_match`, which verifies that a Bloom predicate evaluation error: - retains the row group; - increments `predicate_evaluation_errors`; - does not increment either the Bloom matched or pruned counter. The display tests verify that: - an idle Bloom pruning metric is omitted from aggregated and full Indent, Graphviz, and PostgreSQL JSON output; - unrelated zero-valued metrics remain visible; - genuine nonzero Bloom pruning results remain visible. The following existing test coverage was also run: - Parquet Bloom-filter and row-group-filter unit tests; - `parquet_integration` row-group-pruning tests; - `core_integration` explain-analyze tests; - physical-plan display tests; - affected SQL logic tests for dynamic filtering, dynamic row-group pruning, explain-analyze, limit pruning, and Parquet filter pushdown. Formatting checks also pass. ## Are there any user-facing changes? Yes. Parquet Bloom filter pruning metrics in `EXPLAIN ANALYZE` and other physical-plan displays more accurately represent actual Bloom filter evaluation. An entirely idle `row_groups_pruned_bloom_filter` metric is no longer displayed. There are no public API changes and no changes to query results or row-group access decisions.
1 parent 97c7593 commit 102a162

10 files changed

Lines changed: 617 additions & 217 deletions

File tree

‎datafusion/core/tests/parquet/row_group_pruning.rs‎

Lines changed: 59 additions & 47 deletions
Large diffs are not rendered by default.

‎datafusion/core/tests/sql/explain_analyze.rs‎

Lines changed: 5 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -880,10 +880,7 @@ async fn parquet_explain_analyze() {
880880

881881
// should contain aggregated stats
882882
assert_contains!(&formatted, "output_rows=8");
883-
assert_contains!(
884-
&formatted,
885-
"row_groups_pruned_bloom_filter=1 total \u{2192} 1 matched"
886-
);
883+
assert_not_contains!(&formatted, "row_groups_pruned_bloom_filter");
887884
assert_contains!(
888885
&formatted,
889886
"row_groups_pruned_statistics=1 total \u{2192} 1 matched"
@@ -896,16 +893,14 @@ async fn parquet_explain_analyze() {
896893
// (file-> row-group -> page)
897894
let i_file = formatted.find("files_ranges_pruned_statistics").unwrap();
898895
let i_rowgroup_stat = formatted.find("row_groups_pruned_statistics").unwrap();
899-
let i_rowgroup_bloomfilter =
900-
formatted.find("row_groups_pruned_bloom_filter").unwrap();
901896
let i_page_rows = formatted.find("page_index_rows_pruned").unwrap();
902897
let i_page_pages = formatted.find("page_index_pages_pruned").unwrap();
903898

904899
assert!(
905900
(i_file < i_rowgroup_stat)
906-
&& (i_rowgroup_stat < i_rowgroup_bloomfilter)
907-
&& (i_rowgroup_bloomfilter < i_page_pages && i_page_pages < i_page_rows),
908-
"The parquet pruning metrics should be displayed in an order of: file range -> row group statistics -> row group bloom filter -> page index."
901+
&& (i_rowgroup_stat < i_page_pages)
902+
&& (i_page_pages < i_page_rows),
903+
"The parquet pruning metrics should be displayed in an order of: file range -> row group statistics -> page index."
909904
);
910905
}
911906

@@ -1179,7 +1174,7 @@ async fn parquet_explain_analyze_verbose() {
11791174
.to_string();
11801175

11811176
// should contain the raw per file stats (with the label)
1182-
assert_contains!(&formatted, "row_groups_pruned_bloom_filter{partition=0");
1177+
assert_not_contains!(&formatted, "row_groups_pruned_bloom_filter");
11831178
assert_contains!(&formatted, "row_groups_pruned_statistics{partition=0");
11841179
}
11851180

‎datafusion/datasource-parquet/src/opener/mod.rs‎

Lines changed: 41 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1336,26 +1336,14 @@ impl FiltersPreparedParquetOpen {
13361336
.row_groups_pruned_statistics
13371337
.add_matched(row_groups.remaining_row_group_count());
13381338
}
1339-
1340-
if !prepared.enable_bloom_filter || row_groups.is_empty() {
1341-
// Update metrics: bloom filter unavailable, so all row groups are
1342-
// matched (not pruned)
1343-
prepared
1344-
.file_metrics
1345-
.row_groups_pruned_bloom_filter
1346-
.add_matched(row_groups.remaining_row_group_count());
1347-
}
13481339
} else {
13491340
// Update metrics: no predicate, so all row groups are matched (not pruned)
1341+
// by statistics. Bloom pruning did not run, so its metrics are unchanged.
13501342
let remaining = row_groups.remaining_row_group_count();
13511343
prepared
13521344
.file_metrics
13531345
.row_groups_pruned_statistics
13541346
.add_matched(remaining);
1355-
prepared
1356-
.file_metrics
1357-
.row_groups_pruned_bloom_filter
1358-
.add_matched(remaining);
13591347
}
13601348

13611349
Ok(RowGroupsPrunedParquetOpen {
@@ -3413,6 +3401,32 @@ mod test {
34133401
.expect("row_groups_pruned_statistics metric is emitted")
34143402
}
34153403

3404+
/// Bloom pruning counters after a scan, as `(pruned, matched)`.
3405+
///
3406+
/// Direct `MetricsSet` lookup, not plan display: idle Bloom metrics are
3407+
/// omitted from displayed plans even when the counters remain registered.
3408+
///
3409+
/// Opening one file still registers this name more than once:
3410+
/// `prepare_open_file` creates `ParquetFileMetrics`, and each
3411+
/// `ParquetFileReaderFactory::create_reader` call (initial reader plus
3412+
/// Bloom replacement reader) creates another independent set. Sum the
3413+
/// counters the same way `MetricsSet::sum_by_name` aggregates pruning
3414+
/// metrics. Only the opener's set is incremented by Bloom pruning; the
3415+
/// reader copies stay at zero.
3416+
fn bloom_filter_pruning_metrics(metrics: &ExecutionPlanMetricsSet) -> (usize, usize) {
3417+
use datafusion_physical_plan::metrics::MetricValue;
3418+
match metrics
3419+
.clone_inner()
3420+
.sum_by_name("row_groups_pruned_bloom_filter")
3421+
{
3422+
Some(MetricValue::PruningMetrics {
3423+
pruning_metrics, ..
3424+
}) => (pruning_metrics.pruned(), pruning_metrics.matched()),
3425+
Some(_) => panic!("row_groups_pruned_bloom_filter is not a pruning metric"),
3426+
None => panic!("row_groups_pruned_bloom_filter metric is registered"),
3427+
}
3428+
}
3429+
34163430
#[tokio::test]
34173431
async fn test_prune_all_null_column_equality_from_file_statistics() {
34183432
// Regression: a column whose file statistics say every value is
@@ -4743,6 +4757,16 @@ mod test {
47434757
// contribute to this metric.
47444758
bloom_bytes.push(counter_metric_value(&metrics, "bytes_scanned"));
47454759
assert_eq!(collect_int32_values(stream).await, vec![1, 1, 1, 1]);
4760+
let (pruned, matched) = bloom_filter_pruning_metrics(&metrics);
4761+
assert_eq!(pruned, 0);
4762+
if stats_pruning {
4763+
// [1,1,1] is fully matched by statistics and skips Bloom.
4764+
// [0,1,2] still evaluates Bloom for `a = 1` and cannot prune.
4765+
assert_eq!(matched, 1);
4766+
} else {
4767+
// Statistics pruning is off, so both row groups evaluate Bloom.
4768+
assert_eq!(matched, 2);
4769+
}
47464770
}
47474771
assert!(bloom_bytes[1] > 0, "partial row group needs Bloom I/O");
47484772
assert!(
@@ -4781,6 +4805,10 @@ mod test {
47814805
.unwrap();
47824806
assert_eq!(counter_metric_value(&metrics, "bytes_scanned"), 0);
47834807
assert_eq!(collect_int32_values(stream).await, vec![1, 1, 1]);
4808+
let (pruned, matched) = bloom_filter_pruning_metrics(&metrics);
4809+
// Bloom reads and evaluation are skipped for an entirely fully matched file.
4810+
assert_eq!(pruned, 0);
4811+
assert_eq!(matched, 0);
47844812
}
47854813

47864814
#[test]

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

Lines changed: 51 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -472,7 +472,8 @@ impl RowGroupAccessPlanFilter {
472472
}
473473

474474
if self.access_plan.is_fully_matched(idx) {
475-
metrics.row_groups_pruned_bloom_filter.add_matched(1);
475+
// Statistics already proved every row matches. Bloom did not
476+
// evaluate this group, so do not record a Bloom outcome.
476477
continue;
477478
}
478479

@@ -481,7 +482,6 @@ impl RowGroupAccessPlanFilter {
481482
// evaluation in that case: it runs once per row group and can be expensive for wide
482483
// predicates, a cost files written without bloom filters would pay for nothing.
483484
if stats.is_empty() {
484-
metrics.row_groups_pruned_bloom_filter.add_matched(1);
485485
continue;
486486
}
487487

@@ -493,7 +493,7 @@ impl RowGroupAccessPlanFilter {
493493
"Error evaluating row group predicate on bloom filter: {e}"
494494
);
495495
metrics.predicate_evaluation_errors.add(1);
496-
false
496+
continue;
497497
}
498498
};
499499

@@ -622,9 +622,12 @@ mod tests {
622622

623623
use arrow::datatypes::DataType::Decimal128;
624624
use arrow::datatypes::{DataType, Field};
625-
use datafusion_expr::{cast, col, lit};
626-
use datafusion_physical_expr::PhysicalExpr;
625+
use datafusion_expr::{Operator, cast, col, lit};
627626
use datafusion_physical_expr::planner::logical2physical;
627+
use datafusion_physical_expr::{
628+
PhysicalExpr,
629+
expressions::{BinaryExpr, CastExpr, Column as PhysicalColumn, Literal},
630+
};
628631
use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
629632
use datafusion_pruning::PruningPredicateBuilder;
630633
use parquet::arrow::ArrowSchemaConverter;
@@ -1640,7 +1643,49 @@ mod tests {
16401643
// the row group with a bloom filter is still evaluated and pruned.
16411644
assert_pruned(row_groups, ExpectedPruning::Some(vec![0]));
16421645
assert_eq!(metrics.row_groups_pruned_bloom_filter.pruned(), 1);
1643-
assert_eq!(metrics.row_groups_pruned_bloom_filter.matched(), 1);
1646+
assert_eq!(metrics.row_groups_pruned_bloom_filter.matched(), 0);
1647+
}
1648+
1649+
#[test]
1650+
fn bloom_filter_pruning_error_retains_row_group_without_match() {
1651+
let schema =
1652+
Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
1653+
let cast_expr: Arc<dyn PhysicalExpr> = Arc::new(CastExpr::new(
1654+
Arc::new(PhysicalColumn::new("c1", 0)),
1655+
DataType::FixedSizeBinary(4),
1656+
None,
1657+
));
1658+
let literal: Arc<dyn PhysicalExpr> = Arc::new(Literal::new(
1659+
ScalarValue::FixedSizeBinary(4, Some(vec![0, 0, 0, 1])),
1660+
));
1661+
let expr = Arc::new(BinaryExpr::new(cast_expr, Operator::Eq, literal));
1662+
let pruning_predicate = PruningPredicateBuilder::new()
1663+
.with_file_schema(schema)
1664+
.try_build(expr)
1665+
.unwrap();
1666+
1667+
// A nonempty map bypasses the no-filter path. Evaluating the predicate
1668+
// then fails because Arrow cannot cast Int32 statistics to FixedSizeBinary.
1669+
let mut sbbf = Sbbf::new_with_ndv_fpp(10, 0.01).unwrap();
1670+
sbbf.insert(&1_i32);
1671+
let mut bloom_statistics = BloomFilterStatistics::new();
1672+
bloom_statistics.insert("loaded_filter", sbbf, PhysicalType::INT32, 4);
1673+
assert!(pruning_predicate.prune(&bloom_statistics).is_err());
1674+
1675+
let metrics = parquet_file_metrics();
1676+
let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(1));
1677+
row_groups.prune_by_bloom_filters(
1678+
&pruning_predicate,
1679+
&metrics,
1680+
&[bloom_statistics],
1681+
);
1682+
1683+
// An evaluation failure is conservative: retain the row group, report
1684+
// the error separately, and do not call the result a Bloom match.
1685+
assert_pruned(row_groups, ExpectedPruning::Some(vec![0]));
1686+
assert_eq!(metrics.predicate_evaluation_errors.value(), 1);
1687+
assert_eq!(metrics.row_groups_pruned_bloom_filter.pruned(), 0);
1688+
assert_eq!(metrics.row_groups_pruned_bloom_filter.matched(), 0);
16441689
}
16451690

16461691
fn get_row_group_meta_data(

0 commit comments

Comments
 (0)