feat: unix timestamp codegen dispatch

[Satyr09]

Which issue does this PR close?

Closes #5577.

Closes #5896.

Rationale for this change

unix_timestamp on a string column currently falls back to Spark. Comet's codegen dispatcher can run Spark's existing implementation, keeping the projection and surrounding operators in Comet while preserving Spark's parsing behavior.

This PR also fixes an existing rounding error in the native timestamp path, found while checking the typed-input cases.

What changes are included in this PR?

• Add CodegenDispatchFallback to CometUnixTimestamp. String inputs, including collated strings, use Spark's generated code and preserve parser policy, ANSI mode and time zone behavior. They have no native implementation, even with allowIncompatible=true.
• Keep date, timestamp and timestamp without time zone inputs native. Spark ignores their format argument, so the format's collation should not change how they run.
• Correct the Rust conversion of fractional timestamps to seconds. It now truncates toward zero, matching Spark. At UTC, 1969-12-31 23:59:58.5 produces -1 instead of -2. Casting that timestamp to BIGINT still produces -2, as Spark deliberately uses floor division for the cast.
• Strengthen the SQL and Scala tests to assert native execution or dispatch for the expression itself. Update both collation suites and add tests comparing unix_timestamp(ts) with CAST(ts AS BIGINT) on the same Parquet column, requiring both to run natively.
• Add grouped aggregation cases to the SQL tests and benchmark. Update the expression audit, parser-policy notes and generated compatibility explanation.

The duplicate input check mentioned in #5577 was already removed by #5595.

How are these changes tested?

Validated commit 29a77851e with CometTemporalExpressionSuite and the SQL file filter matching unix_timestamp. Spark 4.0 and 4.1 also ran the full CometCollationSuite.

Spark version	Tests passed	Skipped
3.4.3	55	2
3.5.9	55	2
4.0.4	83	0
4.1.3	78	0
4.2.0, experimental	57	0

The Spark 3 skips require Spark 4 collation support. Tests cover string dispatch, typed native execution, collated formats, parser policies, ANSI errors, time zones and nulls. The new rounding comparisons include negative and positive fractions, whole seconds and zero, with and without nulls. Spark 3.4 through 4.1 results, Spark 4.2 results.

Formatting, style, license, registration and generated documentation checks passed. Semantic Scalafix ran on Spark 4.0, with syntactic checks across all Spark sources. Both formatting patches were empty. Formatting and documentation results.

The initial workflow had two setup errors: SemanticDB is unavailable for the Scala version used by Spark 4.1, and Spark 4.2 does not compile CometCollationSuite. Both were corrected in the fork's CI workflow and the affected checks passed on rerun.

The release native library and all four native unix_timestamp tests passed at eac00028c. The later test runs checked that their native source tree was identical before reusing that library. That run also confirmed that removing the dispatch mixin makes the two string-routing tests fail while typed inputs remain native. Native validation.

Benchmark results

Measured at eac00028c on Spark 4.0.4, JDK 17.0.20 and an Intel Xeon 6973P-C runner, using the release native library. The follow-up commit changes tests and documentation, with no changes to execution routing, native kernels or the benchmark.

The input contains 1,048,576 Parquet rows with one distinct timestamp per row. The grouped queries therefore also have 1,048,576 groups. A temporary benchmark case disables spark.comet.exec.scalaUDF.codegen.enabled to compare dispatch with fallback on the same source, data and runner. Plan assertions check the expected routes. This comparison code is confined to the fork's CI workflow.

Best elapsed time in milliseconds, lower is better:

Query	Spark	Comet, dispatch disabled	Comet, dispatch enabled
String projection, default format	403	367	373
String projection, format column	1,260	1,228	1,269
Native timestamp projection	73	35	39
String grouped aggregation, default format	1,197	1,041	561
String grouped aggregation, format column	2,018	1,873	1,467

The string projection rows are a no-regression check. They show no parsing speedup, and dispatch was slightly slower than fallback in this run. The native timestamp row is a control.

The grouped queries took about 46% and 22% less time with dispatch enabled than with it disabled. Their plans retained CometProject, both CometHashAggregate stages and CometExchange. With dispatch disabled, only the scan remained in Comet.

These are results from one CI run, with two to six measured iterations per string case. Performance will depend on the data and query shape. Full timings, standard deviations, query plans and comparison patch are available via the linked run until 20 September 2026 (due to retention policies).

[sunchao]

Correctness

