fix: TRY_CAST returns NULL for timestamp/date overflow

datafusion/common/src/scalar/mod.rs

@@ -4292,7 +4292,13 @@ impl ScalarValue {
             .or_else(|| timestamp_to_timestamp_multiplier(&source_type, target_type))
             && let Some(value) = self.temporal_scalar_value_as_i64()
         {
-            ensure_timestamp_in_bounds(value, multiplier, &source_type, target_type)?;
+            if cast_options.safe {
+                if multiplier > 1 && value.checked_mul(multiplier).is_none() {
+                    return ScalarValue::try_new_null(target_type);
+                }
+            } else {
+                ensure_timestamp_in_bounds(value, multiplier, &source_type, target_type)?;
+            }
         }
 
         let scalar_array = self.to_array()?;
@@ -10190,6 +10196,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn safe_cast_date_to_timestamp_overflow_returns_null() {
+        let scalar = ScalarValue::Date32(Some(i32::MAX));
+        let safe_options = CastOptions {
+            safe: true,
+            ..DEFAULT_CAST_OPTIONS
+        };
+
+        let casted = scalar
+            .cast_to_with_options(
+                &DataType::Timestamp(TimeUnit::Nanosecond, None),
+                &safe_options,
+            )
+            .expect("expected safe cast to return null");
+
+        assert_eq!(casted, ScalarValue::TimestampNanosecond(None, None));
+    }
+
     #[test]
     fn cast_timestamp_to_timestamp_overflow_returns_error() {
         let scalar = ScalarValue::TimestampSecond(Some(i64::MAX), None);
@@ -10203,6 +10227,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn safe_cast_timestamp_to_timestamp_overflow_returns_null() {
+        let scalar = ScalarValue::TimestampSecond(Some(i64::MAX), None);
+        let safe_options = CastOptions {
+            safe: true,
+            ..DEFAULT_CAST_OPTIONS
+        };
+
+        let casted = scalar
+            .cast_to_with_options(
+                &DataType::Timestamp(TimeUnit::Nanosecond, None),
+                &safe_options,
+            )
+            .expect("expected safe cast to return null");
+
+        assert_eq!(casted, ScalarValue::TimestampNanosecond(None, None));
+    }
+
     #[test]
     fn null_dictionary_scalar_produces_null_dictionary_array() {
         let dictionary_scalar = ScalarValue::Dictionary(

datafusion/expr-common/src/columnar_value.rs

@@ -325,7 +325,9 @@ fn cast_array_by_name(
     ) {
         datafusion_common::nested_struct::cast_column(array, cast_type, cast_options)
     } else {
-        ensure_temporal_array_timestamp_bounds(array, cast_type)?;
+        if !cast_options.safe {
+            ensure_temporal_array_timestamp_bounds(array, cast_type)?;
+        }
         Ok(kernels::cast::cast_with_options(
             array,
             cast_type,
@@ -766,4 +768,32 @@ mod tests {
             "unexpected error: {err}"
         );
     }
+
+    #[test]
+    fn safe_cast_timestamp_array_to_timestamp_overflow_returns_null() {
+        let overflow_value = i64::MAX / 1_000_000_000 + 1;
+        let array: ArrayRef =
+            Arc::new(TimestampSecondArray::from(vec![Some(overflow_value)]));
+        let value = ColumnarValue::Array(array);
+        let safe_options = CastOptions {
+            safe: true,
+            ..DEFAULT_CAST_OPTIONS
+        };
+
+        let casted = value
+            .cast_to(
+                &DataType::Timestamp(TimeUnit::Nanosecond, None),
+                Some(&safe_options),
+            )
+            .expect("expected safe cast to return null");
+
+        let ColumnarValue::Array(array) = casted else {
+            panic!("expected array after cast");
+        };
+        let array = array
+            .as_any()
+            .downcast_ref::<TimestampNanosecondArray>()
+            .expect("expected TimestampNanosecondArray");
+        assert!(array.is_null(0));
+    }
 }

datafusion/sqllogictest/test_files/datetime/timestamps.slt

@@ -5379,6 +5379,25 @@ SELECT to_timestamp(arrow_cast(-9223372036, 'Int64'));
 query error converted value exceeds the representable i64 range
 SELECT to_timestamp(arrow_cast(9223372037, 'Int64'));
 
+# TRY_CAST returns NULL for timestamp/date casts that overflow
+query P
+SELECT TRY_CAST(arrow_cast(9223372037, 'Timestamp(s)') AS TIMESTAMP(9));
+----
+NULL
+
+query P
+SELECT TRY_CAST(DATE '3000-01-01' AS TIMESTAMP(9));
+----
+NULL
+
+query P
+SELECT TRY_CAST(ts AS TIMESTAMP(9)) AS ts
+FROM (
+    VALUES (arrow_cast(9223372037, 'Timestamp(s)'))
+) t(ts);
+----
+NULL
+
 # Float truncation behavior
 query P
 SELECT to_timestamp_seconds(arrow_cast(-1.9, 'Float64'));