[datafusion] Support runtime filters for Paimon scans (#549)

Author: JingsongLi

benchmarks/tpcds/README.md

@@ -164,6 +164,7 @@ target/release/paimon-tpcds-bench run \
   --warmup 1 \
   --iterations 3 \
   --target-partitions 64 \
+  --parquet-pushdown-filters \
   --memory-limit-gib 192 \
   --spill-dir /nvme/datafusion-spill \
   --max-spill-gib 1024
@@ -173,6 +174,10 @@ This is an end-to-end source comparison. Loading the data into Paimon rewrites
 the physical files, so it is not a pure measurement of catalog or manifest
 overhead.
 
+`--parquet-pushdown-filters` only controls DataFusion's Parquet reader. Paimon
+always receives supported predicates for conservative pruning, while exact row
+filtering stays in the parent DataFusion operator for benchmark runs.
+
 ## Cache Protocol
 
 Run and label cold and warm experiments separately:

benchmarks/tpcds/src/cli.rs

@@ -113,6 +113,9 @@ pub struct RuntimeArgs {
     /// DataFusion execution partitions. Defaults to available CPUs.
     #[arg(long)]
     pub target_partitions: Option<usize>,
+    /// Evaluate pushed filters during Parquet scans, in addition to statistics pruning.
+    #[arg(long)]
+    pub parquet_pushdown_filters: bool,
     /// DataFusion memory limit in GiB. Omit for an unbounded pool.
     #[arg(long)]
     pub memory_limit_gib: Option<u64>,
@@ -132,6 +135,7 @@ impl RuntimeArgs {
                 .target_partitions
                 .unwrap_or(defaults.target_partitions)
                 .max(1),
+            parquet_pushdown_filters: self.parquet_pushdown_filters,
             memory_limit_bytes: self.memory_limit_gib.map(gib_to_usize).transpose()?,
             spill_dir: self.spill_dir.clone(),
             max_spill_bytes: self.max_spill_gib.map(gib_to_u64).transpose()?,

benchmarks/tpcds/src/context.rs

@@ -28,6 +28,8 @@ use serde::{Deserialize, Serialize};
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BenchmarkRuntimeConfig {
     pub target_partitions: usize,
+    #[serde(default)]
+    pub parquet_pushdown_filters: bool,
     pub memory_limit_bytes: Option<usize>,
     pub spill_dir: Option<PathBuf>,
     pub max_spill_bytes: Option<u64>,
@@ -39,6 +41,7 @@ impl Default for BenchmarkRuntimeConfig {
             target_partitions: std::thread::available_parallelism()
                 .map(usize::from)
                 .unwrap_or(1),
+            parquet_pushdown_filters: false,
             memory_limit_bytes: None,
             spill_dir: None,
             max_spill_bytes: None,
@@ -60,10 +63,15 @@ pub fn build_sql_context(config: &BenchmarkRuntimeConfig) -> DataFusionResult<SQ
     let sql = SQLContext::new();
     let state_ref = sql.ctx().state_ref();
     let current_state = state_ref.read().clone();
-    let session_config = current_state
+    let mut session_config = current_state
         .config()
         .clone()
         .with_target_partitions(config.target_partitions.max(1));
+    session_config
+        .options_mut()
+        .execution
+        .parquet
+        .pushdown_filters = config.parquet_pushdown_filters;
     let state = SessionStateBuilder::from(current_state)
         .with_config(session_config)
         .with_runtime_env(Arc::new(runtime.build()?))

benchmarks/tpcds/src/lib.rs

@@ -183,6 +183,7 @@ mod tests {
         let spill_dir = TempDir::new().unwrap();
         let ctx = build_sql_context(&BenchmarkRuntimeConfig {
             target_partitions: 3,
+            parquet_pushdown_filters: true,
             memory_limit_bytes: Some(32 * 1024 * 1024),
             spill_dir: Some(spill_dir.path().to_path_buf()),
             max_spill_bytes: Some(64 * 1024 * 1024),
@@ -197,6 +198,14 @@ mod tests {
                 .target_partitions,
             3
         );
+        assert!(
+            ctx.ctx()
+                .state()
+                .config_options()
+                .execution
+                .parquet
+                .pushdown_filters
+        );
         assert!(matches!(
             ctx.ctx().runtime_env().memory_pool.memory_limit(),
             MemoryLimit::Finite(size) if size == 32 * 1024 * 1024

benchmarks/tpcds/tests/smoke.rs

@@ -225,6 +225,7 @@ async fn command_orchestration_loads_and_writes_a_run_report() {
     .unwrap();
     let runtime = RuntimeArgs {
         target_partitions: Some(2),
+        parquet_pushdown_filters: false,
         memory_limit_gib: None,
         spill_dir: None,
         max_spill_gib: None,

crates/integrations/datafusion/src/config.rs

@@ -0,0 +1,61 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::common::config::ConfigExtension;
+use datafusion::common::{config_namespace, extensions_options};
+
+pub const PAIMON_ROW_FILTER: &str = "paimon.read.row_filter";
+
+config_namespace! {
+    /// Paimon read options.
+    pub struct PaimonReadOptions {
+        /// Apply pushed predicates as row filters inside Paimon readers.
+        pub row_filter: bool, default = false
+    }
+}
+
+extensions_options! {
+    /// Paimon-specific DataFusion session options.
+    pub struct PaimonConfig {
+        /// Options that control Paimon reads.
+        pub read: PaimonReadOptions, default = PaimonReadOptions::default()
+    }
+}
+
+impl ConfigExtension for PaimonConfig {
+    const PREFIX: &'static str = "paimon";
+}
+
+#[cfg(test)]
+mod tests {
+    use datafusion::config::ConfigOptions;
+
+    use super::*;
+
+    #[test]
+    fn paimon_row_filter_defaults_to_false_and_can_be_set() {
+        let mut options = ConfigOptions::default();
+        options.extensions.insert(PaimonConfig::default());
+
+        let config = options.extensions.get::<PaimonConfig>().unwrap();
+        assert!(!config.read.row_filter);
+
+        options.set("paimon.read.row_filter", "true").unwrap();
+        let config = options.extensions.get::<PaimonConfig>().unwrap();
+        assert!(config.read.row_filter);
+    }
+}

crates/integrations/datafusion/src/filter_pushdown.rs

@@ -410,7 +410,7 @@ fn reverse_comparison_operator(op: Operator) -> Option<Operator> {
     }
 }
 
-fn scalar_to_datum(scalar: &ScalarValue, data_type: &DataType) -> Option<Datum> {
+pub(crate) fn scalar_to_datum(scalar: &ScalarValue, data_type: &DataType) -> Option<Datum> {
     match data_type {
         DataType::Boolean(_) => match scalar {
             ScalarValue::Boolean(Some(value)) => Some(Datum::Bool(*value)),

crates/integrations/datafusion/src/lib.rs

@@ -40,6 +40,7 @@ mod blob_descriptor_functions;
 mod blob_reader;
 mod blob_view;
 mod catalog;
+pub mod config;
 mod delete;
 mod error;
 mod filter_pushdown;