fix(reader): honour bloom_filter_on_read

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::<usize>(), 1);
+        let rows = batches.iter().map(RecordBatch::num_rows).sum::<usize>();
 
         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

src/datafusion/src/reader/plantime/morselizer.rs

@@ -67,6 +67,7 @@ pub(crate) struct LiquidMorselizer {
     pub(crate) parquet_file_reader_factory: Arc<dyn ParquetFileReaderFactory>,
     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<dyn PhysicalExprAdapterFactory>,
     pub(crate) span: Option<Arc<fastrace::Span>>,
@@ -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<Arc<dyn PhysicalExpr>>,
     reorder_filters: bool,
+    enable_bloom_filter: bool,
     liquid_cache: LiquidCacheParquetRef,
     expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,
     file_identity: ParquetFileIdentity,
@@ -468,7 +471,10 @@ fn prepare_and_prune_by_stats(mut loaded: MetadataLoadedLiquidOpen) -> Result<Li
         },
         row_groups,
     };
-    if prepared.context.pruning_predicate.is_some() && !prepared.row_groups.is_empty() {
+    let read_bloom_filters = prepared.context.pruning_predicate.is_some()
+        && prepared.context.prepared.enable_bloom_filter
+        && !prepared.row_groups.is_empty();
+    if read_bloom_filters {
         Ok(LiquidOpenState::LoadBloomFilters(
             async move {
                 let mut prepared = prepared;
@@ -494,6 +500,17 @@ fn prepare_and_prune_by_stats(mut loaded: MetadataLoadedLiquidOpen) -> Result<Li
             .boxed(),
         ))
     } else {
+        if prepared.context.pruning_predicate.is_some() {
+            // Bloom filters are not consulted, so every remaining row group
+            // counts as matched by them, as in DataFusion's own opener.
+            prepared
+                .context
+                .prepared
+                .metrics
+                .file_metrics
+                .row_groups_pruned_bloom_filter
+                .add_matched(prepared.row_groups.remaining_row_group_count());
+        }
         Ok(LiquidOpenState::PlanRowGroups(Box::new(prune_pages(
             prepared,
         ))))
@@ -1135,6 +1152,7 @@ mod tests {
             )),
             object_store_url: object_store_url.clone(),
             reorder_filters: false,
+            enable_bloom_filter: true,
             liquid_cache: cache.clone(),
             expr_adapter_factory: Arc::new(DefaultPhysicalExprAdapterFactory),
             span: None,
@@ -1277,6 +1295,7 @@ mod tests {
             ))),
             object_store_url: ObjectStoreUrl::parse("data-cache-a:///").unwrap(),
             reorder_filters: false,
+            enable_bloom_filter: true,
             liquid_cache: Arc::clone(&cache),
             expr_adapter_factory: Arc::new(DefaultPhysicalExprAdapterFactory),
             span: None,
@@ -1295,6 +1314,7 @@ mod tests {
             ))),
             object_store_url: ObjectStoreUrl::parse("data-cache-b:///").unwrap(),
             reorder_filters: false,
+            enable_bloom_filter: true,
             liquid_cache: cache,
             expr_adapter_factory: Arc::new(DefaultPhysicalExprAdapterFactory),
             span: None,

src/datafusion/src/reader/plantime/source.rs

@@ -48,6 +48,10 @@ impl LiquidParquetSource {
         self.table_parquet_options.global.reorder_filters
     }
 
+    fn bloom_filter_on_read(&self) -> 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),