non_numeric_to_timestamp

Author: B Vadlamani

native/spark-expr/benches/cast_non_int_numeric_timestamp.rs

@@ -117,7 +117,7 @@ fn create_decimal128_batch() -> RecordBatch {
         if i % 10 == 0 {
             b.append_null();
         } else {
-            b.append_value(rand::random::<i64>() as i128);
+            b.append_value(i as i128 * 1_000_000);
         }
     }
     let array = b.finish().with_precision_and_scale(18, 6).unwrap();

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

@@ -70,9 +70,8 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
       case _ =>
         if (isAlwaysCastToNull(cast.child.dataType, cast.dataType, cometEvalMode)) {
           exprToProtoInternal(Literal.create(null, cast.dataType), inputs, binding)
-        } else if (isAlwaysCastToUTC(cast.child.dataType, cast.dataType, cometEvalMode)) {
-          exprToProtoInternal(Literal.create(0L, cast.dataType), inputs, binding)
-        } else {
+        }
+        else {
           val childExpr = exprToProtoInternal(cast.child, inputs, binding)
           if (childExpr.isDefined) {
             castToProto(cast, cast.timeZoneId, cast.dataType, childExpr.get, cometEvalMode)
@@ -84,17 +83,6 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
     }
   }
 
-  private def isAlwaysCastToUTC(
-      fromType: DataType,
-      toType: DataType,
-      evalMode: CometEvalMode.Value): Boolean = {
-    (fromType, toType) match {
-      case (DataTypes.BooleanType, DataTypes.TimestampType) if evalMode == CometEvalMode.ANSI =>
-        true
-      case _ => false
-    }
-  }
-
 //  Some casts like date -> int/byte / long are always null. Terminate early in planning
   private def isAlwaysCastToNull(
       fromType: DataType,

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

@@ -167,7 +167,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     castTest(generateBools(), DataTypes.StringType)
   }
 
-  ignore("cast BooleanType to TimestampType") {
+  test("cast BooleanType to TimestampType") {
     // Arrow error: Cast error: Casting from Boolean to Timestamp(Microsecond, Some("UTC")) not supported
     castTest(generateBools(), DataTypes.TimestampType)
   }
@@ -507,7 +507,11 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   test("cast FloatType to TimestampType") {
     compatibleTimezones.foreach { tz =>
       withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
-        castTest(generateFloats(), DataTypes.TimestampType)
+        withTable("t1") {
+          generateFloats().write.saveAsTable("t1")
+          val df = spark.sql("select a, cast(a as timestamp) from t1")
+          assertDataFrameEquals(df)
+        }
       }
     }
   }
@@ -564,13 +568,12 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast DoubleType to TimestampType") {
-    // Cast back to double avoids java.sql.Timestamp overflow during collect() for extreme values
     compatibleTimezones.foreach { tz =>
       withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
         withTable("t1") {
-          generateLongs().write.saveAsTable("t1")
-          val df = spark.sql("select a, cast(cast(a as timestamp) as double) from t1")
-          checkSparkAnswerAndOperator(df)
+          generateDoubles().write.saveAsTable("t1")
+          val df = spark.sql("select a, cast(a as timestamp) from t1")
+          assertDataFrameEquals(df)
         }
       }
     }
@@ -638,8 +641,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     castTest(generateDecimalsPrecision10Scale2(), DataTypes.StringType)
   }
 
-  ignore("cast DecimalType(10,2) to TimestampType") {
-    // input: -123456.789000000000000000, expected: 1969-12-30 05:42:23.211, actual: 1969-12-31 15:59:59.876544
+  test("cast DecimalType(10,2) to TimestampType") {
     castTest(generateDecimalsPrecision10Scale2(), DataTypes.TimestampType)
   }
 

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

@@ -1276,4 +1276,40 @@ abstract class CometTestBase
     !usingLegacyNativeCometScan(conf) &&
     CometConf.COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK.get(conf)
   }
+
+  /**
+   * Uses except (difference) to find differences without using collect()
+   * Checks cometDF and sparkDF including schemas
+   */
+  protected def assertDataFrameEquals(
+      df: => DataFrame,
+      assertCometNative: Boolean = true): Unit = {
+
+    var sparkDf: DataFrame = null
+    withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+      sparkDf = datasetOfRows(spark, df.logicalPlan)
+    }
+    val cometDf = datasetOfRows(spark, df.logicalPlan)
+
+    // Compare schemas
+    assert(
+      sparkDf.schema == cometDf.schema,
+      s"Schema mismatch:\nCorrect Answer: ${sparkDf.schema}\nSpark Answer: ${cometDf.schema}")
+
+    // Use except (difference) to compare DataFrames without collect() which error on extremely high Timestamp values
+    val sparkMinusComet = sparkDf.except(cometDf)
+    val cometMinusSpark = cometDf.except(sparkDf)
+
+    val diffCount1 = sparkMinusComet.count()
+    val diffCount2 = cometMinusSpark.count()
+
+    if (diffCount1 > 0 || diffCount2 > 0) {
+      fail(
+        s"DataFrames count doesnt match.\n" )
+    }
+
+    if (assertCometNative) {
+      checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan))
+    }
+  }
 }