address review comments

Author: parthchandra

common/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala

@@ -148,7 +148,7 @@ object Utils extends CometTypeShim with Logging {
         }
       case TimestampNTZType =>
         new ArrowType.Timestamp(TimeUnit.MICROSECOND, null)
-      case dt if dt.getClass.getSimpleName.startsWith("TimeType") =>
+      case dt if isTimeType(dt) =>
         new ArrowType.Time(TimeUnit.NANOSECOND, 64)
       case _ =>
         throw new UnsupportedOperationException(
@@ -401,7 +401,7 @@ object Utils extends CometTypeShim with Logging {
           _: BigIntVector | _: Float4Vector | _: Float8Vector | _: VarCharVector |
           _: DecimalVector | _: DateDayVector | _: TimeStampMicroTZVector | _: VarBinaryVector |
           _: FixedSizeBinaryVector | _: TimeStampMicroVector | _: StructVector | _: ListVector |
-          _: MapVector | _: NullVector) =>
+          _: MapVector | _: NullVector | _: TimeNanoVector) =>
         v.asInstanceOf[FieldVector]
       case _ =>
         throw new SparkException(s"Unsupported Arrow Vector for $reason: ${valueVector.getClass}")

common/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala

@@ -32,4 +32,7 @@ trait CometTypeShim {
 
   @nowarn // Spark 4 feature; Variant shredding doesn't exist in Spark 3.x.
   def isVariantStruct(s: StructType): Boolean = false
+
+  @nowarn // Spark 4.1 feature; TimeType doesn't exist in Spark 3.x.
+  def isTimeType(dt: DataType): Boolean = false
 }

common/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala

@@ -53,4 +53,7 @@ trait CometTypeShim {
   // variant shredding layout, so reading such a struct natively returns nulls. Detect the marker
   // and force scan fallback.
   def isVariantStruct(s: StructType): Boolean = VariantMetadata.isVariantStruct(s)
+
+  def isTimeType(dt: DataType): Boolean =
+    dt.getClass.getSimpleName.startsWith("TimeType")
 }

native/spark-expr/Cargo.toml

@@ -118,3 +118,7 @@ harness = false
 [[bench]]
 name = "map_sort"
 harness = false
+
+[[bench]]
+name = "to_time"
+harness = false

native/spark-expr/benches/to_time.rs

@@ -0,0 +1,73 @@
+// 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.
+
+use arrow::array::StringArray;
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::physical_plan::ColumnarValue;
+use datafusion_comet_spark_expr::spark_to_time;
+use std::sync::Arc;
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("to_time");
+
+    let hh_mm = create_string_array(10000, |i| format!("{}:{:02}", i % 24, i % 60));
+    group.bench_function("hh_mm", |b| {
+        b.iter(|| spark_to_time(std::slice::from_ref(&hh_mm), true).unwrap());
+    });
+
+    let hh_mm_ss = create_string_array(10000, |i| {
+        format!("{}:{:02}:{:02}", i % 24, i % 60, i % 60)
+    });
+    group.bench_function("hh_mm_ss", |b| {
+        b.iter(|| spark_to_time(std::slice::from_ref(&hh_mm_ss), true).unwrap());
+    });
+
+    let fractional = create_string_array(10000, |i| {
+        format!("{}:{:02}:{:02}.{:06}", i % 24, i % 60, i % 60, i * 7 % 1000000)
+    });
+    group.bench_function("fractional", |b| {
+        b.iter(|| spark_to_time(std::slice::from_ref(&fractional), true).unwrap());
+    });
+
+    let am_pm = create_string_array(10000, |i| {
+        let hour = (i % 12) + 1;
+        let suffix = if i % 2 == 0 { "AM" } else { "PM" };
+        format!("{}:{:02}:{:02} {}", hour, i % 60, i % 60, suffix)
+    });
+    group.bench_function("am_pm", |b| {
+        b.iter(|| spark_to_time(std::slice::from_ref(&am_pm), true).unwrap());
+    });
+
+    group.finish();
+}
+
+fn create_string_array(size: usize, f: impl Fn(usize) -> String) -> ColumnarValue {
+    let values: Vec<String> = (0..size).map(&f).collect();
+    let array = StringArray::from(values);
+    ColumnarValue::Array(Arc::new(array))
+}
+
+fn config() -> Criterion {
+    Criterion::default()
+}
+
+criterion_group! {
+    name = benches;
+    config = config();
+    targets = criterion_benchmark
+}
+criterion_main!(benches);

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

@@ -38,5 +38,5 @@ pub use make_date::SparkMakeDate;
 pub use make_time::SparkMakeTime;
 pub use seconds_to_timestamp::SparkSecondsToTimestamp;
 pub use timestamp_trunc::TimestampTruncExpr;
-pub use to_time::{spark_to_time, to_time_return_type};
+pub use to_time::spark_to_time;
 pub use unix_timestamp::SparkUnixTimestamp;

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

@@ -16,7 +16,6 @@
 // under the License.
 
 use arrow::array::{Array, StringArray, Time64NanosecondArray};
