apache · sunchao · Sep 27, 2026 · Sep 24, 2026
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
@@ -297,7 +297,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `timediff` | ✅ | — | Spark 4.0+ grammar alias that parses to `timestampdiff`; runs through codegen dispatch |
 | `timestamp_micros` | ✅ | Codegen dispatch |  |
 | `timestamp_millis` | ✅ | Codegen dispatch |  |
-| `timestamp_seconds` | ✅ | Native |  |
+| `timestamp_seconds` | ✅ | Hybrid | Integer, long, float and double inputs run natively; decimal, byte and short inputs route through the JVM codegen dispatcher |
 | `timestampadd` | ✅ | — | Reached through the grammar rather than the function registry; runs through codegen dispatch |
 | `timestampdiff` | ✅ | — | Reached through the grammar rather than the function registry; runs through codegen dispatch |
 | `to_date` | ✅ | — | Rewrites to `Cast` (or `Cast(GetTimestamp)` with a format) before Comet sees the plan |

diff --git a/spark/src/main/scala/org/apache/comet/serde/datetime.scala b/spark/src/main/scala/org/apache/comet/serde/datetime.scala
@@ -449,12 +449,22 @@ object CometMakeDate extends CometExpressionSerde[MakeDate] {
   }
 }
 
+/**
+ * `timestamp_seconds` lowers to the native `seconds_to_timestamp` kernel for integer, long, float
+ * and double inputs. Decimal, byte and short inputs have no native implementation, so
+ * `CodegenDispatchFallback` keeps them in the Comet pipeline by running Spark's own
+ * `SecondsToTimestamp.doGenCode` in the JVM codegen dispatcher, which matches Spark exactly.
+ *
+ * Decimal input is not a plain multiply: Spark computes `longValueExact()` on the scaled value,
+ * which raises rather than rounds when a nonzero digit remains past microsecond precision, and
+ * raises when the result overflows a long.
+ */
 object CometSecondsToTimestamp
