fix(datafusion): classify filter pushdown accurately

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

@@ -50,6 +50,7 @@ use metadata_table::IcebergMetadataTableProvider;
 
 use crate::error::to_datafusion_error;
 use crate::physical_plan::commit::IcebergCommitExec;
+use crate::physical_plan::expr_to_predicate::classify_filter_pushdown;
 use crate::physical_plan::project::project_with_partition;
 use crate::physical_plan::repartition::repartition;
 use crate::physical_plan::scan::IcebergTableScan;
@@ -146,8 +147,10 @@ impl TableProvider for IcebergTableProvider {
         &self,
         filters: &[&Expr],
     ) -> DFResult<Vec<TableProviderFilterPushDown>> {
-        // Push down all filters, as a single source of truth, the scanner will drop the filters which couldn't be push down
-        Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
+        Ok(filters
+            .iter()
+            .map(|filter| classify_filter_pushdown(filter))
+            .collect())
     }
 
     async fn insert_into(
@@ -326,8 +329,10 @@ impl TableProvider for IcebergStaticTableProvider {
         &self,
         filters: &[&Expr],
     ) -> DFResult<Vec<TableProviderFilterPushDown>> {
-        // Push down all filters, as a single source of truth, the scanner will drop the filters which couldn't be push down
-        Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
+        Ok(filters
+            .iter()
+            .map(|filter| classify_filter_pushdown(filter))
+            .collect())
     }
 
     async fn insert_into(

crates/integrations/datafusion/tests/integration_datafusion_test.rs

@@ -274,7 +274,7 @@ async fn test_table_projection() -> Result<()> {
 }
 
 #[tokio::test]
-async fn test_table_predict_pushdown() -> Result<()> {
+async fn test_table_predicate_pushdown() -> Result<()> {
     let iceberg_catalog = get_iceberg_catalog().await;
     let namespace = NamespaceIdent::new("ns".to_string());
     set_test_namespace(&iceberg_catalog, &namespace).await?;
@@ -315,6 +315,34 @@ async fn test_table_predict_pushdown() -> Result<()> {
     // the first row is logical_plan, the second row is physical_plan
     let expected = "predicate:[(foo > 1) OR (bar IS NULL)]";
     assert!(s.value(1).trim().contains(expected));
+    assert!(
+        s.value(1).contains("FilterExec"),
+        "partially converted filters must be re-applied: {}",
+        s.value(1)
+    );
+
+    let records = ctx
+        .sql("select * from catalog.ns.t1 where foo > 1 and bar is null")
+        .await
+        .unwrap()
+        .explain(false, false)
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    let physical_plan = records[0]
+        .column(1)
+        .as_any()
+        .downcast_ref::<StringArray>()
+        .unwrap()
+        .value(1);
+
+    assert!(physical_plan.contains("predicate:[(foo > 1) AND (bar IS NULL)]"));
+    assert!(
+        !physical_plan.contains("FilterExec"),
+        "exact filters should not be re-applied: {physical_plan}"
+    );
+
     Ok(())
 }
 

crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt

@@ -22,13 +22,10 @@
 query TT
 EXPLAIN SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
 ----
-logical_plan
-01)Filter: default.default.test_binary_table.data = LargeBinary("1,2")
-02)--TableScan: default.default.test_binary_table projection=[id, data], partial_filters=[default.default.test_binary_table.data = LargeBinary("1,2")]
+logical_plan TableScan: default.default.test_binary_table projection=[id, data], full_filters=[default.default.test_binary_table.data = LargeBinary("1,2")]
 physical_plan
-01)FilterExec: data@1 = 0102
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,data] predicate:[data = 0102]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,data] predicate:[data = 0102]
 
 # Verify empty result from empty table
 query I?

crates/sqllogictest/testdata/slts/df_test/boolean_predicate_pushdown.slt

@@ -33,13 +33,10 @@ INSERT INTO default.default.test_boolean_table VALUES
 query TT
 EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active = true
 ----
