adriangb commented on code in PR #23696:
URL: https://github.com/apache/datafusion/pull/23696#discussion_r3837348721
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -519,6 +605,95 @@ impl PushDecoderStreamState {
)
}
+ /// Drop every `rg_plan` entry the dynamic pruner proves cannot contribute,
+ /// returning how many were pruned. The single decoder rebuild that acts on
+ /// the survivors is left to the caller (at most one rebuild per boundary).
+ fn prune_boundary_row_groups(&mut self) -> usize {
+ let Some(pruner) = self.row_group_pruner.as_mut() else {
+ return 0;
+ };
+ let mut pruned_count = 0usize;
+ let mut kept = VecDeque::with_capacity(self.rg_plan.len());
+ while let Some(entry) = self.rg_plan.pop_front() {
+ if pruner.should_prune(&[entry.rg_index]) {
+ pruned_count += 1;
+ self.row_groups_pruned_dynamic.add(1);
+ // The scan is done with this row group's bytes.
+ self.byte_progress.credit(entry.bytes);
+ } else {
+ kept.push_back(entry);
+ }
+ }
+ self.rg_plan = kept;
+ pruned_count
+ }
+
+ /// At a row-group boundary, rebuild the decoder so it reads only the
+ /// surviving `rg_plan` and toggle the per-row `RowFilter` for the upcoming
+ /// RG. Rebuilds only when something changed (`pruned_count > 0` or the
+ /// filter status flips), doing at most one `into_builder` rebuild per
+ /// boundary. Returns `Ok(true)` when the plan is now empty (the stream
+ /// should finish).
+ fn rebuild_decoder_at_boundary(
+ &mut self,
+ pruned_count: usize,
+ ) -> Result<bool, DataFusionError> {
+ // `desired_filter` is `Some(true)` when the next RG needs a real
+ // filter, `Some(false)` when it is fully-matched (filter is a no-op,
so
+ // we suppress it), and `None` when there is no pushdown predicate at
+ // all (toggling is meaningless).
+ let desired_filter: Option<bool> = self
+ .row_filter_context
+ .as_ref()
+ .and_then(|_| self.rg_plan.front().map(|e| !e.fully_matched));
+ let filter_needs_toggle =
+ desired_filter.is_some_and(|want| want != self.filter_installed);
+
+ if pruned_count == 0 && !filter_needs_toggle {
+ return Ok(false);
+ }
+ if self.rg_plan.is_empty() {
+ return Ok(true);
+ }
+
+ let decoder = self.decoder.take().expect("decoder present");
+ let new_indices: Vec<usize> = self.rg_plan.iter().map(|e|
e.rg_index).collect();
+ let mut builder =
decoder.into_builder().map_err(DataFusionError::from)?;
+ builder = builder.with_row_groups(new_indices);
+ if filter_needs_toggle {
+ let want_filter = desired_filter.expect("filter_needs_toggle ⇒
desired Some");
+ if want_filter {
+ let ctx = self
+ .row_filter_context
+ .as_ref()
+ .expect("filter_needs_toggle ⇒ context set");
+ match ctx.build_row_filter() {
+ Some(filter) => {
+ builder = builder.with_row_filter(filter);
+ if let Some(cap) = ctx.max_predicate_cache_size {
+ builder =
builder.with_max_predicate_cache_size(cap);
+ }
+ self.filter_installed = true;
+ }
+ None => {
+ // Filter could not be rebuilt; install an empty filter
+ // so the decoder runs unfiltered for this run rather
+ // than failing.
+ builder =
builder.with_row_filter(RowFilter::new(vec![]));
+ self.filter_installed = false;
+ }
Review Comment:
Could we make `build_row_filter` return a `RowFilter` instead of an `Option`
when `self.prebuilt.is_empty()` and make this arm unreacheable?
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -1504,31 +1522,72 @@ impl RowGroupsPrunedParquetOpen {
// https://github.com/apache/arrow-rs/issues/10624 /
// https://github.com/apache/datafusion/issues/24358.
let has_row_selection =
prepared_access_plan.row_selection.is_some();
+ // Build `rg_plan` parallel to the decoder's view: the
+ // `prepared_access_plan` has already had its empty-selection
+ // row groups stripped, so 1:1 correspondence with the readers
+ // arrow-rs will hand back is restored. We zip with the
+ // `fully_matched` flag so the stream can toggle the per-row
+ // `RowFilter` per RG.
let rg_plan: VecDeque<RgPlanEntry> = prepared_access_plan
.row_group_indexes
.iter()
.copied()
- .map(|rg_index| RgPlanEntry {
+ .zip(prepared_access_plan.fully_matched.iter().copied())
+ .map(|(rg_index, fully_matched)| RgPlanEntry {
rg_index,
+ fully_matched,
bytes: row_group_bytes(&rg_metadata[rg_index]),
})
.collect();
+ // Decide the initial row filter state based on the first RG to
+ // read. If that RG is `fully_matched` the per-row predicate is
+ // a no-op for every row, so we install an empty `RowFilter`
+ // (arrow-rs's `has_predicates` check then short-circuits the
+ // per-row eval) and the stream toggles back to the real filter
+ // at the first non-fully-matched RG boundary.
+ //
+ // `RowFilterContext` carries everything `build_row_filter`
+ // needs so the stream can regenerate the filter later — the
+ // installed filter is owned by the decoder and is not
+ // recoverable once replaced.
+ let first_rg_fully_matched = rg_plan.front().is_some_and(|e|
e.fully_matched);
+ let initial_filter = precomputed_context
+ .as_ref()
+ .and_then(|ctx| ctx.build_row_filter());
+ let row_filter_context = precomputed_context;
Review Comment:
This is discarded if the first row group is fully matched - can we construct
it lazily?
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -401,16 +406,65 @@ pub fn build_row_filter(
reorder_predicates: bool,
file_metrics: &ParquetFileMetrics,
) -> Result<Option<RowFilter>> {
- let rows_pruned = &file_metrics.pushdown_rows_pruned;
- let rows_matched = &file_metrics.pushdown_rows_matched;
- let time = &file_metrics.row_pushdown_eval_time;
+ // Implemented on top of the prebuild split so there is a single place
+ // that splits conjuncts, orders candidates, and wires metrics — callers
+ // that build once per file go through the same code as the per-row-group
+ // rebuild path in `RowFilterContext`.
+ let Some(prebuilt) = prebuild_row_filter_candidates(expr, file_schema,
metadata)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(row_filter_from_prebuilt(
+ &prebuilt,
+ reorder_predicates,
+ file_metrics,
+ )))
+}
+
+/// A precomputed [`FilterCandidate`] with its expression column-reassigned to
+/// the projected schema, ready to be wrapped into a
[`DatafusionArrowPredicate`]
+/// on demand.
+///
+/// Extracting this from [`build_row_filter`] lets callers pay the tree-walk +
+/// column-resolution + `reassign_expr_columns` cost **once per file** instead
+/// of once per row group, which is the hot path for
+/// [`RowFilterContext::build`](crate::push_decoder::RowFilterContext) rebuilds
Review Comment:
I think the method is `build_row_filter` not `build`
--
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]
