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 <[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]
