refactor: Use cast preimages for cast predicate rewrites

[discord9]

Which issue does this PR close?

• Closes Unsafe comparison cast rewriting in ExprSimplifier silently produces wrong query results #22142.

Rationale for this change

The previous cast-unwrap path could only move the original comparison operator
from CAST(expr AS target_type) OP literal to expr OP casted_literal. That is
not correct for many-to-one casts such as timestamp precision narrowing, where
the source-domain preimage of one target value is a range rather than a
singleton.

For example, CAST(ts_ns AS Timestamp(ms)) > 1000ms must not become
ts_ns > 1_000_000_000ns; its exact source boundary is
ts_ns >= 1_001_000_000ns.

Timestamp precision widening has a related ordered-comparison case. A
non-aligned target literal has no singleton equality preimage, but it does have
an exact source-unit boundary for an ordered predicate. For example:

CAST(ts_ms AS Timestamp(ns)) >= 123_456_789ns

becomes:

ts_ms >= 124ms

This PR also makes exact cast rewrites closed-by-default: exact rewrites require
a supported value-preserving cast family. Many-to-one or source-domain-reducing
casts either use an explicit range/boundary preimage or remain unchanged.

The ordered timestamp-widening rewrite deliberately follows the existing
widening policy used by this work. At extreme source values where widening
overflows, ordinary CAST can error and TRY_CAST can return NULL, while the
rewritten source-unit comparison returns a Boolean. This is not a claim of
full-domain equivalence for those overflow cases; a guarded/error-aware
preimage representation is outside this PR's scope.

What changes are included in this PR?

• Add a shared CastPredicatePreimage abstraction in
  datafusion-expr-common:
  • Exact(ScalarValue) for a same-operator source literal or boundary.
  • Range(Interval) for a half-open source-domain interval.
• Share cast-preimage computation between logical and physical simplifiers.
• Organize preimage computation into explicit range, special exact, generic
  exact, and ordered timestamp-widening paths.
• Implement timestamp precision narrowing preimages using half-open buckets
  with truncation-toward-zero semantics, including negative timestamps.
• Implement non-aligned ordered timestamp-widening boundaries using Euclidean
  floor/ceil arithmetic in i128:
  • >= L and < L use ceil(L / q).
  • > L and <= L use floor(L / q).
• Support all timestamp precision-widening unit pairs with matching timezone
  metadata, including CAST, TRY_CAST, and literal-left comparisons.
• Keep non-aligned equality, distinctness, and IN predicates unchanged.
• Keep timestamp precision narrowing out of IN rewrites because its preimage
  is a range rather than a singleton.
• Add conservative exact-cast family gates, including date, integer,
  signedness, decimal precision/scale, and canonical integer/string checks.
• Replace the logical optimizer's cast-unwrap module with a cast-preimage module
  and update the physical simplifier to use the same helper.

Behavior changes compared to main

Expression	Behavior after this PR	Why
CAST(c1:Int32 AS Int64) < 10	c1 < Int32(10)	Integer widening is value-preserving.
CAST(c2:Int64 AS Int32) = 5	kept	Integer narrowing reduces the source domain.
CAST(c1:Int32 AS UInt32) = 5	kept	Signed-to-unsigned is not full-domain safe.
CAST(c1:Int32 AS Utf8) = '123'	c1 = Int32(123)	The literal round-trips canonically.
CAST(c1:Int32 AS Utf8) = '0123'	kept	'0123' -> 123 -> '123' does not round-trip.
CAST(c1:Int32 AS Utf8) < '123'	kept	String ordering is not integer ordering.
CAST(c1:Int32 AS Decimal(12,2)) = 123.00	c1 = Int32(123)	The target decimal represents the full source domain.
CAST(c1:Int32 AS Decimal(10,2)) = 123.00	kept	The target has insufficient integer digits.
CAST(c3:Decimal(10,2) AS Decimal(18,4)) = 123.0000	exact rewrite	Precision/scale widening is value-preserving.
CAST(c3:Decimal(18,2) AS Decimal(18,1)) = 123.0	kept	Scale narrowing is many-to-one.
CAST(ts_ns AS timestamp(ms)) = 1000ms	ts_ns >= 1_000_000_000ns AND ts_ns < 1_001_000_000ns	Equality preimage is a timestamp bucket.
CAST(ts_ns AS timestamp(ms)) > 1000ms	ts_ns >= 1_001_000_000ns	Uses the bucket's upper boundary.
CAST(ts_ns AS timestamp(ms)) <= 0ms	ts_ns < 1_000_000ns	The zero bucket follows truncation toward zero.
CAST(ts_ns AS timestamp(ms)) = -1ms	ts_ns >= -1_999_999ns AND ts_ns < -999_999ns	Negative buckets follow truncation toward zero.
CAST(ts_ns AS timestamp(ms)) != 1000ms	range-complement rewrite	Complements the timestamp bucket.
CAST(ts_ns AS timestamp(ms)) IN (1000ms)	kept	IN currently supports only singleton exact preimages.
CAST(ts_ms AS timestamp(ns)) = 123_000_000ns	ts_ms = 123ms	The aligned widening literal round-trips.
CAST(ts_ms AS timestamp(ns)) = 123_456_789ns	kept	A non-aligned literal has no equality preimage.
CAST(ts_ms AS timestamp(ns)) >= 123_456_789ns	ts_ms >= 124ms	Ordered widening uses the exact Euclidean source boundary under the documented overflow policy.
123_456_789ns < TRY_CAST(ts_ms AS timestamp(ns))	ts_ms > 123ms	Literal-left operators are swapped before computing the same boundary.
CAST(Date64_col AS Date32) = ...	kept	Date precision narrowing is not exact-safe.
CAST(Date32_col AS Date64) = aligned_midnight	exact rewrite	Date widening is injective when the literal is on a day boundary.

