This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 09b44ad6fa feat: unix_timestamp codegen dispatch for strings, and fix
pre-epoch fractional rounding (#5789)
09b44ad6fa is described below
commit 09b44ad6fa17f58f4bbccf958c5cf02d790f7334
Author: Daipayan Mukherjee <[email protected]>
AuthorDate: Mon Sep 21 15:54:52 2026 +0000
feat: unix_timestamp codegen dispatch for strings, and fix pre-epoch
fractional rounding (#5789)
* feat: dispatch unix_timestamp string parsing through codegen
* test: correct unix timestamp fallback fixture and format benchmark
* fix: correct unix_timestamp routing and strengthen regression coverage
* fix: align unix_timestamp collation tests and rounding coverage
---------
Co-authored-by: Andy Grove <[email protected]>
---
.../expression-audits/datetime_funcs.md | 10 +++
.../contributor-guide/spark_configs_support.md | 3 +-
docs/source/user-guide/latest/expressions.md | 2 +-
.../src/datetime_funcs/unix_timestamp.rs | 62 +++++++++++++++--
.../scala/org/apache/comet/serde/datetime.scala | 19 ++---
.../expressions/datetime/unix_timestamp.sql | 33 ++++++++-
...olicy_corrected.sql => unix_timestamp_ansi.sql} | 34 +++++----
.../datetime/unix_timestamp_fallback.sql | 42 ++++++++++++
.../datetime/unix_timestamp_strings.sql | 65 ++++++++++++++++++
.../datetime/unix_timestamp_time_parser_policy.sql | 10 +--
...unix_timestamp_time_parser_policy_corrected.sql | 17 +++--
...unix_timestamp_time_parser_policy_exception.sql | 17 ++++-
.../unix_timestamp_time_parser_policy_legacy.sql | 17 +++--
.../comet/CometTemporalExpressionSuite.scala | 80 ++++++++++++++++------
.../benchmark/CometUnixTimestampBenchmark.scala | 67 ++++++++++++++++++
.../org/apache/spark/sql/CometCollationSuite.scala | 32 +++++----
.../org/apache/spark/sql/CometCollationSuite.scala | 32 +++++----
17 files changed, 437 insertions(+), 105 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/datetime_funcs.md
b/docs/source/contributor-guide/expression-audits/datetime_funcs.md
index b285c4067b..242fd7e6c3 100644
--- a/docs/source/contributor-guide/expression-audits/datetime_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/datetime_funcs.md
@@ -111,6 +111,16 @@
- Rewrites to `Cast(..., EvalMode.LEGACY)` (no format, native) or
`GetTimestamp(..., failOnError = false)` (with format, via the codegen
dispatcher) before Comet sees the plan. In non-ANSI mode the rewritten tree is
identical to `to_timestamp`; invalid inputs return NULL to match Spark.
+## unix_timestamp
+
+- Spark 3.4.3 (audited 2026-09-13): baseline. String inputs accept literal or
column formats. Date, timestamp, and timestamp without time zone inputs ignore
the format argument.
+- Spark 3.5.8 (audited 2026-09-13): parsing failures use structured timestamp
parsing errors.
+- Spark 4.0.1 (audited 2026-09-13): `inputTypes` widened to
`StringTypeWithCollation` for the input and format arguments.
+- Spark 4.1.1 (audited 2026-09-13): same input types and parsing behavior as
Spark 4.0.1.
+- String inputs use Spark's generated parser through codegen dispatch,
including collated strings and formats. Literal and column formats preserve
null handling, ANSI errors, parser policy, and session time zone.
+- Date, timestamp, and timestamp without time zone inputs retain native
execution and ignore the format, including its collation. String input stays
unsupported by the native serializer so `allowIncompatible=true` cannot send it
to the native kernel.
+- Native timestamp conversion truncates fractional seconds toward zero,
matching Spark's `ToTimestamp`. This fixes the previous use of floor division
for negative fractional timestamps: at UTC, `1969-12-31 23:59:58.5` produces
`-1`, not `-2`. Casting a timestamp to `BIGINT` deliberately uses floor
division in Spark and Comet, so that cast still produces `-2`.
+
[Spark Expression Support]: ../../user-guide/latest/expressions.md
## weekday
diff --git a/docs/source/contributor-guide/spark_configs_support.md
b/docs/source/contributor-guide/spark_configs_support.md
index b2c55bc146..92107975ba 100644
--- a/docs/source/contributor-guide/spark_configs_support.md
+++ b/docs/source/contributor-guide/spark_configs_support.md
@@ -124,7 +124,8 @@ and fallback paths:
`spark.comet.expression.FromUnixTime.allowIncompatible=true` is set;
otherwise
it routes through the codegen dispatcher.
- `unix_timestamp(<timestamp_or_date>)` does not call the formatter at all; the
- string-input overload falls back.
+ string-input overload routes through the codegen dispatcher and preserves
Spark's
+ selected parser policy.
- `to_unix_timestamp` routes through the codegen dispatcher.
If a Comet contributor adds native string-format parsing or extends the
date_format
diff --git a/docs/source/user-guide/latest/expressions.md
b/docs/source/user-guide/latest/expressions.md
index 501ce468c4..ab2794071a 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -317,7 +317,7 @@ The type-name conversion functions (`bigint`, `binary`,
`boolean`, `date`, `deci
| `unix_micros` | ✅ | Codegen dispatch | |
| `unix_millis` | ✅ | Codegen dispatch | |
| `unix_seconds` | ✅ | Codegen dispatch | |
-| `unix_timestamp` | ✅ | Native | |
+| `unix_timestamp` | ✅ | Hybrid | String parsing uses Spark's codegen and
honors the time parser policy, ANSI mode, and session time zone. Date and
timestamp inputs ignore the format and use native execution. |
| `weekday` | ✅ | Native | |
| `weekofyear` | ✅ | Native | |
| `window` | ✅ | — | Batch tumbling and sliding time-window grouping runs
natively |
diff --git a/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
b/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
index bd62563a6b..9602dba5fe 100644
--- a/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
+++ b/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
@@ -23,7 +23,6 @@ use datafusion::common::{internal_datafusion_err,
DataFusionError};
use datafusion::logical_expr::{
ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
};
-use num::integer::div_floor;
use std::{fmt::Debug, sync::Arc};
const MICROS_PER_SECOND: i64 = 1_000_000;
@@ -84,12 +83,12 @@ impl ScalarUDFImpl for SparkUnixTimestamp {
timestamp_array
.values()
.iter()
- .map(|µs| div_floor(micros,
MICROS_PER_SECOND))
+ .map(|µs| micros / MICROS_PER_SECOND)
.collect()
} else {
timestamp_array
.iter()
- .map(|v| v.map(|micros| div_floor(micros,
MICROS_PER_SECOND)))
+ .map(|v| v.map(|micros| micros /
MICROS_PER_SECOND))
.collect()
};
@@ -116,12 +115,12 @@ impl ScalarUDFImpl for SparkUnixTimestamp {
timestamp_array
.values()
.iter()
- .map(|µs| div_floor(micros,
MICROS_PER_SECOND))
+ .map(|µs| micros / MICROS_PER_SECOND)
.collect()
} else {
timestamp_array
.iter()
- .map(|v| v.map(|micros| div_floor(micros,
MICROS_PER_SECOND)))
+ .map(|v| v.map(|micros| micros /
MICROS_PER_SECOND))
.collect()
};
@@ -153,7 +152,7 @@ impl ScalarUDFImpl for SparkUnixTimestamp {
} else {
timestamp_array
.iter()
- .map(|v| v.map(|micros| div_floor(micros,
MICROS_PER_SECOND)))
+ .map(|v| v.map(|micros| micros /
MICROS_PER_SECOND))
.collect()
};
@@ -208,6 +207,57 @@ mod tests {
}
}
+ #[test]
+ fn test_unix_timestamp_truncates_fractional_seconds_toward_zero() {
+ for timezone in [None, Some("UTC")] {
+ for with_null in [false, true] {
+ let mut values = vec![
+ Some(-1_500_000),
+ Some(-1_000_000),
+ Some(-999_999),
+ Some(-1),
+ Some(0),
+ Some(1),
+ Some(999_999),
+ Some(1_000_000),
+ Some(1_500_000),
+ ];
+ let mut expected = vec![
+ Some(-1),
+ Some(-1),
+ Some(0),
+ Some(0),
+ Some(0),
+ Some(0),
+ Some(0),
+ Some(1),
+ Some(1),
+ ];
+ if with_null {
+ values.push(None);
+ expected.push(None);
+ }
+ let input =
TimestampMicrosecondArray::from(values).with_timezone_opt(timezone);
+ let number_rows = input.len();
+ let udf = SparkUnixTimestamp::new("UTC".to_string());
+ let result = udf
+ .invoke_with_args(ScalarFunctionArgs {
+ args: vec![ColumnarValue::Array(Arc::new(input))],
+ number_rows,
+ return_field: Arc::new(Field::new("unix_timestamp",
DataType::Int64, true)),
+ config_options: Arc::new(ConfigOptions::default()),
+ arg_fields: vec![],
+ })
+ .unwrap();
+ let ColumnarValue::Array(result) = result else {
+ panic!("Expected array result");
+ };
+ let actual = result.as_primitive::<Int64Type>();
+ assert_eq!(actual.iter().collect::<Vec<_>>(), expected);
+ }
+ }
+ }
+
#[test]
fn test_unix_timestamp_from_date() {
// Test with Date32
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 ab40b70657..d1ca21b886 100644
--- a/spark/src/main/scala/org/apache/comet/serde/datetime.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/datetime.scala
@@ -257,15 +257,12 @@ private[serde] object DatetimeCollation extends
CometTypeShim {
expr.children.exists(c => hasNonDefaultStringCollation(c.dataType))
}
-object CometUnixTimestamp extends CometExpressionSerde[UnixTimestamp] {
-
- private val collationReason = DatetimeCollation.reason("unix_timestamp")
+object CometUnixTimestamp
+ extends CometExpressionSerde[UnixTimestamp]
+ with CodegenDispatchFallback {
override def getUnsupportedReasons(): Seq[String] = Seq(
- "Only `DateType`, `TimestampType`, and `TimestampNTZType` inputs are
supported.")
-
- override def getIncompatibleReasons(): Seq[String] =
- DatetimeCollation.incompatibleReasons("unix_timestamp")
+ "String inputs, including collated strings, have no native
implementation.")
private def isSupportedInputType(expr: UnixTimestamp): Boolean = {
expr.children.head.dataType match {
@@ -276,16 +273,10 @@ object CometUnixTimestamp extends
CometExpressionSerde[UnixTimestamp] {
}
override def getSupportLevel(expr: UnixTimestamp): SupportLevel = {
- // The input type is screened ahead of the collation check on purpose. A
non-date/timestamp
- // input has no native path at all, so it must report `Unsupported` rather
than
- // `Incompatible`: the latter is waved straight through to `convert` when
- // `spark.comet.expression.UnixTimestamp.allowIncompatible=true`, and the
native kernel then
- // raises an execution error on the string child instead of falling back
to Spark.
+ // Strings have no native path, even when incompatible expressions are
allowed.
if (!isSupportedInputType(expr)) {
val inputType = expr.children.head.dataType
Unsupported(Some(s"unix_timestamp does not support input type:
$inputType"))
- } else if (DatetimeCollation.hasNonDefaultCollation(expr)) {
- Incompatible(Some(collationReason))
} else {
Compatible()
}
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp.sql
index 3d09626035..99ee7efd35 100644
--- a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp.sql
+++ b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp.sql
@@ -15,15 +15,46 @@
-- specific language governing permissions and limitations
-- under the License.
+-- Config: spark.sql.session.timeZone=UTC
+
statement
CREATE TABLE test_unix_ts(ts timestamp) USING parquet
statement
INSERT INTO test_unix_ts VALUES (timestamp('1970-01-01 00:00:00')),
(timestamp('2024-06-15 10:30:45')), (NULL)
-query
+query expect_native(unix_timestamp)
SELECT unix_timestamp(ts) FROM test_unix_ts
-- literal arguments
query ignore(https://github.com/apache/datafusion-comet/issues/3336)
SELECT unix_timestamp(timestamp('1970-01-01 00:00:00')),
unix_timestamp(timestamp('2024-06-15 10:30:45'))
+
+-- Native timestamp conversion truncates fractional seconds toward zero,
including before epoch.
+statement
+CREATE TABLE test_unix_ts_fractional(ts timestamp, ntz timestamp_ntz) USING
parquet
+
+statement
+INSERT INTO test_unix_ts_fractional VALUES
+ (CAST('1969-12-31 23:59:58.500000' AS TIMESTAMP), CAST('1969-12-31
23:59:58.500000' AS TIMESTAMP_NTZ)),
+ (CAST('1969-12-31 23:59:59.000000' AS TIMESTAMP), CAST('1969-12-31
23:59:59.000000' AS TIMESTAMP_NTZ)),
+ (CAST('1969-12-31 23:59:59.500000' AS TIMESTAMP), CAST('1969-12-31
23:59:59.500000' AS TIMESTAMP_NTZ)),
+ (CAST('1969-12-31 23:59:59.999999' AS TIMESTAMP), CAST('1969-12-31
23:59:59.999999' AS TIMESTAMP_NTZ)),
+ (CAST('1970-01-01 00:00:00.000000' AS TIMESTAMP), CAST('1970-01-01
00:00:00.000000' AS TIMESTAMP_NTZ)),
+ (CAST('1970-01-01 00:00:00.000001' AS TIMESTAMP), CAST('1970-01-01
00:00:00.000001' AS TIMESTAMP_NTZ)),
+ (CAST('1970-01-01 00:00:01.500000' AS TIMESTAMP), CAST('1970-01-01
00:00:01.500000' AS TIMESTAMP_NTZ)),
+ (NULL, NULL)
+
+query expect_native(unix_timestamp)
+SELECT unix_timestamp(ts), unix_timestamp(ntz) FROM test_unix_ts_fractional
+
+query expect_native(unix_timestamp)
+SELECT unix_timestamp(ts), unix_timestamp(ntz) FROM test_unix_ts_fractional
WHERE ts IS NOT NULL
+
+-- unix_timestamp truncates toward zero, while casting a timestamp to BIGINT
floors.
+-- At -1.5 seconds the results are -1 and -2; at -0.5 seconds they are 0 and
-1.
+query expect_native(unix_timestamp, cast)
+SELECT unix_timestamp(ts), CAST(ts AS BIGINT) FROM test_unix_ts_fractional
+
+query expect_native(unix_timestamp, cast)
+SELECT unix_timestamp(ts), CAST(ts AS BIGINT) FROM test_unix_ts_fractional
WHERE ts IS NOT NULL
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_ansi.sql
similarity index 51%
copy from
spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
copy to
spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_ansi.sql
index 71a4c6d380..70211380cb 100644
---
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_ansi.sql
@@ -15,22 +15,28 @@
-- specific language governing permissions and limitations
-- under the License.
--- unix_timestamp() under CORRECTED timeParserPolicy.
--- The new java.time formatter is strict: lenient inputs return null without
raising
--- SparkUpgradeException.
+-- Config: spark.sql.ansi.enabled=true
-- Config: spark.sql.legacy.timeParserPolicy=CORRECTED
--- Config: spark.sql.session.timeZone=UTC
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
statement
-CREATE TABLE test_unix_ts_strict(s string) USING parquet
+CREATE TABLE test_unix_ts_ansi(s string, fmt string) USING parquet
statement
-INSERT INTO test_unix_ts_strict VALUES
- ('2024-1-1'),
- ('2024-13-01'),
- ('2024-02-30'),
- ('2024-01-01garbage'),
- ('2024')
-
-query spark_answer_only
-SELECT s, unix_timestamp(s, 'yyyy-MM-dd') FROM test_unix_ts_strict ORDER BY s
+INSERT INTO test_unix_ts_ansi VALUES ('not a date', 'yyyy-MM-dd')
+
+query expect_error(could not be parsed)
+SELECT unix_timestamp(s, 'yyyy-MM-dd') FROM test_unix_ts_ansi
+
+query expect_error(could not be parsed)
+SELECT unix_timestamp(s, fmt) FROM test_unix_ts_ansi
+
+query expect_error(could not be parsed)
+SELECT unix_timestamp('2024-13-99', 'yyyy-MM-dd')
+
+-- Valid queries require Comet execution, so fallback cannot hide an
error-path regression.
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp('2024-06-15', 'yyyy-MM-dd'), unix_timestamp(CAST(NULL AS
STRING))
+
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp('2024-06-15', fmt) FROM test_unix_ts_ansi
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_fallback.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_fallback.sql
new file mode 100644
index 0000000000..7a531df056
--- /dev/null
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_fallback.sql
@@ -0,0 +1,42 @@
+-- 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.comet.exec.scalaUDF.codegen.enabled=false
+
+statement
+CREATE TABLE test_unix_ts_fallback(s string, fmt string, d date, ts timestamp,
ntz timestamp_ntz) USING parquet
+
+statement
+INSERT INTO test_unix_ts_fallback VALUES
+ ('2024-06-15', 'yyyy-MM-dd', date('2024-06-15'), timestamp('2024-06-15
10:30:45'), CAST('2024-06-15 10:30:45' AS TIMESTAMP_NTZ)),
+ (NULL, NULL, NULL, NULL, NULL)
+
+query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled)
+SELECT unix_timestamp(s) FROM test_unix_ts_fallback
+
+query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled)
+SELECT unix_timestamp(s, fmt) FROM test_unix_ts_fallback
+
+query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled)
+SELECT unix_timestamp('2024-06-15', 'yyyy-MM-dd')
+
+-- Date and timestamp inputs keep their native path and ignore the format.
+query expect_native(unix_timestamp)
+SELECT unix_timestamp(d), unix_timestamp(ts), unix_timestamp(ntz) FROM
test_unix_ts_fallback
+
+query expect_native(unix_timestamp)
+SELECT unix_timestamp(d, fmt), unix_timestamp(ts, fmt), unix_timestamp(ntz,
fmt) FROM test_unix_ts_fallback
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_strings.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_strings.sql
new file mode 100644
index 0000000000..9c18497be1
--- /dev/null
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_strings.sql
@@ -0,0 +1,65 @@
+-- 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.legacy.timeParserPolicy=CORRECTED
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+-- ConfigMatrix: spark.sql.session.timeZone=UTC,America/Los_Angeles
+
+statement
+CREATE TABLE test_unix_ts_strings(s string, fmt string) USING parquet
+
+statement
+INSERT INTO test_unix_ts_strings VALUES
+ ('1970-01-01 00:00:00', 'yyyy-MM-dd HH:mm:ss'),
+ ('1969-12-31 23:59:59', 'yyyy-MM-dd HH:mm:ss'),
+ ('2024-02-29 12:30:45', 'yyyy-MM-dd HH:mm:ss'),
+ ('2024-03-10 02:30:00', 'yyyy-MM-dd HH:mm:ss'),
+ ('2024-11-03 01:30:00', 'yyyy-MM-dd HH:mm:ss'),
+ ('1969-12-31 23:59:59.999999', 'yyyy-MM-dd HH:mm:ss.SSSSSS'),
+ ('2024/06/15', 'yyyy/MM/dd'),
+ ('2024-06-15T10:30:45+05:30', "yyyy-MM-dd'T'HH:mm:ssXXX"),
+ ('1582-10-04', 'yyyy-MM-dd'),
+ ('0001-01-01', 'yyyy-MM-dd'),
+ ('9999-12-31', 'yyyy-MM-dd'),
+ ('not a date', 'yyyy-MM-dd'),
+ ('2024-02-30', 'yyyy-MM-dd'),
+ ('', 'yyyy-MM-dd'),
+ (NULL, 'yyyy-MM-dd'),
+ ('2024-06-15', NULL),
+ ('2024-06-15', ''),
+ (NULL, NULL)
+
+-- Exercise both the cached literal formatter and the per-row formatter.
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp(s), unix_timestamp(s, 'yyyy-MM-dd HH:mm:ss') FROM
test_unix_ts_strings
+
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp(s, fmt) FROM test_unix_ts_strings
+
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp('2024-06-15', fmt) FROM test_unix_ts_strings
+
+-- Constant folding is disabled by the SQL test harness.
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp('2024-06-15', 'yyyy-MM-dd'), unix_timestamp(''),
unix_timestamp(CAST(NULL AS STRING)), unix_timestamp('2024-06-15', CAST(NULL AS
STRING))
+
+-- Parsing inside grouping must keep both the expression and aggregation in
Comet.
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp(s) AS u, count(*) FROM test_unix_ts_strings GROUP BY u
+
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp(s, fmt) AS u, count(*) FROM test_unix_ts_strings GROUP
BY u
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy.sql
index 25737128e4..5ccddce2e9 100644
---
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy.sql
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy.sql
@@ -26,10 +26,10 @@ CREATE TABLE test_unix_ts_policy(s string) USING parquet
statement
INSERT INTO test_unix_ts_policy VALUES ('2024-06-15 10:30:45'), ('1970-01-01
00:00:00'), (NULL), ('')
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT unix_timestamp(s, 'yyyy-MM-dd HH:mm:ss') FROM test_unix_ts_policy
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT unix_timestamp(s) FROM test_unix_ts_policy
-- date-only input with date-only pattern
@@ -39,12 +39,12 @@ CREATE TABLE test_unix_ts_date_policy(s string) USING
parquet
statement
INSERT INTO test_unix_ts_date_policy VALUES ('2024-06-15'), ('1970-01-01'),
(NULL)
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT unix_timestamp(s, 'yyyy-MM-dd') FROM test_unix_ts_date_policy
-- literal arguments
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT unix_timestamp('2024-06-15', 'yyyy-MM-dd')
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT unix_timestamp(NULL, 'yyyy-MM-dd')
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
index 71a4c6d380..f7dab3b393 100644
---
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_corrected.sql
@@ -22,15 +22,18 @@
-- Config: spark.sql.session.timeZone=UTC
statement
-CREATE TABLE test_unix_ts_strict(s string) USING parquet
+CREATE TABLE test_unix_ts_strict(s string, fmt string) USING parquet
statement
INSERT INTO test_unix_ts_strict VALUES
- ('2024-1-1'),
- ('2024-13-01'),
- ('2024-02-30'),
- ('2024-01-01garbage'),
- ('2024')
+ ('2024-1-1', 'yyyy-MM-dd'),
+ ('2024-13-01', 'yyyy-MM-dd'),
+ ('2024-02-30', 'yyyy-MM-dd'),
+ ('2024-01-01garbage', 'yyyy-MM-dd'),
+ ('2024', 'yyyy-MM-dd')
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT s, unix_timestamp(s, 'yyyy-MM-dd') FROM test_unix_ts_strict ORDER BY s
+
+query expect_dispatch(unix_timestamp)
+SELECT s, unix_timestamp(s, fmt) FROM test_unix_ts_strict ORDER BY s
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_exception.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_exception.sql
index 7a90b44fe6..7a9cacca82 100644
---
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_exception.sql
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_exception.sql
@@ -15,17 +15,28 @@
-- specific language governing permissions and limitations
-- under the License.
--- unix_timestamp() under EXCEPTION timeParserPolicy (the default).
+-- unix_timestamp() under EXCEPTION timeParserPolicy.
-- New parser fails on lenient inputs; legacy parser would have succeeded;
-- DateTimeFormatterHelper.checkParsedDiff converts the failure to
SparkUpgradeException.
-- Config: spark.sql.legacy.timeParserPolicy=EXCEPTION
-- Config: spark.sql.session.timeZone=UTC
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
statement
-CREATE TABLE test_unix_ts_exception(s string) USING parquet
+CREATE TABLE test_unix_ts_exception(s string, fmt string) USING parquet
statement
-INSERT INTO test_unix_ts_exception VALUES ('2024-1-1')
+INSERT INTO test_unix_ts_exception VALUES ('2024-1-1', 'yyyy-MM-dd')
query expect_error(INCONSISTENT_BEHAVIOR_CROSS_VERSION)
SELECT unix_timestamp(s, 'yyyy-MM-dd') FROM test_unix_ts_exception
+
+query expect_error(INCONSISTENT_BEHAVIOR_CROSS_VERSION)
+SELECT unix_timestamp(s, fmt) FROM test_unix_ts_exception
+
+-- Require Comet execution under EXCEPTION as well as checking the errors.
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp('2024-06-15', 'yyyy-MM-dd')
+
+query expect_dispatch(unix_timestamp)
+SELECT unix_timestamp('2024-06-15', fmt) FROM test_unix_ts_exception
diff --git
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_legacy.sql
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_legacy.sql
index 259a06d1f2..553be95e44 100644
---
a/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_legacy.sql
+++
b/spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_time_parser_policy_legacy.sql
@@ -22,15 +22,18 @@
-- Config: spark.sql.session.timeZone=UTC
statement
-CREATE TABLE test_unix_ts_lenient(s string) USING parquet
+CREATE TABLE test_unix_ts_lenient(s string, fmt string) USING parquet
statement
INSERT INTO test_unix_ts_lenient VALUES
- ('2024-1-1'),
- ('2024-13-01'),
- ('2024-02-30'),
- ('2024-01-01garbage'),
- ('2024')
+ ('2024-1-1', 'yyyy-MM-dd'),
+ ('2024-13-01', 'yyyy-MM-dd'),
+ ('2024-02-30', 'yyyy-MM-dd'),
+ ('2024-01-01garbage', 'yyyy-MM-dd'),
+ ('2024', 'yyyy-MM-dd')
-query spark_answer_only
+query expect_dispatch(unix_timestamp)
SELECT s, unix_timestamp(s, 'yyyy-MM-dd') FROM test_unix_ts_lenient ORDER BY s
+
+query expect_dispatch(unix_timestamp)
+SELECT s, unix_timestamp(s, fmt) FROM test_unix_ts_lenient ORDER BY s
diff --git
a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala
b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala
index 9b2b73f97e..4d950b8718 100644
--- a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala
@@ -386,7 +386,7 @@ class CometTemporalExpressionSuite extends CometTestBase
with AdaptiveSparkPlanH
}
}
- test("unix_timestamp - string input falls back to Spark") {
+ test("unix_timestamp - string input uses codegen dispatch") {
withTempView("string_tbl") {
// Create test data with timestamp strings
val schema = StructType(Seq(StructField("ts_str", DataTypes.StringType,
true)))
@@ -399,19 +399,28 @@ class CometTemporalExpressionSuite extends CometTestBase
with AdaptiveSparkPlanH
.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")
+ withSQLConf(
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") {
+ for (allowIncompatible <- Seq("false", "true")) {
+ withSQLConf(
+ CometConf.getExprAllowIncompatConfigKey("UnixTimestamp") ->
allowIncompatible) {
+ for (query <- Seq(
+ "SELECT unix_timestamp(ts_str) FROM string_tbl",
+ "SELECT unix_timestamp(ts_str, 'yyyy-MM-dd HH:mm:ss') FROM
string_tbl",
+ "SELECT unix_timestamp('2024-06-15', 'yyyy-MM-dd') FROM
string_tbl")) {
+ checkSparkAnswerAndImpl(
+ query,
+ native = Seq.empty,
+ dispatched = Seq("unix_timestamp"))
+ }
+ }
+ }
+ }
}
}
- test("unix_timestamp - string input falls back even when a collated format
opts into native") {
+ test("unix_timestamp - collated strings use codegen even when native is
opted into") {
assume(isSpark40Plus, "string collation requires Spark 4.0+")
withTempView("string_tbl") {
val schema = StructType(Seq(StructField("ts_str", DataTypes.StringType,
true)))
@@ -420,15 +429,46 @@ class CometTemporalExpressionSuite extends CometTestBase
with AdaptiveSparkPlanH
.createDataFrame(spark.sparkContext.parallelize(data), schema)
.createOrReplaceTempView("string_tbl")
- // A collated format argument makes the expression `Incompatible`, which
- // `allowIncompatible=true` waves straight through to `convert`. The
input type has to be
- // rejected ahead of that opt-in: the native kernel accepts only
date/timestamp input, so
- // serializing a string child raises an execution error instead of
falling back to Spark.
- withSQLConf(CometConf.getExprAllowIncompatConfigKey("UnixTimestamp") ->
"true") {
- checkSparkAnswerAndFallbackReason(
- "SELECT ts_str, unix_timestamp(ts_str, 'yyyy-MM-dd HH:mm:ss' COLLATE
UTF8_LCASE) " +
- "from string_tbl order by ts_str",
- "unix_timestamp does not support input type: StringType")
+ // Strings have no native path, even when incompatible expressions are
allowed.
+ for (allowIncompatible <- Seq("false", "true")) {
+ withSQLConf(
+ CometConf.getExprAllowIncompatConfigKey("UnixTimestamp") ->
allowIncompatible) {
+ for (query <- Seq(
+ "SELECT unix_timestamp(ts_str, 'yyyy-MM-dd HH:mm:ss' COLLATE
UTF8_LCASE) " +
+ "FROM string_tbl",
+ "SELECT unix_timestamp(ts_str COLLATE UTF8_LCASE) FROM
string_tbl")) {
+ checkSparkAnswerAndImpl(query, native = Seq.empty, dispatched =
Seq("unix_timestamp"))
+ }
+ }
+ }
+ }
+ }
+
+ test("unix_timestamp - date and timestamp inputs ignore collated formats and
stay native") {
+ assume(isSpark40Plus, "string collation requires Spark 4.0+")
+ val data = Seq(
+ (
+ java.sql.Date.valueOf("2024-06-15"),
+ java.sql.Timestamp.valueOf("2024-06-15 10:30:45"),
+ java.time.LocalDateTime.parse("2024-06-15T10:30:45")),
+ (
+ java.sql.Date.valueOf("1969-12-31"),
+ java.sql.Timestamp.valueOf("1969-12-31 23:59:59.500000"),
+ java.time.LocalDateTime.parse("1969-12-31T23:59:59.500000")),
+ (null, null, null))
+ withParquetTable(data, "tbl") {
+ for {
+ column <- Seq("_1", "_2", "_3")
+ format <- Seq("'unused'", "CAST(NULL AS STRING)")
+ allowIncompatible <- Seq("false", "true")
+ } {
+ withSQLConf(
+ CometConf.getExprAllowIncompatConfigKey("UnixTimestamp") ->
allowIncompatible) {
+ checkSparkAnswerAndImpl(
+ s"SELECT unix_timestamp($column, $format COLLATE UTF8_LCASE) FROM
tbl",
+ native = Seq("unix_timestamp"),
+ dispatched = Seq.empty)
+ }
}
}
}
diff --git
a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnixTimestampBenchmark.scala
b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnixTimestampBenchmark.scala
new file mode 100644
index 0000000000..85f2a30b14
--- /dev/null
+++
b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnixTimestampBenchmark.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.spark.sql.benchmark
+
+/**
+ * Compares string parsing through codegen dispatch with Spark and native
timestamp input. Run
+ * with:
+ * {{{
+ * make benchmark-org.apache.spark.sql.benchmark.CometUnixTimestampBenchmark
+ * }}}
+ */
+object CometUnixTimestampBenchmark extends CometBenchmarkBase {
+ override def runCometBenchmark(mainArgs: Array[String]): Unit = {
+ val rows = 1024 * 1024
+ withTempPath { dir =>
+ withTempTable("parquetV1Table") {
+ prepareTable(
+ dir,
+ spark
+ .range(rows)
+ .selectExpr(
+ "timestamp_seconds(id) AS ts",
+ "date_format(timestamp_seconds(id), 'yyyy-MM-dd HH:mm:ss') AS s",
+ "CASE WHEN id % 2 = 0 THEN 'yyyy-MM-dd HH:mm:ss' " +
+ "ELSE 'yyyy-MM-dd H:m:s' END AS fmt"))
+ for ((shape, arguments) <- Seq(
+ "default format" -> "s",
+ "column format" -> "s, fmt",
+ "native timestamp" -> "ts")) {
+ val name = s"unix_timestamp ($shape)"
+ runBenchmark(name) {
+ runExpressionBenchmark(
+ name,
+ rows,
+ s"SELECT unix_timestamp($arguments) FROM parquetV1Table")
+ }
+ }
+ for ((shape, arguments) <- Seq("default format" -> "s", "column
format" -> "s, fmt")) {
+ val name = s"unix_timestamp ($shape, grouped aggregation)"
+ runBenchmark(name) {
+ runExpressionBenchmark(
+ name,
+ rows,
+ s"SELECT unix_timestamp($arguments) AS u, count(*) FROM
parquetV1Table GROUP BY u")
+ }
+ }
+ }
+ }
+ }
+}
diff --git
a/spark/src/test/spark-4.0/org/apache/spark/sql/CometCollationSuite.scala
b/spark/src/test/spark-4.0/org/apache/spark/sql/CometCollationSuite.scala
index 76df61a515..2f1a605296 100644
--- a/spark/src/test/spark-4.0/org/apache/spark/sql/CometCollationSuite.scala
+++ b/spark/src/test/spark-4.0/org/apache/spark/sql/CometCollationSuite.scala
@@ -250,14 +250,11 @@ class CometCollationSuite extends CometTestBase {
// ---- datetime expression collation guards (issue #4646)
--------------------------------
//
- // Comet's native datetime functions use string arguments (format patterns,
timezones,
- // day-of-week) as raw bytes, so non-default collations on those arguments
must not reach
- // the native path silently. `unix_timestamp` has no codegen-dispatcher
fallback and falls
- // back to Spark entirely. Expressions with CodegenDispatchFallback
- // (next_day, trunc, date_trunc, date_format, from_unixtime, make_timestamp,
- // to_unix_timestamp, convert_timezone) fall back to Spark when
- // COMET_SCALA_UDF_CODEGEN_ENABLED is false or route through Spark codegen
inside the Comet
- // pipeline when it is true.
+ // Native datetime functions that interpret string arguments must account
for their
+ // collation. Expressions with CodegenDispatchFallback use Spark codegen for
unsupported
+ // cases when COMET_SCALA_UDF_CODEGEN_ENABLED is true, or fall back to Spark
when it is false.
+ // unix_timestamp uses codegen for string inputs. Date and timestamp inputs
ignore the
+ // format argument, including its collation, and stay native even with
codegen disabled.
private def withDatetimeCollationTable(f: => Unit): Unit = {
withParquetTable(
@@ -299,11 +296,20 @@ class CometCollationSuite extends CometTestBase {
"next_day does not support non-UTF8_BINARY collations")
}
- test("unix_timestamp rejects non-UTF8_BINARY collated format (issue #4646)")
{
- checkDatetimeFallback(
- "SELECT unix_timestamp(CAST(_2 AS TIMESTAMP), _7 COLLATE utf8_lcase) " +
- "FROM datetime_collation_tbl",
- "unix_timestamp does not support non-UTF8_BINARY collations")
+ test("unix_timestamp stays native with a collated format for timestamp input
(issue #4646)") {
+ withDatetimeCollationTable {
+ for (codegenEnabled <- Seq("false", "true")) {
+ withSQLConf(
+ CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> codegenEnabled,
+ CometConf.getExprAllowIncompatConfigKey("UnixTimestamp") -> "false")
{
+ checkSparkAnswerAndImpl(
+ "SELECT unix_timestamp(CAST(_2 AS TIMESTAMP), _7 COLLATE
utf8_lcase) " +
+ "FROM datetime_collation_tbl",
+ native = Seq("unix_timestamp"),
+ dispatched = Seq.empty)
+ }
+ }
+ }
}
test("from_unixtime rejects non-UTF8_BINARY collated format (issue #4646)") {
diff --git
a/spark/src/test/spark-4.1/org/apache/spark/sql/CometCollationSuite.scala
b/spark/src/test/spark-4.1/org/apache/spark/sql/CometCollationSuite.scala
index bb1c223455..2d6fe81dfc 100644
--- a/spark/src/test/spark-4.1/org/apache/spark/sql/CometCollationSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/spark/sql/CometCollationSuite.scala
@@ -71,14 +71,11 @@ class CometCollationSuite extends CometTestBase {
// ---- datetime expression collation guards (issue #4646)
--------------------------------
//
- // Comet's native datetime functions use string arguments (format patterns,
timezones,
- // day-of-week) as raw bytes, so non-default collations on those arguments
must not reach
- // the native path silently. `unix_timestamp` has no codegen-dispatcher
fallback and falls
- // back to Spark entirely. Expressions with CodegenDispatchFallback
- // (next_day, trunc, date_trunc, date_format, from_unixtime, make_timestamp,
- // to_unix_timestamp, convert_timezone) fall back to Spark when
- // COMET_SCALA_UDF_CODEGEN_ENABLED is false or route through Spark codegen
inside the Comet
- // pipeline when it is true.
+ // Native datetime functions that interpret string arguments must account
for their
+ // collation. Expressions with CodegenDispatchFallback use Spark codegen for
unsupported
+ // cases when COMET_SCALA_UDF_CODEGEN_ENABLED is true, or fall back to Spark
when it is false.
+ // unix_timestamp uses codegen for string inputs. Date and timestamp inputs
ignore the
+ // format argument, including its collation, and stay native even with
codegen disabled.
private def withDatetimeCollationTable(f: => Unit): Unit = {
withParquetTable(
@@ -120,11 +117,20 @@ class CometCollationSuite extends CometTestBase {
"next_day does not support non-UTF8_BINARY collations")
}
- test("unix_timestamp rejects non-UTF8_BINARY collated format (issue #4646)")
{
- checkDatetimeFallback(
- "SELECT unix_timestamp(CAST(_2 AS TIMESTAMP), _7 COLLATE utf8_lcase) " +
- "FROM datetime_collation_tbl",
- "unix_timestamp does not support non-UTF8_BINARY collations")
+ test("unix_timestamp stays native with a collated format for timestamp input
(issue #4646)") {
+ withDatetimeCollationTable {
+ for (codegenEnabled <- Seq("false", "true")) {
+ withSQLConf(
+ CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> codegenEnabled,
+ CometConf.getExprAllowIncompatConfigKey("UnixTimestamp") -> "false")
{
+ checkSparkAnswerAndImpl(
+ "SELECT unix_timestamp(CAST(_2 AS TIMESTAMP), _7 COLLATE
utf8_lcase) " +
+ "FROM datetime_collation_tbl",
+ native = Seq("unix_timestamp"),
+ dispatched = Seq.empty)
+ }
+ }
+ }
}
test("from_unixtime rejects non-UTF8_BINARY collated format (issue #4646)") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