Reviewed 7b2dfcd8cc06bd3db78579acfffc4fb42050a7d3 against base 4eeb1f80f0541f72389a11e6e2d0ee269d648c23. I found no new P1/P2 issue. Previously, string unix_timestamp inputs left Comet because the serializer only supported native date/timestamp inputs. Adding CodegenDispatchFallback lets the existing dispatcher evaluate Spark's generated expression inside the Comet plan. The input-type guard still precedes the collation check, so enabling incompatible native expressions cannot send strings to the native timestamp implementation. Compatible date, timestamp, and timestamp-NTZ inputs retain their existing native route. Disabling JVM codegen retains fallback.

The dispatcher binds and serializes the complete Spark expression, retaining its format expression, timezone, and ANSI flag. The maintained Spark 3.5/4.0 implementations confirm literal and per-row format handling, null evaluation order, parsing exceptions, and truncation of fractional seconds on the new string route. The JNI bridge installs task context before evaluation, allowing parser-policy lookup to use task configuration. Canonical maintained Spark 3.4/4.1 branches were unavailable for source comparison.

The author's test run explicitly checked out this PR head: 57 tests passed on each Spark 4.0/4.1/4.2 job, and 55 passed with two collation tests canceled on each 3.4/3.5 job. Coverage includes string formats, nulls, invalid inputs, ANSI errors, parser policies, timezone/DST cases, collations, dispatch controls, and retained native inputs. The native artifact's recorded source tree matches this head. Removing the mixin intentionally failed the three dispatch tests while the three native controls passed. I inspected this evidence. I did not run local suites. Apache test workflows still awaited approval at 2026-09-09T07:35:26Z, and these author jobs did not execute the current base/head merge.

Performance

The change reuses the existing per-task generated kernel and Spark formatter handling. It adds no new parsing loop or cache and preserves the native fast path. Literal formats can reuse a formatter, while dynamic formats retain Spark's per-row formatter creation cost.

The executed benchmark uses 1,048,576 persisted rows, consumes the projected expression, disables constant folding in both arms, and verifies that all three cases avoid trivial-plan and Comet-fallback warnings. On Spark 4.0.4/JDK 17, best times were 739 → 672 ms for the default string format and 1,841 → 1,782 ms for a format column. These are modest observations from one runner with three and two timed iterations per arm, respectively. The 91 → 52 ms native-timestamp control measures pre-existing behavior. This supplies useful focused performance evidence without establishing a general speedup.

Design

The marker mixin is a small, consistent extension of the existing serializer contract and the sibling datetime dispatch paths. Unsupported string inputs reuse Spark's parser, while compatibility checks continue to govern native conversion. Serializing the whole expression keeps format and error semantics together. Implementing another native string parser would introduce substantially more compatibility work. The documentation's hybrid classification accurately reflects the resulting execution paths.

Abstraction & complexity

No additional abstraction or native implementation is introduced. The SQL fixtures exercise semantic behavior, and the Scala tests reuse the existing codegen assertions to establish actual dispatch and native-route selection. The mutation control strengthens those assertions by demonstrating that the new routing tests detect removal of the production change. This division makes the implementation and its validation straightforward to maintain. I found no further actionable complexity issue.

[rich7420]

LGTM , @Satyr09 thanks for the patch!

[andygrove]

I checked this out and ran it locally against Spark 4.1.3 before commenting. All 22 unix_timestamp SQL fixtures pass, CometTemporalExpressionSuite is 35 of 35, and I added ten probe tests of my own that all pass as well, covering routing, aggregate context where the serde is called with binding = false, dictionary-encoded Parquet input, ANSI error class parity rather than just the message text, invalid per-row format patterns, and awkward strings such as leading whitespace, Arabic-Indic digits, hour 24 and second 60. I also repeated your mutation experiment on the SQL side. Removing the mixin fails 12 of the 22 fixtures as well as the three suite tests, so the fixtures really do pin this.

One thing I wanted to rule out specifically. CometScalaUDFCodegen carries a TODO(dict-encoded) saying dictionary-encoded vectors would fail the cast in specFor, and this change routes a Parquet string column into exactly that path. It turns out the column arrives already materialized as a plain VarCharVector, so there is no hazard, and your parquet.enable.dictionary matrix confirms it end to end. Good instinct adding that matrix.

I did not find a correctness problem. The questions I have left inline are about the collation gate, about pinning the routing now that #5610 has landed, and about what the benchmark is actually measuring.

[andygrove]

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.

[Satyr09]

Good catch!
Fixed - the timestamp/date types with a collate attached now execute natively.

[andygrove]

#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.

[Satyr09]

Aah, thanks for that context. Updated.

[andygrove]

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.

[Satyr09]

