From a4002e0912832739708d8d209d209530f2867544 Mon Sep 17 00:00:00 2001 From: Anoop Narang Date: Tue, 6 Oct 2026 23:36:57 +0530 Subject: [PATCH] fix(reader): honour bloom_filter_on_read The morsel reader loaded every row group's bloom filters whenever the scan had a pruning predicate, ignoring the session's `datafusion.execution.parquet.bloom_filter_on_read`. DataFusion's own parquet opener skips them when the option is off, so a caller that turns it off to save the per-file bloom reads got them anyway once the scan went through liquid cache. Carry the option from the source's TableParquetOptions into the morselizer, skip the bloom-filter load when it is off, and count the remaining row groups as matched by bloom filters, as DataFusion does. --- src/datafusion/src/optimizers/mod.rs | 37 ++++++++++++++++--- .../src/reader/plantime/morselizer.rs | 22 ++++++++++- src/datafusion/src/reader/plantime/source.rs | 5 +++ 3 files changed, 58 insertions(+), 6 deletions(-) diff --git a/src/datafusion/src/optimizers/mod.rs b/src/datafusion/src/optimizers/mod.rs index d8950b11..f6c47068 100644 --- a/src/datafusion/src/optimizers/mod.rs +++ b/src/datafusion/src/optimizers/mod.rs @@ -230,7 +230,7 @@ mod tests { physical_plan::{ PhysicalExpr, collect, display::DisplayableExecutionPlan, filter_pushdown::PushedDown, }, - prelude::SessionContext, + prelude::{SessionConfig, SessionContext}, }; use liquid_cache::{ cache::{AlwaysHydrate, TranscodeEvict}, @@ -391,13 +391,21 @@ mod tests { writer.close().unwrap(); } - #[tokio::test] - async fn prunes_row_group_with_bloom_filter() { + /// Scan `a = 2` over a file whose two row groups both span `1..=4`, so + /// statistics keep both and only the bloom filter can rule one out. Returns + /// the rows produced and the bloom-filter pruning metric. + async fn scan_bloom_file( + bloom_filter_on_read: bool, + ) -> (usize, datafusion::physical_plan::metrics::PruningMetrics) { let tmp_dir = tempfile::tempdir().unwrap(); let parquet_path = tmp_dir.path().join("bloom.parquet"); write_bloom_file(&parquet_path); - let ctx = SessionContext::new(); + let config = SessionConfig::new().set_bool( + "datafusion.execution.parquet.bloom_filter_on_read", + bloom_filter_on_read, + ); + let ctx = SessionContext::new_with_config(config); ctx.register_parquet("t", parquet_path.to_str().unwrap(), Default::default()) .await .unwrap(); @@ -413,7 +421,7 @@ mod tests { let metrics = liquid_source(&rewritten).metrics().clone(); let batches = collect(rewritten, ctx.task_ctx()).await.unwrap(); - assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 1); + let rows = batches.iter().map(RecordBatch::num_rows).sum::(); let metric = metrics .clone_inner() @@ -425,7 +433,26 @@ mod tests { else { panic!("unexpected metric: {metric:?}"); }; + (rows, pruning_metrics) + } + + #[tokio::test] + async fn prunes_row_group_with_bloom_filter() { + let (rows, pruning_metrics) = scan_bloom_file(true).await; + assert_eq!(rows, 1); assert_eq!(pruning_metrics.pruned(), 1); + assert_eq!(pruning_metrics.matched(), 1); + } + + /// `bloom_filter_on_read = false` must keep the reader off the bloom + /// filters entirely: the row group the filter would have pruned is read, + /// and the predicate alone still produces the right answer. + #[tokio::test] + async fn skips_bloom_filter_when_bloom_filter_on_read_is_off() { + let (rows, pruning_metrics) = scan_bloom_file(false).await; + assert_eq!(rows, 1); + assert_eq!(pruning_metrics.pruned(), 0); + assert_eq!(pruning_metrics.matched(), 2); } /// Declining a scan costs it the cache, so the guard must key on what the scan diff --git a/src/datafusion/src/reader/plantime/morselizer.rs b/src/datafusion/src/reader/plantime/morselizer.rs index a8ff2b80..7e75f1b0 100644 --- a/src/datafusion/src/reader/plantime/morselizer.rs +++ b/src/datafusion/src/reader/plantime/morselizer.rs @@ -67,6 +67,7 @@ pub(crate) struct LiquidMorselizer { pub(crate) parquet_file_reader_factory: Arc, pub(crate) object_store_url: ObjectStoreUrl, pub(crate) reorder_filters: bool, + pub(crate) enable_bloom_filter: bool, pub(crate) liquid_cache: LiquidCacheParquetRef, pub(crate) expr_adapter_factory: Arc, pub(crate) span: Option>, @@ -188,6 +189,7 @@ impl Morselizer for LiquidMorselizer { projection, predicate, reorder_filters: self.reorder_filters, + enable_bloom_filter: self.enable_bloom_filter, liquid_cache: self.liquid_cache.clone(), expr_adapter_factory: Arc::clone(&self.expr_adapter_factory), file_identity, @@ -238,6 +240,7 @@ struct PreparedLiquidOpen { projection: ProjectionExprs, predicate: Option>, reorder_filters: bool, + enable_bloom_filter: bool, liquid_cache: LiquidCacheParquetRef, expr_adapter_factory: Arc, file_identity: ParquetFileIdentity, @@ -468,7 +471,10 @@ fn prepare_and_prune_by_stats(mut loaded: MetadataLoadedLiquidOpen) -> Result
  • Result
  • bool { + self.table_parquet_options.global.bloom_filter_on_read + } + /// Set the span for the LiquidParquetSource pub fn with_span(&self, span: fastrace::Span) -> Self { Self { @@ -175,6 +179,7 @@ impl FileSource for LiquidParquetSource { parquet_file_reader_factory, object_store_url: base_config.object_store_url.clone(), reorder_filters: self.reorder_filters(), + enable_bloom_filter: self.bloom_filter_on_read(), expr_adapter_factory, span: execution_span.map(Arc::new), lineages: Arc::clone(&self.lineages),