feat: Support Spark expression window time

Author: 0lai0

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -215,7 +215,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[WeekDay] -> CometWeekDay,
     classOf[DayOfYear] -> CometDayOfYear,
     classOf[WeekOfYear] -> CometWeekOfYear,
-    classOf[Quarter] -> CometQuarter)
+    classOf[Quarter] -> CometQuarter,
+    classOf[PreciseTimestampConversion] -> CometPreciseTimestampConversion)
 
   private val conversionExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = Map(
     classOf[Cast] -> CometCast)

spark/src/main/scala/org/apache/comet/serde/datetime.scala

@@ -21,7 +21,7 @@ package org.apache.comet.serde
 
 import java.util.Locale
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, DateAdd, DateDiff, DateFormatClass, DateSub, DayOfMonth, DayOfWeek, DayOfYear, GetDateField, Hour, LastDay, Literal, MakeDate, Minute, Month, NextDay, Quarter, Second, TruncDate, TruncTimestamp, UnixDate, UnixTimestamp, WeekDay, WeekOfYear, Year}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, DateAdd, DateDiff, DateFormatClass, DateSub, DayOfMonth, DayOfWeek, DayOfYear, GetDateField, Hour, LastDay, Literal, MakeDate, Minute, Month, NextDay, PreciseTimestampConversion, Quarter, Second, TruncDate, TruncTimestamp, UnixDate, UnixTimestamp, WeekDay, WeekOfYear, Year}
 import org.apache.spark.sql.types.{DateType, IntegerType, StringType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -586,3 +586,28 @@ object CometDateFormat extends CometExpressionSerde[DateFormatClass] {
     }
   }
 }
+
+object CometPreciseTimestampConversion extends CometExpressionSerde[PreciseTimestampConversion] {
+  override def convert(
+      expr: PreciseTimestampConversion,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    // PreciseTimestampConversion reinterprets between LongType and TimestampType
+    // without changing the underlying microsecond value, so a simple cast suffices.
+    for {
+      childExpr <- exprToProtoInternal(expr.child, inputs, binding)
+      dt <- serializeDataType(expr.toType)
+    } yield {
+      ExprOuterClass.Expr
+        .newBuilder()
+        .setCast(
+          ExprOuterClass.Cast
+            .newBuilder()
+            .setChild(childExpr)
+            .setDatatype(dt)
+            .setEvalMode(ExprOuterClass.EvalMode.LEGACY)
+            .setAllowIncompat(false))
+        .build()
+    }
+  }
+}

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

@@ -2992,4 +2992,27 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("window_time") {
+    withTable("t1") {
+      sql("create table t1(time timestamp, value int) using parquet")
+      sql(
+        "insert into t1 values" +
+          "(cast('2023-01-01 12:00:00' as timestamp), 1)," +
+          "(cast('2023-01-01 12:05:00' as timestamp), 2)," +
+          "(cast('2023-01-01 12:15:00' as timestamp), 3)")
+
+      // basic window_time with aggregation
+      checkSparkAnswer(
+        "select max(window_time(window)), sum(value) " +
+          "from (select window(time, '10 minutes') as window, value from t1) " +
+          "group by window")
+
+      // window_time with sliding window
+      checkSparkAnswer(
+        "select max(window_time(window)), count(value) " +
+          "from (select window(time, '10 minutes', '5 minutes') as window, value from t1) " +
+          "group by window")
+    }
+  }
+
 }