feat: push down filters to parquet read path (#208)

Author: QuakeWang

Cargo.toml

@@ -32,7 +32,8 @@ arrow = "57.0"
 arrow-array = { version = "57.0", features = ["ffi"] }
 arrow-schema = "57.0"
 arrow-cast = "57.0"
+arrow-ord = "57.0"
 datafusion = "52.3.0"
 datafusion-ffi = "52.3.0"
 parquet = "57.0"
-tokio = "1.39.2"
+tokio = "1.39.2"

bindings/c/src/table.rs

@@ -349,8 +349,10 @@ pub unsafe extern "C" fn paimon_table_read_to_arrow(
     let end = (offset.saturating_add(length)).min(all_splits.len());
     let selected = &all_splits[start..end];
 
-    // Create TableRead with the stored read_type (projection)
-    let table_read = paimon::table::TableRead::new(&state.table, state.read_type.clone());
+    // C bindings currently persist only the projection, so reconstructing the
+    // read uses an empty predicate set.
+    let table_read =
+        paimon::table::TableRead::new(&state.table, state.read_type.clone(), Vec::new());
 
     match table_read.to_arrow(selected) {
         Ok(stream) => {

crates/integrations/datafusion/Cargo.toml

@@ -35,4 +35,10 @@ futures = "0.3"
 tokio = { workspace = true, features = ["rt", "time", "fs"] }
 
 [dev-dependencies]
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+parquet = { workspace = true }
+serde = "1"
+serde_json = "1"
+tempfile = "3"
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

crates/integrations/datafusion/src/physical_plan/scan.rs

@@ -26,6 +26,7 @@ use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties};
 use futures::{StreamExt, TryStreamExt};
+use paimon::spec::Predicate;
 use paimon::table::Table;
 use paimon::DataSplit;
 
@@ -41,6 +42,9 @@ pub struct PaimonTableScan {
     table: Table,
     /// Projected column names (if None, reads all columns).
     projected_columns: Option<Vec<String>>,
+    /// Filter translated from DataFusion expressions and reused during execute()
+    /// so reader-side pruning reaches the actual read path.
+    pushed_predicate: Option<Predicate>,
     /// Pre-planned partition assignments: `planned_partitions[i]` contains the
     /// Paimon splits that DataFusion partition `i` will read.
     /// Wrapped in `Arc` to avoid deep-cloning `DataSplit` metadata in `execute()`.
@@ -55,6 +59,7 @@ impl PaimonTableScan {
         schema: ArrowSchemaRef,
         table: Table,
         projected_columns: Option<Vec<String>>,
+        pushed_predicate: Option<Predicate>,
         planned_partitions: Vec<Arc<[DataSplit]>>,
         limit: Option<usize>,
     ) -> Self {
@@ -67,6 +72,7 @@ impl PaimonTableScan {
         Self {
             table,
             projected_columns,
+            pushed_predicate,
             planned_partitions,
             plan_properties,
             limit,
@@ -82,6 +88,11 @@ impl PaimonTableScan {
         &self.planned_partitions
     }
 
+    #[cfg(test)]
+    pub(crate) fn pushed_predicate(&self) -> Option<&Predicate> {
+        self.pushed_predicate.as_ref()
+    }
+
     pub fn limit(&self) -> Option<usize> {
         self.limit
     }
@@ -126,6 +137,7 @@ impl ExecutionPlan for PaimonTableScan {
         let table = self.table.clone();
         let schema = self.schema();
         let projected_columns = self.projected_columns.clone();
+        let pushed_predicate = self.pushed_predicate.clone();
 
         let fut = async move {
             let mut read_builder = table.new_read_builder();
@@ -134,6 +146,9 @@ impl ExecutionPlan for PaimonTableScan {
                 let col_refs: Vec<&str> = columns.iter().map(|s| s.as_str()).collect();
                 read_builder.with_projection(&col_refs);
             }
+            if let Some(filter) = pushed_predicate {
+                read_builder.with_filter(filter);
+            }
 
             let read = read_builder.new_read().map_err(to_datafusion_error)?;
             let stream = read.to_arrow(&splits).map_err(to_datafusion_error)?;
@@ -173,11 +188,26 @@ impl DisplayAs for PaimonTableScan {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
+    mod test_utils {
+        include!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../test_utils.rs"));
+    }
+
+    use datafusion::arrow::array::Int32Array;
+    use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
     use datafusion::physical_plan::ExecutionPlan;
+    use datafusion::prelude::SessionContext;
+    use futures::TryStreamExt;
+    use paimon::catalog::Identifier;
+    use paimon::io::FileIOBuilder;
+    use paimon::spec::{
+        BinaryRow, DataType, Datum, IntType, PredicateBuilder, Schema as PaimonSchema, TableSchema,
+    };
+    use std::fs;
+    use tempfile::tempdir;
+    use test_utils::{local_file_path, test_data_file, write_int_parquet_file};
 
     fn test_schema() -> ArrowSchemaRef {
-        Arc::new(Schema::new(vec![Field::new(
+        Arc::new(ArrowSchema::new(vec![Field::new(
             "id",
             ArrowDataType::Int32,
             false,
@@ -191,6 +221,7 @@ mod tests {
             schema,
             dummy_table(),
             None,
+            None,
             vec![Arc::from(Vec::new())],
             None,
         );
@@ -205,19 +236,16 @@ mod tests {
             Arc::from(Vec::new()),
             Arc::from(Vec::new()),
         ];
-        let scan = PaimonTableScan::new(schema, dummy_table(), None, planned_partitions, None);
+        let scan =
+            PaimonTableScan::new(schema, dummy_table(), None, None, planned_partitions, None);
         assert_eq!(scan.properties().output_partitioning().partition_count(), 3);
     }
 
     /// Constructs a minimal Table for testing (no real files needed since we
     /// only test PlanProperties, not actual reads).
     fn dummy_table() -> Table {
-        use paimon::catalog::Identifier;
-        use paimon::io::FileIOBuilder;
-        use paimon::spec::{Schema, TableSchema};
-
         let file_io = FileIOBuilder::new("file").build().unwrap();
-        let schema = Schema::builder().build().unwrap();
+        let schema = PaimonSchema::builder().build().unwrap();
         let table_schema = TableSchema::new(0, &schema);
         Table::new(
             file_io,
@@ -226,4 +254,83 @@ mod tests {
             table_schema,
         )
     }
+
+    #[tokio::test]
+    async fn test_execute_applies_pushed_filter_during_read() {
+        let tempdir = tempdir().unwrap();
+        let table_path = local_file_path(tempdir.path());
+        let bucket_dir = tempdir.path().join("bucket-0");
+        fs::create_dir_all(&bucket_dir).unwrap();
+
+        write_int_parquet_file(
+            &bucket_dir.join("data.parquet"),
+            vec![("id", vec![1, 2, 3, 4]), ("value", vec![5, 20, 30, 40])],
+            Some(2),
+        );
+
+        let file_io = FileIOBuilder::new("file").build().unwrap();
+        let table_schema = TableSchema::new(
+            0,
+            &paimon::spec::Schema::builder()
+                .column("id", DataType::Int(IntType::new()))
+                .column("value", DataType::Int(IntType::new()))
+                .build()
+                .unwrap(),
+        );
+        let table = Table::new(
+            file_io,
+            Identifier::new("default", "t"),
+            table_path,
+            table_schema,
+        );
+
+        let split = paimon::DataSplitBuilder::new()
+            .with_snapshot(1)
+            .with_partition(BinaryRow::new(0))
+            .with_bucket(0)
+            .with_bucket_path(local_file_path(&bucket_dir))
+            .with_total_buckets(1)
+            .with_data_files(vec![test_data_file("data.parquet", 4)])
+            .with_raw_convertible(true)
+            .build()
+            .unwrap();
+
+        let pushed_predicate = PredicateBuilder::new(table.schema().fields())
+            .greater_or_equal("value", Datum::Int(10))
+            .unwrap();
+
+        let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+            "id",
+            ArrowDataType::Int32,
+            false,
+        )]));
+        let scan = PaimonTableScan::new(
+            schema,
+            table,
+            Some(vec!["id".to_string()]),
+            Some(pushed_predicate),
+            vec![Arc::from(vec![split])],
+            None,
+        );
+
+        let ctx = SessionContext::new();
+        let stream = scan
+            .execute(0, ctx.task_ctx())
+            .expect("execute should succeed");
+        let batches = stream.try_collect::<Vec<_>>().await.unwrap();
+
+        let actual_ids: Vec<i32> = batches
+            .iter()
+            .flat_map(|batch| {
+                let ids = batch
+                    .column(0)
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .expect("id column should be Int32Array");
+                (0..ids.len()).map(|idx| ids.value(idx)).collect::<Vec<_>>()
+            })
+            .collect();
+
+        assert_eq!(actual_ids, vec![2, 3, 4]);
+    }
 }

crates/integrations/datafusion/src/table/mod.rs

@@ -36,8 +36,13 @@ use crate::runtime::await_with_runtime;
 
 /// Read-only table provider for a Paimon table.
 ///
-/// Supports full table scan, column projection, and partition predicate pushdown.
-/// Data-level filtering remains a residual DataFusion filter.
+/// Supports full table scan, column projection, and predicate pushdown for
+/// planning. Partition predicates prune splits eagerly, while supported
+/// non-partition data predicates may also be reused by the Parquet read path
+/// for row-group pruning and partial decode-time filtering.
+///
+/// DataFusion still treats pushed filters as inexact because unsupported
+/// predicates and non-Parquet reads remain residual filters.
 #[derive(Debug, Clone)]
 pub struct PaimonTableProvider {
     table: Table,
@@ -103,8 +108,9 @@ impl TableProvider for PaimonTableProvider {
         };
 
         // Plan splits eagerly so we know partition count upfront.
+        let pushed_predicate = build_pushed_predicate(filters, self.table.schema().fields());
         let mut read_builder = self.table.new_read_builder();
-        if let Some(filter) = build_pushed_predicate(filters, self.table.schema().fields()) {
+        if let Some(filter) = pushed_predicate.clone() {
             read_builder.with_filter(filter);
         }
         // Push the limit hint to paimon-core planning to reduce splits when possible.
@@ -141,6 +147,7 @@ impl TableProvider for PaimonTableProvider {
             projected_schema,
             self.table.clone(),
             projected_columns,
+            pushed_predicate,
             planned_partitions,
             limit,
         )))
@@ -318,4 +325,27 @@ mod tests {
             BTreeSet::from([("2024-01-01".to_string(), 10)]),
         );
     }
+
+    #[tokio::test]
+    async fn test_scan_keeps_pushed_predicate_for_execute() {
+        let provider = create_provider("partitioned_log_table").await;
+        let filter = col("id").gt(lit(1));
+
+        let config = SessionConfig::new().with_target_partitions(8);
+        let ctx = SessionContext::new_with_config(config);
+        let state = ctx.state();
+        let plan = provider
+            .scan(&state, None, std::slice::from_ref(&filter), None)
+            .await
+            .expect("scan() should succeed");
+        let scan = plan
+            .as_any()
+            .downcast_ref::<PaimonTableScan>()
+            .expect("Expected PaimonTableScan");
+
+        let expected = build_pushed_predicate(&[filter], provider.table().schema().fields())
+            .expect("data filter should translate");
+
+        assert_eq!(scan.pushed_predicate(), Some(&expected));
+    }
 }

crates/paimon/Cargo.toml

@@ -57,6 +57,7 @@ indexmap = "2.5.0"
 roaring = "0.11"
 arrow-array = { workspace = true }
 arrow-cast = { workspace = true }
+arrow-ord = { workspace = true }
 arrow-schema = { workspace = true }
 futures = "0.3"
 parquet = { workspace = true, features = ["async", "zstd"] }

crates/paimon/src/arrow/filtering.rs

@@ -0,0 +1,74 @@
+// 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 crate::arrow::schema_evolution::create_index_mapping;
+pub(crate) use crate::predicate_stats::{predicates_may_match_with_schema, StatsAccessor};
+use crate::spec::{DataField, Predicate, PredicateOperator};
+
+pub(crate) fn reader_pruning_predicates(data_predicates: Vec<Predicate>) -> Vec<Predicate> {
+    data_predicates
+        .into_iter()
+        .filter(predicate_supported_for_reader_pruning)
+        .collect()
+}
+
+pub(crate) fn build_field_mapping(
+    table_fields: &[DataField],
+    file_fields: &[DataField],
+) -> Vec<Option<usize>> {
+    normalize_field_mapping(
+        create_index_mapping(table_fields, file_fields),
+        table_fields.len(),
+    )
+}
+
+fn predicate_supported_for_reader_pruning(predicate: &Predicate) -> bool {
+    match predicate {
+        Predicate::AlwaysFalse => true,
+        Predicate::Leaf { op, .. } => {
+            matches!(
+                op,
+                PredicateOperator::IsNull
+                    | PredicateOperator::IsNotNull
+                    | PredicateOperator::Eq
+                    | PredicateOperator::NotEq
+                    | PredicateOperator::Lt
+                    | PredicateOperator::LtEq
+                    | PredicateOperator::Gt
+                    | PredicateOperator::GtEq
+                    | PredicateOperator::In
+                    | PredicateOperator::NotIn
+            )
+        }
+        Predicate::AlwaysTrue | Predicate::And(_) | Predicate::Or(_) | Predicate::Not(_) => false,
+    }
+}
+
+fn identity_field_mapping(num_fields: usize) -> Vec<Option<usize>> {
+    (0..num_fields).map(Some).collect()
+}
+
+fn normalize_field_mapping(mapping: Option<Vec<i32>>, num_fields: usize) -> Vec<Option<usize>> {
+    mapping
+        .map(|field_mapping| {
+            field_mapping
+                .into_iter()
+                .map(|index| usize::try_from(index).ok())
+                .collect()
+        })
+        .unwrap_or_else(|| identity_field_mapping(num_fields))
+}

crates/paimon/src/arrow/mod.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+pub(crate) mod filtering;
 mod reader;
 pub(crate) mod schema_evolution;