int_timestamp_cast_support

Author: coderfender

native/spark-expr/src/conversion_funcs/cast.rs

@@ -620,6 +620,9 @@ macro_rules! cast_int_to_timestamp_impl {
             if arr.is_null(i) {
                 $builder.append_null();
             } else {
+                // saturating_mul limits to i64::MIN/MAX on overflow instead of panicking,
+                // which could occur when converting extreme values (e.g., Long.MIN_VALUE)
+                // matching spark behavior (irrespective of EvalMode)
                 let micros = (arr.value(i) as i64).saturating_mul(MICROS_PER_SECOND);
                 $builder.append_value(micros);
             }

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

@@ -65,6 +65,23 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   lazy val usingParquetExecWithIncompatTypes: Boolean =
     hasUnsignedSmallIntSafetyCheck(conf)
 
+  // Timezone list to check temporal type casts
+  private val compatibleTimezones = Seq(
+    "UTC",
+    "America/New_York",
+    "America/Chicago",
+    "America/Denver",
+    "America/Los_Angeles",
+    "Europe/London",
+    "Europe/Paris",
+    "Europe/Berlin",
+    "Asia/Tokyo",
+    "Asia/Shanghai",
+    "Asia/Singapore",
+    "Asia/Kolkata",
+    "Australia/Sydney",
+    "Pacific/Auckland")
+
   test("all valid cast combinations covered") {
     val names = testNames
 
@@ -224,13 +241,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast ByteType to TimestampType") {
-    val compatibleTimezones = Seq(
-      "UTC",
-      "America/New_York",
-      "America/Los_Angeles",
-      "Europe/London",
-      "Asia/Tokyo",
-      "Australia/Sydney")
     compatibleTimezones.foreach { tz =>
       withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
         castTest(
@@ -311,13 +321,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast ShortType to TimestampType") {
-    val compatibleTimezones = Seq(
-      "UTC",
-      "America/New_York",
-      "America/Los_Angeles",
-      "Europe/London",
-      "Asia/Tokyo",
-      "Australia/Sydney")
     compatibleTimezones.foreach { tz =>
       withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
         castTest(
@@ -384,13 +387,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast IntegerType to TimestampType") {
-    val compatibleTimezones = Seq(
-      "UTC",
-      "America/New_York",
-      "America/Los_Angeles",
-      "Europe/London",
-      "Asia/Tokyo",
-      "Australia/Sydney")
     compatibleTimezones.foreach { tz =>
       withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
         castTest(generateInts(), DataTypes.TimestampType)
@@ -443,19 +439,10 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   test("cast LongType to TimestampType") {
     // Use assertDataFrameEquals because extreme Long values (Long.MIN_VALUE, Long.MAX_VALUE)
     // overflow when converted to java.sql.Timestamp during collect(), but the cast itself works.
-    val compatibleTimezones = Seq(
-      "UTC",
-      "America/New_York",
-      "America/Los_Angeles",
-      "Europe/London",
-      "Asia/Tokyo",
-      "Australia/Sydney")
     compatibleTimezones.foreach { tz =>
       withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
-        withTempPath { dir =>
-          val input = generateLongs()
-          val data = roundtripParquet(input, dir).coalesce(1)
-          val df = data.withColumn("ts", col("a").cast(DataTypes.TimestampType))
+        withParquetTable(generateLongs(), "t1") {
+          val df = spark.table("t1").withColumn("ts", col("a").cast(DataTypes.TimestampType))
           assertDataFrameEquals(df)
         }
       }

spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala

@@ -332,7 +332,7 @@ abstract class CometTestBase
     }
   }
 
-//  inspired from spark-testing-base
+  // inspired from spark-testing-base
   protected def assertDataFrameEquals(
       df: => DataFrame,
       checkNativeOperators: Boolean = true): Unit = {
@@ -343,12 +343,12 @@ abstract class CometTestBase
     }
     val cometDf = datasetOfRows(spark, df.logicalPlan)
 
-//    schema match check
+    // schema match check
     assert(
       sparkDf.schema == cometDf.schema,
       s"Schemas do not match.\nCorrect Answer: ${sparkDf.schema}\n Spark Answer: ${cometDf.schema}")
 
-//    diff check using except instead of collect (collect errors out on spark driver for long -> timestamp conv)
+    // diff check using except instead of collect (collect errors out on spark driver for long -> timestamp conv)
     val sparkMinusComet = sparkDf.except(cometDf)
     val cometMinusSpark = cometDf.except(sparkDf)