diff --git a/crates/datafusion/src/read/scan.rs b/crates/datafusion/src/read/scan.rs index a3259fd..2fd3174 100644 --- a/crates/datafusion/src/read/scan.rs +++ b/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) -> 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 { + 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) -> Result { + 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 { + 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, diff --git a/crates/datafusion/src/table_provider/catalog_table_provider.rs b/crates/datafusion/src/table_provider/catalog_table_provider.rs index b804f56..1691b35 100644 --- a/crates/datafusion/src/table_provider/catalog_table_provider.rs +++ b/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, NamespaceIdent, String, TempDir) { let temp_dir = TempDir::new().unwrap(); diff --git a/crates/sqllogictest/testdata/schedules/df_nested_null_predicate_pushdown.toml b/crates/sqllogictest/testdata/schedules/df_nested_null_predicate_pushdown.toml new file mode 100644 index 0000000..7103bd3 --- /dev/null +++ b/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" diff --git a/crates/sqllogictest/testdata/slts/df_test/nested_null_predicate_pushdown.slt b/crates/sqllogictest/testdata/slts/df_test/nested_null_predicate_pushdown.slt new file mode 100644 index 0000000..12c1b51 --- /dev/null +++ b/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) + +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