zhuqi-lucas commented on code in PR #23696:
URL: https://github.com/apache/datafusion/pull/23696#discussion_r3764787778
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -337,11 +423,36 @@ impl PushDecoderStreamState {
// been handed back yet), step 3 drives it forward and we get
// another chance at the next boundary — the pruner is stateful
// and idempotent, so deferring loses nothing.
- let at_boundary = self
- .decoder
- .as_ref()
- .expect("decoder present")
- .is_at_row_group_boundary();
+ let decoder_ref = self.decoder.as_ref().expect("decoder present");
+ let at_boundary = decoder_ref.is_at_row_group_boundary();
+ // Sync `rg_plan` with the row group the decoder will actually
+ // emit next. arrow-rs's `try_next_reader` silently advances
+ // past row groups whose row selection is empty (e.g. when
+ // page-index pruning has already eliminated every page of
+ // that RG via the `ColumnIndex` path inside `try_build`).
+ // Without this peek, `rg_plan.front()` would drift off-by-one
+ // from the decoder's frontier and the per-RG toggle below
+ // would target the wrong row group.
+ if at_boundary {
+ match decoder_ref.peek_next_row_group() {
+ Ok(Some(actual)) => {
+ while let Some(front) = self.rg_plan.front() {
Review Comment:
Done — extracted into `advance_rg_plan_to()` (e0fbd24).
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -337,11 +423,36 @@ impl PushDecoderStreamState {
// been handed back yet), step 3 drives it forward and we get
// another chance at the next boundary — the pruner is stateful
// and idempotent, so deferring loses nothing.
- let at_boundary = self
- .decoder
- .as_ref()
- .expect("decoder present")
- .is_at_row_group_boundary();
+ let decoder_ref = self.decoder.as_ref().expect("decoder present");
+ let at_boundary = decoder_ref.is_at_row_group_boundary();
Review Comment:
Done — moved into `sync_rg_plan_to_decoder_frontier` +
`prune_boundary_row_groups` (e0fbd24).
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -356,15 +467,70 @@ impl PushDecoderStreamState {
}
self.rg_plan = kept;
}
- if pruned_count > 0 {
+
+ // Decide whether the per-row `RowFilter` needs to be
+ // toggled for the upcoming RG. `desired_filter` is
+ // `Some(true)` when the next RG needs a real filter,
+ // `Some(false)` when it's 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 {
if self.rg_plan.is_empty() {
return None;
}
let decoder = self.decoder.take().expect("decoder
present");
let new_indices: Vec<usize> =
self.rg_plan.iter().map(|e| e.rg_index).collect();
let rebuilt = match decoder.into_builder() {
- Ok(b) => b.with_row_groups(new_indices).build(),
+ Ok(mut builder) => {
+ builder = builder.with_row_groups(new_indices);
Review Comment:
Pulled the advance/prune helpers out (e0fbd24). Left the rebuild+toggle
block inline for now (early returns + multi-field borrows) — can extract it too
as a follow-up if you prefer.
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -337,11 +423,36 @@ impl PushDecoderStreamState {
// been handed back yet), step 3 drives it forward and we get
// another chance at the next boundary — the pruner is stateful
// and idempotent, so deferring loses nothing.
- let at_boundary = self
- .decoder
- .as_ref()
- .expect("decoder present")
- .is_at_row_group_boundary();
+ let decoder_ref = self.decoder.as_ref().expect("decoder present");
+ let at_boundary = decoder_ref.is_at_row_group_boundary();
+ // Sync `rg_plan` with the row group the decoder will actually
+ // emit next. arrow-rs's `try_next_reader` silently advances
+ // past row groups whose row selection is empty (e.g. when
+ // page-index pruning has already eliminated every page of
Review Comment:
Trimmed to the point (e0fbd24).
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -272,6 +282,82 @@ pub(crate) struct PushDecoderStreamState {
pub(crate) row_group_pruner: Option<RowGroupPruner>,
/// Count of row groups skipped at runtime by [`Self::row_group_pruner`].
pub(crate) row_groups_pruned_dynamic: Count,
+ /// Side-channel state for regenerating the parquet [`RowFilter`] when
+ /// the per-RG `fully_matched` toggle flips from skip → install. `None`
+ /// when the scan has no pushdown predicate, so no filter can be
+ /// installed (and the toggle is a no-op).
+ pub(crate) row_filter_context: Option<RowFilterContext>,
+ /// Whether the currently-installed decoder is running with a non-empty
+ /// row filter. Toggled per RG by the `fully_matched` skip path.
+ pub(crate) filter_installed: bool,
+ /// Count of row groups for which the per-row [`RowFilter`] was
+ /// suppressed because the upcoming RG is `fully_matched`.
+ pub(crate) row_filter_skipped_fully_matched: Count,
+}
+
+/// Side-channel state that lets [`PushDecoderStreamState`] **rebuild** the
+/// parquet [`RowFilter`] mid-scan.
+///
+/// The decoder owns the filter once installed, but `Box<dyn ArrowPredicate>`
+/// has no clone path, so a filter that was replaced at a previous boundary
+/// cannot be reinstalled later. This struct keeps a pre-built candidate list
+/// alongside the stream so the next non-fully-matched row group can be
+/// wrapped into a fresh [`RowFilter`] without redoing the tree walks and
+/// column resolution that the initial build did.
+pub(crate) struct RowFilterContext {
+ /// Prebuilt candidates: expression already column-reassigned, projection
+ /// mask already resolved. Shared across the file's row groups. `Arc` so
+ /// cloning into stream state is cheap.
+ pub(crate) prebuilt: Arc<Vec<PrebuiltRowFilterCandidate>>,
Review Comment:
Done — added `PrebuiltRowFilterCandidateList` (e0fbd24).
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -272,6 +282,82 @@ pub(crate) struct PushDecoderStreamState {
pub(crate) row_group_pruner: Option<RowGroupPruner>,
/// Count of row groups skipped at runtime by [`Self::row_group_pruner`].
pub(crate) row_groups_pruned_dynamic: Count,
+ /// Side-channel state for regenerating the parquet [`RowFilter`] when
+ /// the per-RG `fully_matched` toggle flips from skip → install. `None`
+ /// when the scan has no pushdown predicate, so no filter can be
+ /// installed (and the toggle is a no-op).
+ pub(crate) row_filter_context: Option<RowFilterContext>,
+ /// Whether the currently-installed decoder is running with a non-empty
+ /// row filter. Toggled per RG by the `fully_matched` skip path.
+ pub(crate) filter_installed: bool,
+ /// Count of row groups for which the per-row [`RowFilter`] was
+ /// suppressed because the upcoming RG is `fully_matched`.
+ pub(crate) row_filter_skipped_fully_matched: Count,
+}
+
+/// Side-channel state that lets [`PushDecoderStreamState`] **rebuild** the
Review Comment:
Renamed to a cache (e0fbd24).
--
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]
