fix: refuse to look inside casts in expr_to_predicate

AGENTS.md

@@ -73,7 +73,7 @@ Tests are self-contained (in-memory catalogs, temp dirs, checked-in metadata JSO
 
 `IcebergTableScan` converts the projection to column names and the filters to a single Iceberg `Predicate` (`physical_plan/expr_to_predicate.rs`). It applies `limit` in-stream, has one output partition, and delegates the actual read to iceberg's `TableScan::to_arrow()`.
 
-Both providers report every filter as `Inexact`, so DataFusion re-applies the original filters after the scan. The pushed-down predicate therefore only has to be *implied by* the original filter. It may match extra rows but must never drop a row that matches. Unconvertible parts are dropped: an `AND` keeps whichever side converted, while an `OR` needs both sides. Preserve this soundness rule when extending `expr_to_predicate.rs`; for example, it is why date casts and some NaN arithmetic are deliberately not pushed down.
+Both providers report every filter as `Inexact`, so DataFusion re-applies the original filters after the scan. The pushed-down predicate therefore only has to be *implied by* the original filter. It may match extra rows but must never drop a row that matches. Unconvertible parts are dropped: an `AND` keeps whichever side converted, while an `OR` needs both sides. Preserve this soundness rule when extending `expr_to_predicate.rs`; for example, it is why casts and some NaN arithmetic are deliberately not pushed down. Removing a cast is left to DataFusion's simplifier (`unwrap_cast_in_comparison`), which rewrites a comparison only when the result doesn't change.
 
 ### Write path (`IcebergTableProvider::insert_into`)
 

crates/datafusion/src/read/predicate.rs

@@ -145,16 +145,6 @@ fn to_iceberg_predicate(expr: &Expr) -> TransformedResult {
                 _ => TransformedResult::NotTransformed,
             }
         }
