feat: Add from_unixtime support (#1943)

Added from_unixtime support
Added a unit test

Author: kazuyukitanimura

docs/source/user-guide/expressions.md

@@ -222,3 +222,4 @@ The following Spark expressions are currently available. Any known compatibility
 | Coalesce                |                                                                                 |
 | NormalizeNaNAndZero     |                                                                                 |
 | ToPrettyString          |                                                                                 |
+| FromUnixTime            | Does not support format, supports only -8334601211038 <= sec <= 8210266876799   |

docs/spark_expressions_support.md

@@ -175,7 +175,7 @@
  - [ ] dayofweek
  - [ ] dayofyear
  - [x] extract
- - [ ] from_unixtime
+ - [x] from_unixtime
  - [ ] from_utc_timestamp
  - [ ] hour
  - [ ] last_day

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

@@ -84,6 +84,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[Chr] -> CometChr,
     classOf[InitCap] -> CometInitCap,
     classOf[BitLength] -> CometBitLength,
+    classOf[FromUnixTime] -> CometFromUnixTime,
     classOf[Length] -> CometLength,
     classOf[StringInstr] -> CometStringInstr,
     classOf[StringRepeat] -> CometStringRepeat,

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

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+package org.apache.comet.serde
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, FromUnixTime, Literal}
+import org.apache.spark.sql.catalyst.util.TimestampFormatter
+
+import org.apache.comet.CometSparkSessionExtensions.withInfo
+import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto}
+
+// TODO: DataFusion supports only -8334601211038 <= sec <= 8210266876799
+// https://github.com/apache/datafusion/issues/16594
+object CometFromUnixTime extends CometExpressionSerde with IncompatExpr {
+  override def convert(
+      expr: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    expr match {
+      case FromUnixTime(sec, format, timeZoneId) =>
+        val secExpr = exprToProtoInternal(sec, inputs, binding)
+        // TODO: DataFusion toChar does not support Spark datetime pattern format
+        // https://github.com/apache/datafusion/issues/16577
+        // https://github.com/apache/datafusion/issues/14536
+        // After fixing these issues, use provided `format` instead of the manual replacement below
+        val formatExpr = exprToProtoInternal(Literal("%Y-%m-%d %H:%M:%S"), inputs, binding)
+        val timeZone = exprToProtoInternal(Literal(timeZoneId.orNull), inputs, binding)
+
+        if (format != Literal(TimestampFormatter.defaultPattern)) {
+          withInfo(expr, "Datetime pattern format is unsupported")
+          None
+        } else if (secExpr.isDefined && formatExpr.isDefined) {
+          val timestampExpr =
+            scalarFunctionExprToProto("from_unixtime", Seq(secExpr, timeZone): _*)
+          val optExpr = scalarFunctionExprToProto("to_char", Seq(timestampExpr, formatExpr): _*)
+          optExprWithInfo(optExpr, expr, sec, format)
+        } else {
+          withInfo(expr, sec, format)
+          None
+        }
+    }
+  }
+}

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

@@ -1605,6 +1605,45 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("from_unixtime") {
+    Seq(false, true).foreach { dictionary =>
+      withSQLConf(
+        "parquet.enable.dictionary" -> dictionary.toString,
+        CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key -> "true") {
+        val table = "test"
+        withTempDir { dir =>
+          val path = new Path(dir.toURI.toString, "test.parquet")
+          makeParquetFileAllPrimitiveTypes(
+            path,
+            dictionaryEnabled = dictionary,
+            -128,
+            128,
+            randomSize = 100)
+          withParquetTable(path.toString, table) {
+            // TODO: DataFusion supports only -8334601211038 <= sec <= 8210266876799
+            // https://github.com/apache/datafusion/issues/16594
+            // After fixing this issue, remove the where clause below
+            val where = "where _5 BETWEEN -8334601211038 AND 8210266876799"
+            checkSparkAnswerAndOperator(s"SELECT from_unixtime(_5) FROM $table $where")
+            checkSparkAnswerAndOperator(s"SELECT from_unixtime(_8) FROM $table $where")
+            // TODO: DataFusion toChar does not support Spark datetime pattern format
+            // https://github.com/apache/datafusion/issues/16577
+            // https://github.com/apache/datafusion/issues/14536
+            // After fixing these issues, change checkSparkAnswer to checkSparkAnswerAndOperator
+            checkSparkAnswer(s"SELECT from_unixtime(_5, 'yyyy') FROM $table $where")
+            checkSparkAnswer(s"SELECT from_unixtime(_8, 'yyyy') FROM $table $where")
+            withSQLConf(SESSION_LOCAL_TIMEZONE.key -> "Asia/Kathmandu") {
+              checkSparkAnswerAndOperator(s"SELECT from_unixtime(_5) FROM $table $where")
+              checkSparkAnswerAndOperator(s"SELECT from_unixtime(_8) FROM $table $where")
+              checkSparkAnswer(s"SELECT from_unixtime(_5, 'yyyy') FROM $table $where")
+              checkSparkAnswer(s"SELECT from_unixtime(_8, 'yyyy') FROM $table $where")
+            }
+          }
+        }
+      }
+    }
+  }
+
   test("Decimal binary ops multiply is aligned to Spark") {
     Seq(true, false).foreach { allowPrecisionLoss =>
       withSQLConf(