fix: align time parsing and native second extraction with Spark

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 (`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 |  |

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:?}"
                                 )))
                             }
                         },

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 {

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;
@@ -265,6 +266,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)
         }

native/spark-expr/src/datetime_funcs/extract_date_part.rs

@@ -16,13 +16,18 @@
 // under the License.
 
 use crate::utils::array_with_timezone;
+use arrow::array::Decimal128Array;
 use arrow::compute::{date_part, DatePart};
 use arrow::datatypes::{DataType, 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,
 };
 use std::fmt::Debug;
+use std::sync::Arc;
 
 /// Returns true when the type is a timestamp without a timezone (Spark's TimestampNTZType),
 /// including when wrapped in a dictionary. Such values store local wall-clock time and must not
@@ -117,6 +122,38 @@ extract_date_part!(SparkHour, "hour", Hour);
 extract_date_part!(SparkMinute, "minute", Minute);
 extract_date_part!(SparkSecond, "second", Second);
 
+/// 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::*;
@@ -125,6 +162,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;
 

native/spark-expr/src/datetime_funcs/mod.rs

@@ -34,6 +34,7 @@ pub use date_diff::SparkDateDiff;
 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 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;

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

native/spark-expr/src/lib.rs

@@ -77,10 +77,10 @@ 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, SparkHour, SparkHoursTransform, SparkMakeDate, SparkMakeInterval,
-    SparkMakeTime, SparkMinute, SparkNextDay, SparkSecond, SparkSecondsToTimestamp,
-    SparkUnixTimestamp, TimestampTruncExpr,
+    spark_day_name, spark_month_name, spark_seconds_of_time, spark_to_time, SparkDateDiff,
+    SparkDateFromUnixDate, SparkDateTrunc, SparkHour, SparkHoursTransform, SparkMakeDate,
+    SparkMakeInterval, SparkMakeTime, SparkMinute, SparkNextDay, SparkSecond,
+    SparkSecondsToTimestamp, SparkUnixTimestamp, TimestampTruncExpr,
 };
 pub use error::{decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult};
 pub use hash_funcs::*;

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 EXTRACT(SECOND FROM TIME) returns 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)