Are these changes tested?

Yes. Tests cover:

• exact and range preimages and conservative cast-family gating,
• timestamp narrowing for positive, zero, and negative values,
• all timestamp widening unit pairs,
• aligned and non-aligned widening literals,
• all four ordered operators with positive and negative values,
• CAST, TRY_CAST, and literal-left forms,
• timezone/type/NULL rejection and i64 boundary literals,
• equality, distinctness, and IN remaining unchanged where required,
• logical, physical, and sqllogictest plan output.

Validated locally with:

cargo test -p datafusion-expr-common --lib cast_predicate
cargo test -p datafusion-optimizer --lib cast_preimage
cargo test -p datafusion-physical-expr --lib unwrap_cast
cargo test -p datafusion-sqllogictest --test sqllogictests -- simplify_expr
cargo fmt --all -- --check
cargo clippy --all-targets --all-features -- -D warnings

Are there any user-facing changes?

There are no public API changes. Optimized plans may now use exact source-domain
ranges or boundaries for cast predicates, and previously unsafe exact rewrites
may remain unchanged. Ordered timestamp-widening comparisons also follow the
explicit overflow policy described above.

[2010YOUY01]

This PR looks like a very nice solution for the cast pattern. I'm comfortable proceeding with it, but please forgive me for briefly advocating an alternative approach (that I'm to happy to help reviewing or implementing):

I believe the fundamental goal here is to enable pruning through nested expressions, and the propagation based approach could be a better long term solution.

• Proposal: Prune complex predicates by propagating column statistics #19487

My concern with the preimage approach is that it requires introducing and maintaining an ever-growing set of reverse-transformation rules. Even with additional rules, there will likely still be cases that cannot be handled. If this becomes a supported pattern, I worry that the long-term maintenance burden could be significant.

In contrast, the propagation approach seems both more general and easier to reason about. The key intuition is that it follows a forward-evaluation model, similar to normal expression evaluation, whereas the preimage approach attempts to reverse complex expressions back into a simpler form. In many cases, the latter is inherently more difficult and may require expression-specific logic.

[alamb]

I think this idea shows promise -- I will review it more carefully shortly

[discord9]

Hi @alamb, quick update: I rebased this PR onto latest main and all CI checks are green now. No rush, but when you get a chance I'd appreciate your review.

[alamb]

Thank you -- I will try and review it shorlty.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 96.66091% with 58 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.95%. Comparing base (55b2f09) to head (1b81710).
⚠️ Report is 119 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/expr-common/src/casts.rs	97.14%	16 Missing and 13 partials ⚠️
...ptimizer/src/simplify_expressions/cast_preimage.rs	95.65%	12 Missing and 4 partials ⚠️
...fusion/physical-expr/src/simplifier/unwrap_cast.rs	96.74%	8 Missing and 3 partials ⚠️
datafusion/physical-expr/src/simplifier/mod.rs	66.66%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #22906      +/-   ##
==========================================
+ Coverage   82.74%   82.95%   +0.20%     
==========================================
  Files        1147     1155       +8     
  Lines      449755   459335    +9580     
  Branches   449755   459335    +9580     
