feat: Support Spark expression: convert_timezone (#4369)

Author: YutaLin

docs/source/contributor-guide/spark_expressions_support.md

@@ -215,7 +215,7 @@
 ### datetime_funcs
 
 - [ ] add_months
-- [ ] convert_timezone
+- [x] convert_timezone
 - [ ] curdate
 - [ ] current_date
 - [ ] current_time

docs/source/user-guide/latest/expressions.md

@@ -101,6 +101,7 @@ of expressions that be disabled.
 
 | Expression       | SQL                          |
 | ---------------- | ---------------------------- |
+| ConvertTimezone  | `convert_timezone`           |
 | CurrentTimeZone  | `current_timezone`           |
 | DateAdd          | `date_add`                   |
 | DateDiff         | `datediff`                   |

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

@@ -217,6 +217,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
 
   private[comet] val temporalExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
     Map(
+      classOf[ConvertTimezone] -> CometConvertTimezone,
       classOf[DateAdd] -> CometDateAdd,
       classOf[DateDiff] -> CometDateDiff,
       classOf[DateFormatClass] -> CometDateFormat,

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, DateFromUnixDate, DateSub, DayOfMonth, DayOfWeek, DayOfYear, Days, FromUTCTimestamp, GetDateField, Hour, Hours, LastDay, Literal, MakeDate, Minute, Month, NextDay, Quarter, Second, SecondsToTimestamp, ToUTCTimestamp, TruncDate, TruncTimestamp, UnixDate, UnixTimestamp, WeekDay, WeekOfYear, Year}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, ConvertTimezone, DateAdd, DateDiff, DateFormatClass, DateFromUnixDate, DateSub, DayOfMonth, DayOfWeek, DayOfYear, Days, FromUTCTimestamp, GetDateField, Hour, Hours, LastDay, Literal, MakeDate, Minute, Month, NextDay, Quarter, Second, SecondsToTimestamp, ToUTCTimestamp, TruncDate, TruncTimestamp, UnixDate, UnixTimestamp, WeekDay, WeekOfYear, Year}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DateType, DoubleType, FloatType, IntegerType, LongType, StringType, TimestampNTZType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -409,6 +409,27 @@ object CometToUTCTimestamp extends CometExpressionSerde[ToUTCTimestamp] {
   }
 }
 
+object CometConvertTimezone extends CometExpressionSerde[ConvertTimezone] {
+
+  override def getSupportLevel(expr: ConvertTimezone): SupportLevel =
+    Incompatible(Some(UTCTimestampSerde.tzParseIncompatReason))
+
+  override def getIncompatibleReasons(): Seq[String] =
+    Seq(UTCTimestampSerde.tzParseIncompatReason)
+
+  override def convert(
+      expr: ConvertTimezone,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    val srcTz = exprToProtoInternal(expr.sourceTz, inputs, binding)
+    val tgtTz = exprToProtoInternal(expr.targetTz, inputs, binding)
+    val ts = exprToProtoInternal(expr.sourceTs, inputs, binding)
+    val toUtc = scalarFunctionExprToProto("to_utc_timestamp", ts, srcTz)
+    val fromUtc = scalarFunctionExprToProto("from_utc_timestamp", toUtc, tgtTz)
+    optExprWithInfo(fromUtc, expr, expr.children: _*)
+  }
+}
+
 object CometNextDay extends CometScalarFunction[NextDay]("next_day")
 
 object CometMakeDate extends CometScalarFunction[MakeDate]("make_date")

spark/src/test/resources/sql-tests/expressions/datetime/convert_timezone.sql

@@ -0,0 +1,49 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Config: spark.comet.expression.ConvertTimezone.allowIncompatible=true
+-- ConfigMatrix: spark.sql.session.timeZone=UTC,America/Los_Angeles
+
+statement
+CREATE TABLE test_convert_timezone(ts timestamp_ntz, src string, tgt string) USING parquet
+
+statement
+INSERT INTO test_convert_timezone VALUES
+  (timestamp_ntz'2021-12-06 08:00:00', 'UTC', 'America/Los_Angeles'),
+  (timestamp_ntz'2021-07-01 12:00:00', 'America/New_York', 'Asia/Tokyo'),
+  (timestamp_ntz'2023-01-15 09:30:00', 'America/Los_Angeles', 'UTC'),
+  (NULL, 'UTC', 'Asia/Tokyo'),
+  (timestamp_ntz'2021-12-06 08:00:00', NULL, 'Asia/Tokyo'),
+  (timestamp_ntz'2021-12-06 08:00:00', 'UTC', NULL)
+
+query
+SELECT convert_timezone('UTC', 'America/Los_Angeles', timestamp_ntz'2021-12-06 08:00:00')
+
+query
+SELECT convert_timezone('Asia/Tokyo', 'Europe/Berlin', timestamp_ntz'2021-12-06 12:00:00')
+
+query
+SELECT convert_timezone('America/Los_Angeles', 'Asia/Tokyo', timestamp_ntz'2023-01-15 20:00:00')
+
+query
+SELECT convert_timezone(CAST(NULL AS STRING), 'Asia/Tokyo', timestamp_ntz'2021-12-06 08:00:00')
+
+query
+SELECT convert_timezone('UTC', CAST(NULL AS STRING), timestamp_ntz'2021-12-06 08:00:00')
+
+query
+SELECT convert_timezone(src, tgt, ts) FROM test_convert_timezone