andygrove commented on code in PR #5789:
URL: https://github.com/apache/datafusion-comet/pull/5789#discussion_r3996916794
##########
spark/src/main/scala/org/apache/comet/serde/datetime.scala:
##########
@@ -290,12 +290,14 @@ private[serde] object DatetimeCollation extends
CometTypeShim {
expr.children.exists(c => hasNonDefaultStringCollation(c.dataType))
}
-object CometUnixTimestamp extends CometExpressionSerde[UnixTimestamp] {
+object CometUnixTimestamp
+ extends CometExpressionSerde[UnixTimestamp]
+ with CodegenDispatchFallback {
private val collationReason = DatetimeCollation.reason("unix_timestamp")
override def getUnsupportedReasons(): Seq[String] = Seq(
- "Only `DateType`, `TimestampType`, and `TimestampNTZType` inputs are
supported.")
+ "Native execution only supports `DateType`, `TimestampType`, and
`TimestampNTZType` inputs.")
override def getIncompatibleReasons(): Seq[String] =
DatetimeCollation.incompatibleReasons("unix_timestamp")
Review Comment:
Now that string input always reports `Unsupported`, this collation check can
only be reached by a collated *format* argument over a date, timestamp or
timestamp-NTZ input. Spark ignores the format entirely for those types, since
`ToTimestamp.eval` and `doGenCode` only read `right` in the `StringType` branch
on 3.4 through 4.1, so the collation of the format cannot change the answer.
I confirmed that `SELECT unix_timestamp(d, 'unused' COLLATE UTF8_LCASE)` is
dispatched rather than run on the native kernel, and that the user also gets a
`[COMET-INFO]` suggesting
`spark.comet.expression.UnixTimestamp.allowIncompatible=true` for a difference
that cannot exist. That is still an improvement on the full fallback we had
before, so nothing regresses here. But would it be better to scope the check to
the input child, something like
`hasNonDefaultStringCollation(expr.children.head.dataType)`, so those inputs
stay native? The new "collated formats dispatch date and timestamp inputs by
default" test would then flip to asserting native execution, which reads as the
stronger claim.
##########
spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala:
##########
@@ -394,19 +398,27 @@ 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")) {
+ assertCodegenRan {
Review Comment:
#5610 landed `checkSparkAnswerAndImpl` after you branched, and it fits these
three tests better than `assertCodegenRan`. `checkSparkAnswerAndOperator`
passes whether the expression ran natively or through the dispatcher, so on its
own it cannot say which one happened, and `assertCodegenRan` only says that
something somewhere in the JVM dispatched.
Would you replace these with `checkSparkAnswerAndImpl(query, native =
Seq.empty, dispatched = Seq("unix_timestamp"))`? I tried it locally and it
passes as written. The same applies to the
`CometScalaUDFCodegen.stats().totalLookups == 0` assertion at line 462, where
`native = Seq("unix_timestamp"), dispatched = Seq.empty` makes the same
statement per plan instead of leaning on a JVM-wide counter that unrelated
dispatcher activity could perturb. Doing both would also let the
`CometCodegenAssertions` mixin and the `CometScalaUDFCodegen` import come back
out of this file.
##########
docs/source/contributor-guide/expression-audits/datetime_funcs.md:
##########
@@ -95,4 +95,12 @@
- 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-09): string input uses Spark's generated parser
through codegen dispatch. Literal and column formats preserve null handling,
ANSI errors, parser policy, and session time zone.
Review Comment:
The version bullets in this file record what changed in Spark between
versions, with the Comet-side behavior on the unversioned bullets.
`from_utc_timestamp` and `to_utc_timestamp` just above are the pattern. Three
of the four bullets here describe Comet's routing rather than a Spark
difference, and they say the same thing for every version.
Could the routing move down into the unversioned bullet, leaving the version
bullets for the things that genuinely differ, such as 3.5's structured parsing
errors and 4.0's collated `inputTypes`?
##########
spark/src/test/resources/sql-tests/expressions/datetime/unix_timestamp_strings.sql:
##########
@@ -0,0 +1,58 @@
+-- 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
Review Comment:
Routing is what this PR changes, so would you use `query
expect_dispatch(unix_timestamp)` here and on the parser-policy queries? A plain
`query` only asserts that the projection stayed in Comet. A later change that
kept the expression in Comet but moved it back onto a native kernel with
different parsing would still pass, as long as the answers happened to agree on
these particular inputs.
To be clear about the severity, I mutation-tested the fixtures as they stand
and they do catch removal of the mixin, 12 of 22 fail. This is about locking
the direction in rather than filling a hole.
##########
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.
Review Comment:
The comment claims a native path but the assertion below cannot see it.
Because this file disables the dispatcher, a flip to dispatch would surface as
a fallback and the test would fail anyway, so it is covered today. `query
expect_native(unix_timestamp)` would state it directly and keep holding if
someone later changes the config header at the top of the file.
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnixTimestampBenchmark.scala:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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(
Review Comment:
I ran this locally on an M3 Max at `local[1]`, release build, same row
count, and got 351 vs 340 ms for the default format and 1036 vs 1002 ms for the
format column, with the native timestamp control at 37 vs 24 ms. That lines up
with your runner. The two string cases are at parity rather than faster, which
makes sense, because Spark's per-row `TimestampFormatter.parse` dominates both
arms and the dispatcher neither helps nor hurts it.
The benefit of this change is structural instead. The projection and
everything above it stay on Comet rather than paying a columnar-to-row round
trip and demoting the surrounding operators. A scan into project into noop
shape cannot show that, because the old fallback cost exactly one
`CometColumnarToRow` here and nothing downstream. I measured the pre-change
routing by accident at one point and got 361 and 995 ms for the same two cases,
which is inside the noise.
Could you add a case where the projection feeds a Comet operator, something
like `SELECT unix_timestamp(s) AS u, count(*) FROM parquetV1Table GROUP BY u`?
It would also be worth saying in the description that the string cases are a
no-regression check rather than a speedup, because the table as written reads
as evidence that dispatch is faster than Spark's whole-stage codegen for this
expression, and it is not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
