fix: fix string to timestamp cast for UTC timestamps

parthchandra · parthchandra · commit ff025161ba9b · 2026-03-20T16:47:08.000-07:00
diff --git a/native/spark-expr/src/conversion_funcs/string.rs b/native/spark-expr/src/conversion_funcs/string.rs
@@ -37,19 +37,29 @@ macro_rules! cast_utf8_to_timestamp {
     ($array:expr, $eval_mode:expr, $array_type:ty, $cast_method:ident, $tz:expr) => {{
         let len = $array.len();
         let mut cast_array = PrimitiveArray::<$array_type>::builder(len).with_timezone("UTC");
+        let mut cast_err: Option<SparkError> = None;
         for i in 0..len {
             if $array.is_null(i) {
                 cast_array.append_null()
-            } else if let Ok(Some(cast_value)) =
-                $cast_method($array.value(i).trim(), $eval_mode, $tz)
-            {
-                cast_array.append_value(cast_value);
             } else {
-                cast_array.append_null()
+                match $cast_method($array.value(i).trim(), $eval_mode, $tz) {
+                    Ok(Some(cast_value)) => cast_array.append_value(cast_value),
+                    Ok(None) => cast_array.append_null(),
+                    Err(e) => {
+                        if $eval_mode == EvalMode::Ansi {
+                            cast_err = Some(e);
+                            break;
+                        }
+                        cast_array.append_null()
+                    }
+                }
             }
         }
-        let result: ArrayRef = Arc::new(cast_array.finish()) as ArrayRef;
-        result
+        if let Some(e) = cast_err {
+            Err(e)
+        } else {
+            Ok(Arc::new(cast_array.finish()) as ArrayRef)
+        }
     }};
 }
 
