You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/16 23:02:36 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-5738-29075580a9c01243cbf555736abab69391c84edb created (now 36146a87bf)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5738-29075580a9c01243cbf555736abab69391c84edb
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 36146a87bf fix: align time parsing and native second extraction with Spark (#5738)
This branch includes the following new commits:
new 36146a87bf fix: align time parsing and native second extraction with Spark (#5738)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: align time parsing and native second extraction with Spark (#5738)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5738-29075580a9c01243cbf555736abab69391c84edb
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 36146a87bf9ca9ca9e211b4372628ed2f9d8c8c6
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Wed Sep 16 23:02:14 2026 +0000
fix: align time parsing and native second extraction with Spark (#5738)
* fix: align time parsing and native second extraction with Spark
* Add benchmarks for fractional TIME extraction
* Test fractional TIME extraction on Spark 4.2
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
---
docs/source/user-guide/latest/expressions.md | 2 +-
native/core/src/execution/planner.rs | 5 +-
native/spark-expr/benches/to_time.rs | 40 +++-
native/spark-expr/src/comet_scalar_funcs.rs | 5 +
.../src/datetime_funcs/extract_date_part.rs | 102 ++++++++-
native/spark-expr/src/datetime_funcs/mod.rs | 1 +
native/spark-expr/src/datetime_funcs/to_time.rs | 60 ++++-
native/spark-expr/src/lib.rs | 6 +-
.../org/apache/comet/shims/CometExprShim.scala | 19 +-
.../expressions/datetime/extract_time.sql | 62 +++++
.../sql-tests/expressions/datetime/to_time.sql | 39 ++++
.../sql/benchmark/CometTimeExtractBenchmark.scala | 253 +++++++++++++++++++++
12 files changed, 577 insertions(+), 17 deletions(-)
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index 2c648c7ff1..501ce468c4 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -269,7 +269,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
| `dayofmonth` | ✅ | Native | |
| `dayofweek` | ✅ | Native | |
| `dayofyear` | ✅ | Native | |
-| `extract` | ✅ | — | |
+| `extract` | ✅ | — | `SECOND FROM TIME` is native for Spark 4.1+ with `Decimal(8,6)` output |
| `from_unixtime` | ✅ | Hybrid | |
| `from_utc_timestamp` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default (handles all timezone forms); the native path is opt-in via allowIncompatible ([details](compatibility/expressions/datetime.md)) |
| `hour` | ✅ | Native | |
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 4c174e032d..50f708a283 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -590,9 +590,12 @@ impl PhysicalPlanner {
DataType::Duration(TimeUnit::Microsecond) => {
ScalarValue::DurationMicrosecond(Some(*value))
}
+ DataType::Time64(TimeUnit::Nanosecond) => {
+ ScalarValue::Time64Nanosecond(Some(*value))
+ }
dt => {
return Err(GeneralError(format!(
- "Expected 'Int64', 'Timestamp', or 'Duration(Microsecond)' for LongVal, but found {dt:?}"
+ "Expected 'Int64', 'Timestamp', 'Duration(Microsecond)', or 'Time64(Nanosecond)' for LongVal, but found {dt:?}"
)))
}
},
diff --git a/native/spark-expr/benches/to_time.rs b/native/spark-expr/benches/to_time.rs
index 81815fd980..6e0ab8efe2 100644
--- a/native/spark-expr/benches/to_time.rs
+++ b/native/spark-expr/benches/to_time.rs
@@ -15,9 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use arrow::array::StringArray;
+use arrow::array::{Decimal128Array, StringArray, Time64NanosecondArray};
use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::common::ScalarValue;
use datafusion::physical_plan::ColumnarValue;
+use datafusion_comet_spark_expr::spark_seconds_of_time;
use datafusion_comet_spark_expr::spark_to_time;
use std::sync::Arc;
@@ -58,6 +60,42 @@ fn criterion_benchmark(c: &mut Criterion) {
});
group.finish();
+
+ let mut group = c.benchmark_group("seconds_of_time");
+ for precision in [0_u32, 3, 6] {
+ for nulls in [false, true] {
+ let divisor = 10_i64.pow(9 - precision);
+ let values = Time64NanosecondArray::from_iter((0..10000_i64).map(|i| {
+ if nulls && i % 4 == 0 {
+ None
+ } else {
+ Some(
+ (45_240_000_000_000
+ + (i % 60) * 1_000_000_000
+ + (i * 7919 % 1_000_000) * 1000)
+ / divisor
+ * divisor,
+ )
+ }
+ }));
+ let args = [
+ ColumnarValue::Array(Arc::new(values)),
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(precision as i32))),
+ ];
+ group.bench_function(format!("p={precision},nulls={nulls}"), |b| {
+ b.iter(|| {
+ let ColumnarValue::Array(result) =
+ spark_seconds_of_time(std::hint::black_box(&args)).unwrap()
+ else {
+ unreachable!()
+ };
+ let result = result.as_any().downcast_ref::<Decimal128Array>().unwrap();
+ std::hint::black_box(result.iter().flatten().sum::<i128>());
+ });
+ });
+ }
+ }
+ group.finish();
}
fn create_string_array(size: usize, f: impl Fn(usize) -> String) -> ColumnarValue {
diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs
index c9b50a02a0..8fe19f0aad 100644
--- a/native/spark-expr/src/comet_scalar_funcs.rs
+++ b/native/spark-expr/src/comet_scalar_funcs.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::datetime_funcs::spark_seconds_of_time;
use crate::hash_funcs::*;
use crate::json_funcs::JsonArrayLength;
use crate::map_funcs::spark_map_sort;
@@ -266,6 +267,10 @@ pub fn create_comet_physical_fun_with_eval_mode(
let func = Arc::new(spark_map_sort);
make_comet_scalar_udf!("spark_map_sort", func, without data_type)
}
+ "seconds_of_time" => {
+ let func = Arc::new(spark_seconds_of_time);
+ make_comet_scalar_udf!("seconds_of_time", func, without data_type)
+ }
"to_time" => {
make_comet_scalar_udf!("to_time", spark_to_time, without data_type, fail_on_error)
}
diff --git a/native/spark-expr/src/datetime_funcs/extract_date_part.rs b/native/spark-expr/src/datetime_funcs/extract_date_part.rs
index 7d6e2a50ea..7733dcc3aa 100644
--- a/native/spark-expr/src/datetime_funcs/extract_date_part.rs
+++ b/native/spark-expr/src/datetime_funcs/extract_date_part.rs
@@ -16,10 +16,13 @@
// under the License.
use crate::utils::array_with_timezone;
-use arrow::array::{Array, ArrayRef, Int32Array, TimestampMicrosecondArray};
+use arrow::array::{Array, ArrayRef, Decimal128Array, Int32Array, TimestampMicrosecondArray};
use arrow::compute::{date_part, DatePart};
use arrow::datatypes::{DataType, Int32Type, TimeUnit::Microsecond};
-use datafusion::common::{internal_datafusion_err, DataFusionError};
+use datafusion::common::cast::as_time64_nanosecond_array;
+use datafusion::common::{
+ internal_datafusion_err, utils::take_function_args, DataFusionError, Result, ScalarValue,
+};
use datafusion::logical_expr::{
ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
};
@@ -187,6 +190,38 @@ extract_date_part!(SparkHour, "hour", Hour, hour_of_day);
extract_date_part!(SparkMinute, "minute", Minute, minute_of_hour);
extract_date_part!(SparkSecond, "second", Second, second_of_minute);
+/// Spark 4.1 EXTRACT(SECOND FROM TIME): truncate to the input precision and return
+/// Decimal(8,6). The precision is a literal supplied by Spark's TimeType lowering.
+pub fn spark_seconds_of_time(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+ let [time, precision] = take_function_args("seconds_of_time", args)?;
+ let precision = match precision {
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(p))) if (0..=6).contains(p) => *p as u32,
+ _ => {
+ return Err(internal_datafusion_err!(
+ "seconds_of_time requires a literal precision from 0 to 6"
+ ))
+ }
+ };
+ let divisor = 10_i64.pow(9 - precision);
+ let multiplier = 10_i128.pow(6 - precision);
+ let extract = |nanos: i64| i128::from((nanos % 60_000_000_000) / divisor) * multiplier;
+ match time {
+ 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)?,
+ )))
+ }
+ ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(nanos)) => Ok(ColumnarValue::Scalar(
+ ScalarValue::Decimal128(nanos.map(extract), 8, 6),
+ )),
+ _ => Err(internal_datafusion_err!(
+ "seconds_of_time requires Time64(Nanosecond)"
+ )),
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -195,6 +230,69 @@ mod tests {
use datafusion::config::ConfigOptions;
use std::sync::Arc;
+ #[test]
+ fn seconds_of_time_truncates_to_input_precision() {
+ for (precision, expected) in [
+ 45_000_000, 45_100_000, 45_120_000, 45_123_000, 45_123_400, 45_123_450, 45_123_456,
+ ]
+ .into_iter()
+ .enumerate()
+ {
+ let precision = ColumnarValue::Scalar(ScalarValue::Int32(Some(precision as i32)));
+ for nanos in [Some(45_045_123_456_789), None] {
+ let args = [
+ ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(nanos)),
+ precision.clone(),
+ ];
+ let ColumnarValue::Scalar(result) = spark_seconds_of_time(&args).unwrap() else {
+ panic!("expected scalar")
+ };
+ assert_eq!(
+ result,
+ ScalarValue::Decimal128(nanos.map(|_| expected), 8, 6)
+ );
+ }
+ let times =
+ arrow::array::Time64NanosecondArray::from(vec![Some(45_045_123_456_789), None]);
+ let args = [ColumnarValue::Array(Arc::new(times)), precision];
+ let ColumnarValue::Array(result) = spark_seconds_of_time(&args).unwrap() else {
+ panic!("expected array")
+ };
+ let expected = Decimal128Array::from(vec![Some(expected), None])
+ .with_precision_and_scale(8, 6)
+ .unwrap();
+ assert_eq!(result.as_ref(), &expected);
+ }
+ }
+
+ #[test]
+ fn seconds_of_time_boundaries() {
+ for (nanos, expected) in [
+ (0, 0),
+ (999, 0),
+ (1_000, 1),
+ (59_999_999_999, 59_999_999),
+ (60_000_000_000, 0),
+ (86_399_999_999_999, 59_999_999),
+ ] {
+ let args = [
+ ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(Some(nanos))),
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(6))),
+ ];
+ let ColumnarValue::Scalar(result) = spark_seconds_of_time(&args).unwrap() else {
+ panic!("expected scalar")
+ };
+ assert_eq!(result, ScalarValue::Decimal128(Some(expected), 8, 6));
+ }
+ for precision in [-1, 7, i32::MAX] {
+ assert!(spark_seconds_of_time(&[
+ ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(Some(0))),
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(precision)))
+ ])
+ .is_err());
+ }
+ }
+
// 2024-01-15 18:30:45 UTC, in microseconds since the epoch.
const MICROS: i64 = 1_705_343_445_000_000;
diff --git a/native/spark-expr/src/datetime_funcs/mod.rs b/native/spark-expr/src/datetime_funcs/mod.rs
index 6c8c33b1ef..5330ba1bf1 100644
--- a/native/spark-expr/src/datetime_funcs/mod.rs
+++ b/native/spark-expr/src/datetime_funcs/mod.rs
@@ -36,6 +36,7 @@ pub use date_from_unix_date::SparkDateFromUnixDate;
pub use date_trunc::SparkDateTrunc;
pub use day_month_name::{spark_day_name, spark_month_name};
pub use day_of_week::{SparkDayOfWeek, SparkWeekDay};
+pub use extract_date_part::spark_seconds_of_time;
pub use extract_date_part::SparkHour;
pub use extract_date_part::SparkMinute;
pub use extract_date_part::SparkSecond;
diff --git a/native/spark-expr/src/datetime_funcs/to_time.rs b/native/spark-expr/src/datetime_funcs/to_time.rs
index 249be04a4f..deb8802b95 100644
--- a/native/spark-expr/src/datetime_funcs/to_time.rs
+++ b/native/spark-expr/src/datetime_funcs/to_time.rs
@@ -165,7 +165,7 @@ fn string_to_time(s: &str) -> Option<i64> {
Some(nanos)
}
-/// Parse time components from a string like "HH:mm:ss.ffffff" or "T HH:mm:ss".
+/// Parse time components from a string like "HH:mm:ss.ffffff" or "T12".
/// Returns (hour, minute, second, microseconds) or None if invalid.
fn parse_time_components(s: &str) -> Option<(i32, i32, i32, i32)> {
let bytes = s.as_bytes();
@@ -187,6 +187,11 @@ fn parse_time_components(s: &str) -> Option<(i32, i32, i32, i32)> {
return None;
}
+ // Only a T prefix identifies an hour without a colon as a time in Spark.
+ if pos == bytes.len() && bytes[0] == b'T' {
+ return Some((hour, 0, 0, 0));
+ }
+
// Expect ':'
if pos >= bytes.len() || bytes[pos] != b':' {
return None;
@@ -223,7 +228,7 @@ fn parse_time_components(s: &str) -> Option<(i32, i32, i32, i32)> {
pos += 1;
// Parse fractional seconds (up to 6 digits, pad with zeros)
- let (micros, new_pos) = parse_fractional(bytes, pos)?;
+ let (micros, new_pos) = parse_fractional(bytes, pos);
pos = new_pos;
// Nothing should follow the fractional seconds (timezone not allowed for time)
@@ -260,8 +265,9 @@ fn parse_digits(bytes: &[u8], start: usize) -> Option<(i32, usize)> {
}
/// Parse fractional seconds (microseconds). Up to 6 digits, padded with zeros.
+/// Spark allows an empty fraction after the decimal point.
/// Returns (microseconds, new_pos).
-fn parse_fractional(bytes: &[u8], start: usize) -> Option<(i32, usize)> {
+fn parse_fractional(bytes: &[u8], start: usize) -> (i32, usize) {
let mut pos = start;
let mut value: i32 = 0;
let mut count = 0;
@@ -277,10 +283,6 @@ fn parse_fractional(bytes: &[u8], start: usize) -> Option<(i32, usize)> {
}
}
- if count == 0 {
- return None;
- }
-
// Skip any remaining digits beyond 6 (truncation)
while pos < bytes.len() && bytes[pos].is_ascii_digit() {
pos += 1;
@@ -292,7 +294,7 @@ fn parse_fractional(bytes: &[u8], start: usize) -> Option<(i32, usize)> {
count += 1;
}
- Some((value, pos))
+ (value, pos)
}
#[cfg(test)]
@@ -390,6 +392,48 @@ mod tests {
);
}
+ #[test]
+ fn test_hour_only_t_prefix() {
+ for (input, hour) in [
+ ("T0", 0),
+ ("T1", 1),
+ ("T12", 12),
+ ("T23", 23),
+ ("T12 AM", 0),
+ ("T12 PM", 12),
+ ("T1pm", 13),
+ ("T1\tAM", 1),
+ ] {
+ assert_eq!(
+ string_to_time(input),
+ Some(hour * NANOS_PER_HOUR),
+ "{input:?}"
+ );
+ }
+ for input in [
+ "12", "12 AM", "T", "T001", "T24", "T0 AM", "T13 PM", "T12:", "T12.", " T12",
+ "T12 AM\t",
+ ] {
+ assert_eq!(string_to_time(input), None, "{input:?}");
+ }
+ }
+
+ #[test]
+ fn test_empty_fraction() {
+ for (input, expected) in [
+ ("12:30:45.", "12:30:45"),
+ ("T1:2:3.", "T1:2:3"),
+ ("12:30:45. AM", "12:30:45 AM"),
+ ("12:30:45.PM", "12:30:45 PM"),
+ ("12:30:45.\t", "12:30:45"),
+ ] {
+ assert_eq!(string_to_time(input), string_to_time(expected), "{input:?}");
+ }
+ for input in ["12:30.", "12:30:45..", "12:30:45.x", "12:30:45.Z"] {
+ assert_eq!(string_to_time(input), None, "{input:?}");
+ }
+ }
+
#[test]
fn test_am_pm() {
// 12:00:00 AM = midnight
diff --git a/native/spark-expr/src/lib.rs b/native/spark-expr/src/lib.rs
index e3b77cefab..ee1edf481e 100644
--- a/native/spark-expr/src/lib.rs
+++ b/native/spark-expr/src/lib.rs
@@ -77,9 +77,9 @@ pub use comet_scalar_funcs::{
};
pub use csv_funcs::*;
pub use datetime_funcs::{
- spark_day_name, spark_month_name, spark_to_time, SparkDateDiff, SparkDateFromUnixDate,
- SparkDateTrunc, SparkDayOfWeek, SparkHour, SparkHoursTransform, SparkMakeDate,
- SparkMakeInterval, SparkMakeTime, SparkMinute, SparkNextDay, SparkSecond,
+ spark_day_name, spark_month_name, spark_seconds_of_time, spark_to_time, SparkDateDiff,
+ SparkDateFromUnixDate, SparkDateTrunc, SparkDayOfWeek, SparkHour, SparkHoursTransform,
+ SparkMakeDate, SparkMakeInterval, SparkMakeTime, SparkMinute, SparkNextDay, SparkSecond,
SparkSecondsToTimestamp, SparkUnixTimestamp, SparkWeekDay, TimestampTruncExpr,
};
pub use error::{decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult};
diff --git a/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala b/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala
index a2380e4008..5573d04583 100644
--- a/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala
+++ b/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.TimeType
+import org.apache.spark.sql.types.{DecimalType, IntegerType, TimeType}
import org.apache.comet.expressions.CometEvalMode
import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
@@ -66,6 +66,23 @@ trait CometExprShim extends Spark4xCometExprShim {
scalarFunctionExprToProtoWithReturnType("make_time", s.dataType, true, childExprs: _*)
optExpr
+ // Spark 4.1 and 4.2 EXTRACT(SECOND FROM TIME) return Decimal(8,6), even for TIME(p < 6).
+ case s: StaticInvoke
+ if s.staticObject == classOf[DateTimeUtils.type] &&
+ s.functionName == "getSecondsOfTimeWithFraction" &&
+ s.dataType == DecimalType(8, 6) && s.arguments.size == 2 &&
+ s.arguments.head.dataType.isInstanceOf[TimeType] &&
+ (s.arguments(1) match {
+ case Literal(p: Int, IntegerType) => p >= 0 && p <= 6
+ case _ => false
+ }) =>
+ val childExprs = s.arguments.map(exprToProtoInternal(_, inputs, binding))
+ scalarFunctionExprToProtoWithReturnType(
+ "seconds_of_time",
+ s.dataType,
+ false,
+ childExprs: _*)
+
case i: Invoke =>
(i.targetObject, i.functionName, i.arguments) match {
case (Literal(parser: ToTimeParser, _), "parse", args)
diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/extract_time.sql b/spark/src/test/resources/sql-tests/expressions/datetime/extract_time.sql
new file mode 100644
index 0000000000..6ddffaefa8
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/datetime/extract_time.sql
@@ -0,0 +1,62 @@
+-- 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.
+
+-- MinSparkVersion: 4.1
+-- MaxSparkVersion: 4.2
+-- Config: spark.sql.timeType.enabled=true
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false
+
+-- Spark 4.1 and 4.2 return Decimal(8,6); Spark master uses precision-dependent result types.
+statement
+CREATE TABLE test_extract_time(s STRING) USING parquet
+
+statement
+INSERT INTO test_extract_time VALUES
+ ('00:00:00'),
+ ('00:00:00.000001'),
+ ('00:01:00'),
+ ('12:30:45.123456789'),
+ ('23:59:59.999999'),
+ ('T12'),
+ ('T12 AM'),
+ ('12:30:45.'),
+ (NULL)
+
+-- No JVM dispatcher: both parsing and fractional-second extraction must execute in Rust.
+query
+SELECT extract(second from to_time(s)), extract(second from try_to_time(s))
+FROM test_extract_time
+
+-- The result is Decimal(8,6) for all input precisions, including trailing zeroes.
+-- Literal casts establish TIME(p) inputs without depending on native TIME-to-TIME casts.
+query
+SELECT
+ extract(second from cast(TIME '12:30:45.123456' as TIME(0))),
+ extract(second from cast(TIME '12:30:45.123456' as TIME(1))),
+ extract(second from cast(TIME '12:30:45.123456' as TIME(2))),
+ extract(second from cast(TIME '12:30:45.123456' as TIME(3))),
+ extract(second from cast(TIME '12:30:45.123456' as TIME(4))),
+ extract(second from cast(TIME '12:30:45.123456' as TIME(5))),
+ extract(second from cast(TIME '12:30:45.123456' as TIME(6)))
+FROM test_extract_time
+
+-- Invalid strings become NULL before extraction.
+query
+SELECT extract(second from try_to_time('T24')),
+ extract(second from try_to_time('12:30:45..')),
+ extract(second from cast(NULL as TIME(6)))
+FROM test_extract_time
diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql b/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql
index 30c9d5483f..6cc0e4c724 100644
--- a/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql
+++ b/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql
@@ -33,10 +33,21 @@ INSERT INTO test_to_time VALUES
('00:00:00.001'),
('00:00:00.000001'),
('00:00:00.1234567'),
+ ('12:30:45.123456789'),
+ ('12:30:45.1234567891'),
('23:59:59.999999'),
('1:2:3'),
('1:2:3.04'),
('T12:30:45'),
+ ('T0'),
+ ('T1'),
+ ('T12'),
+ ('T23'),
+ ('T12 AM'),
+ ('T1pm'),
+ ('12:30:45.'),
+ ('12:30:45. AM'),
+ ('12:30:45.PM'),
('T1:02:3.04'),
('12:00:00 AM'),
('1:00:00 AM'),
@@ -55,6 +66,34 @@ INSERT INTO test_to_time VALUES
query spark_answer_only
SELECT s, to_time(s) FROM test_to_time ORDER BY s
+-- Native column evaluation of both error modes, without a TimeType shuffle.
+query
+SELECT to_time(s), try_to_time(s) FROM test_to_time
+
+-- Hour-only T prefixes and empty fractions (issue #5366).
+query
+SELECT to_time('T12'), to_time('T1'), to_time('T12 AM'), to_time('12:30:45.')
+
+query
+SELECT try_to_time('T12'), try_to_time('T1pm'), try_to_time('12:30:45.PM')
+
+-- Missing components and malformed suffixes remain invalid.
+query
+SELECT try_to_time('12'), try_to_time('12 AM'), try_to_time('T'),
+ try_to_time('T001'), try_to_time('T24'), try_to_time('T0 AM'),
+ try_to_time('T13 PM'), try_to_time('T12:'), try_to_time('T12.'),
+ try_to_time('12:30.'), try_to_time('12:30:45..'), try_to_time('12:30:45.x')
+
+query expect_error(cannot be parsed to a TIME value)
+SELECT to_time('T24')
+
+query expect_error(cannot be parsed to a TIME value)
+SELECT to_time('12:30:45..')
+
+-- Spark 4.1 truncates to microseconds during parsing; EXTRACT must see the same precision.
+query
+SELECT extract(second from to_time(s)), extract(second from try_to_time(s)) FROM test_to_time
+
-- literal HH:mm
query
SELECT to_time('00:00')
diff --git a/spark/src/test/spark-4.1/org/apache/spark/sql/benchmark/CometTimeExtractBenchmark.scala b/spark/src/test/spark-4.1/org/apache/spark/sql/benchmark/CometTimeExtractBenchmark.scala
new file mode 100644
index 0000000000..de5e64190d
--- /dev/null
+++ b/spark/src/test/spark-4.1/org/apache/spark/sql/benchmark/CometTimeExtractBenchmark.scala
@@ -0,0 +1,253 @@
+/*
+ * 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
+
+import org.apache.arrow.vector.{TimeNanoVector, VarBinaryVector}
+import org.apache.spark.SparkEnv
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Alias, BoundReference, Literal, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
+import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.classic.{Dataset, SparkSession}
+import org.apache.spark.sql.comet.{CometProjectExec, SerializedPlan}
+import org.apache.spark.sql.execution.{ProjectExec, SparkPlan}
+import org.apache.spark.sql.types.{DecimalType, TimeType}
+
+import org.apache.comet.{CometArrowAllocator, CometConf}
+import org.apache.comet.serde.CometScalaUDF
+import org.apache.comet.udf.codegen.CometScalaUDFCodegen
+import org.apache.comet.vector.CometVector
+
+/**
+ * Spark 4.1 EXTRACT(SECOND FROM TIME), including the previous JVM dispatcher path. Run with `make
+ * benchmark-org.apache.spark.sql.benchmark.CometTimeExtractBenchmark PROFILES=-Pspark-4.1`.
+ * Optional argument: SQL row count (default 1M). Direct kernels use 10000 rows and checksum every
+ * output; the corresponding Rust cases run with `cargo bench -p datafusion-comet-spark-expr
+ * --bench to_time -- seconds_of_time` from native/. The direct dispatcher excludes JNI overhead.
+ * SQL inputs are materialized before timing. SQL timings include scanning, parsing, projection,
+ * row conversion and Spark task overhead, but exclude query planning. Its 0/3/6-digit strings all
+ * parse to TIME(6); the direct kernels exercise TIME(0), TIME(3) and TIME(6).
+ */
+object CometTimeExtractBenchmark extends CometBenchmarkBase {
+ override def runCometBenchmark(args: Array[String]): Unit = {
+ require(spark.version.startsWith("4.1."), "This benchmark targets Spark 4.1 TIME semantics")
+ val rows = args.headOption.map(_.toInt).getOrElse(1024 * 1024)
+ withSQLConf(
+ "spark.sql.timeType.enabled" -> "true",
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.optimizer.excludedRules" ->
+ "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ // Run the small kernels before allocating the million-row SQL inputs.
+ Seq(0, 3, 6).foreach(p => Seq(false, true).foreach(runKernels(p, _)))
+ Seq(0, 3, 6).foreach { precision =>
+ Seq(false, true).foreach { nulls =>
+ withTempTable("timeInput") {
+ val time = "concat('12:34:', lpad(cast(id % 60 as string), 2, '0'), '.', " +
+ "lpad(cast((id * 7919) % 1000000 as string), 6, '0'))"
+ val trimmed = s"substring($time, 1, ${if (precision == 0) 8 else 9 + precision})"
+ val nullable = if (nulls) s"if(id % 4 = 0, null, $trimmed)" else trimmed
+ val input = spark.sql(s"select $nullable as s from range($rows)")
+ val data = input.queryExecution.executedPlan.executeCollect().toSeq
+ Dataset
+ .ofRows(
+ spark.asInstanceOf[SparkSession],
+ LocalRelation(input.queryExecution.analyzed.output, data))
+ .createOrReplaceTempView("timeInput")
+ runCase(s"to_time + extract, digits=$precision, nulls=$nulls", rows)
+ }
+ }
+ }
+ }
+ }
+
+ // Comet's standard scan paths cannot ingest TIME columns yet. Exercise the exact Spark and
+ // dispatcher kernels directly; benches/to_time.rs supplies the corresponding native cases.
+ private def runKernels(precision: Int, nulls: Boolean): Unit = {
+ val size = 10000
+ val time = new TimeNanoVector("t", CometArrowAllocator)
+ val serialized = new VarBinaryVector("expr", CometArrowAllocator)
+ try {
+ time.allocateNew(size)
+ val divisor = math.pow(10, 9 - precision).toLong
+ val inputProjection =
+ UnsafeProjection.create(Seq(BoundReference(0, TimeType(precision), nullable = true)))
+ val rows = (0 until size)
+ .map { i =>
+ if (nulls && i % 4 == 0) {
+ time.setNull(i)
+ InternalRow(null)
+ } else {
+ val nanos = (45240000000000L + (i % 60) * 1000000000L +
+ (i * 7919L % 1000000) * 1000) / divisor * divisor
+ time.setSafe(i, nanos)
+ InternalRow(nanos)
+ }
+ }
+ .map(row => inputProjection(row).copy())
+ time.setValueCount(size)
+ val expr = StaticInvoke(
+ classOf[DateTimeUtils.type],
+ DecimalType(8, 6),
+ "getSecondsOfTimeWithFraction",
+ Seq(BoundReference(0, TimeType(precision), nullable = true), Literal(precision)))
+ val projection = UnsafeProjection.create(Seq(expr))
+ val buffer = SparkEnv.get.closureSerializer.newInstance().serialize(expr)
+ val bytes = new Array[Byte](buffer.remaining())
+ buffer.get(bytes)
+ serialized.allocateNew()
+ serialized.setSafe(0, bytes)
+ serialized.setValueCount(1)
+ val dispatcher = new CometScalaUDFCodegen()
+ val inputs = Array[org.apache.arrow.vector.ValueVector](serialized, time)
+ val result = dispatcher.evaluate(inputs, size)
+ try {
+ val vector =
+ CometVector.getVector(result.asInstanceOf[org.apache.arrow.vector.FieldVector], null)
+ rows.indices.foreach { i =>
+ val expected = projection(rows(i))
+ require(vector.isNullAt(i) == expected.isNullAt(0))
+ if (!expected.isNullAt(0))
+ require(vector.getDecimal(i, 8, 6) == expected.getDecimal(0, 8, 6))
+ }
+ } finally result.close()
+ runBenchmark(s"seconds_of_time p=$precision nulls=$nulls (10000 rows per batch)") {
+ val benchmark =
+ new Benchmark(s"seconds_of_time p=$precision nulls=$nulls", size, output = output)
+ // Consume every result and publish one checksum per batch to prevent dead-code removal.
+ benchmark.addCase("Spark generated projection") { _ =>
+ var sum = 0L
+ var i = 0
+ while (i < size) {
+ val result = projection(rows(i))
+ if (!result.isNullAt(0)) sum += result.getLong(0)
+ i += 1
+ }
+ sink = sum
+ }
+ benchmark.addCase("JVM dispatcher (direct)") { _ =>
+ val result = dispatcher
+ .evaluate(inputs, size)
+ .asInstanceOf[org.apache.arrow.vector.DecimalVector]
+ try {
+ var sum = 0L
+ var i = 0
+ while (i < size) {
+ if (!result.isNull(i)) sum += result.getDataBuffer.getLong(i * 16L)
+ i += 1
+ }
+ sink = sum
+ } finally result.close()
+ }
+ val cases = benchmark.benchmarks.toSeq :+
+ benchmark.benchmarks.head.copy(name = "Spark generated projection (repeat)")
+ cases.foreach { c =>
+ val result = benchmark.measure(size, c.numIters)(c.fn)
+ benchmark.out.println(
+ f"seconds_of_time p=$precision nulls=$nulls, ${c.name}: " +
+ f"mean=${result.avgMs * 1e6 / size}%.2f ns/row, " +
+ f"best=${result.bestMs * 1e6 / size}%.2f ns/row, " +
+ f"stdev=${result.stdevMs * 1e6 / size}%.2f ns/row")
+ }
+ }
+ } finally {
+ serialized.close()
+ time.close()
+ }
+ }
+
+ @volatile private var sink: Long = 0L
+
+ private def plan(query: String, arm: String): SparkPlan = {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> (arm != "Spark").toString,
+ CometConf.COMET_EXEC_ENABLED.key -> (arm != "Spark").toString,
+ CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> (arm == "JVM dispatcher").toString) {
+ val original = spark.sql(query).queryExecution.executedPlan
+ if (arm == "Spark") {
+ require(original.find(_.isInstanceOf[ProjectExec]).isDefined, original.toString)
+ require(original.find(_.isInstanceOf[CometProjectExec]).isEmpty, original.toString)
+ original
+ } else {
+ val projects = original.collect { case p: CometProjectExec => p }
+ require(projects.size == 1, original.toString)
+ require(findFirstNonCometOperator(original).isEmpty, original.toString)
+ require(projects.head.nativeOp.toString.contains("seconds_of_time"))
+ if (arm == "Native") original
+ else
+ original.transformDown { case p: CometProjectExec =>
+ // Reproduce the old catch-all: serialize the entire original expression, including
+ // to_time in the composed case, through the existing JVM codegen dispatcher.
+ val expr = p.projectList.head.asInstanceOf[Alias].child
+ val dispatch = CometScalaUDF.emitJvmCodegenDispatch(expr, p.child.output, true).get
+ require(dispatch.hasJvmScalarUdf)
+ val projection = p.nativeOp.getProjection.toBuilder
+ .clearProjectList()
+ .addProjectList(dispatch)
+ val op = p.nativeOp.toBuilder.setProjection(projection).build()
+ p.copy(nativeOp = op, serializedPlanOpt = SerializedPlan(None)).convertBlock()
+ }
+ }
+ }
+ }
+
+ private def consume(plan: SparkPlan): Unit = {
+ sink = plan
+ .execute()
+ .mapPartitions { rows =>
+ var sum = 0L
+ rows.foreach { row =>
+ if (!row.isNullAt(0)) sum += row.getDecimal(0, 8, 6).toUnscaledLong
+ }
+ Iterator.single(sum)
+ }
+ .collect()
+ .sum
+ }
+
+ private def runCase(name: String, rows: Int): Unit = {
+ runBenchmark(name) {
+ val query = "select extract(second from to_time(s)) as seconds from timeInput"
+ val arms = Seq("Spark", "JVM dispatcher", "Native").map(arm => arm -> plan(query, arm))
+ val expected = arms.head._2.executeCollect().map(_.copy())
+ arms.foreach { case (arm, physical) =>
+ CometScalaUDFCodegen.resetStats()
+ val actual = physical.executeCollect()
+ require(actual.sameElements(expected), s"$name: $arm differs from Spark")
+ val stats = CometScalaUDFCodegen.stats()
+ require(
+ (stats.compileCount + stats.cacheHitCount > 0) == (arm == "JVM dispatcher"),
+ s"$name: $arm unexpected dispatcher activity: $stats")
+ }
+ val benchmark = new Benchmark(name, rows, minNumIters = 5, output = output)
+ benchmark.out.println(
+ s"Spark ${spark.version}; Java ${System.getProperty("java.version")}; " +
+ s"rows=$rows; local[1]; batchSize=${CometConf.COMET_BATCH_SIZE.get()}")
+ arms.foreach { case (arm, physical) =>
+ benchmark.out.println(s"Verified $arm: $physical")
+ benchmark.addCase(arm)(_ => consume(physical))
+ }
+ benchmark.addCase("Spark (repeat)")(_ => consume(arms.head._2))
+ benchmark.run()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org