alamb commented on code in PR #23696:
URL: https://github.com/apache/datafusion/pull/23696#discussion_r3761788165
##########
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:
Can this be handled at a higher layer or encapsulated somehow? Like in a
method? The state machine is already complicated enough and so moving this plan
manipulation into methods will help hopefully
##########
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:
same thing here -- can we somehow encapsulate this logic into a method or
function to keep the lengh of the state machine down?
##########
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:
maybe this would be a good method on rg_plan -- like
`rg_plan.advance_to_index`(actual)` or something 🤔
##########
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:
I think this comment could be shortened significantly -- the point is that
try_next_reader can silently advance, so we have to advance the plan
accordingly. The details about ColumnIndex and rg_pla.front() seems overly
focused on the details
##########
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:
why is it a "side channel"? It seems like it is more like a "cache"
##########
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>`
Review Comment:
"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."
seems like an implementation detail -- keeping the pre-built filters seems
like the key point
##########
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:
Another thought I had was make this a strucgture of
```rust
pub(crate) prebuilt: PrebuiltRowFilterCandidateList,
```
with
```rust
struct PrebuiltRowFilterCandidateList {
inner : Arc<Vec<PrebuiltRowFilterCandidate>>,
}
```
Again as a way to try and encapsualte the complexity of this optimization
more
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -109,6 +113,12 @@ impl DecoderBuilderConfig<'_> {
#[derive(Debug, Clone)]
pub(crate) struct RgPlanEntry {
pub(crate) rg_index: usize,
+ /// `true` when the static pruning predicate proved every row of this
Review Comment:
I think the first sentence here would be eough -- the rest is details of the
implementation that I don't think adds mich here. The point is that we can skip
row filtering when we know it won't filter anything
--
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]
