From 7903c38763780993b7aba34bcb11b7e9caf8c134 Mon Sep 17 00:00:00 2001 From: Colin Marc Date: Thu, 29 Oct 2026 10:30:10 +0100 Subject: [PATCH] fix: refuse to look inside casts in `expr_to_predicate` The purpose of `expr_to_predicate` is to take an expr and return a valid, equivalent push-down predicate. The existing behavior of just dropping the cast ignored many boundary conditions that would lead to incorrectly excluding partitions. There was already a carveout for dates, because that is the most obvious example. The following expressions are obviously not equivalent: -- x is a date, eg '2026-01-22' arrow_cast(x, 'timestamp(ns)') < '2026-01-22T12:00' -- After dropping the cast, iceberg coerces the literal, so the -- resulting expr is: x < '2026-01-22T12:00' Whether it's a bug depends on the type of `x`; if x is a date, or broadly coarser than the literal, then dropping the cast creates a bad predicate (specifically: the partition '2026-01-22' would be dropped, even though the original expression should keep it). Once I started looking, though, I found a few more weird cases: -- x is int32. '10' < '5' (lexicographically) arrow_cast(x, 'utf8') < '5' -- Drop the cast and Datum::to('5', int32) gives you: x < 5 And here's another: -- x is f32. 0.1f32 > 0.1f64 (0.100000001490116 > 0.100000000000000005) arrow_cast(x, 'float64') > 0.1 -- Dropping the cast, plus coercion to f32: x > 0.1f32 It seems like there might be more, any time two conditions are fulfilled: - The cast is to a coarser type - The literal would be coerced in a lossy way by `Datum::to` Fixes #36. --- AGENTS.md | 2 +- crates/datafusion/src/read/predicate.rs | 41 +---- .../schedules/df_cast_predicate_pushdown.toml | 23 +++ .../slts/df_test/cast_predicate_pushdown.slt | 160 ++++++++++++++++++ 4 files changed, 192 insertions(+), 34 deletions(-) create mode 100644 crates/sqllogictest/testdata/schedules/df_cast_predicate_pushdown.toml create mode 100644 crates/sqllogictest/testdata/slts/df_test/cast_predicate_pushdown.slt diff --git a/AGENTS.md b/AGENTS.md index 103ff4c..af27d25 100644 --- a/AGENTS.md +++ b/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`) diff --git a/crates/datafusion/src/read/predicate.rs b/crates/datafusion/src/read/predicate.rs index 33bf71b..115085b 100644 --- a/crates/datafusion/src/read/predicate.rs +++ b/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 { 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] diff --git a/crates/sqllogictest/testdata/schedules/df_cast_predicate_pushdown.toml b/crates/sqllogictest/testdata/schedules/df_cast_predicate_pushdown.toml new file mode 100644 index 0000000..e57f2ca --- /dev/null +++ b/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" diff --git a/crates/sqllogictest/testdata/slts/df_test/cast_predicate_pushdown.slt b/crates/sqllogictest/testdata/slts/df_test/cast_predicate_pushdown.slt new file mode 100644 index 0000000..53bfa4b --- /dev/null +++ b/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