-        Expr::Cast(c) => {
-            if *c.field.data_type() == DataType::Date32
-                || *c.field.data_type() == DataType::Date64
-            {
-                // Casts to date truncate the expression, we cannot simply extract it as it
-                // can create erroneous predicates.
-                return TransformedResult::NotTransformed;
-            }
-            to_iceberg_predicate(&c.expr)
-        }
         Expr::Like(Like {
             negated,
             expr,
@@ -273,7 +263,6 @@ fn scalar_function_to_iceberg_predicate(
 ///
 /// * negation: `-x` is NaN iff `x` is NaN
 /// * `abs(x)`: `abs(x)` is NaN iff `x` is NaN
-/// * casts between numeric types preserve NaN
 /// * `x + c`, `c + x`, `x - c`, `c - x` for a finite literal `c`
 /// * `x * c`, `c * x`, `x / c` for a finite, non-zero literal `c`
 ///
@@ -285,16 +274,6 @@ fn resolve_nan_preserving_reference(expr: &Expr) -> Option<Reference> {
     match expr {
         Expr::Column(column) => Some(Reference::new(column.name())),
         Expr::Negative(inner) => resolve_nan_preserving_reference(inner),
-        Expr::Cast(cast) => {
-            // Casts to date truncate the value and are not numeric, so they
-            // cannot be treated as NaN-preserving.
-            if *cast.field.data_type() == DataType::Date32
-                || *cast.field.data_type() == DataType::Date64
-            {
-                return None;
-            }
-            resolve_nan_preserving_reference(&cast.expr)
-        }
         Expr::ScalarFunction(ScalarFunction { func, args })
             if func.name() == "abs" && args.len() == 1 =>
         {
@@ -673,18 +652,14 @@ mod tests {
 
     #[test]
     fn test_predicate_conversion_with_cast() {
-        let sql = "ts >= timestamp '2023-01-05T00:00:00'";
-        let predicate = convert_to_iceberg_predicate(sql).unwrap();
-        let expected_predicate = Reference::new("ts")
-            .greater_than_or_equal_to(Datum::string("2023-01-05T00:00:00"));
-        assert_eq!(predicate, expected_predicate);
-    }
-
-    #[test]
-    fn test_predicate_conversion_with_date_cast() {
-        let sql = "ts >= date '2023-01-05T11:00:00'";
-        let predicate = convert_to_iceberg_predicate(sql);
-        assert_eq!(predicate, None);
+        for sql in [
+            "ts >= timestamp '2023-01-05T00:00:00'",
+            "ts >= date '2023-01-05T11:00:00'",
+            "cast(foo as varchar) < '5'",
+            "not isnan(cast(bar as double))",
+        ] {
+            assert_eq!(convert_to_iceberg_predicate(sql), None, "sql: {sql}");
+        }
     }
 
     #[test]

crates/sqllogictest/testdata/schedules/df_cast_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/cast_predicate_pushdown.slt"

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

@@ -0,0 +1,160 @@
+# 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.
+
+# This is a regression test for:
+# https://github.com/apache/datafusion-iceberg/issues/36.
+
+statement ok
+CREATE TABLE default.default.test_cast_table (id INT NOT NULL, d DATE, n INT)
+
+query I
+INSERT INTO default.default.test_cast_table VALUES
+  (1, DATE '2023-01-04', 3),
+  (2, DATE '2023-01-05', 10),
+  (3, DATE '2023-01-06', 3),
+  (4, DATE '1969-12-30', 10),
+  (5, DATE '1969-12-31', 3)
+----
+5
+
+query TT
+EXPLAIN SELECT id FROM default.default.test_cast_table WHERE d < TIMESTAMP '2023-01-05 12:00:00'
+----
+logical_plan
+01)Projection: default.default.test_cast_table.id
+02)--Filter: CAST(default.default.test_cast_table.d AS Timestamp(ns)) < TimestampNanosecond(1672920000000000000, None)
+03)----TableScan: default.default.test_cast_table projection=[id, d], partial_filters=[CAST(default.default.test_cast_table.d AS Timestamp(ns)) < TimestampNanosecond(1672920000000000000, None)]
+physical_plan
+01)FilterExec: CAST(d@1 AS Timestamp(ns)) < 1672920000000000000, projection=[id@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----DataSourceExec: format=iceberg, projection=[id, d]
+
+query I rowsort
+SELECT id FROM default.default.test_cast_table WHERE d < TIMESTAMP '2023-01-05 12:00:00'
+----
+1
+2
+4
+5
+
+query I rowsort
+SELECT id FROM default.default.test_cast_table WHERE d > DATE_TRUNC('day', TIMESTAMP '2023-01-05 12:00:00')
+----
+3
+
+query I rowsort
+SELECT id FROM default.default.test_cast_table WHERE d > TIMESTAMP '1969-12-30 12:00:00'
+----
+1
+2
+3
+5
+
+query TT
+EXPLAIN SELECT id FROM default.default.test_cast_table WHERE arrow_cast(n, 'Utf8') < arrow_cast('5', 'Utf8')
+----
+logical_plan
+01)Projection: default.default.test_cast_table.id
+02)--Filter: CAST(default.default.test_cast_table.n AS Utf8) < Utf8("5")
+03)----TableScan: default.default.test_cast_table projection=[id, n], partial_filters=[CAST(default.default.test_cast_table.n AS Utf8) < Utf8("5")]
+physical_plan
+01)FilterExec: CAST(n@1 AS Utf8) < 5, projection=[id@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----DataSourceExec: format=iceberg, projection=[id, n]
+
+query I rowsort
+SELECT id FROM default.default.test_cast_table WHERE arrow_cast(n, 'Utf8') < arrow_cast('5', 'Utf8')
+----
+1
+2
+3
+4
+5
+
+statement ok
+DROP TABLE default.default.test_cast_table
+
+statement ok
+CREATE TABLE default.default.test_float_cast_table (id INT NOT NULL, x REAL)
+
+query I
+INSERT INTO default.default.test_float_cast_table VALUES
+  (1, CAST(0.1 AS REAL)),
+  (2, CAST(0.2 AS REAL)),
+  (3, CAST(0.05 AS REAL))
+----
+3
+
+query TT
+EXPLAIN SELECT id FROM default.default.test_float_cast_table WHERE x > 0.1
+----
+logical_plan
+01)Projection: default.default.test_float_cast_table.id
+02)--Filter: CAST(default.default.test_float_cast_table.x AS Float64) > Float64(0.1)
+03)----TableScan: default.default.test_float_cast_table projection=[id, x], partial_filters=[CAST(default.default.test_float_cast_table.x AS Float64) > Float64(0.1)]
+physical_plan
+01)FilterExec: CAST(x@1 AS Float64) > 0.1, projection=[id@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----DataSourceExec: format=iceberg, projection=[id, x]
+
+query I rowsort
+SELECT id FROM default.default.test_float_cast_table WHERE x > 0.1
+----
+1
+2
+
+statement ok
+DROP TABLE default.default.test_float_cast_table
+
+statement ok
+CREATE TABLE default.default.test_isnan_cast_table (id INT NOT NULL, s VARCHAR, n INT)
+
+query I
+INSERT INTO default.default.test_isnan_cast_table VALUES (1, 'NaN', 1), (2, '1.5', 2)
+----
+2
+
+query TT
+EXPLAIN SELECT id FROM default.default.test_isnan_cast_table WHERE NOT isnan(CAST(s AS DOUBLE))
+----
+logical_plan
+01)Projection: default.default.test_isnan_cast_table.id
+02)--Filter: NOT isnan(CAST(default.default.test_isnan_cast_table.s AS Float64))
+03)----TableScan: default.default.test_isnan_cast_table projection=[id, s], partial_filters=[NOT isnan(CAST(default.default.test_isnan_cast_table.s AS Float64))]
+physical_plan
+01)FilterExec: NOT isnan(CAST(s@1 AS Float64)), projection=[id@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----DataSourceExec: format=iceberg, projection=[id, s]
+
+query I rowsort
+SELECT id FROM default.default.test_isnan_cast_table WHERE NOT isnan(CAST(s AS DOUBLE))
+----
+2
+
+query I rowsort
+SELECT id FROM default.default.test_isnan_cast_table WHERE isnan(CAST(s AS DOUBLE))
+----
+1
+
+query I rowsort
+SELECT id FROM default.default.test_isnan_cast_table WHERE NOT isnan(CAST(n AS DOUBLE))
+----
+1
+2
+
+statement ok
+DROP TABLE default.default.test_isnan_cast_table