-    extends CometScalarFunction[SecondsToTimestamp]("seconds_to_timestamp") {
+    extends CometScalarFunction[SecondsToTimestamp]("seconds_to_timestamp")
+    with CodegenDispatchFallback {
 
   override def getUnsupportedReasons(): Seq[String] = Seq(
-    "Only `IntegerType`, `LongType`, `FloatType`, and `DoubleType` inputs are supported." +
-      " `DecimalType`, `ByteType`, and `ShortType` fall back to Spark.")
+    "`DecimalType`, `ByteType`, and `ShortType` inputs")
 
   override def getSupportLevel(expr: SecondsToTimestamp): SupportLevel =
     expr.child.dataType match {

diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/timestamp_seconds.sql b/spark/src/test/resources/sql-tests/expressions/datetime/timestamp_seconds.sql
@@ -16,6 +16,7 @@
 -- under the License.
 
 -- Config: spark.sql.session.timeZone=UTC
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
 -- ConfigMatrix: parquet.enable.dictionary=false,true
 
 -- bigint column
@@ -25,7 +26,7 @@ CREATE TABLE test_ts_seconds_bigint(c0 bigint) USING parquet
 statement
 INSERT INTO test_ts_seconds_bigint VALUES (0), (1640995200), (-86400), (4102444800), (-2208988800), (NULL)
 
-query
+query expect_native(timestamp_seconds)
 SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_bigint
 
 -- int column
@@ -35,7 +36,7 @@ CREATE TABLE test_ts_seconds_int(c0 int) USING parquet
 statement
 INSERT INTO test_ts_seconds_int VALUES (0), (1640995200), (-86400), (NULL)
 
-query
+query expect_native(timestamp_seconds)
 SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_int
 
 -- double column
@@ -45,10 +46,121 @@ CREATE TABLE test_ts_seconds_double(c0 double) USING parquet
 statement
 INSERT INTO test_ts_seconds_double VALUES (0.0), (1640995200.123), (-86400.5), (NULL)
 
-query
+query expect_native(timestamp_seconds)
 SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_double
 
+-- float column
+statement
+CREATE TABLE test_ts_seconds_float(c0 float) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_float VALUES (0.0), (1.5), (-86400.5), (1640995200.0), (CAST('NaN' AS FLOAT)), (CAST('Infinity' AS FLOAT)), (CAST('-Infinity' AS FLOAT)), (NULL)
+
+query expect_native(timestamp_seconds)
+SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_float
+
+-- tinyint, smallint and decimal have no native implementation and run through the JVM codegen
+-- dispatcher, which executes Spark's own doGenCode (issue #5588).
+
+-- tinyint column
+statement
+CREATE TABLE test_ts_seconds_tinyint(c0 tinyint) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_tinyint VALUES (0), (1), (127), (-128), (NULL)
+
+query expect_dispatch(timestamp_seconds)
+SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_tinyint
+
+-- smallint column
+statement
+CREATE TABLE test_ts_seconds_smallint(c0 smallint) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_smallint VALUES (0), (1), (32767), (-32768), (NULL)
+
+query expect_dispatch(timestamp_seconds)
+SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_smallint
+
+-- decimal column, whole seconds
+statement
+CREATE TABLE test_ts_seconds_dec10_0(c0 decimal(10, 0)) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_dec10_0 VALUES (0), (1640995200), (-86400), (9999999999), (-9999999999), (NULL)
+
+query expect_dispatch(timestamp_seconds)
+SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_dec10_0
+
+-- decimal column at exactly microsecond precision
+statement
+CREATE TABLE test_ts_seconds_dec20_6(c0 decimal(20, 6)) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_dec20_6 VALUES (0), (1640995200.123456), (-86400.5), (-0.000001), (0.000001), (4102444800.999999), (-2208988800.000001), (NULL)
+
+query expect_dispatch(timestamp_seconds)
+SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_dec20_6
+
+-- decimal column with a wider scale whose digits past the sixth are zero
+statement
+CREATE TABLE test_ts_seconds_dec38_18(c0 decimal(38, 18)) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_dec38_18 VALUES (0), (1.5), (1640995200.123456), (-86400.000001), (NULL)
+
+query expect_dispatch(timestamp_seconds)
+SELECT c0, timestamp_seconds(c0) FROM test_ts_seconds_dec38_18
+
+-- Spark converts decimal input with longValueExact, so a nonzero digit past microsecond precision
+-- raises rather than rounds. Spark raises a bare java.lang.ArithmeticException with no error class
+-- on 3.4 through 4.1, so the expect_error patterns below match the JDK message.
+statement
+CREATE TABLE test_ts_seconds_dec20_7(c0 decimal(20, 7)) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_dec20_7 VALUES (1.1234567)
+
+query expect_error(Rounding necessary)
+SELECT timestamp_seconds(c0) FROM test_ts_seconds_dec20_7
+
+-- and a result outside the long range raises as well
+statement
+CREATE TABLE test_ts_seconds_dec38_0(c0 decimal(38, 0)) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_dec38_0 VALUES (99999999999999999999)
+
+query expect_error(Overflow)
+SELECT timestamp_seconds(c0) FROM test_ts_seconds_dec38_0
+
+-- The row that raises must not be evaluated when a conditional does not select it, matching
+-- Spark, which never evaluates an unselected branch
+statement
+CREATE TABLE test_ts_seconds_conditional(c0 decimal(20, 7), k int) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_conditional VALUES (1.5, 1), (1.1234567, 2), (NULL, 3)
+
+query expect_dispatch(timestamp_seconds)
+SELECT k, CASE WHEN k = 1 THEN timestamp_seconds(c0) END FROM test_ts_seconds_conditional
+
+query expect_dispatch(timestamp_seconds)
+SELECT k, CASE WHEN k <> 2 THEN timestamp_seconds(c0) ELSE timestamp('2020-01-01 00:00:00') END FROM test_ts_seconds_conditional
+
+query expect_dispatch(timestamp_seconds)
+SELECT k, IF(k = 1, timestamp_seconds(c0), NULL) FROM test_ts_seconds_conditional
+
+query expect_dispatch(timestamp_seconds)
+SELECT k, coalesce(timestamp('2020-01-01 00:00:00'), timestamp_seconds(c0)) FROM test_ts_seconds_conditional
+
 -- literal arguments
+query expect_dispatch(timestamp_seconds)
+SELECT timestamp_seconds(CAST(1.5 AS DECIMAL(10, 1))), timestamp_seconds(CAST(127 AS TINYINT)), timestamp_seconds(CAST(-32768 AS SMALLINT))
+
+query expect_error(Rounding necessary)
+SELECT timestamp_seconds(CAST(1.1234567 AS DECIMAL(20, 7)))
+
 query
 SELECT timestamp_seconds(0)
 

diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/timestamp_seconds_fallback.sql b/spark/src/test/resources/sql-tests/expressions/datetime/timestamp_seconds_fallback.sql
@@ -0,0 +1,38 @@
+-- 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.sql.session.timeZone=UTC
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false
+
+statement
+CREATE TABLE test_ts_seconds_fallback(b tinyint, s smallint, d decimal(20, 6), l bigint) USING parquet
+
+statement
+INSERT INTO test_ts_seconds_fallback VALUES (1, 1, 1640995200.123456, 1640995200), (NULL, NULL, NULL, NULL)
+
+query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled)
+SELECT timestamp_seconds(b) FROM test_ts_seconds_fallback
+
+query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled)
+SELECT timestamp_seconds(s) FROM test_ts_seconds_fallback
+
+query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled)
+SELECT timestamp_seconds(d) FROM test_ts_seconds_fallback
+
+-- The native input types do not require the dispatcher.
+query expect_native(timestamp_seconds)
+SELECT timestamp_seconds(l) FROM test_ts_seconds_fallback
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCodegenDispatchBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCodegenDispatchBenchmark.scala
@@ -113,6 +113,14 @@ object CometCodegenDispatchBenchmark extends CometBenchmarkBase {
       "select to_time(c_time, 'HH:mm:ss') from parquetV1Table",
       isSpark41Plus,
       Seq("spark.sql.timeType.enabled" -> "true")),