-logical_plan
-01)Filter: default.default.test_boolean_table.is_active
-02)--TableScan: default.default.test_boolean_table projection=[id, is_active, description], partial_filters=[default.default.test_boolean_table.is_active]
+logical_plan TableScan: default.default.test_boolean_table projection=[id, is_active, description], full_filters=[default.default.test_boolean_table.is_active]
 physical_plan
-01)FilterExec: is_active@1
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,is_active,description] predicate:[is_active = true]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,is_active,description] predicate:[is_active = true]
 
 # Query with is_active = true
 query ITT rowsort
@@ -53,13 +50,10 @@ SELECT * FROM default.default.test_boolean_table WHERE is_active = true
 query TT
 EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active = false
 ----
-logical_plan
-01)Filter: NOT default.default.test_boolean_table.is_active
-02)--TableScan: default.default.test_boolean_table projection=[id, is_active, description], partial_filters=[NOT default.default.test_boolean_table.is_active]
+logical_plan TableScan: default.default.test_boolean_table projection=[id, is_active, description], full_filters=[NOT default.default.test_boolean_table.is_active]
 physical_plan
-01)FilterExec: NOT is_active@1
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
 
 # Query with is_active = false
 query ITT rowsort
@@ -72,13 +66,10 @@ SELECT * FROM default.default.test_boolean_table WHERE is_active = false
 query TT
 EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active != true
 ----
-logical_plan
-01)Filter: NOT default.default.test_boolean_table.is_active
-02)--TableScan: default.default.test_boolean_table projection=[id, is_active, description], partial_filters=[NOT default.default.test_boolean_table.is_active]
+logical_plan TableScan: default.default.test_boolean_table projection=[id, is_active, description], full_filters=[NOT default.default.test_boolean_table.is_active]
 physical_plan
-01)FilterExec: NOT is_active@1
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
 
 # Query with is_active != true (includes false and NULL)
 query ITT rowsort

crates/sqllogictest/testdata/slts/df_test/like_predicate_pushdown.slt

@@ -31,13 +31,10 @@ INSERT INTO default.default.test_unpartitioned_table VALUES (5, 'alice'), (6, 'A
 query TT
 EXPLAIN SELECT * FROM default.default.test_unpartitioned_table WHERE name LIKE 'Al%'
 ----
-logical_plan
-01)Filter: default.default.test_unpartitioned_table.name LIKE Utf8("Al%")
-02)--TableScan: default.default.test_unpartitioned_table projection=[id, name], partial_filters=[default.default.test_unpartitioned_table.name LIKE Utf8("Al%")]
+logical_plan TableScan: default.default.test_unpartitioned_table projection=[id, name], full_filters=[default.default.test_unpartitioned_table.name LIKE Utf8("Al%")]
 physical_plan
-01)FilterExec: name@1 LIKE Al%
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,name] predicate:[name STARTS WITH "Al"]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,name] predicate:[name STARTS WITH "Al"]
 
 # Test LIKE filtering with case-sensitive match
 query IT rowsort

crates/sqllogictest/testdata/slts/df_test/timestamp_predicate_pushdown.slt

@@ -44,13 +44,10 @@ VALUES
 query TT
 EXPLAIN SELECT * FROM default.default.test_timestamp_table WHERE ts = CAST('2023-01-05 12:30:00' AS TIMESTAMP)
 ----
-logical_plan
-01)Filter: default.default.test_timestamp_table.ts = TimestampNanosecond(1672921800000000000, None)
-02)--TableScan: default.default.test_timestamp_table projection=[id, ts], partial_filters=[default.default.test_timestamp_table.ts = TimestampNanosecond(1672921800000000000, None)]
+logical_plan TableScan: default.default.test_timestamp_table projection=[id, ts], full_filters=[default.default.test_timestamp_table.ts = TimestampNanosecond(1672921800000000000, None)]
 physical_plan
