fix: apply support-level consistency fixes surfaced by datetime audit

Mechanical fixes for the support-level / reason alignment issues found
during the datetime audit. No behavioural changes; the only observable
effect is that EXPLAIN-time fallback messages now include the specific
reason instead of a generic "not fully compatible with Spark".

- TimeFieldSerde companion (new) hoists the shared TimestampNTZ reason
  string used by CometHour, CometMinute, and CometSecond, mirroring
  the existing UTCTimestampSerde pattern. The three serdes now share
  one reason and one support-level helper.
- CometTruncDate extracts the duplicated reason strings into private
  vals and corrects the wording drift (the inline reason said "Invalid"
  while the docs reason said "Non-literal"; they now match).
- CometTruncTimestamp adds the missing non-literal-format reason to
  getIncompatibleReasons, adds the missing getUnsupportedReasons
  override for unsupported format literals, and extracts both reasons
  into private vals.
- CometSecondsToTimestamp adds the missing getUnsupportedReasons
  override so the Compatibility Guide reflects which input types are
  supported.
- CometHours and CometDays add getSupportLevel and getUnsupportedReasons
  overrides so the unsupported-input-type fallback surfaces in EXPLAIN
  output and the Compatibility Guide; the dispatcher now handles the
  fall-back uniformly and the withInfo call in convert is no longer
  needed for those branches.
- CometFromUnixTime moves the format-pattern check out of convert into
  getSupportLevel (returning Unsupported for non-default patterns and
  Incompatible for the DataFusion timestamp-range issue on default
  patterns). Reasons are shared via private vals; getUnsupportedReasons
  and getIncompatibleReasons both populated. As a side effect the
  fallback message for non-default formats now includes the specific
  reason ("Only the default datetime format pattern...") rather than
  the generic "not fully compatible with Spark"; updated the existing
  from_unix_time.sql expect_fallback assertions accordingly.

The CometDateFormat and CometUnixTimestamp findings need deeper
semantics analysis and are left for follow-up.

Author: andygrove

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

@@ -23,7 +23,7 @@ import java.util.Locale
 
 import org.apache.spark.sql.catalyst.expressions.{AddMonths, Attribute, ConvertTimezone, DateAdd, DateDiff, DateFormatClass, DateFromUnixDate, DateSub, DayOfMonth, DayOfWeek, DayOfYear, Days, FromUTCTimestamp, GetDateField, Hour, Hours, LastDay, Literal, MakeDate, MakeTimestamp, MicrosToTimestamp, MillisToTimestamp, Minute, Month, MonthsBetween, NextDay, Quarter, Second, SecondsToTimestamp, ToUnixTimestamp, ToUTCTimestamp, TruncDate, TruncTimestamp, UnixDate, UnixMicros, UnixMillis, UnixSeconds, 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.sql.types.{DataType, DateType, DoubleType, FloatType, IntegerType, LongType, StringType, TimestampNTZType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
 
 import org.apache.comet.CometConf
@@ -179,23 +179,24 @@ object CometQuarter extends CometExpressionSerde[Quarter] with CometExprGetDateF
   }
 }
 
