fix: avoid query failures from inferred numeric cast bounds

datafusion/expr-common/src/interval_arithmetic.rs

@@ -417,11 +417,23 @@ impl Interval {
     }
 
     /// Casts this interval to `data_type` using `cast_options`.
+    /// Numeric endpoints that fail to cast become unbounded,
+    /// regardless of `cast_options.safe`. Other cast errors are propagated.
     pub fn cast_to(
         &self,
         data_type: &DataType,
         cast_options: &CastOptions,
     ) -> Result<Self> {
+        // Estimated endpoints may overflow even when all runtime values fit.
+        let bound_options = CastOptions {
+            safe: true,
+            ..cast_options.clone()
+        };
+        let cast_options = if self.data_type().is_numeric() && data_type.is_numeric() {
+            &bound_options
+        } else {
+            cast_options
+        };
         Self::try_new(
             cast_scalar_value(&self.lower, data_type, cast_options)?,
             cast_scalar_value(&self.upper, data_type, cast_options)?,
@@ -2261,16 +2273,220 @@ impl NullableInterval {
 mod tests {
     use crate::{
         interval_arithmetic::{
-            Interval, handle_overflow, next_value, prev_value, satisfy_greater,
+            Interval, cast_scalar_value, handle_overflow, next_value, prev_value,
+            satisfy_greater,
         },
         operator::Operator,
     };
 
     use crate::interval_arithmetic::NullableInterval;
+    use arrow::compute::CastOptions;
     use arrow::datatypes::DataType;
     use datafusion_common::rounding::{next_down, next_up};
     use datafusion_common::{Result, ScalarValue};
 
+    #[test]
+    fn test_numeric_cast_out_of_range_bounds() -> Result<()> {
+        for safe in [false, true] {
+            let options = CastOptions {
+                safe,
+                ..Default::default()
+            };
+            for source in [
+                DataType::Int64,
+                DataType::Float64,
+                DataType::Decimal128(10, 0),
+            ] {
+                for (lower, upper, expected_lower, expected_upper) in [
+                    (Some(-129i64), Some(42), None, Some(42i8)),
+                    (Some(-42), Some(128), Some(-42), None),
+                    (Some(-129), Some(128), None, None),
+                    (Some(-128), Some(127), Some(-128i8), Some(127i8)),
+                    (Some(128), Some(129), None, None),
+                    (Some(-130), Some(-129), None, None),
+                    (None, Some(42), None, Some(42)),
+                    (Some(-42), None, Some(-42), None),
+                ] {
+                    let input =
+                        Interval::make(lower, upper)?.cast_to(&source, &options)?;
+                    assert_eq!(
+                        input.cast_to(&DataType::Int8, &options)?,
+                        Interval::make(expected_lower, expected_upper)?,
+                        "{source}: {input}, safe={safe}"
+                    );
+                }
+            }
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn test_numeric_cast_bounds_across_types() -> Result<()> {
+        for safe in [false, true] {
+            let options = CastOptions {
+                safe,
+                ..Default::default()
+            };
+            // Cover unsigned, floating-point and decimal sources and targets.
+            for source in [
+                DataType::Int64,
+                DataType::UInt64,
+                DataType::Float64,
+                DataType::Decimal128(10, 2),
+                DataType::Decimal256(30, 3),
+            ] {
+                let input = Interval::make(Some(42i64), Some(1000))?
+                    .cast_to(&source, &options)?;
+                for target in
+                    [DataType::Int8, DataType::UInt8, DataType::Decimal128(3, 1)]
+                {
+                    let lower = ScalarValue::Int64(Some(42)).cast_to(&target)?;
+                    assert_eq!(
+                        input.cast_to(&target, &options)?,
+                        Interval::try_new(lower, ScalarValue::try_from(&target)?)?,
+                        "{source} -> {target}, safe={safe}"
+                    );
+                }
+            }
+            assert_eq!(
+                Interval::make(Some(-1i64), Some(42))?
+                    .cast_to(&DataType::UInt8, &options)?,
+                Interval::make(Some(0u8), Some(42))?
+            );
+
+            // Fractional inputs retain truncation, including with one overflowing bound.
+            for lower in [-129.5f64, -42.5] {
+                for source in [DataType::Float64, DataType::Decimal128(10, 2)] {
+                    let input = Interval::make(Some(lower), Some(42.5))?
+                        .cast_to(&source, &options)?;
+                    let expected_lower = if lower < -128.0 { None } else { Some(-42i8) };
+                    assert_eq!(
+                        input.cast_to(&DataType::Int8, &options)?,
+                        Interval::make(expected_lower, Some(42))?
+                    );
+                }
+            }
+            // Float overflow is normalized by Interval::try_new.
+            assert_eq!(
+                Interval::make(Some(0.0f64), Some(f64::MAX))?
+                    .cast_to(&DataType::Float32, &options)?,
+                Interval::make(Some(0.0f32), None)?
+            );
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn test_numeric_cast_bounds_contain_values() -> Result<()> {
+        let types = [
+            DataType::Int8,
+            DataType::Int16,
+            DataType::Int32,
+            DataType::Int64,
+            DataType::UInt8,
+            DataType::UInt16,
+            DataType::UInt32,
+            DataType::UInt64,
+            DataType::Float32,
+            DataType::Float64,
+            DataType::Decimal128(3, 0),
+            DataType::Decimal128(10, 2),
+            DataType::Decimal128(20, -2),
+            DataType::Decimal256(30, 3),
+        ];
+        let samples = [
+            i64::MIN,
+            -16777217,
+            -65537,
+            -129,
+            -128,
+            -1,
+            0,
+            1,
+            42,
+            127,
+            128,
+            255,
+            256,
+            65536,
+            16777217,
+            i64::MAX,
+        ];
+        let safe_options = CastOptions {
+            safe: true,
+            ..Default::default()
+        };
+        let cast = |value: &ScalarValue, target: &DataType| {
+            cast_scalar_value(value, target, &safe_options)
+        };
+        for source in &types {
+            let mut values = samples
+                .iter()
+                .map(|value| cast(&ScalarValue::from(*value), source))
+                .collect::<Result<Vec<_>>>()?;
+            values.retain(|value| !value.is_null());
+            for target in &types {
+                let converted = values
+                    .iter()
+                    .map(|value| cast(value, target))
+                    .collect::<Result<Vec<_>>>()?;
+                for safe in [false, true] {
+                    let options = CastOptions {
+                        safe,
+                        ..Default::default()
+                    };
+                    for start in 0..values.len() {
+                        for end in start..values.len() {
+                            let input = Interval::try_new(
+                                values[start].clone(),
+                                values[end].clone(),
+                            )?;
+                            let bounds = input.cast_to(target, &options)?;
+                            for value in &converted[start..=end] {
+                                if !value.is_null() {
+                                    assert!(
+                                        (bounds.lower.is_null()
+                                            || bounds.lower <= *value)
+                                            && (bounds.upper.is_null()
+                                                || *value <= bounds.upper),
+                                        "{source} -> {target}, {input} -> {bounds}, value={value}, safe={safe}"
+                                    );
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn test_non_numeric_cast_retains_error_policy() -> Result<()> {
+        let value = ScalarValue::Utf8(Some("not a number".into()));
+        let input = Interval::try_new(value.clone(), value)?;
+        let strict = CastOptions {
+            safe: false,
+            ..Default::default()
+        };
+        assert!(input.cast_to(&DataType::Int8, &strict).is_err());
+        let safe = CastOptions {
+            safe: true,
+            ..strict
+        };
+        assert_eq!(
+            input.cast_to(&DataType::Int8, &safe)?,
+            Interval::make_unbounded(&DataType::Int8)?
+        );
+        let numeric = Interval::make(Some(0i64), Some(1))?;
+        assert!(
+            numeric
+                .cast_to(&DataType::Struct(Default::default()), &safe)
+                .is_err()
+        );
+        Ok(())
+    }
+
     #[test]
     fn test_next_prev_value() -> Result<()> {
         let zeros = vec![

datafusion/sqllogictest/test_files/cast.slt

@@ -125,3 +125,59 @@ set datafusion.optimizer.expand_views_at_output = false;
 
 statement ok
 drop table t;
+
+# Regression for https://github.com/apache/datafusion/issues/25524
+
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+statement ok
+CREATE TABLE narrow_column (a INT, z TINYINT) AS VALUES (1, 0), (2, 0), (3, 0);
+
+# Inferred upper bound 1000 is out of range, but all runtime values are zero.
+query I
+SELECT a FROM narrow_column
+WHERE CAST(CAST(a < 0 AS INT) * 1000 AS TINYINT) = z
+ORDER BY a;
+----
+1
+2
+3
+
+# The same applies to the inferred lower bound -1000.
+query I
+SELECT a FROM narrow_column
+WHERE CAST(CAST(a < 0 AS INT) * -1000 AS TINYINT) = z
+ORDER BY a;
+----
+1
+2
+3
+
+# Real overflow must still fail at execution.
+statement error Can't cast value 1000 to type Int8
+SELECT CAST(CAST(a > 0 AS INT) * 1000 AS TINYINT) FROM narrow_column;
+
+statement error Can't cast value -1000 to type Int8
+SELECT CAST(CAST(a > 0 AS INT) * -1000 AS TINYINT) FROM narrow_column;
+
+# TRY_CAST must still return NULL for real overflow.
+query I
+SELECT TRY_CAST(CAST(a > 0 AS INT) * 1000 AS TINYINT) FROM narrow_column;
+----
+NULL
+NULL
+NULL
+
+query I
+SELECT TRY_CAST(CAST(a > 0 AS INT) * -1000 AS TINYINT) FROM narrow_column;
+----
+NULL
+NULL
+NULL
+
+statement ok
+DROP TABLE narrow_column;
+
+statement ok
+SET datafusion.execution.target_partitions = 4;