From 4dbdc57d1020c2fe44aa050784d759f72b7a3457 Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Thu, 24 Sep 2026 19:30:43 +0300 Subject: [PATCH] route timestamp_seconds decimal, byte and short input through codegen dispatch --- docs/source/user-guide/latest/expressions.md | 2 +- .../org/apache/comet/serde/datetime.scala | 16 ++- .../datetime/timestamp_seconds.sql | 118 +++++++++++++++++- .../datetime/timestamp_seconds_fallback.sql | 38 ++++++ .../CometCodegenDispatchBenchmark.scala | 15 +++ 5 files changed, 182 insertions(+), 7 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/datetime/timestamp_seconds_fallback.sql diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 700d027d0da..0d004ac6e12 100644 --- 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 index d1ca21b8869..ef0ff48dde3 100644 --- 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 index b1605b32844..a7cd622119e 100644 --- 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 new file mode 100644 index 00000000000..18328258d72 --- /dev/null +++ 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 index ab2c15affd9..8ced35556a2 100644 --- 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'))"))