You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/22 21:22:15 UTC
(datafusion) branch gh-readonly-queue/main/pr-26476-d090d2948824e3ee1c283f3fdd21e9af31b3e793 created (now c0ec5127ac)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26476-d090d2948824e3ee1c283f3fdd21e9af31b3e793
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at c0ec5127ac perf: don't build a parquet stream for a file range with no row group left (#26476)
This branch includes the following new commits:
new c0ec5127ac perf: don't build a parquet stream for a file range with no row group left (#26476)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: perf: don't build a parquet stream for a file range with no row group left (#26476)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26476-d090d2948824e3ee1c283f3fdd21e9af31b3e793
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit c0ec5127ac99be38e10c1d10d4a9cd4060b71dde
Author: Braedon Wooding <br...@displayr.com>
AuthorDate: Thu Oct 22 19:45:20 2026 +0000
perf: don't build a parquet stream for a file range with no row group left (#26476)
I've got a few fixes to upstream so just starting with small ones.
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #26475
## Rationale for this change
If row-group statistics or Bloom filters prune every row group in a file
range, the opener still builds the decoder read plans, the decoder and
the row filter, then hands back a stream morsel that yields nothing.
`PruneFile` already ends early when file statistics prune the range;
this does the same once row-group pruning has.
## What changes are included in this PR?
`PruneWithStatistics` and `PruneWithBloomFilters` return
`ParquetOpenState::Done` when no row group survives.
## What is the testing strategy for this PR?
Testing to ensure that we do actually skip them!
## Are there any user-facing changes?
Nope!
---------
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
Co-authored-by: Claude Fable 5.1 <no...@anthropic.com>
---
datafusion/datasource-parquet/src/opener/mod.rs | 74 +++++++++++++++++++++-
.../datasource-parquet/src/row_group_filter.rs | 25 ++++++++
2 files changed, 96 insertions(+), 3 deletions(-)
diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs
index b048a23eea..3a1766869d 100644
--- a/datafusion/datasource-parquet/src/opener/mod.rs
+++ b/datafusion/datasource-parquet/src/opener/mod.rs
@@ -660,6 +660,9 @@ impl ParquetOpenState {
}
ParquetOpenState::PruneWithStatistics(prepared) => {
let mut prepared_row_groups = (*prepared).prune_row_groups()?;
+ if prepared_row_groups.row_groups.is_fully_pruned() {
+ return Ok(ParquetOpenState::Done);
+ }
if prepared_row_groups.should_load_page_index()? {
Ok(ParquetOpenState::LoadPageIndex(
prepared_row_groups.load_page_index().boxed(),
@@ -690,9 +693,13 @@ impl ParquetOpenState {
ParquetOpenState::LoadBloomFilters(future) => {
Ok(ParquetOpenState::LoadBloomFilters(future))
}
- ParquetOpenState::PruneWithBloomFilters(loaded) => Ok(
- ParquetOpenState::BuildStream(Box::new(loaded.prune_bloom_filters())),
- ),
+ ParquetOpenState::PruneWithBloomFilters(loaded) => {
+ let pruned = loaded.prune_bloom_filters();
+ if pruned.row_groups.is_fully_pruned() {
+ return Ok(ParquetOpenState::Done);
+ }
+ Ok(ParquetOpenState::BuildStream(Box::new(pruned)))
+ }
ParquetOpenState::BuildStream(prepared) => {
Ok(ParquetOpenState::Ready(prepared.build_stream()?))
}
@@ -2690,6 +2697,26 @@ mod test {
}
}
+ /// Drives a [`ParquetMorselizer`] to completion and returns how many
+ /// morsels it produced.
+ async fn count_morsels(
+ morselizer: &dyn Morselizer,
+ file: PartitionedFile,
+ ) -> Result<usize> {
+ let mut planners = VecDeque::from([morselizer.plan_file(file)?]);
+ let mut morsels = 0;
+ while let Some(planner) = planners.pop_front() {
+ if let Some(mut plan) = planner.plan()? {
+ morsels += plan.take_morsels().len();
+ planners.extend(plan.take_ready_planners());
+ if let Some(pending_planner) = plan.take_pending_planner() {
+ planners.push_front(pending_planner.await?);
+ }
+ }
+ }
+ Ok(morsels)
+ }
+
fn constant_int_stats() -> (Statistics, SchemaRef) {
let schema = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, false),
@@ -3234,6 +3261,47 @@ mod test {
assert_eq!(num_rows, 0);
}
+ #[tokio::test]
+ async fn test_no_morsel_when_every_row_group_is_pruned() {
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batches = vec![
+ record_batch!(("a", Int32, vec![0, 2, 4])).unwrap(),
+ record_batch!(("a", Int32, vec![0, 2, 4])).unwrap(),
+ ];
+ let schema = batches[0].schema();
+ let props = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(3))
+ .set_bloom_filter_enabled(true)
+ .set_statistics_enabled(EnabledStatistics::Chunk)
+ .build();
+ let data_size = write_parquet_batches(
+ Arc::clone(&store),
+ "test.parquet",
+ batches,
+ Some(props),
+ )
+ .await;
+ let file = PartitionedFile::new("test.parquet".to_string(), data_size as u64);
+
+ let morsels = |expr: Expr| {
+ let morselizer = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_schema(Arc::clone(&schema))
+ .with_predicate(logical2physical(&expr, &schema))
+ .with_row_group_stats_pruning(true)
+ .with_enable_bloom_filter(true)
+ .build();
+ let file = file.clone();
+ async move { count_morsels(&morselizer, file).await.unwrap() }
+ };
+
+ // Statistics prune both row groups.
+ assert_eq!(morsels(col("a").gt(lit(10))).await, 0);
+ // Statistics keep both row groups; their Bloom filters prune both.
+ assert_eq!(morsels(col("a").eq(lit(1))).await, 0);
+ assert_eq!(morsels(col("a").eq(lit(2))).await, 1);
+ }
+
#[tokio::test]
async fn test_prune_on_partition_statistics_with_dynamic_expression() {
let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs
index 02fc2700d1..feba768955 100644
--- a/datafusion/datasource-parquet/src/row_group_filter.rs
+++ b/datafusion/datasource-parquet/src/row_group_filter.rs
@@ -99,6 +99,12 @@ impl RowGroupAccessPlanFilter {
self.access_plan.is_empty()
}
+ /// Return true if no row groups remain to be scanned, either because the
+ /// file has none or because pruning has skipped every one of them.
+ pub fn is_fully_pruned(&self) -> bool {
+ self.access_plan.row_group_index_iter().next().is_none()
+ }
+
/// Return the number of row groups that are currently expected to be scanned
pub fn remaining_row_group_count(&self) -> usize {
self.access_plan.row_group_index_iter().count()
@@ -706,6 +712,25 @@ mod tests {
assert_eq!(filter.remaining_row_group_count(), 2);
}
+ #[test]
+ fn is_fully_pruned_only_when_every_row_group_is_skipped() {
+ let mut filter = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(2));
+ assert!(!filter.is_fully_pruned());
+
+ filter.access_plan.skip(0);
+ assert!(!filter.is_fully_pruned());
+
+ filter.access_plan.skip(1);
+ assert!(filter.is_fully_pruned());
+ // Skipped row groups are still part of the plan
+ assert!(!filter.is_empty());
+
+ assert!(
+ RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(0))
+ .is_fully_pruned()
+ );
+ }
+
/// A row group with one column chunk of `size` bytes at the given offsets
fn row_group_at(
dictionary_page_offset: Option<i64>,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org