-use arrow::datatypes::{DataType, TimeUnit};
 use datafusion::common::{DataFusionError, Result};
 use datafusion::physical_plan::ColumnarValue;
 use std::sync::Arc;
@@ -81,11 +80,18 @@ pub fn spark_to_time(args: &[ColumnarValue], fail_on_error: bool) -> Result<Colu
 /// Parse a time string to nanoseconds from midnight, matching Spark's stringToTime behavior.
 /// Returns None for invalid input.
 fn string_to_time(s: &str) -> Option<i64> {
-    let trimmed = s.trim_end();
+    let trimmed = s.trim();
     if trimmed.is_empty() {
         return None;
     }
 
+    // Spark's parseTimestampString gates the T-prefix branch on j == 0 (start of
+    // the trimmed string), so " T12:30" is rejected even though leading whitespace
+    // is trimmed: the original segment start differs from the trimmed position.
+    if trimmed.as_bytes()[0] == b'T' && s.as_bytes()[0].is_ascii_whitespace() {
+        return None;
+    }
+
     let bytes = trimmed.as_bytes();
     let num_chars = bytes.len();
 
@@ -286,11 +292,6 @@ fn parse_fractional(bytes: &[u8], start: usize) -> Option<(i32, usize)> {
     Some((value, pos))
 }
 
-/// Return type for to_time
-pub fn to_time_return_type() -> DataType {
-    DataType::Time64(TimeUnit::Nanosecond)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -334,6 +335,11 @@ mod tests {
         );
         // 6 digits
         assert_eq!(string_to_time("00:00:00.000001"), Some(NANOS_PER_MICRO));
+        // >6 digits truncated to microseconds
+        assert_eq!(
+            string_to_time("00:00:00.1234567"),
+            Some(123_456 * NANOS_PER_MICRO)
+        );
         // Full precision
         assert_eq!(
             string_to_time("23:59:59.999999"),
@@ -459,8 +465,18 @@ mod tests {
     }
 
     #[test]
-    fn test_leading_space_with_t_prefix() {
-        // Leading space before T should be rejected (Spark only right-trims)
+    fn test_leading_whitespace() {
+        assert_eq!(string_to_time("  12:30"), string_to_time("12:30"));
+        assert_eq!(string_to_time(" 12:30:45"), string_to_time("12:30:45"));
+        assert_eq!(string_to_time(" 12:30:45 "), string_to_time("12:30:45"));
+        assert_eq!(string_to_time("  1:00:00 AM"), string_to_time("1:00:00 AM"));
+        // Tabs and newlines are also trimmed (Spark's isWhitespaceOrISOControl)
+        assert_eq!(string_to_time("\t12:30:45"), string_to_time("12:30:45"));
+        assert_eq!(string_to_time("\n12:30:45"), string_to_time("12:30:45"));
+        // T-prefix is rejected when preceded by whitespace because Spark's
+        // parseTimestampString gates the T-prefix branch on j == 0 (start of
+        // the already-trimmed segment), so leading whitespace moves j past 0.
+        assert_eq!(string_to_time("  T12:30:45"), None);
         assert_eq!(string_to_time(" T12:30"), None);
     }
 }

native/spark-expr/src/lib.rs

@@ -75,7 +75,7 @@ pub use comet_scalar_funcs::{
 };
 pub use csv_funcs::*;
 pub use datetime_funcs::{
-    spark_to_time, to_time_return_type, SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc,
+    spark_to_time, SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc,
     SparkHour, SparkHoursTransform, SparkMakeDate, SparkMakeTime, SparkMinute, SparkSecond,
     SparkSecondsToTimestamp, SparkUnixTimestamp, TimestampTruncExpr,
 };

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -363,7 +363,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType |
         _: DecimalType | _: DateType | _: BooleanType | _: NullType =>
       true
-    case dt if dt.getClass.getSimpleName.startsWith("TimeType") =>
+    case dt if isTimeType(dt) =>
       true
     case s: StructType if allowComplex =>
       s.fields.nonEmpty && s.fields.map(_.dataType).forall(supportedDataType(_, allowComplex))
@@ -399,7 +399,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       case _: ArrayType => 14
       case _: MapType => 15
       case _: StructType => 16
-      case dt if dt.getClass.getSimpleName.startsWith("TimeType") => 17
+      case dt if isTimeType(dt) => 17
       case dt =>
         logWarning(s"Cannot serialize Spark data type: $dt")
         return None

spark/src/test/resources/sql-tests/expressions/datetime/make_time.sql

@@ -33,7 +33,8 @@ INSERT INTO test_make_time VALUES
   (12, 30, NULL),
   (NULL, NULL, NULL)
 
--- column arguments (spark_answer_only: shuffle does not support TimeType yet)
+-- column arguments (spark_answer_only: shuffle does not support TimeType yet; TODO: promote to
+-- full native-verification once SPARK-51779 lands)
 query spark_answer_only
 SELECT hours, minutes, secs, make_time(hours, minutes, secs) FROM test_make_time ORDER BY hours, minutes, secs