@@ -668,15 +678,13 @@ pub(crate) fn cast_string_to_timestamp(
     let tz = &timezone::Tz::from_str(timezone_str).unwrap();
 
     let cast_array: ArrayRef = match to_type {
-        DataType::Timestamp(_, _) => {
-            cast_utf8_to_timestamp!(
-                string_array,
-                eval_mode,
-                TimestampMicrosecondType,
-                timestamp_parser,
-                tz
-            )
-        }
+        DataType::Timestamp(_, _) => cast_utf8_to_timestamp!(
+            string_array,
+            eval_mode,
+            TimestampMicrosecondType,
+            timestamp_parser,
+            tz
+        )?,
         _ => unreachable!("Invalid data type {:?} in cast from string", to_type),
     };
     Ok(cast_array)
@@ -961,6 +969,12 @@ fn get_timestamp_values<T: TimeZone>(
 ) -> SparkResult<Option<i64>> {
     let values: Vec<_> = value.split(['T', '-', ':', '.']).collect();
     let year = values[0].parse::<i32>().unwrap_or_default();
+
+    // NaiveDate (used internally by chrono's with_ymd_and_hms) is bounded to ±262142.
+    if !(-262143..=262142).contains(&year) {
+        return Ok(None);
+    }
+
     let month = values.get(1).map_or(1, |m| m.parse::<u32>().unwrap_or(1));
     let day = values.get(2).map_or(1, |d| d.parse::<u32>().unwrap_or(1));
     let hour = values.get(3).map_or(0, |h| h.parse::<u32>().unwrap_or(0));
@@ -1004,7 +1018,7 @@ fn get_timestamp_values<T: TimeZone>(
             .with_second(second)
             .with_microsecond(microsecond),
         _ => {
-            return Err(SparkError::CastInvalidValue {
+            return Err(SparkError::InvalidInputInCastToDatetime {
                 value: value.to_string(),
                 from_type: "STRING".to_string(),
                 to_type: "TIMESTAMP".to_string(),
@@ -1082,7 +1096,6 @@ fn parse_str_to_microsecond_timestamp<T: TimeZone>(
     get_timestamp_values(value, "microsecond", tz)
 }
 
-// used in tests only
 fn timestamp_parser<T: TimeZone>(
     value: &str,
     eval_mode: EvalMode,
@@ -1095,31 +1108,31 @@ fn timestamp_parser<T: TimeZone>(
     // Define regex patterns and corresponding parsing functions
     let patterns = &[
         (
-            Regex::new(r"^\d{4,5}$").unwrap(),
+            Regex::new(r"^\d{4,7}$").unwrap(),
             parse_str_to_year_timestamp as fn(&str, &T) -> SparkResult<Option<i64>>,
         ),
         (
-            Regex::new(r"^\d{4,5}-\d{2}$").unwrap(),
+            Regex::new(r"^\d{4,7}-\d{2}$").unwrap(),
             parse_str_to_month_timestamp,
         ),
         (
-            Regex::new(r"^\d{4,5}-\d{2}-\d{2}$").unwrap(),
+            Regex::new(r"^\d{4,7}-\d{2}-\d{2}$").unwrap(),
             parse_str_to_day_timestamp,
         ),
         (
-            Regex::new(r"^\d{4,5}-\d{2}-\d{2}T\d{1,2}$").unwrap(),
+            Regex::new(r"^\d{4,7}-\d{2}-\d{2}T\d{1,2}$").unwrap(),
             parse_str_to_hour_timestamp,
         ),
         (
-            Regex::new(r"^\d{4,5}-\d{2}-\d{2}T\d{2}:\d{2}$").unwrap(),
+            Regex::new(r"^\d{4,7}-\d{2}-\d{2}T\d{2}:\d{2}$").unwrap(),
             parse_str_to_minute_timestamp,
         ),
         (
-            Regex::new(r"^\d{4,5}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$").unwrap(),
+            Regex::new(r"^\d{4,7}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$").unwrap(),
             parse_str_to_second_timestamp,
         ),
         (
-            Regex::new(r"^\d{4,5}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{1,6}$").unwrap(),
+            Regex::new(r"^\d{4,7}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{1,6}$").unwrap(),
             parse_str_to_microsecond_timestamp,
         ),
         (
@@ -1140,7 +1153,7 @@ fn timestamp_parser<T: TimeZone>(
 
     if timestamp.is_none() {
         return if eval_mode == EvalMode::Ansi {
-            Err(SparkError::CastInvalidValue {
+            Err(SparkError::InvalidInputInCastToDatetime {
                 value: value.to_string(),
                 from_type: "STRING".to_string(),
                 to_type: "TIMESTAMP".to_string(),
@@ -1150,12 +1163,7 @@ fn timestamp_parser<T: TimeZone>(
         };
     }
 
-    match timestamp {
-        Some(ts) => Ok(Some(ts)),
-        None => Err(SparkError::Internal(
-            "Failed to parse timestamp".to_string(),
-        )),
-    }
+    Ok(timestamp)
 }
 
 fn parse_str_to_time_only_timestamp<T: TimeZone>(value: &str, tz: &T) -> SparkResult<Option<i64>> {
@@ -1202,17 +1210,20 @@ fn date_parser(date_str: &str, eval_mode: EvalMode) -> SparkResult<Option<i32>>
     }
 
     fn is_valid_digits(segment: i32, digits: usize) -> bool {
-        // An integer is able to represent a date within [+-]5 million years.
+        // NaiveDate is bounded to [-262142, 262142] (6 digits). We allow up to 7 digits to support
+        // leading-zero year strings like "0002020" (= year 2020), matching Spark's
+        // isValidDigits. Values outside the bounds are caught by an explicit bounds
+        // check below.
         let max_digits_year = 7;
-        //year (segment 0) can be between 4 to 7 digits,
-        //month and day (segment 1 and 2) can be between 1 to 2 digits
+        // year (segment 0) can be between 4 to 7 digits,
+        // month and day (segment 1 and 2) can be between 1 to 2 digits
         (segment == 0 && digits >= 4 && digits <= max_digits_year)
             || (segment != 0 && digits > 0 && digits <= 2)
     }
 
     fn return_result(date_str: &str, eval_mode: EvalMode) -> SparkResult<Option<i32>> {
         if eval_mode == EvalMode::Ansi {
-            Err(SparkError::CastInvalidValue {
+            Err(SparkError::InvalidInputInCastToDatetime {
                 value: date_str.to_string(),
                 from_type: "STRING".to_string(),
                 to_type: "DATE".to_string(),
@@ -1285,11 +1296,13 @@ fn date_parser(date_str: &str, eval_mode: EvalMode) -> SparkResult<Option<i32>>
 
     date_segments[current_segment as usize] = current_segment_value.0;
 
-    match NaiveDate::from_ymd_opt(
-        sign * date_segments[0],
-        date_segments[1] as u32,
-        date_segments[2] as u32,
-    ) {
+    // Reject out-of-range years explicitly
+    let year = sign * date_segments[0];
+    if !(-262143..=262142).contains(&year) {
+        return Ok(None);
+    }
+
+    match NaiveDate::from_ymd_opt(year, date_segments[1] as u32, date_segments[2] as u32) {
         Some(date) => {
             let duration_since_epoch = date
                 .signed_duration_since(DateTime::UNIX_EPOCH.naive_utc().date())
@@ -1341,7 +1354,8 @@ mod tests {
             TimestampMicrosecondType,
             timestamp_parser,
             tz
-        );
+        )
+        .unwrap();
 
         assert_eq!(
             result.data_type(),
@@ -1350,6 +1364,33 @@ mod tests {
         assert_eq!(result.len(), 4);
     }
 
+    #[test]
+    fn test_cast_string_to_timestamp_ansi_error() {
+        // In ANSI mode, an invalid timestamp string must produce an error rather than null.
+        let array: ArrayRef = Arc::new(StringArray::from(vec![
+            Some("2020-01-01T12:34:56.123456"),
+            Some("not_a_timestamp"),
+        ]));
+        let tz = &timezone::Tz::from_str("UTC").unwrap();
+        let string_array = array
+            .as_any()
+            .downcast_ref::<GenericStringArray<i32>>()
+            .expect("Expected a string array");
+
+        let eval_mode = EvalMode::Ansi;
+        let result = cast_utf8_to_timestamp!(
+            &string_array,
+            eval_mode,
+            TimestampMicrosecondType,
+            timestamp_parser,
+            tz
+        );
+        assert!(
+            result.is_err(),
+            "ANSI mode should return Err for an invalid timestamp string"
+        );
+    }
+
     #[test]
     fn test_cast_dict_string_to_timestamp() -> DataFusionResult<()> {
         // prepare input data
diff --git a/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala b/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala
@@ -100,7 +100,7 @@ object SparkErrorConverter extends ShimSparkErrorConverter {
       case None => Array.empty[QueryContext] // No context
     }
 
-    val summary: String = errorJson.summary.orNull
+    val summary: String = errorJson.summary.getOrElse("")
 
     // Delegate to version-specific shim - let conversion exceptions propagate
     val optEx = convertErrorType(errorJson.errorType, errorClass, params, sparkContext, summary)
diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -217,10 +217,7 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
         Compatible(Some("Only supports years between 262143 BC and 262142 AD"))
       case DataTypes.TimestampType if timeZoneId.exists(tz => tz != "UTC") =>
         Incompatible(Some(s"Cast will use UTC instead of $timeZoneId"))
-      case DataTypes.TimestampType if evalMode == CometEvalMode.ANSI =>
-        Incompatible(Some("ANSI mode not supported"))
       case DataTypes.TimestampType =>
-        // https://github.com/apache/datafusion-comet/issues/328
         Incompatible(Some("Not all valid formats are supported"))
       case _ =>
         unsupported(DataTypes.StringType, toType)
diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala
@@ -23,7 +23,7 @@ import java.io.FileNotFoundException
 
 import scala.util.matching.Regex
 
-import org.apache.spark.{QueryContext, SparkException}
+import org.apache.spark.{QueryContext, SparkDateTimeException, SparkException}
 import org.apache.spark.sql.catalyst.trees.SQLQueryContext
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.types._
@@ -172,6 +172,22 @@ trait ShimSparkErrorConverter {
           QueryExecutionErrors
             .invalidInputInCastToNumberError(targetType, str, sqlCtx(context)))
 
+      case "InvalidInputInCastToDatetime" =>
+        val expression =
+          s"'${params("value").toString.replace("\\", "\\\\").replace("'", "\\'")}'"
+        val sourceType = s""""${params("fromType").toString}""""
+        val targetType = s""""${params("toType").toString}""""
+        Some(
+          new SparkDateTimeException(
+            errorClass = "CAST_INVALID_INPUT",
+            messageParameters = Map(
+              "expression" -> expression,
+              "sourceType" -> sourceType,
+              "targetType" -> targetType,
+              "ansiConfig" -> "\"spark.sql.ansi.enabled\""),
+            context = context,
+            summary = summary))
+
       case "CastOverFlow" =>
         val fromType = getDataType(params("fromType").toString)
         val toType = getDataType(params("toType").toString)
diff --git a/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala
@@ -23,7 +23,7 @@ import java.io.FileNotFoundException
 
 import scala.util.matching.Regex
 
-import org.apache.spark.{QueryContext, SparkException}
+import org.apache.spark.{QueryContext, SparkDateTimeException, SparkException}
 import org.apache.spark.sql.catalyst.trees.SQLQueryContext
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.types._
@@ -170,6 +170,22 @@ trait ShimSparkErrorConverter {
           QueryExecutionErrors
             .invalidInputInCastToNumberError(targetType, str, sqlCtx(context)))
 
+      case "InvalidInputInCastToDatetime" =>
+        val expression =
+          s"'${params("value").toString.replace("\\", "\\\\").replace("'", "\\'")}'"
+        val sourceType = s""""${params("fromType").toString}""""
+        val targetType = s""""${params("toType").toString}""""
+        Some(
+          new SparkDateTimeException(
+            errorClass = "CAST_INVALID_INPUT",
+            messageParameters = Map(
+              "expression" -> expression,
+              "sourceType" -> sourceType,
+              "targetType" -> targetType,
+              "ansiConfig" -> "\"spark.sql.ansi.enabled\""),
+            context = context,
+            summary = summary))
+
       case "CastOverFlow" =>
         val fromType = getDataType(params("fromType").toString)
         val toType = getDataType(params("toType").toString)
diff --git a/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala
@@ -182,6 +182,13 @@ trait ShimSparkErrorConverter {
           QueryExecutionErrors
             .invalidInputInCastToNumberError(targetType, str, context.headOption.orNull))
 
+      case "InvalidInputInCastToDatetime" =>
+        val str = UTF8String.fromString(params("value").toString)
+        val targetType = getDataType(params("toType").toString)
+        Some(
+          QueryExecutionErrors
+            .invalidInputInCastToDatetimeError(str, targetType, context.headOption.orNull))
+
       case "CastOverFlow" =>
         val fromType = getDataType(params("fromType").toString)
         val toType = getDataType(params("toType").toString)
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala

Original file line number	Diff line number	Diff line change
`@@ -100,7 +100,7 @@ object SparkErrorConverter extends ShimSparkErrorConverter {`
`100`	`100`	`case None => Array.empty[QueryContext] // No context`
`101`	`101`	`}`
`102`	`102`
`103`		`- val summary: String = errorJson.summary.orNull`
	`103`	`+ val summary: String = errorJson.summary.getOrElse("")`
`104`	`104`
`105`	`105`	`// Delegate to version-specific shim - let conversion exceptions propagate`
`106`	`106`	`val optEx = convertErrorType(errorJson.errorType, errorClass, params, sparkContext, summary)`