non_numeric_to_timestamp

Author: coderfender

native/spark-expr/benches/cast_non_int_numeric_timestamp.rs

@@ -50,16 +50,23 @@ fn criterion_benchmark(c: &mut Criterion) {
     // Boolean -> Timestamp
     let batch_bool = create_boolean_batch();
     let expr_bool = Arc::new(Column::new("a", 0));
-    let cast_bool_to_ts = Cast::new(expr_bool, timestamp_type.clone(), spark_cast_options.clone());
+    let cast_bool_to_ts = Cast::new(
+        expr_bool,
+        timestamp_type.clone(),
+        spark_cast_options.clone(),
+    );
     group.bench_function("cast_bool_to_timestamp", |b| {
         b.iter(|| cast_bool_to_ts.evaluate(&batch_bool).unwrap());
     });
 
     // Decimal128 -> Timestamp
     let batch_decimal = create_decimal128_batch();
     let expr_decimal = Arc::new(Column::new("a", 0));
-    let cast_decimal_to_ts =
-        Cast::new(expr_decimal, timestamp_type.clone(), spark_cast_options.clone());
+    let cast_decimal_to_ts = Cast::new(
+        expr_decimal,
+        timestamp_type.clone(),
+        spark_cast_options.clone(),
+    );
     group.bench_function("cast_decimal_to_timestamp", |b| {
         b.iter(|| cast_decimal_to_ts.evaluate(&batch_decimal).unwrap());
     });
@@ -117,7 +124,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();

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

@@ -3644,11 +3644,8 @@ mod tests {
         ];
 
         for tz in &timezones {
-            let bool_array: ArrayRef = Arc::new(BooleanArray::from(vec![
-                Some(true),
-                Some(false),
-                None,
-            ]));
+            let bool_array: ArrayRef =
+                Arc::new(BooleanArray::from(vec![Some(true), Some(false), None]));
 
             let result = cast_boolean_to_timestamp(&bool_array, tz).unwrap();
             let ts_array = result.as_primitive::<TimestampMicrosecondType>();

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

@@ -70,8 +70,6 @@ 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 {
           val childExpr = exprToProtoInternal(cast.child, inputs, binding)
           if (childExpr.isDefined) {
@@ -84,17 +82,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,39 @@ 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("DataFrames count doesnt match.\n")
+    }
+
+    if (assertCometNative) {
+      checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan))
+    }
+  }
 }