You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/21 15:55:16 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-5789-5ca149928f7743bfe7a96feadea5e0f9bed1412f created (now 09b44ad6fa)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5789-5ca149928f7743bfe7a96feadea5e0f9bed1412f
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 09b44ad6fa feat: unix_timestamp codegen dispatch for strings, and fix pre-epoch fractional rounding (#5789)
This branch includes the following new commits:
new 09b44ad6fa feat: unix_timestamp codegen dispatch for strings, and fix pre-epoch fractional rounding (#5789)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: feat: unix_timestamp codegen dispatch for strings, and fix pre-epoch fractional rounding (#5789)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5789-5ca149928f7743bfe7a96feadea5e0f9bed1412f
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 09b44ad6fa17f58f4bbccf958c5cf02d790f7334
Author: Daipayan Mukherjee <da...@gmail.com>
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 <ag...@apache.org>
---
.../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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org