-object CometHour extends CometExpressionSerde[Hour] {
+private object TimeFieldSerde {
+  val timestampNtzIncompatReason: String =
+    "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
+      " (https://github.com/apache/datafusion-comet/issues/3180)"
 
-  val incompatReason: String = "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
-    " (https://github.com/apache/datafusion-comet/issues/3180)"
+  def supportLevelForChild(childType: DataType): SupportLevel = childType match {
+    case TimestampNTZType => Incompatible(Some(timestampNtzIncompatReason))
+    case _ => Compatible()
+  }
+}
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason)
+object CometHour extends CometExpressionSerde[Hour] {
 
-  override def getSupportLevel(expr: Hour): SupportLevel = {
-    if (expr.child.dataType == TimestampNTZType) {
-      Incompatible(
-        Some(
-          "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
-            " (https://github.com/apache/datafusion-comet/issues/3180)"))
-    } else {
-      Compatible()
-    }
-  }
+  override def getIncompatibleReasons(): Seq[String] =
+    Seq(TimeFieldSerde.timestampNtzIncompatReason)
+
+  override def getSupportLevel(expr: Hour): SupportLevel =
+    TimeFieldSerde.supportLevelForChild(expr.child.dataType)
 
   override def convert(
       expr: Hour,
@@ -224,20 +225,11 @@ object CometHour extends CometExpressionSerde[Hour] {
 
 object CometMinute extends CometExpressionSerde[Minute] {
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(
-    "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
-      " (https://github.com/apache/datafusion-comet/issues/3180)")
-
-  override def getSupportLevel(expr: Minute): SupportLevel = {
-    if (expr.child.dataType == TimestampNTZType) {
-      Incompatible(
-        Some(
-          "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
-            " (https://github.com/apache/datafusion-comet/issues/3180)"))
-    } else {
-      Compatible()
-    }
-  }
+  override def getIncompatibleReasons(): Seq[String] =
+    Seq(TimeFieldSerde.timestampNtzIncompatReason)
+
+  override def getSupportLevel(expr: Minute): SupportLevel =
+    TimeFieldSerde.supportLevelForChild(expr.child.dataType)
 
   override def convert(
       expr: Minute,
@@ -266,20 +258,11 @@ object CometMinute extends CometExpressionSerde[Minute] {
 
 object CometSecond extends CometExpressionSerde[Second] {
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(
-    "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
-      " (https://github.com/apache/datafusion-comet/issues/3180)")
-
-  override def getSupportLevel(expr: Second): SupportLevel = {
-    if (expr.child.dataType == TimestampNTZType) {
-      Incompatible(
-        Some(
-          "Incorrectly applies timezone conversion to TimestampNTZ inputs" +
-            " (https://github.com/apache/datafusion-comet/issues/3180)"))
-    } else {
-      Compatible()
-    }
-  }
+  override def getIncompatibleReasons(): Seq[String] =
+    Seq(TimeFieldSerde.timestampNtzIncompatReason)
+
+  override def getSupportLevel(expr: Second): SupportLevel =
+    TimeFieldSerde.supportLevelForChild(expr.child.dataType)
 
   override def convert(
       expr: Second,
@@ -437,6 +420,11 @@ object CometMakeDate extends CometScalarFunction[MakeDate]("make_date")
 
 object CometSecondsToTimestamp
     extends CometScalarFunction[SecondsToTimestamp]("seconds_to_timestamp") {
+
+  override def getUnsupportedReasons(): Seq[String] = Seq(
+    "Only `IntegerType`, `LongType`, `FloatType`, and `DoubleType` inputs are supported." +
+      " `DecimalType`, `ByteType`, and `ShortType` fall back to Spark.")
+
   override def getSupportLevel(expr: SecondsToTimestamp): SupportLevel =
     expr.child.dataType match {
       case IntegerType | LongType | FloatType | DoubleType => Compatible()
@@ -482,8 +470,14 @@ object CometTruncDate extends CometExpressionSerde[TruncDate] {
   val supportedFormats: Seq[String] =
     Seq("year", "yyyy", "yy", "quarter", "mon", "month", "mm", "week")
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(
-    "Non-literal format strings will throw an exception instead of returning NULL")
+  private val nonLiteralFormatIncompatReason: String =
+    "Non-literal format strings will throw an exception instead of returning NULL"
+
+  private def unsupportedFormatReason(fmt: Any): String =
+    s"Format $fmt is not supported. Only the following formats are supported: " +
+      supportedFormats.mkString(", ")
+
+  override def getIncompatibleReasons(): Seq[String] = Seq(nonLiteralFormatIncompatReason)
 
   override def getUnsupportedReasons(): Seq[String] = Seq(
     "Only the following formats are supported: " + supportedFormats.mkString(", "))
@@ -494,11 +488,10 @@ object CometTruncDate extends CometExpressionSerde[TruncDate] {
         if (supportedFormats.contains(fmt.toString.toLowerCase(Locale.ROOT))) {
           Compatible()
         } else {
-          Unsupported(Some(s"Format $fmt is not supported"))
+          Unsupported(Some(unsupportedFormatReason(fmt)))
         }
       case _ =>
-        Incompatible(
-          Some("Invalid format strings will throw an exception instead of returning NULL"))
+        Incompatible(Some(nonLiteralFormatIncompatReason))
     }
   }
 
@@ -521,10 +514,6 @@ object CometTruncDate extends CometExpressionSerde[TruncDate] {
 
 object CometTruncTimestamp extends CometExpressionSerde[TruncTimestamp] {
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(
-    "Produces incorrect results when used with non-UTC timezones. Compatible when timezone is" +
-      " UTC. (https://github.com/apache/datafusion-comet/issues/2649)")
-
   val supportedFormats: Seq[String] =
     Seq(
       "year",
@@ -543,6 +532,23 @@ object CometTruncTimestamp extends CometExpressionSerde[TruncTimestamp] {
       "millisecond",
       "microsecond")
 
+  private val nonUtcIncompatReason: String =
+    "Produces incorrect results when used with non-UTC timezones. Compatible when timezone is" +
+      " UTC. (https://github.com/apache/datafusion-comet/issues/2649)"
+
+  private val nonLiteralFormatIncompatReason: String =
+    "Non-literal format strings will throw an exception instead of returning NULL"
+
+  private def unsupportedFormatReason(fmt: Any): String =
+    s"Format $fmt is not supported. Only the following formats are supported: " +
+      supportedFormats.mkString(", ")
+
+  override def getIncompatibleReasons(): Seq[String] =
+    Seq(nonUtcIncompatReason, nonLiteralFormatIncompatReason)
+
+  override def getUnsupportedReasons(): Seq[String] = Seq(
+    "Only the following formats are supported: " + supportedFormats.mkString(", "))
+
   override def getSupportLevel(expr: TruncTimestamp): SupportLevel = {
     val timezone = expr.timeZoneId.getOrElse("UTC")
     val isUtc = timezone == "UTC" || timezone == "Etc/UTC"
@@ -552,17 +558,13 @@ object CometTruncTimestamp extends CometExpressionSerde[TruncTimestamp] {
           if (isUtc) {
             Compatible()
           } else {
-            Incompatible(
-              Some(
-                s"Incorrect results in non-UTC timezone '$timezone'" +
-                  " (https://github.com/apache/datafusion-comet/issues/2649)"))
+            Incompatible(Some(nonUtcIncompatReason))
           }
         } else {
-          Unsupported(Some(s"Format $fmt is not supported"))
+          Unsupported(Some(unsupportedFormatReason(fmt)))
         }
       case _ =>
-        Incompatible(
-          Some("Invalid format strings will throw an exception instead of returning NULL"))
+        Incompatible(Some(nonLiteralFormatIncompatReason))
     }
   }
 
@@ -700,24 +702,27 @@ object CometDateFormat extends CometExpressionSerde[DateFormatClass] {
  * without applying any session timezone offset.
  */
 object CometHours extends CometExpressionSerde[Hours] {
+
+  override def getUnsupportedReasons(): Seq[String] = Seq(
+    "Only `TimestampType` and `TimestampNTZType` inputs are supported.")
+
+  override def getSupportLevel(expr: Hours): SupportLevel = expr.child.dataType match {
+    case TimestampType | TimestampNTZType => Compatible()
+    case other => Unsupported(Some(s"Hours does not support input type: $other"))
+  }
+
   override def convert(
       expr: Hours,
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
-    val optExpr = expr.child.dataType match {
-      case TimestampType | TimestampNTZType =>
-        exprToProtoInternal(expr.child, inputs, binding).map { childExpr =>
-          val builder = ExprOuterClass.HoursTransform.newBuilder()
-          builder.setChild(childExpr)
+    val optExpr = exprToProtoInternal(expr.child, inputs, binding).map { childExpr =>
+      val builder = ExprOuterClass.HoursTransform.newBuilder()
+      builder.setChild(childExpr)
 
-          ExprOuterClass.Expr
-            .newBuilder()
-            .setHoursTransform(builder)
-            .build()
-        }
-      case other =>
-        withInfo(expr, s"Hours does not support input type: $other")
-        None
+      ExprOuterClass.Expr
+        .newBuilder()
+        .setHoursTransform(builder)
+        .build()
     }
     optExprWithInfo(optExpr, expr, expr.child)
   }
@@ -734,6 +739,16 @@ object CometHours extends CometExpressionSerde[Hours] {
  * The first cast respects the session timezone to correctly determine the date boundary.
  */
 object CometDays extends CometExpressionSerde[Days] {
+
+  override def getUnsupportedReasons(): Seq[String] = Seq(
+    "Only `DateType` and `TimestampType` inputs are supported." +
+      " `TimestampNTZType` is not supported.")
+
+  override def getSupportLevel(expr: Days): SupportLevel = expr.child.dataType match {
+    case DateType | TimestampType => Compatible()
+    case other => Unsupported(Some(s"Days does not support input type: $other"))
+  }
+
   override def convert(
       expr: Days,
       inputs: Seq[Attribute],
@@ -748,9 +763,7 @@ object CometDays extends CometExpressionSerde[Days] {
         childExpr.flatMap { child =>
           CometCast.castToProto(expr, Some(timezone), DateType, child, CometEvalMode.LEGACY)
         }
-      case other =>
-        withInfo(expr, s"Days does not support input type: $other")
-        None
+      case _ => None
     }
 
     // Convert DateType to IntegerType (days since epoch)

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

@@ -29,12 +29,26 @@ import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithIn
 // https://github.com/apache/datafusion/issues/16594
 object CometFromUnixTime extends CometExpressionSerde[FromUnixTime] {
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(
-    "Only supports the default datetime format pattern `yyyy-MM-dd HH:mm:ss`." +
-      " DataFusion's valid timestamp range differs from Spark" +
-      " (https://github.com/apache/datafusion/issues/16594)")
+  private val incompatReason: String =
+    "DataFusion's valid timestamp range differs from Spark" +
+      " (https://github.com/apache/datafusion/issues/16594)"
 
-  override def getSupportLevel(expr: FromUnixTime): SupportLevel = Incompatible(None)
+  private val unsupportedFormatReason: String =
+    "Only the default datetime format pattern `yyyy-MM-dd HH:mm:ss` is supported;" +
+      " other patterns fall back to Spark" +
+      " (https://github.com/apache/datafusion/issues/16577)"
+
+  override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason)
+
+  override def getUnsupportedReasons(): Seq[String] = Seq(unsupportedFormatReason)
+
+  override def getSupportLevel(expr: FromUnixTime): SupportLevel = {
+    if (expr.format != Literal(TimestampFormatter.defaultPattern)) {
+      Unsupported(Some(unsupportedFormatReason))
+    } else {
+      Incompatible(Some(incompatReason))
+    }
+  }
 
   override def convert(
       expr: FromUnixTime,
@@ -48,10 +62,7 @@ object CometFromUnixTime extends CometExpressionSerde[FromUnixTime] {
     val formatExpr = exprToProtoInternal(Literal("%Y-%m-%d %H:%M:%S"), inputs, binding)
     val timeZone = exprToProtoInternal(Literal(expr.timeZoneId.orNull), inputs, binding)
 
-    if (expr.format != Literal(TimestampFormatter.defaultPattern)) {
-      withInfo(expr, "Datetime pattern format is unsupported")
-      None
-    } else if (secExpr.isDefined && formatExpr.isDefined) {
+    if (secExpr.isDefined && formatExpr.isDefined) {
       val timestampExpr =
         scalarFunctionExprToProto("from_unixtime", Seq(secExpr, timeZone): _*)
       val optExpr = scalarFunctionExprToProto("to_char", Seq(timestampExpr, formatExpr): _*)

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

@@ -24,12 +24,12 @@ INSERT INTO test_from_unix_time VALUES (0), (1718451045), (-1), (NULL), (2147483
 query expect_fallback(not fully compatible with Spark)
 SELECT from_unixtime(t) FROM test_from_unix_time
 
-query expect_fallback(not fully compatible with Spark)
+query expect_fallback(Only the default datetime format pattern)
 SELECT from_unixtime(t, 'yyyy-MM-dd') FROM test_from_unix_time
 
 -- literal arguments
 query expect_fallback(not fully compatible with Spark)
 SELECT from_unixtime(0)
 
-query expect_fallback(not fully compatible with Spark)
+query expect_fallback(Only the default datetime format pattern)
 SELECT from_unixtime(1718451045, 'yyyy-MM-dd')