==========================================
+ Hits       372168   381029    +8861     
- Misses      54915    55243     +328     
- Partials    22672    23063     +391     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[discord9]

One scope question before review: the latest push includes the ordered
timestamp-widening case from
GreptimeTeam/datafusion#23,
but I am unsure whether it is better to keep that case in this PR or split it
into a follow-up.

The motivating shape appears after mixed timestamp coercion, for example:

CAST(ts_ms AS Timestamp(ns)) >= TimestampNanosecond(123456789)

The non-aligned literal has no singleton equality preimage, so equality and
IN remain unchanged. Ordered comparisons do have an exact source-unit bound:

ts_ms >= TimestampMillisecond(124)

More generally, for widening ratio q and target literal L:

• >= L and < L use ceil(L / q);
• > L and <= L use floor(L / q).

The implementation uses Euclidean i128 arithmetic, supports all timestamp
unit-widening pairs with matching timezone metadata, and covers positive and
negative literals, CAST / TRY_CAST, literal-left comparisons, and boundary
values. Keeping it here is attractive because both logical and physical paths
can reuse the shared cast-preimage abstraction introduced by this PR rather
than implementing separate rewrite logic.

There is an important policy caveat: for extreme source values where widening
overflows, ordinary CAST can error and TRY_CAST can return NULL, while the
rewritten source-unit comparison returns a Boolean. The fork PR deliberately
accepts and documents that existing widening policy; a fully equivalent
upstream solution would require a richer guarded/error-aware preimage model.

So I see two reasonable choices:

1. Keep the ordered widening support here because it is a natural use of the
   new preimage abstraction and fixes the motivating predicate-pushdown case.
2. Keep this already-large PR focused on the abstraction/narrowing work and
   split ordered widening (and its explicit overflow policy) into a follow-up.

I am happy to keep the latest commits or split them back out, depending on what
maintainers would find easier to review and merge.

[jayzhan211]

Thanks @discord9 , I left some suggestions

[jayzhan211]

Casting a timezone-naive timestamp to a timezone-aware one isn't a pure unit change: arrow-cast calls adjust_timestamp_to_timezone for (None, Some(tz)), which reads the stored value as local time and shifts it to UTC. timestamp_narrowing_range_preimage builds the bucket from the raw literal and ignores this, so the range is off by the zone offset. On main this query wasn't rewritten. The exact path (is_exact_cast_safe, same-unit and widening) has the same hole, and #22142 lists it as "timezone silently dropped".

create table t(ts timestamp, l timestamp) as values
  ('2024-01-01T00:00:00.5'::timestamp, '2024-01-01T00:00:00.5'::timestamp);
-- literal moved into a column, no rewrite: 1
select count(*) from t where arrow_cast(ts, 'Timestamp(Millisecond, Some("+07:00"))')
  = arrow_cast(l, 'Timestamp(Millisecond, Some("+07:00"))');
-- literal, rewritten to `ts >= 1704042000500000000 AND ts < 1704042000501000000`: 0
select count(*) from t where arrow_cast(ts, 'Timestamp(Millisecond, Some("+07:00"))')
  = arrow_cast('2024-01-01T00:00:00.5'::timestamp, 'Timestamp(Millisecond, Some("+07:00"))');

Fix: reject this timezone pair in both places (the second check also covers the IN-list path):

/// Arrow shifts values when casting a timezone-naive timestamp to a
/// timezone-aware one (local wall clock -> UTC), so the cast has no
/// literal-independent preimage.
fn is_naive_to_tz_timestamp_cast(source_type: &DataType, target_type: &DataType) -> bool {
    matches!(
        (source_type, target_type),
        (DataType::Timestamp(_, None), DataType::Timestamp(_, Some(_)))
    )
}

 pub fn cast_predicate_preimage(
@@
 ) -> Result<Option<CastPredicatePreimage>> {
+    if is_naive_to_tz_timestamp_cast(source_type, target_type) {
+        return Ok(None);
+    }
     if let Some(preimage) = maybe_range_preimage(source_type, target_type, lit_value)? {
@@ fn is_exact_cast_safe
     if is_timestamp_cast(src, tgt) {
-        return !is_timestamp_precision_narrowing_cast(src, tgt);
+        return !is_timestamp_precision_narrowing_cast(src, tgt)
+            && !is_naive_to_tz_timestamp_cast(src, tgt);
     }

Please add a result-level slt for this case, not just EXPLAIN. With the fix, optimizer_ine) -> ms("UTC")) goes back to keeping the cast, unless you explicitly allow zero-offset"UTC"/"+00:00".

