fix: leave predicates Iceberg cannot bind to DataFusion instead of failing the scan

crates/datafusion/src/read/scan.rs

@@ -31,7 +31,8 @@ use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{DisplayFormatType, Partitioning};
 use datafusion::prelude::Expr;
 use futures::{Stream, TryStreamExt};
-use iceberg::expr::Predicate;
+use iceberg::expr::{Bind, Predicate};
+use iceberg::spec::SchemaRef;
 use iceberg::table::Table;
 
 use super::predicate::convert_filters_to_predicate;
@@ -127,6 +128,11 @@ impl IcebergDataSourceBuilder {
     }
 
     /// Filters the scan with `predicate`, or not at all if `None` (the default).
+    ///
+    /// The top-level conjuncts of `predicate` that do not bind to the schema of the
+    /// snapshot being scanned are dropped in [`Self::build`], as Iceberg would fail
+    /// the scan on them. A null test on a struct, list or map column is one:
+    /// Iceberg binds references to primitive fields only.
     pub fn with_predicate(mut self, predicate: Option<Predicate>) -> Self {
         self.predicate = predicate;
         self
@@ -149,8 +155,16 @@ impl IcebergDataSourceBuilder {
     ///
     /// # Errors
     ///
-    /// Returns an error if the projection is out of bounds for the schema.
+    /// Returns an error if the projection is out of bounds for the schema, or if
+    /// the scanned snapshot's schema cannot be read.
     pub fn build(self) -> Result<IcebergDataSource> {
+        let predicate = match self.predicate {
+            Some(predicate) => bindable_conjuncts(
+                predicate,
+                &scan_schema(&self.table, self.snapshot_id)?,
+            ),
+            None => None,
+        };
         let schema = match &self.projection {
             None => self.schema,
             Some(projection) => Arc::new(self.schema.project(projection)?),
@@ -165,12 +179,52 @@ impl IcebergDataSourceBuilder {
             snapshot_id: self.snapshot_id,
             schema,
             projection,
-            predicate: self.predicate,
+            predicate,
             fetch: self.fetch,
         })
     }
 }
 
+/// The schema the scan binds its predicate to: that of the snapshot it reads, or
+/// the table's current schema if the snapshot is missing (the scan reports that).
+fn scan_schema(table: &Table, snapshot_id: Option<i64>) -> Result<SchemaRef> {
+    let metadata = table.metadata();
+    let snapshot = match snapshot_id {
+        Some(id) => metadata.snapshot_by_id(id),
+        None => metadata.current_snapshot(),
+    };
+    match snapshot {
+        Some(snapshot) => snapshot.schema(metadata).map_err(to_datafusion_error),
+        None => Ok(metadata.current_schema().clone()),
+    }
+}
+
+/// The top-level conjuncts of `predicate` that bind to `schema`, combined with
+/// `AND`, or `None` if none do.
+///
+/// Dropping a conjunct only lets the scan return more rows, and every filter is
+/// pushed down as `Inexact`, so DataFusion still applies it above the scan.
+fn bindable_conjuncts(predicate: Predicate, schema: &SchemaRef) -> Option<Predicate> {
+    let mut conjuncts = Vec::new();
+    split_conjunction(&predicate, &mut conjuncts);
+    conjuncts
+        .into_iter()
+        .filter(|conjunct| conjunct.bind(schema.clone(), true).is_ok())
+        .cloned()
+        .reduce(Predicate::and)
+}
+
+fn split_conjunction<'a>(predicate: &'a Predicate, conjuncts: &mut Vec<&'a Predicate>) {
+    match predicate {
+        Predicate::And(and) => {
+            for input in and.inputs() {
+                split_conjunction(input, conjuncts);
+            }
+        }
+        _ => conjuncts.push(predicate),
+    }
+}
+
 impl DataSource for IcebergDataSource {
     fn apply_expressions(
         &self,

crates/datafusion/src/table_provider/catalog_table_provider.rs

@@ -479,6 +479,41 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_scan_drops_conjuncts_that_do_not_bind() {
+        use iceberg::expr::{Predicate, Reference};
+        use iceberg::spec::Datum;
+
+        let (catalog, namespace, table_name, _temp_dir) =
+            get_test_catalog_and_table().await;
+        let table_ident = TableIdent::new(namespace, table_name);
+        let table = catalog.load_table(&table_ident).await.unwrap();
+        let schema =
+            Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap());
+        let id = Reference::new("id").equal_to(Datum::int(1));
+        let name = Reference::new("name").equal_to(Datum::string("a"));
+        // Stands for any predicate Iceberg cannot bind, such as a null test on a
+        // struct column.
+        let unbound = Reference::new("missing").is_null();
+        let source = |predicate: Predicate| {
+            IcebergDataSource::builder(table.clone(), schema.clone())
+                .with_predicate(Some(predicate))
+                .build()
+                .unwrap()
+        };
+
+        let both = id.clone().and(unbound.clone()).and(name.clone());
+        assert_eq!(
+            source(both).predicate(),
+            Some(&id.clone().and(name.clone()))
+        );
+        assert_eq!(source(unbound.clone()).predicate(), None);
+        // Only the top level is split: an `OR` with a side that does not bind is
+        // dropped whole.
+        assert_eq!(source(id.clone().or(unbound)).predicate(), None);
+        assert_eq!(source(id.clone()).predicate(), Some(&id));
+    }
+
     async fn get_test_catalog_and_table()
     -> (Arc<dyn Catalog>, NamespaceIdent, String, TempDir) {
         let temp_dir = TempDir::new().unwrap();

crates/sqllogictest/testdata/schedules/df_nested_null_predicate_pushdown.toml

@@ -0,0 +1,23 @@
+# 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.
+
+[engines]
+df = { type = "datafusion" }
+
+[[steps]]
+engine = "df"
+slt = "df_test/nested_null_predicate_pushdown.slt"

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

@@ -0,0 +1,46 @@
+# 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.
+
+# IS NULL / IS NOT NULL on a column that is not a primitive (a struct, list or map)
+# must not fail the scan: Iceberg can only bind a reference to a primitive field.
+
+statement ok
+CREATE TABLE default.default.test_nested_null_table (id INT NOT NULL, s STRUCT<a INT, b VARCHAR>)
+
+query I
+INSERT INTO default.default.test_nested_null_table VALUES
+  (1, named_struct('a', 1, 'b', 'x')),
+  (2, NULL),
+  (3, named_struct('a', NULL, 'b', 'y'))
+----
+3
+
+query I
+SELECT count(*) FROM default.default.test_nested_null_table WHERE s IS NOT NULL
+----
+2
+
+query I
+SELECT id FROM default.default.test_nested_null_table WHERE s IS NULL
+----
+2
+
+query I
+SELECT id FROM default.default.test_nested_null_table WHERE s.a IS NULL ORDER BY id
+----
+2
+3