Add input type checking and tests for unix_timestamp

- Add getSupportLevel() to validate input types before conversion
- Only support TimestampType and DateType inputs
- Fall back to Spark for StringType (with or without format argument)
- Fall back to Spark for TimestampNTZType (timezone handling is incorrect)
- Add tests to verify fallback behavior for unsupported types

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: andygrove

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

@@ -22,7 +22,7 @@ package org.apache.comet.serde
 import java.util.Locale
 
 import org.apache.spark.sql.catalyst.expressions.{Attribute, DateAdd, DateSub, DayOfMonth, DayOfWeek, DayOfYear, GetDateField, Hour, Literal, Minute, Month, Quarter, Second, TruncDate, TruncTimestamp, UnixTimestamp, WeekDay, WeekOfYear, Year}
-import org.apache.spark.sql.types.{DateType, IntegerType}
+import org.apache.spark.sql.types.{DateType, IntegerType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
 
 import org.apache.comet.CometSparkSessionExtensions.withInfo
@@ -255,10 +255,36 @@ object CometSecond extends CometExpressionSerde[Second] {
 }
 
 object CometUnixTimestamp extends CometExpressionSerde[UnixTimestamp] {
+
+  private def isSupportedInputType(expr: UnixTimestamp): Boolean = {
+    // Note: TimestampNTZType is not supported because Comet incorrectly applies
+    // timezone conversion to TimestampNTZ values. TimestampNTZ stores local time
+    // without timezone, so no conversion should be applied.
+    expr.children.head.dataType match {
+      case TimestampType | DateType => true
+      case _ => false
+    }
+  }
+
+  override def getSupportLevel(expr: UnixTimestamp): SupportLevel = {
+    if (isSupportedInputType(expr)) {
+      Compatible()
+    } else {
+      val inputType = expr.children.head.dataType
+      Unsupported(Some(s"unix_timestamp does not support input type: $inputType"))
+    }
+  }
+
   override def convert(
       expr: UnixTimestamp,
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
+    if (!isSupportedInputType(expr)) {
+      val inputType = expr.children.head.dataType
+      withInfo(expr, s"unix_timestamp does not support input type: $inputType")
+      return None
+    }
+
     val childExpr = exprToProtoInternal(expr.children.head, inputs, binding)
 
     if (childExpr.isDefined) {

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

@@ -21,7 +21,7 @@ package org.apache.comet
 
 import scala.util.Random
 
-import org.apache.spark.sql.{CometTestBase, SaveMode}
+import org.apache.spark.sql.{CometTestBase, Row, SaveMode}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
@@ -135,6 +135,44 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH
     }
   }
 
+  test("unix_timestamp - timestamp_ntz input falls back to Spark") {
+    // TimestampNTZ is not supported because Comet incorrectly applies timezone
+    // conversion. TimestampNTZ stores local time without timezone, so the unix
+    // timestamp should just be the value divided by microseconds per second.
+    val r = new Random(42)
+    val ntzSchema = StructType(Seq(StructField("ts_ntz", DataTypes.TimestampNTZType, true)))
+    val ntzDF = FuzzDataGenerator.generateDataFrame(r, spark, ntzSchema, 100, DataGenOptions())
+    ntzDF.createOrReplaceTempView("ntz_tbl")
+    checkSparkAnswerAndFallbackReason(
+      "SELECT ts_ntz, unix_timestamp(ts_ntz) from ntz_tbl order by ts_ntz",
+      "unix_timestamp does not support input type: TimestampNTZType")
+  }
+
+  test("unix_timestamp - string input falls back to Spark") {
+    withTempView("string_tbl") {
+      // Create test data with timestamp strings
+      val schema = StructType(Seq(StructField("ts_str", DataTypes.StringType, true)))
+      val data = Seq(
+        Row("2020-01-01 00:00:00"),
+        Row("2021-06-15 12:30:45"),
+        Row("2022-12-31 23:59:59"),
+        Row(null))
+      spark
+        .createDataFrame(spark.sparkContext.parallelize(data), schema)
+        .createOrReplaceTempView("string_tbl")
+
+      // String input should fall back to Spark
+      checkSparkAnswerAndFallbackReason(
+        "SELECT ts_str, unix_timestamp(ts_str) from string_tbl order by ts_str",
+        "unix_timestamp does not support input type: StringType")
+
+      // String input with custom format should also fall back
+      checkSparkAnswerAndFallbackReason(
+        "SELECT ts_str, unix_timestamp(ts_str, 'yyyy-MM-dd HH:mm:ss') from string_tbl",
+        "unix_timestamp does not support input type: StringType")
+    }
+  }
+
   private def createTimestampTestData = {
     val r = new Random(42)
     val schema = StructType(