[jayzhan211]

The is_date_narrowing_cast early returns here, at casts.rs:159 and at physical-expr/src/simplifier/unwrap_cast.rs:134 are dead code. is_exact_cast_safe already rejects Date64 -> Date32 (including the dictionary-wrapped form, which this bare-type check misses), and the range, int→string and widening paths can't match date types. Dropping all three leaves the allowlist as the single gate:

-    if is_date_narrowing_cast(source_type, target_type) {
-        return Ok(None);
-    }

[jayzhan211]

Suggested change

	_ => unreachable!("preimage only supports comparison operators"),
	_ => return internal_err!("Expect comparison operators, got {op}"),

[jayzhan211]

Thanks @discord9 , overall LGTM!

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion v55.1.0 (current)
       Built [  57.040s] (current)
     Parsing datafusion v55.1.0 (current)
      Parsed [   0.033s] (current)
    Building datafusion v55.1.0 (baseline)
       Built [  55.532s] (baseline)
     Parsing datafusion v55.1.0 (baseline)
      Parsed [   0.033s] (baseline)
    Checking datafusion v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.629s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [ 115.209s] datafusion
    Building datafusion-expr-common v55.1.0 (current)
       Built [  18.549s] (current)
     Parsing datafusion-expr-common v55.1.0 (current)
      Parsed [   0.018s] (current)
    Building datafusion-expr-common v55.1.0 (baseline)
       Built [  18.461s] (baseline)
     Parsing datafusion-expr-common v55.1.0 (baseline)
      Parsed [   0.018s] (baseline)
    Checking datafusion-expr-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.233s] 229 checks: 228 pass, 1 fail, 0 warn, 31 skip

--- failure function_marked_deprecated: function #[deprecated] added ---

Description:
A function is now #[deprecated]. Downstream crates will get a compiler warning when using this function.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.51.0/src/lints/function_marked_deprecated.ron

Failed in:
  function datafusion_expr_common::casts::is_date_narrowing_cast in /home/runner/work/datafusion/datafusion/datafusion/expr-common/src/casts.rs:674

     Summary semver requires new minor version: 0 major and 1 minor checks failed
    Finished [  38.043s] datafusion-expr-common
    Building datafusion-optimizer v55.1.0 (current)
       Built [  25.118s] (current)
     Parsing datafusion-optimizer v55.1.0 (current)
      Parsed [   0.033s] (current)
    Building datafusion-optimizer v55.1.0 (baseline)
       Built [  25.531s] (baseline)
     Parsing datafusion-optimizer v55.1.0 (baseline)
      Parsed [   0.035s] (baseline)
    Checking datafusion-optimizer v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.174s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [  51.675s] datafusion-optimizer
    Building datafusion-physical-expr v55.1.0 (current)
       Built [  27.584s] (current)
     Parsing datafusion-physical-expr v55.1.0 (current)
      Parsed [   0.049s] (current)
    Building datafusion-physical-expr v55.1.0 (baseline)
       Built [  27.166s] (baseline)
     Parsing datafusion-physical-expr v55.1.0 (baseline)
      Parsed [   0.050s] (baseline)
    Checking datafusion-physical-expr v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.361s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [  56.254s] datafusion-physical-expr
    Building datafusion-sqllogictest v55.1.0 (current)
       Built [  93.383s] (current)
     Parsing datafusion-sqllogictest v55.1.0 (current)
      Parsed [   0.015s] (current)
    Building datafusion-sqllogictest v55.1.0 (baseline)
       Built [  95.958s] (baseline)
     Parsing datafusion-sqllogictest v55.1.0 (baseline)
      Parsed [   0.016s] (baseline)
    Checking datafusion-sqllogictest v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.112s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [ 191.799s] datafusion-sqllogictest

[discord9]

Thanks @discord9 , overall LGTM!

@jayzhan211 fixed the CI(due to decimal overflow forbid pushdown hence prevent bloom filtering)

[jayzhan211]

There are new issue found

[jayzhan211]

@discord9 , I found another one in latest commit

[jayzhan211]

Coercion caps the target at the variant's max precision whatever the source precision is, so requiring the source to be max too still misses the case. decimal(30,2) / (28,0) / (36,4) vs 4.00 all coerce to CAST(col AS Decimal128(38,15)), cast_predicate_preimage returns None, and the cast stays (main unwraps all three → Bloom pruning lost). The overflow hazard is the same as for a (38,2) source, so the source check adds no safety.