Sure!

[andygrove]

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.

[Satyr09]

Fair, updated

[andygrove]

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.

[andygrove]

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?

[Satyr09]

Yeah that makes sense to me. Updated.

[sunchao]

Re-reviewed 0249715725 against 3810936b40 after the merge from main. All twelve files retain the same authored changes as the previously reviewed 7b2dfcd8cc; the timestamp serializer, SQL fixtures, temporal suite and benchmark are byte-identical. I also checked the updated shared serializer and test harness, plus Spark's maintained 3.5/4.0 implementations. No additional P1/P2 findings; my existing approval stands.

Andy's open comments already cover the unnecessary collation gate on an ignored format argument, stronger per-expression routing assertions, and benchmark/documentation improvements. I have no additional inline findings.

Validation is source review and provenance checks, with no new local Spark/JNI execution. The earlier fork runs exercised 7b2dfcd8cc; the merged head has a different native tree, so those runs do not establish a current-head pass. Andy reports passing Spark 4.1.3 tests in his review. Apache CI, CodeQL and the Delta gate still await approval. Maintained Spark 3.4/4.1 source branches remain unavailable for independent source verification.

[andygrove]

Dropping the collation branch from getSupportLevel is the right call, but it leaves CometCollationSuite asserting the old behavior. The test at spark/src/test/spark-4.1/org/apache/spark/sql/CometCollationSuite.scala:123 and its twin at spark/src/test/spark-4.0/org/apache/spark/sql/CometCollationSuite.scala:302 run SELECT unix_timestamp(CAST(_2 AS TIMESTAMP), _7 COLLATE utf8_lcase) and require the fallback reason unix_timestamp does not support non-UTF8_BINARY collations. That string came from DatetimeCollation.reason("unix_timestamp") through the collationReason val this PR deletes, so nothing emits it any more and both tests fail. CometCollationSuite is in the suite lists in pr_build_linux.yml and pr_build_macos.yml, so that is red CI on both Spark 4 profiles rather than a local artifact. Would you flip those two to assert native execution with a collated format, mirroring the new unix_timestamp - date and timestamp inputs ignore collated formats and stay native test you added to CometTemporalExpressionSuite? The comment block above them needs the same treatment, since it still says unix_timestamp falls back to Spark entirely.

Separately, the description says the native implementation is reused without Rust changes, but native/spark-expr/src/datetime_funcs/unix_timestamp.rs swaps div_floor for plain integer division on all three timestamp branches. That is a real answer change for sub-second values before the epoch. At UTC, unix_timestamp(TIMESTAMP '1969-12-31 23:59:58.5') used to return -2 and now returns -1. Spark's ToTimestamp.eval does t.asInstanceOf[Long] / downScaleFactor and doGenCode emits the same division, so -1 is right and the old value was a bug. It is a subtle one to get backwards, because Cast(timestamp, LongType) really does use Math.floorDiv, which is why conversion_funcs/utils.rs keeps div_floor. Could you call the fix out in the description and in the unix_timestamp bullet in docs/source/contributor-guide/expression-audits/datetime_funcs.md, and open an issue so it shows up in the release notes? Someone reading the title has no way to know the native timestamp path changed.

While you are editing the description, the benchmark table still reads as evidence that dispatch is faster than Spark on the two string cases, and both your numbers and mine put them at parity. Could you add the grouped-aggregation numbers from the new cases and say plainly that the string rows are a no-regression check?

Two smaller things. docs/source/contributor-guide/spark_configs_support.md:127 still says the string-input overload of unix_timestamp falls back, inside the audit notes for spark.sql.legacy.timeParserPolicy. That is the line a contributor checks before touching parser-policy behavior and #5577 points at it, so it should read like the to_unix_timestamp bullet below it now. And the unsupported reason in datetime.scala now renders under the header "The following cases have no native implementation and always run in the JVM using Spark's code-generated implementation", but the bullet lists the types that do have a native path, so the page reads backwards. Would something like "String inputs, including collated strings, have no native implementation" work better?

[Satyr09]

Thanks @andygrove , @sunchao for the reviews.
Updated the PR based on your comments.

Also discovered a pre-existing in the course of writing some regression tests - documented it here: #5896

[Satyr09]

aah, sorry @andygrove just saw your new comment.
Thanks for the super quick review!
I've already updated the description and I'm currently fixing the test coverage issue you mentioned.

Re: the newly discovered negative epoch timestamp bug - I've documented it here : #5896

Would you prefer that that fix go in a separate PR?

[Satyr09]

Test coverage and documentation gaps fixed now.