fix: restrict reverse cast constraint propagation to safe conversions (#25531)

## Which issue does this PR close?

Closes #25523.
Related to #25407.

## Rationale for this change

A filter on a cast can incorrectly make its input appear constant and
remove a required sort. For example, `CAST(x AS INT) = 0` matches both
-0.5 and 0.5, but casting the singleton result interval back to DOUBLE
yields [0.0, 0.0]. The optimizer can then return the wrong order for
`ORDER BY x DESC`.

## What changes are included in this PR?

Gate reverse Cast constraint propagation through a common allowlist for
all types. Preserve propagation for conversions recognized by
`check_bigger_cast`, integer-to-integer casts, and Float32-to-Float64
casts. For other conversions, keep the existing input range instead of
treating a cast back as an inverse.

The conservative allowlist may reduce range refinement and pruning for
safe conversions that are not yet recognized. Runtime CAST behavior and
forward interval evaluation are unchanged.

## What is the testing strategy for this PR?

- Extend existing `cast.slt` with float-to-integer, integer-to-Float32,
timestamp-to-date, and string-to-integer cases. The first three
reproduce incorrect sort elimination before the corresponding guards;
the string case checks compatibility.
- Add a table-driven test for permitted and rejected reverse propagation
paths.
- Check Float32 widening against actual Cast results at NaN, infinity,
signed zero, subnormal, and rounding boundaries.
- Passed 24 Cast unit tests (one existing test ignored), 46 filter
tests, and four relevant SLT files.

## Are there any user-facing changes?

Queries retain required sorting when cast constraints cannot safely
determine the input range. No public API changes.

Author: haohuaijin

datafusion/physical-expr/src/expressions/cast.rs

@@ -386,11 +386,13 @@ impl PhysicalExpr for CastExpr {
         interval: &Interval,
         children: &[&Interval],
     ) -> Result<Option<Vec<Interval>>> {
-        let child_interval = children[0];
-        // Get child's datatype:
-        let cast_type = child_interval.data_type();
+        let source_type = children[0].data_type();
+        let target_type = self.cast_type();
+        if !can_propagate_cast_constraints(&source_type, target_type) {
+            return Ok(Some(vec![]));
+        }
         Ok(Some(vec![
-            interval.cast_to(&cast_type, &DEFAULT_SAFE_CAST_OPTIONS)?,
+            interval.cast_to(&source_type, &DEFAULT_SAFE_CAST_OPTIONS)?,
         ]))
     }
 
@@ -427,6 +429,19 @@ impl PhysicalExpr for CastExpr {
     }
 }
 