Fix (only the (37,0) -> (38,2) blocked assertion flips; move it to the admitted cases):

-    let both_max_same_variant = match (src, tgt) {
-        (DataType::Decimal32(p_src, _), DataType::Decimal32(p_tgt, _)) => {
-            *p_src == DECIMAL32_MAX_PRECISION && *p_tgt == DECIMAL32_MAX_PRECISION
-        }
-        (DataType::Decimal64(p_src, _), DataType::Decimal64(p_tgt, _)) => {
-            *p_src == DECIMAL64_MAX_PRECISION && *p_tgt == DECIMAL64_MAX_PRECISION
-        }
-        (DataType::Decimal128(p_src, _), DataType::Decimal128(p_tgt, _)) => {
-            *p_src == DECIMAL128_MAX_PRECISION && *p_tgt == DECIMAL128_MAX_PRECISION
-        }
-        _ => false,
-    };
-    s_tgt >= s_src && (tgt_int >= src_int || both_max_same_variant)
+    // Coercion caps the target at the variant's maximum precision, which can
+    // lose integer digits; same overflow trade-off as timestamp widening.
+    let capped = match (src, tgt) {
+        (DataType::Decimal32(_, _), DataType::Decimal32(p, _)) => {
+            *p == DECIMAL32_MAX_PRECISION
+        }
+        (DataType::Decimal64(_, _), DataType::Decimal64(p, _)) => {
+            *p == DECIMAL64_MAX_PRECISION
+        }
+        (DataType::Decimal128(_, _), DataType::Decimal128(p, _)) => {
+            *p == DECIMAL128_MAX_PRECISION
+        }
+        _ => false,
+    };
+    s_tgt >= s_src && (tgt_int >= src_int || capped)

Worth adding a decimal(30,2) column to the new simplify_expr.slt block (explain ... where price = 4.00 should show price = Decimal128(4.00,30,2)).

[jayzhan211]

Maybe we could try this 🤔

Requiring both sides at max precision still misses decimal(30,2) / (28,0) / (36,4) vs 4.00: they coerce to CAST(col AS Decimal128(38,15)), cast_predicate_preimage returns None, and the cast stays (main unwraps). Since this PR already accepts that overflowing values compare as booleans after the rewrite, precision adds nothing: a non-decreasing, nonnegative scale is exact and ordered for every value the cast accepts. One rule covers decimal and int → decimal and drops is_safe_decimal_widening, is_safe_integer_to_decimal, integer_decimal_digits (a differential check against arrow cast + cmp, 5.4M rows, found 0 mismatches on cast-accepted rows):

if is_decimal_type(tgt) && (is_decimal_type(src) || is_integer_type(src)) {
    // A non-decreasing, nonnegative scale multiplies by 10^k: exact and
    // ordered for every value the cast accepts. Precision only decides
    // overflow, which follows the timestamp-widening policy above.
    let s_src = if is_decimal_type(src) {
        decimal_precision_scale(src).1
    } else {
        0
    };
    return s_src >= 0 && decimal_precision_scale(tgt).1 >= s_src;
}

Keep the s_src >= 0 guard: a negative scale reaches 10_i128.pow(*scale as u32) in try_cast_numeric_literal and panics.

[jayzhan211]

BTW, the LoC has grown quite a bit during review. It might be worth seeing whether there's a simpler way to get the same result — but that can be a follow-up.

[jayzhan211]

The decimal change looks good, thanks for adding the Bloom-filter test. Every timestamp unit pair can use one interval. Narrowing takes the truncation bucket and widening takes [ceil(v/r), floor(v/r) + 1). A singleton becomes Exact, an empty interval with =/!=/IN returns None, and anything else becomes Range. That replaces maybe_range_preimage, timestamp_widening_ordered_preimage and the round-trip check, and in my local try it removed about 100 non-test lines. The dispatcher needs to strip one Dictionary level first so dictionary-encoded timestamps keep their rewrites.

let (lower, upper) = if source_scale > target_scale {
    let Some(bucket) = trunc_toward_zero_bucket(value, source_scale / target_scale) else {
        return Ok(None);
    };
    bucket
} else {
    let ratio = target_scale / source_scale;
    let floor = value.div_euclid(ratio);
    let ceil = floor + i128::from(value.rem_euclid(ratio) != 0);
    (ceil, floor + 1)
};