Repository navigation
fix: align time parsing and native second extraction with Spark - #5738
Conversation
andygrove
added
bug
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Prior state, approach, and behavior
Reviewed head cb4e5d6e against base and merge-base 7e198439. The native parser rejected hour-only T12 and an empty fractional part such as 12:30:45.. The changes accept those forms while retaining component-range checks, complete-input validation, AM/PM handling, and six-digit fractional truncation. to_time keeps its throwing behavior for invalid input; try_to_time returns NULL.
The new Spark 4.1+ shim recognizes only getSecondsOfTimeWithFraction with a TIME argument, literal precision 0–6, and Decimal(8,6) output. Other shapes retain the existing dispatcher/fallback route. The native kernel preserves scalar versus array shape and NULLs, extracts the seconds within a minute, truncates to the requested precision, and returns scale 6. For valid TIME values, its arithmetic remains within Decimal(8,6) bounds; the precision guard also bounds both powers. It performs no timezone conversion. The planner addition decodes non-NULL TIME literals using the existing long-valued representation. I found no new correctness defect in the reviewed diff.
Spark comparison and validation limits
The maintained Spark 3.5/4.0 branches have no TIME type or fractional TIME extraction implementation. Their related timestamp parsers accept the two newly supported forms and truncate fractional input to microseconds; that is auxiliary parser evidence only. Maintained Spark 3.4 and 4.1 sources were unavailable, so this review does not claim independent source qualification for those versions. The Spark 4.1 compatibility evidence comes from the inspected CI tests.
The successful CI merge 3e12010d has the assigned head/base as parents and a tree identical to the reviewed head. The Rust job passed all 18 parser tests and both new extraction tests, including precisions 0–6, scalar/array NULLs, and boundary values. The Spark 4.1 expression job passed both SQL fixtures; its native library is traceable to the same reviewed source. The extraction fixture disables the JVM dispatcher and the harness disables constant folding. Its literal casts establish TIME precision during serialization, so these tests cover native literal/extraction handling rather than native TIME-to-TIME casts. The fixture is restricted to Spark 4.1; its skipped Spark 4.2 entry provides no extraction coverage. No local tests or performance runs were performed for this review.
Performance
[P2] Add the required microbenchmark and results for the new native extraction path. The existing parser benchmarks do not measure it, and all PR benchmark checks are skipped. Compare matched Spark, the previous Comet JVM dispatcher, and native execution with plans verified; include column inputs, NULLs, representative precisions, and parsing/extraction composition. The inline comment describes the requested evidence. No slowdown is established by the current data.
The parser edits add no allocation. Extraction computes its scale factors once per call and produces one decimal output array using per-value remainder, division, and multiplication. Source inspection reveals no unnecessary input copy or repeated scale computation, but cannot establish the benefit of replacing the dispatcher.
Design
The change keeps the parser fixes shared between throwing and non-throwing calls and gives fractional TIME extraction a separate function from integer timestamp second. Passing precision explicitly preserves information that the Arrow nanosecond type alone does not carry. The exact method, argument, and result-type guard keeps version-specific behavior localized. Supporting the existing TIME literal encoding completes the native path used by the precision tests. I found no actionable design defect; the performance comparison is the remaining pre-merge request for the new execution path.
Abstraction & complexity
The implementation reuses the scalar-function factory, datetime module, and version shim. A short extraction function and a targeted literal-decoding branch fit the change without introducing a new framework or intermediate representation. The scalar and array cases share the arithmetic closure, which keeps precision and NULL handling straightforward to audit. No additional abstraction or simplification is required by the reviewed scope.
| ColumnarValue::Array(array) => { | ||
| let times = as_time64_nanosecond_array(array.as_ref())?; | ||
| let seconds: Decimal128Array = times.iter().map(|nanos| nanos.map(extract)).collect(); | ||
| Ok(ColumnarValue::Array(Arc::new( | ||
| seconds.with_precision_and_scale(8, 6)?, |
There was a problem hiding this comment.
Performance
[P2] Add a focused microbenchmark and results for this new native EXTRACT(SECOND FROM TIME) path before enabling it. The existing to_time benchmarks measure parsing, and this PR supplies correctness tests but no extraction benchmark or measurements. The new shim replaces the existing Comet JVM codegen route for eligible expressions, so a Spark-only comparison would also miss that baseline. Please compare matched Spark, existing Comet dispatcher, and native runs on column input, including NULLs and representative TIME precisions, and verify the execution plans. Isolate extraction from parsing where possible and include the composed to_time case. If native is slower, document the acceptance rationale or a concrete optimization plan.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
The update from cb4e5d6e to ce9ec965 adds benchmarks and exports the existing kernel for the Rust benchmark. The parser, extraction arithmetic, precision guard and literal handling are unchanged. I rechecked the units: TIME values use nanoseconds, while the parser truncates fractional input to six digits before scaling microseconds to nanoseconds. Arrow 59.3 and the DataFusion 55 downcast preserve that representation. The extraction kernel truncates at the declared TIME precision and returns Decimal(8,6). The SQL fixture's literal casts are evaluated during serialization, so they do not establish native TIME-to-TIME cast coverage.
The Rust job passed 1,192 tests, including the parser and extraction tests. The Spark 4.1 expression job passed 1,329 tests, including both TIME fixtures, with its downloaded native artifact matched to the producer by ID and digest. These jobs tested merge 2241450b with newer base 92ad99e9, rather than the assigned base 7e198439. All 11 authored code, test and benchmark files match the reviewed head. Maintained Spark 4.1 source remains unavailable, so this is qualified CI evidence, not independent Spark 4.1 source qualification. At the 2026-09-08 04:53 UTC refresh, 65 checks had succeeded, 8 were skipped and one Iceberg job remained running. The Spark 4.0 shuffle job had failed on a Maven dependency HTTP 403 before tests.
Performance
The previous P2 benchmark request is addressed. The added benchmarks compare Spark, the previous Comet dispatcher and native execution on matched inputs, including precisions 0/3/6 and 0%/25% NULLs. Direct extraction uses aligned TIME values and consumes outputs through checksums. The composed SQL cases use materialized string columns, compare answers and verify execution paths. Their fractional-digit variants all parse to TIME(6), which the PR now states explicitly.
The author reports faster native extraction and better composed SQL times than the previous dispatcher. The nullable zero-digit SQL case has a higher mean than Spark: 474 ms native versus 437 ms for the repeated Spark control, with a reported native standard deviation of 200 ms. That variance and the repeat controls do not establish a repeatable regression. The acceptance rationale addresses that uncertainty. Direct timings include different output representations and exclude JNI for the dispatcher, so they should not be read as pure arithmetic comparisons. I inspected the benchmark implementation and reported results but did not independently rerun the measurements.
Design
Calling kernels directly is justified by the current limitation on scanning TIME columns. The SQL benchmark separately exercises conversion, parsing and extraction. Its dispatcher baseline uses the existing serializer for the whole expression tree, clears the serialized plan after replacing the projection, and checks dispatcher counters. This gives the requested comparison with the previous Comet path without adding a production execution mode.
Abstraction & complexity
The public kernel re-export is needed by the external Rust benchmark. The remaining additions stay in benchmark code and reuse existing projection, serialization and checksum machinery. I found no new or remaining P1/P2 issue in this follow-up.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for this. I traced the new parser inputs through Spark's parseTimestampString and stringToTime and they line up, including all of the invalid cases in the new to_time.sql query. The spark_seconds_of_time kernel also matches Spark's getSecondsOfTimeWithFraction for every precision, since the double detour Spark takes cannot change the rounding at scale 6. The one failing check (Spark 4.0 shuffle) is an expired native-lib-linux artifact on a re-run and is unrelated.
Two small comments inline about the Spark version range. Spark 4.2.0 still uses Decimal(8,6) here, so the fixture is skipping coverage on 4.2 that it could have.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked unchanged ce9ec965 against 7e198439 after Andy’s review. The earlier benchmark P2 remains addressed. No new verified P1/P2 finding; the existing approval stands.
The Spark 4.2 profile pins 4.2.0 and includes the shim, whose native eligibility requires Decimal(8,6) and literal precision 0–6. The fixture’s MaxSparkVersion: 4.1 excludes its entire body on 4.2. Its zero-duration CI entry therefore provides no TIME(0)–TIME(5) coverage. Andy’s existing coverage/documentation comments remain open. I could not independently verify the claimed 4.2 result type or double-conversion semantics because the prescribed maintained Spark 4.1/4.2 sources are unavailable.
The current Spark 4.0 shuffle failure occurs while downloading native-lib-linux, before Java tests. Artifact metadata confirms expiration before the rerun. This differs from the earlier attempt’s Maven HTTP 403 failure and establishes no code regression. Current checks: 66 successful, 8 skipped, 1 failed.
These CI jobs checked out 2241450b, with newer base 92ad99e9, rather than the assigned base. Eleven authored code/test/benchmark files and the inspected dependency/dispatcher context match head. The successful native and Spark 4.1 logs remain byte-identical to the prior review. No local build, test or benchmark was run.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed bf31269c against 481aefea. The Spark 4.2 coverage request is addressed: extract_time.sql now runs through 4.2, and the documentation retains the Decimal(8,6) qualification. The native parser, extraction arithmetic, and eligibility checks are unchanged from the previous review.
Current CI ran both TIME fixtures successfully on Spark 4.1 and 4.2, with the JVM dispatcher disabled for fractional extraction. The two native extraction tests and all 18 parser tests also passed. CI's merge checkout has the same complete source tree as this head. The four canceled Spark 4.2 tests concern unrelated integer-overflow cases.
No new or remaining verified P1/P2 issue. The earlier benchmark P2 remains addressed; its timing results are author-reported and were not rerun here. Required local Spark 3.4/4.1/4.2 branches remain unavailable, so this review does not claim independent canonical-source verification for them.
…compatibility Conflicts with apache#5771, which rewrote the hour/minute/second path in extract_date_part.rs and added native dayofweek/weekday kernels. Both sides were additive: this branch adds spark_seconds_of_time for EXTRACT(SECOND FROM TIME), upstream adds the integer clock fast path and SparkDayOfWeek/SparkWeekDay. - extract_date_part.rs: union of the two import blocks; the extract_date_part! invocations keep upstream's four-argument form. - datetime_funcs/mod.rs: keep both re-exports. - lib.rs: union of the two export lists. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
@sunchao I resolved the conflict. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 8153b82d against 8c229a70. The conflict resolution preserves the TIME parser, fractional-second kernel and tests, precision/type guard, SQL fixtures and benchmarks from the prior approved head, while retaining main's separate timestamp optimization. No new or remaining verified P1/P2 findings. The earlier benchmark, Spark 4.2 coverage and documentation requests remain addressed.
The current Spark 4.1 build passed on fcaf240d, whose complete tree equals the reviewed head, with tests explicitly skipped. At 08:34 UTC on September 16, CI had 15 successful checks, 13 skipped and the native build still running. No local runtime or benchmark was rerun. Prior author timings are unchanged evidence, not a newly measured speedup. Maintained Spark 3.4/4.1/4.2 source gaps remain recorded.
The 4.1+ shim matched the TIME family's StaticInvoke forms and sent them to the codegen dispatcher, and CometCast reported every TIME cast as Compatible so that convert could dispatch it. Main does both already. Since apache#5692, CometStaticInvoke sends any StaticInvoke it has no handler for to the dispatcher. CometCast mixes in CodegenDispatchFallback, and every TIME cast falls through to an Unsupported arm in isSupported, which the mixin dispatches. With both removed, every TIME fixture passes and reports the same native and dispatched expressions. Removing the shim arm also keeps it out of the way of apache#4997, which adds native arms for getHoursOfTime, getMinutesOfTime and getSecondsOfTime to the same match. Its comment had also gone stale: getSecondsOfTimeWithFraction runs natively since apache#5738.
Every fixture's comments said its queries ran through the codegen dispatcher, but the plain `query` mode cannot tell the dispatcher from a native kernel. Queries that do dispatch now say so with `expect_dispatch`. Those are the column arguments, and the literal arguments to the TIME functions. A cast of a literal is folded at planning time and a NULL argument records no expression, so those stay plain queries. The comments now credit CometStaticInvoke and CodegenDispatchFallback, which do the routing since the shim and cast wiring were dropped. - subtract_times.sql pinned `time - time` with `spark_answer_only` on the premise that it fell back. It runs in Comet now. - second_time.sql drops its EXTRACT(SECOND FROM TIME) block. That expression runs natively since apache#5738, and extract_time.sql covers it. - time_add_interval.sql adds an interval computed from a column, and the DATETIME_OVERFLOW error for a result past midnight. - The local-scan TIME test adds a NULL, TIME(3), and TIME inside an array and a struct, with two rows to a batch so each scan writes several batches.
apache#4821) * feat: support Spark 4.1 TIME type and expressions via codegen dispatch Wires the Spark 4.1 TIME data type end-to-end and adds JVM codegen dispatch for the new TIME expressions. No native implementations are added: everything routes through the existing emitJvmCodegenDispatch path so the enclosing projection stays native while Spark's own doGenCode supplies the semantics. Adds TimeType to DataTypeSupport, teaches the batch-kernel input and output emitters, CometScalaUDFCodegen, and ArrowWriters to handle TimeNanoVector as a Long-backed vector, and updates the shim so HoursOfTime, MinutesOfTime, SecondsOfTime, SecondsOfTimeWithFraction, TimeAddInterval, SubtractTimes, TimeDiff, and TimeTrunc reach the dispatcher via their StaticInvoke(DateTimeUtils, ...) replacement form. Casts touching TimeType follow the same route via CometCast. Parquet scans still opt out of TimeType (no native decoder yet); the fallback reason is now explicit. * test: cover Spark 4.1 TIME expressions via SQL-file tests Replaces the initial CometTimeTypeSuite Scala tests with per-expression sql-tests fixtures under expressions/datetime and expressions/cast, so coverage runs through the same CometSqlFileTestSuite pipeline as other datetime tests (constant folding disabled, unified reporting). Also accept Time64(Nanosecond) in the native LongVal literal path so constant-folded TIME literals from CometCast (cast literal string -> TIME evaluated at planning time) round-trip through the native scan. * review: gate nested TIME at canHandle; add DurationVector spec; scope to_time docs Restricts isSupportedDataType so ArrayType/StructType/MapType containing TIME return false at plan time, and the operator falls back cleanly instead of crashing when the array/struct/map getters do not yet route TIME to getLong. Adds DurationVector (DayTimeIntervalType) and IntervalYearVector (YearMonthIntervalType) to CometScalaUDFCodegen.specFor and to the getLong ordinal switch in CometBatchKernelCodegenInput so a non-foldable interval attribute passed into timeAddInterval (or any long-backed dispatcher call) evaluates through the dispatcher instead of throwing UnsupportedOperationException. Narrows the to_time and try_to_time docs to state that only the default-format form is supported; formatted forms fall back to Spark. * test: record that the local scan admits TIME at its supportedDataType boundary The merge of main gave CometLocalTableScanExec the default allowTimeType, since TimeNanoWriter makes TIME writable there. The boundary table in QueryPlanSerdeSuite mirrors each caller's flags, so it still passed while listing TIME as rejected, and the supportedDataType scaladoc said the same. * fix: keep TIME columns out of the native CSV scan The TIME arm in DataTypeSupport reached every operator that inherits it, not just the local scan. CometBatchScanExec gates the native CSV V2 scan with it, so a CSV read with a TIME column went native. Spark 4.1's own CSV reader rejects TIME, and Spark 4.2 parses it with the `timeFormat` option, which the native reader does not take. The arm also opened CometSparkToColumnarExec to TIME, which nothing tested. Since the merge of main, the local scan admits TIME through its own supportedDataType flags, so the arm has no remaining purpose. The new test asserts on the plan, because Spark 4.1 has no answer to compare against. * refactor: drop TIME dispatch wiring that main already provides The 4.1+ shim matched the TIME family's StaticInvoke forms and sent them to the codegen dispatcher, and CometCast reported every TIME cast as Compatible so that convert could dispatch it. Main does both already. Since apache#5692, CometStaticInvoke sends any StaticInvoke it has no handler for to the dispatcher. CometCast mixes in CodegenDispatchFallback, and every TIME cast falls through to an Unsupported arm in isSupported, which the mixin dispatches. With both removed, every TIME fixture passes and reports the same native and dispatched expressions. Removing the shim arm also keeps it out of the way of apache#4997, which adds native arms for getHoursOfTime, getMinutesOfTime and getSecondsOfTime to the same match. Its comment had also gone stale: getSecondsOfTimeWithFraction runs natively since apache#5738. * perf: write TIME through the fixed-width Arrow writer paths TimeNanoWriter extended plain ArrowFieldWriter, so it missed the unchecked row write and the bulk column copy that apache#6566 gave every other fixed-width writer. It now extends FixedWidthArrowFieldWriter like TimestampNTZWriter, and TimeNanoVector joins the 8-byte arm of copyValues. CometArrowWriterSuite compares the columnar paths with the row path for TIME on Spark 4.1+. * test: pin the TIME fixtures to the dispatcher and widen their coverage Every fixture's comments said its queries ran through the codegen dispatcher, but the plain `query` mode cannot tell the dispatcher from a native kernel. Queries that do dispatch now say so with `expect_dispatch`. Those are the column arguments, and the literal arguments to the TIME functions. A cast of a literal is folded at planning time and a NULL argument records no expression, so those stay plain queries. The comments now credit CometStaticInvoke and CodegenDispatchFallback, which do the routing since the shim and cast wiring were dropped. - subtract_times.sql pinned `time - time` with `spark_answer_only` on the premise that it fell back. It runs in Comet now. - second_time.sql drops its EXTRACT(SECOND FROM TIME) block. That expression runs natively since apache#5738, and extract_time.sql covers it. - time_add_interval.sql adds an interval computed from a column, and the DATETIME_OVERFLOW error for a result past midnight. - The local-scan TIME test adds a NULL, TIME(3), and TIME inside an array and a struct, with two rows to a batch so each scan writes several batches. * docs: bring the TIME rows up to date current_time, time_diff and time_trunc gain the spark.sql.timeType.enabled note that the other TIME rows carry. The two-argument (date, time) forms of make_timestamp_ltz and make_timestamp_ntz run through codegen dispatch rather than falling back, and so does to_time with a format argument. The TimeType row in the data types page says that Parquet scans of TIME columns fall back.
3 participants
Which issue does this PR close?
Closes #5366.
Rationale for this change
Spark accepts
to_time('T12')andto_time('12:30:45.'), but Comet rejects them or returns NULL withtry_to_time. This PR fixes those parser differences and adds native fractional-second extraction soEXTRACT(SECOND FROM TIME)can run with JVM codegen dispatch disabled.What changes are included in this PR?
The shared time parser accepts a
Tprefix on hour-only values and an empty fractional part, matching Spark 4.1's six-digit precision. The native planner also supports non-NULL TIME literals.Native second extraction returns
Decimal(8,6), preserves NULLs, and truncates to the input TIME precision from 0 through 6. The Spark mapping checks the result type to avoid selecting this implementation for incompatible Spark versions. SQL tests and expression documentation cover the new support.How are these changes tested?
Validation passed: 18 Rust parser tests, two Rust extraction tests, and the Spark 4.1.3
to_time.sqlandextract_time.sqlfixtures. These cover precision, NULLs, boundaries, and malformed strings; the extraction fixture disables JVM codegen dispatch. The release native build, JVM test compilation, Scala formatting and style checks, Rust formatting, and benchmark Clippy checks also passed.The benchmarks compare pure Spark, Comet's previous JVM dispatcher, and native Comet. All result comparisons and 18 SQL execution-path checks passed. Measurements used an Apple M4 with 24 GiB RAM, macOS 26.6.2, Java 21.0.6, Spark 4.1.3, a 4 GiB JVM heap,
local[1], batch size 8192, and a verified release native library.Comet scans currently reject TIME columns, so the isolated benchmark calls each kernel directly on matching 10,000-row inputs and consumes results through a checksum. Spark uses generated
UnsafeProjectionoverUnsafeRow; the dispatcher and native kernel use Arrow. Timings include their respective output representations, and the direct dispatcher excludes JNI overhead.Direct extraction, mean ns/row:
The SQL benchmark evaluates
extract(second from to_time(s))over materialized string columns. The previous Comet path uses the existingemitJvmCodegenDispatchto evaluate the whole expression tree. Plan assertions, serialized expressions, and dispatcher counters verify the three paths. Timings include conversion, parsing, extraction, checksumming, and task overhead, with at least five iterations after warmup. All strings parse to TIME(6).Composed SQL, best / mean milliseconds per 1,048,576 rows:
Native's best SQL times were 1.32–2.37x faster than the previous dispatcher. Variance limits comparisons with pure Spark: the nullable zero-digit case averaged 474 ms natively versus Spark's 437 ms, with a native standard deviation of 200 ms. The first Spark extraction control at precision 0 also varied from 120.06 to 57.03 ns/row on repetition. The lower extraction cost and improvement over the previous Comet path support retaining the native implementation.
Reproduce the benchmarks
Run from the repository root, with compilation complete before timing and the two benchmarks running sequentially:
Scala output is generated under
spark/benchmarks; Criterion estimates are undernative/spark-expr/target/criterion/seconds_of_time.