+    // `timestamp_seconds` lowers to a native kernel for int, long, float and double input, but
+    // decimal, byte and short input have no native implementation and are dispatched instead.
+    DispatchCase(
+      "timestamp_seconds(decimal)",
+      "select timestamp_seconds(c_dec) from parquetV1Table"),
+    DispatchCase(
+      "timestamp_seconds(tinyint)",
+      "select timestamp_seconds(c_byte) from parquetV1Table"),
     // The case the catch-all is really about: one unhandled expression used to cost the whole
     // projection, including the three expressions next to it that do have native kernels.
     DispatchCase(
@@ -382,6 +390,13 @@ object CometCodegenDispatchBenchmark extends CometBenchmarkBase {
       // Short, so `lpad` / `rpad` have padding to do on most rows.
       "c_pad" -> "CAST(CAST(PMOD(id, 100) AS STRING) AS BINARY)",
       "c_long" -> "id",
+      // `timestamp_seconds` has no native path for decimal, byte or short input, so these are
+      // dispatched. The scale stops at microseconds because Spark's decimal branch is
+      // `longValueExact`, which raises on a nonzero digit past the sixth. The cast is applied to
+      // the sum so that the column is exactly `DECIMAL(20, 6)`: casting each operand instead
+      // would leave `Add.resultDecimalType` to widen the result to `DECIMAL(21, 6)`.
+      "c_dec" -> "CAST(id + 0.123456 AS DECIMAL(20, 6))",
+      "c_byte" -> "CAST(PMOD(id, 100) AS TINYINT)",
       "c_time" -> ("CONCAT(LPAD(CAST(PMOD(id, 24) AS STRING), 2, '0'), ':', " +
         "LPAD(CAST(PMOD(id, 60) AS STRING), 2, '0'), ':', " +
         "LPAD(CAST(PMOD(id * 7, 60) AS STRING), 2, '0'))"))