jayzhan211 commented on code in PR #25811:
URL: https://github.com/apache/datafusion/pull/25811#discussion_r4539357139
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -1559,53 +1517,26 @@ impl RowGroupsPrunedParquetOpen {
continue;
}
};
- row_group_filters.insert(
- column_name,
- bf,
- *physical_type,
- *type_length,
- );
+ any_filter_loaded = true;
+
+ if self.row_groups.prune_row_group_with_bloom_filter(
+ idx,
+ col_guarantee,
+ &bf,
+ &prepared.file_metrics,
+ ) {
+ break;
+ }
}
- row_group_bloom_filters[idx] = row_group_filters;
- }
- }
-
- Ok(BloomFiltersLoadedParquetOpen {
- prepared: self,
- row_group_bloom_filters,
- })
- }
-}
-impl BloomFiltersLoadedParquetOpen {
- /// Apply bloom filter pruning using already loaded bloom filters.
- fn prune_bloom_filters(mut self) -> RowGroupsPrunedParquetOpen {
- let bloom_filter_eval_time = self
- .prepared
- .prepared
- .loaded
- .prepared
- .file_metrics
- .bloom_filter_eval_time
- .clone();
- let _timer_guard = bloom_filter_eval_time.timer();
- if let Some(predicate) = self
- .prepared
- .prepared
- .pruning_predicate
- .as_ref()
- .map(|p| p.as_ref())
- && self.prepared.prepared.loaded.prepared.enable_bloom_filter
- && !self.prepared.row_groups.is_empty()
- {
- self.prepared.row_groups.prune_by_bloom_filters(
- predicate,
- &self.prepared.prepared.loaded.prepared.file_metrics,
- &self.row_group_bloom_filters,
- );
+ if any_filter_loaded &&
self.row_groups.access_plan().should_scan(idx) {
+ self.row_groups
+ .record_bloom_filter_matched(&prepared.file_metrics);
Review Comment:
Optional cleanup: `record_bloom_filter_matched` doesn't use `self`, so the
metric can be bumped directly here and the method in `row_group_filter.rs`
dropped.
```suggestion
prepared
.file_metrics
.row_groups_pruned_bloom_filter
.add_matched(1);
```
##########
datafusion/datasource-parquet/src/row_group_filter.rs:
##########
@@ -493,12 +496,83 @@ impl RowGroupAccessPlanFilter {
if prune_group {
metrics.row_groups_pruned_bloom_filter.add_pruned(1);
- self.access_plan.skip(idx)
+ self.access_plan.skip(idx);
} else {
metrics.row_groups_pruned_bloom_filter.add_matched(1);
}
}
}
+
+ /// Extracts column guarantees from a [`PruningPredicate`] that can be
evaluated against
+ /// Parquet Bloom filters.
+ pub fn prepare_bloom_column_guarantees(
+ predicate: &PruningPredicate,
+ parquet_schema: &SchemaDescriptor,
+ file_schema: &Schema,
+ ) -> Vec<BloomColumnGuarantee> {
+ predicate
+ .literal_columns()
+ .into_iter()
+ .filter_map(|column_name| {
+ let (column_idx, _) = parquet::arrow::parquet_column(
+ parquet_schema,
+ file_schema,
+ &column_name,
+ )?;
+ let in_guarantees: Vec<HashSet<ScalarValue>> = predicate
+ .literal_guarantees()
+ .iter()
+ .filter(|lg| {
+ lg.column.name == column_name && lg.guarantee ==
Guarantee::In
+ })
+ .map(|lg| lg.literals.clone())
+ .collect();
+ Some(BloomColumnGuarantee {
Review Comment:
A Bloom filter can only prove a value absent, so a column whose only
guarantee is `NotIn` (`a <> 999`) can never prune, yet its filter is still
read, and read first when it comes first. On the `bloom_filter_io` fixture,
`WHERE a <> 999 AND b = 999` reads `a`'s filter (request 4) before `b`'s
(request 5). Skipping such columns removes that request. It also changes the
Bloom `matched` count in 7 `row_group_pruning` tests (`*_neq`,
`*_in_list_negated`; e.g. `prune_string_neq` 2 → 0), since nothing is evaluated
for those groups anymore. Fine to handle in a follow-up PR if you'd rather keep
the metric change separate.
```suggestion
.collect();
// A Bloom filter can only prove a value absent, so a column
// with only `NotIn` guarantees (`a != 5`) can never prune.
if in_guarantees.is_empty() {
return None;
}
Some(BloomColumnGuarantee {
```
<details>
<summary>Snapshot test for <code>bloom_filter_io.rs</code> (output from a
local run with the change)</summary>
```rs
/// `a <> 999` can never be pruned by a Bloom filter, so only `b`'s is read.
#[tokio::test]
async fn test_parquet_bloom_filter_io_not_eq_skipped() {
let test = bloom_filter_test().await;
assert_snapshot!(test.query("SELECT * FROM t WHERE a <> 999 AND b =
999").await, @r"
--- Query: SELECT * FROM t WHERE a <> 999 AND b = 999 (0 rows) ---
Request 1
Head (object size and metadata)
Request 2
Read Parquet footer (offsets 213..678)
Request 3
Read OffsetIndex for columns 'a', 'b' (rg 0) (offsets 192..213)
Request 4
Read Bloom filter for column 'b' (rg 0) (offsets 145..192)
");
}
```
</details>
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -4880,6 +4811,547 @@ mod test {
assert_eq!(matched, 0);
}
+ #[derive(Clone, Default, Debug)]
+ struct RangeTracker {
+ ranges: Arc<std::sync::Mutex<Vec<std::ops::Range<u64>>>>,
+ }
+
+ struct TrackingBloomAsyncReader {
+ inner: Box<dyn AsyncFileReader + Send>,
+ tracker: RangeTracker,
+ }
+
+ impl AsyncFileReader for TrackingBloomAsyncReader {
+ fn get_bytes(
+ &mut self,
+ range: std::ops::Range<u64>,
+ ) -> BoxFuture<'_, parquet::errors::Result<bytes::Bytes>> {
+ self.tracker.ranges.lock().unwrap().push(range.clone());
+ self.inner.get_bytes(range)
+ }
+
+ fn get_byte_ranges(
+ &mut self,
+ ranges: Vec<std::ops::Range<u64>>,
+ ) -> BoxFuture<'_, parquet::errors::Result<Vec<bytes::Bytes>>> {
+ self.tracker.ranges.lock().unwrap().extend(ranges.clone());
+ self.inner.get_byte_ranges(ranges)
+ }
+
+ fn get_metadata<'a>(
+ &'a mut self,
+ options: Option<&'a ArrowReaderOptions>,
+ ) -> BoxFuture<'a, parquet::errors::Result<Arc<ParquetMetaData>>> {
+ self.inner.get_metadata(options)
+ }
+ }
+
+ #[derive(Debug)]
+ struct TrackingBloomReaderFactory {
+ base_factory: Arc<dyn ParquetFileReaderFactory>,
+ tracker: RangeTracker,
+ }
+
+ impl ParquetFileReaderFactory for TrackingBloomReaderFactory {
+ fn create_reader(
+ &self,
+ partition_index: usize,
+ file: PartitionedFile,
+ metadata_size_hint: Option<usize>,
+ metrics: &ExecutionPlanMetricsSet,
+ ) -> Result<Box<dyn AsyncFileReader + Send>> {
+ let inner = self.base_factory.create_reader(
+ partition_index,
+ file,
+ metadata_size_hint,
+ metrics,
+ )?;
+ Ok(Box::new(TrackingBloomAsyncReader {
+ inner,
+ tracker: self.tracker.clone(),
+ }))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_bloom_filter_incremental_io_pattern() {
Review Comment:
Optional cleanup: this test still passes with the `break` removed (so with
main's I/O pattern), because case 1 also reads data pages and case 2 doesn't,
so `requests2 < requests1` holds either way. The `bloom_filter_io.rs` snapshot
already shows the skipped request, and
`incremental_bloom_filter_pruning_short_circuits_remaining_columns` does fail
without the `break`. Could you drop this test along with `RangeTracker` and
`TrackingBloomAsyncReader`/`TrackingBloomReaderFactory` (about 145 lines)?
##########
datafusion/datasource-parquet/src/row_group_filter.rs:
##########
@@ -493,12 +496,83 @@ impl RowGroupAccessPlanFilter {
if prune_group {
metrics.row_groups_pruned_bloom_filter.add_pruned(1);
- self.access_plan.skip(idx)
+ self.access_plan.skip(idx);
} else {
metrics.row_groups_pruned_bloom_filter.add_matched(1);
}
}
}
+
+ /// Extracts column guarantees from a [`PruningPredicate`] that can be
evaluated against
+ /// Parquet Bloom filters.
+ pub fn prepare_bloom_column_guarantees(
Review Comment:
Optional cleanup: these three methods are `pub` on the exported
`RowGroupAccessPlanFilter`, but `BloomColumnGuarantee` isn't re-exported, so
downstream can call `prepare_bloom_column_guarantees` yet can't name its return
type. Only the opener uses them, so `pub(crate)` on the three methods, the
struct and its fields, and `BloomFilterStatistics::is_known_not_present` keeps
them out of the public API (clippy and the crate tests pass with that change).
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -1517,38 +1486,27 @@ impl RowGroupsPrunedParquetOpen {
mem::replace(&mut prepared.async_file_reader,
replacement_reader),
reader_metadata,
);
- let parquet_columns: Vec<(String, usize, Type, i32)> = predicate
- .literal_columns()
- .into_iter()
- .filter_map(|column_name| {
- let parquet_schema = builder.parquet_schema();
- let (column_idx, _) = parquet_column(
- parquet_schema,
- &prepared.physical_file_schema,
- &column_name,
- )?;
- Some((
- column_name,
- column_idx,
- parquet_schema.column(column_idx).physical_type(),
- parquet_schema.column(column_idx).type_length(),
- ))
- })
- .collect();
- for idx in self.row_groups.row_group_indexes() {
- // Statistics have already proved that every row in this group
- // matches the predicate, so its Bloom filters cannot prune it.
- if self.row_groups.access_plan().is_fully_matched(idx) {
+ let parquet_columns =
+ RowGroupAccessPlanFilter::prepare_bloom_column_guarantees(
+ predicate,
+ builder.parquet_schema(),
+ &prepared.physical_file_schema,
+ );
+
+ let row_group_indexes: Vec<usize> =
+ self.row_groups.row_group_indexes().collect();
+ for idx in row_group_indexes {
+ if !self.row_groups.access_plan().should_scan(idx)
+ || self.row_groups.access_plan().is_fully_matched(idx)
+ {
Review Comment:
Optional cleanup: `row_group_indexes()` only yields row groups that are
still scanned, and the loop only ever skips the current `idx`, so this
`should_scan` check can't be true (removing it keeps all tests green).
```suggestion
if self.row_groups.access_plan().is_fully_matched(idx) {
```
--
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]