+/// Whether output bounds can be cast back to `source` without excluding valid inputs.
+///
+/// Used for reverse constraint propagation through a cast from `source` to `target`.
+/// Many-to-one casts, such as Float64 to Int32, cannot generally be inverted this way:
+/// an output of 0 does not imply an input of 0.0.
+/// Returns false for unrecognized conversions so the input range remains unchanged.
+fn can_propagate_cast_constraints(source: &DataType, target: &DataType) -> bool {
+    CastExpr::check_bigger_cast(target, source)
+        || (source.is_integer() && target.is_integer())
+        // NaN bounds are unbounded; finite Float32 values widen exactly.
+        || (*source == Float32 && *target == Float64)
+}
+
 #[cfg(feature = "proto")]
 impl CastExpr {
     /// Reconstruct a [`CastExpr`] from its protobuf representation.
@@ -595,10 +610,120 @@ mod tests {
         as_boolean_array, as_int64_array, as_string_array, as_struct_array,
         as_uint8_array,
     };
+    use datafusion_common::rounding::{next_down, next_up};
     use datafusion_physical_expr_common::physical_expr::fmt_sql;
     use insta::assert_snapshot;
     use std::collections::HashMap;
 
+    #[test]
+    fn test_cast_constraint_propagation() -> Result<()> {
+        for (source, target, propagates) in [
+            (Utf8, Int32, false),
+            (Utf8View, Int32, false),
+            (Timestamp(TimeUnit::Nanosecond, None), Date32, false),
+            (Int32, Date32, true),
+            (Date32, Int32, true),
+            (Utf8, LargeUtf8, true),
+            (Utf8, Utf8, true),
+            (Float64, Int32, false),
+            (Int64, Float32, false),
+            (Float64, Float32, false),
+            (Decimal128(4, 1), Decimal128(4, 0), false),
+            (Decimal128(4, 1), Int32, false),
+            (Float64, Decimal128(4, 1), false),
+            (Int8, Int64, true),
+            (Int64, Int8, true),
+            (Int32, UInt32, true),
+            (UInt32, Int32, true),
+            (Int32, Float64, true),
+            (Float32, Float64, true),
+            (Decimal128(4, 1), Decimal128(4, 1), true),
+        ] {
+            let schema = Schema::new(vec![Field::new("x", source.clone(), true)]);
+            let expr = CastExpr::new(col("x", &schema)?, target.clone(), None);
+            let input = Interval::make_unbounded(&source)?;
+            let value = ScalarValue::Int32(Some(0)).cast_to(&target)?;
+            let output = Interval::from(&value);
+            let expected = if propagates {
+                vec![Interval::from(&value.cast_to(&source)?)]
+            } else {
+                vec![]
+            };
+            assert_eq!(
+                expr.propagate_constraints(&output, &[&input])?,
+                Some(expected),
+                "{source} -> {target}"
+            );
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn test_float_widening_constraint_boundaries() -> Result<()> {
+        let mut values = vec![
+            f32::NEG_INFINITY,
+            -f32::MAX,
+            -1.0,
+            -f32::MIN_POSITIVE,
+            -f32::from_bits(1),
+            -0.0,
+            0.0,
+            f32::from_bits(1),
+            f32::MIN_POSITIVE,
+            1.0,
+            f32::MAX,
+            f32::INFINITY,
+        ];
+        // Include both signs of signaling and quiet NaNs with distinct payloads.
+        values.extend(
+            [0x7f800001, 0x7f800002, 0x7fc00001, 0xff800001, 0xffc00001]
+                .map(f32::from_bits),
+        );
+        values.extend([next_down(1.0f32), next_up(1.0f32)]);
+        let schema = Arc::new(Schema::new(vec![Field::new("x", Float32, false)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Float32Array::from(values.clone()))],
+        )?;
+        let expr = CastExpr::new(col("x", &schema)?, Float64, None);
+        let array = expr.evaluate(&batch)?.into_array(values.len())?;
+        let widened = array.as_any().downcast_ref::<Float64Array>().unwrap();
+        let mut bounds = vec![-f64::MAX, f64::MAX];
+        for value in widened.values() {
+            bounds.extend([next_down(*value), *value, next_up(*value)]);
+        }
+        // Midpoints exercise rounding back to Float32 in both directions.
+        bounds.extend([
+            f64::from(f32::from_bits(1)) / 2.0,
+            -f64::from(f32::from_bits(1)) / 2.0,
+            f64::midpoint(1.0, f64::from(next_up(1.0f32))),
+        ]);
+        bounds.sort_by(f64::total_cmp);
+        bounds.dedup_by(|a, b| a.to_bits() == b.to_bits());
+        let input = Interval::make_unbounded(&Float32)?;
+        for (i, lower) in bounds.iter().enumerate() {
+            for upper in &bounds[i..] {
+                let output = Interval::make(Some(*lower), Some(*upper))?;
+                let propagated = expr.propagate_constraints(&output, &[&input])?.unwrap();
+                assert_eq!(propagated.len(), 1);
+                for (index, value) in values.iter().enumerate() {
+                    if output.contains_value(ScalarValue::Float64(Some(
+                        widened.value(index),
+                    )))? {
+                        assert!(
+                            propagated[0]
+                                .contains_value(ScalarValue::Float32(Some(*value)))?,
+                            "input bits={:08x}, output={output}, propagated={:?}",
+                            value.to_bits(),
+                            propagated[0]
+                        );
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
     fn make_struct_array(fields: Fields, arrays: Vec<ArrayRef>) -> StructArray {
         StructArray::new(fields, arrays, None)
     }

datafusion/sqllogictest/test_files/cast.slt

@@ -91,6 +91,54 @@ statement ok
 drop table t0;
 
 
+# Regression for https://github.com/apache/datafusion/issues/25523.
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+statement ok
+CREATE TABLE lossy_cast (id INT, x DOUBLE, n BIGINT)
+AS VALUES (1, -0.5, 16777216), (2, 0.5, 16777217), (3, NULL, NULL);
+
+# A singleton cast result does not imply a singleton input.
+query I
+SELECT id FROM lossy_cast WHERE CAST(x AS INT) = 0 ORDER BY x DESC;
+----
+2
+1
+
+query I
+SELECT id FROM lossy_cast WHERE CAST(n AS REAL) = 16777216::REAL ORDER BY n DESC;
+----
+2
+1
+
+statement ok
+DROP TABLE lossy_cast;
+
+statement ok
+CREATE TABLE lossy_non_numeric (id INT, t TIMESTAMP, s VARCHAR)
+AS VALUES (1, TIMESTAMP '2026-01-01 01:00:00', '01'),
+          (2, TIMESTAMP '2026-01-01 02:00:00', '1');
+
+query I
+SELECT id FROM lossy_non_numeric
+WHERE CAST(t AS DATE) = DATE '2026-01-01' ORDER BY t DESC;
+----
+2
+1
+
+query I
+SELECT id FROM lossy_non_numeric WHERE CAST(s AS INT) = 1 ORDER BY s DESC;
+----
+2
+1
+
+statement ok
+DROP TABLE lossy_non_numeric;
+
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
 # ensure that automatically casting with "datafusion.optimizer.expand_views_at_output" does not
 # change the column name