-01)FilterExec: ts@1 = 1672921800000000000
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[ts = 2023-01-05 12:30:00]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,ts] predicate:[ts = 2023-01-05 12:30:00]
 
 # Verify timestamp equality filtering works
 query I?
@@ -62,13 +59,10 @@ SELECT * FROM default.default.test_timestamp_table WHERE ts = CAST('2023-01-05 1
 query TT
 EXPLAIN SELECT * FROM default.default.test_timestamp_table WHERE ts > CAST('2023-01-10 00:00:00' AS TIMESTAMP)
 ----
-logical_plan
-01)Filter: default.default.test_timestamp_table.ts > TimestampNanosecond(1673308800000000000, None)
-02)--TableScan: default.default.test_timestamp_table projection=[id, ts], partial_filters=[default.default.test_timestamp_table.ts > TimestampNanosecond(1673308800000000000, None)]
+logical_plan TableScan: default.default.test_timestamp_table projection=[id, ts], full_filters=[default.default.test_timestamp_table.ts > TimestampNanosecond(1673308800000000000, None)]
 physical_plan
-01)FilterExec: ts@1 > 1673308800000000000
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-10 00:00:00]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-10 00:00:00]
 
 # Verify timestamp greater than filtering
 query I? rowsort
@@ -92,10 +86,10 @@ WHERE ts >= CAST('2023-01-05 00:00:00' AS TIMESTAMP)
   AND ts <= CAST('2023-01-15 23:59:59' AS TIMESTAMP)
 ----
 logical_plan
-01)Filter: default.default.test_timestamp_table.ts >= TimestampNanosecond(1672876800000000000, None) AND default.default.test_timestamp_table.ts <= TimestampNanosecond(1673827199000000000, None)
-02)--TableScan: default.default.test_timestamp_table projection=[id, ts], partial_filters=[default.default.test_timestamp_table.ts >= TimestampNanosecond(1672876800000000000, None), default.default.test_timestamp_table.ts <= TimestampNanosecond(1673827199000000000, None)]
+01)Filter: default.default.test_timestamp_table.ts <= TimestampNanosecond(1673827199000000000, None)
+02)--TableScan: default.default.test_timestamp_table projection=[id, ts], full_filters=[default.default.test_timestamp_table.ts >= TimestampNanosecond(1672876800000000000, None)], partial_filters=[default.default.test_timestamp_table.ts <= TimestampNanosecond(1673827199000000000, None)]
 physical_plan
-01)FilterExec: ts@1 >= 1672876800000000000 AND ts@1 <= 1673827199000000000
+01)FilterExec: ts@1 <= 1673827199000000000
 02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 03)----IcebergTableScan projection:[id,ts] predicate:[(ts >= 2023-01-05 00:00:00) AND (ts <= 2023-01-15 23:59:59)]
 
@@ -156,13 +150,10 @@ VALUES
 query TT
 EXPLAIN SELECT * FROM default.default.test_timestamp_micros WHERE ts > CAST('2023-01-01 00:00:00' AS TIMESTAMP)
 ----
-logical_plan
-01)Filter: default.default.test_timestamp_micros.ts > TimestampMicrosecond(1672531200000000, None)
-02)--TableScan: default.default.test_timestamp_micros projection=[id, ts], partial_filters=[default.default.test_timestamp_micros.ts > TimestampMicrosecond(1672531200000000, None)]
+logical_plan TableScan: default.default.test_timestamp_micros projection=[id, ts], full_filters=[default.default.test_timestamp_micros.ts > TimestampMicrosecond(1672531200000000, None)]
 physical_plan
-01)FilterExec: ts@1 > 1672531200000000
-02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-01 00:00:00]
+01)CooperativeExec
+02)--IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-01 00:00:00]
 
 query I?
 SELECT * FROM default.default.test_timestamp_micros WHERE ts > CAST('2023-01-01 00:00:00' AS TIMESTAMP)