Enable for UTC

Author: parthchandra

spark/src/main/scala/org/apache/comet/expressions/CometCast.scala

@@ -217,11 +217,8 @@ 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"))
+        Compatible(Some("Only supports years between 262143 BC and 262142 AD"))
       case _ =>
         unsupported(DataTypes.StringType, toType)
     }

spark/src/test/scala/org/apache/comet/CometCastSuite.scala

@@ -954,19 +954,40 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
-  test("cast StringType to TimestampType disabled by default") {
-    withSQLConf((SQLConf.SESSION_LOCAL_TIMEZONE.key, "UTC")) {
-      val values = Seq("2020-01-01T12:34:56.123456", "T2").toDF("a")
-      castFallbackTest(
-        values.toDF("a"),
-        DataTypes.TimestampType,
-        "Not all valid formats are supported")
+  test("cast StringType to TimestampType - UTC") {
+    withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+      val values = Seq(
+        "2020",
+        "2020-01",
+        "2020-01-01",
+        "2020-01-01T12",
+        "2020-01-01T12:34",
+        "2020-01-01T12:34:56",
+        "2020-01-01T12:34:56.123456",
+        "T2",
+        "-9?",
+        "0100",
+        "0100-01",
+        "0100-01-01",
+        "0100-01-01T12",
+        "0100-01-01T12:34",
+        "0100-01-01T12:34:56",
+        "0100-01-01T12:34:56.123456",
+        "10000",
+        "10000-01",
+        "10000-01-01",
+        "10000-01-01T12",
+        "10000-01-01T12:34",
+        "10000-01-01T12:34:56",
+        "10000-01-01T12:34:56.123456")
+      castTimestampTest(values.toDF("a"), DataTypes.TimestampType, assertNative = true)
     }
   }
 
   ignore("cast StringType to TimestampType") {
-    // https://github.com/apache/datafusion-comet/issues/328
-    withSQLConf((CometConf.getExprAllowIncompatConfigKey(classOf[Cast]), "true")) {
+    // TODO: enable once all Spark timestamp formats are supported natively
+    // (e.g. time-only formats like "T12:34" are accepted by Spark but not yet by Comet)
+    withSQLConf((SQLConf.SESSION_LOCAL_TIMEZONE.key, "UTC")) {
       val values = Seq("2020-01-01T12:34:56.123456", "T2") ++ gen.generateStrings(
         dataSize,
         timestampPattern,
@@ -1021,12 +1042,10 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
       castTimestampTest(values.toDF("a"), DataTypes.TimestampType)
     }
 
-    // test for invalid inputs
-    withSQLConf(
-      SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
-      CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) -> "true") {
+    // test for invalid inputs with UTC - asserts native execution
+    withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
       val values = Seq("-9?", "1-", "0.5")
-      castTimestampTest(values.toDF("a"), DataTypes.TimestampType)
+      castTimestampTest(values.toDF("a"), DataTypes.TimestampType, assertNative = true)
     }
   }
 
@@ -1492,25 +1511,30 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
-  private def castTimestampTest(input: DataFrame, toType: DataType) = {
+  private def castTimestampTest(
+      input: DataFrame,
+      toType: DataType,
+      assertNative: Boolean = false) = {
     withTempPath { dir =>
       val data = roundtripParquet(input, dir).coalesce(1)
       data.createOrReplaceTempView("t")
 
       withSQLConf((SQLConf.ANSI_ENABLED.key, "false")) {
         // cast() should return null for invalid inputs when ansi mode is disabled
         val df = data.withColumn("converted", col("a").cast(toType))
-        checkSparkAnswer(df)
+        if (assertNative) {
+          checkSparkAnswerAndOperator(df)
+        } else {
+          checkSparkAnswer(df)
+        }
 
         // try_cast() should always return null for invalid inputs
         val df2 = spark.sql(s"select try_cast(a as ${toType.sql}) from t")
         checkSparkAnswer(df2)
       }
 
       // with ANSI enabled, we should produce the same exception as Spark
-      withSQLConf(
-        SQLConf.ANSI_ENABLED.key -> "true",
-        CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) -> "true") {
+      withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
         val df = data.withColumn("converted", col("a").cast(toType))
         checkSparkAnswerMaybeThrows(df) match {
           case (None, None) =>