[SPARK-52779][SQL][CONNECT] Support TimeType literal in Connect

python/pyspark/sql/connect/proto/expressions_pb2.pyi

@@ -654,6 +654,36 @@ class Expression(google.protobuf.message.Message):
                 | None
             ): ...
 
+        class Time(google.protobuf.message.Message):
+            DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+            NANO_FIELD_NUMBER: builtins.int
+            PRECISION_FIELD_NUMBER: builtins.int
+            nano: builtins.int
+            precision: builtins.int
+            """The precision of this time, if omitted, uses the default value of MICROS_PRECISION."""
+            def __init__(
+                self,
+                *,
+                nano: builtins.int = ...,
+                precision: builtins.int | None = ...,
+            ) -> None: ...
+            def HasField(
+                self,
+                field_name: typing_extensions.Literal[
+                    "_precision", b"_precision", "precision", b"precision"
+                ],
+            ) -> builtins.bool: ...
+            def ClearField(
+                self,
+                field_name: typing_extensions.Literal[
+                    "_precision", b"_precision", "nano", b"nano", "precision", b"precision"
+                ],
+            ) -> None: ...
+            def WhichOneof(
+                self, oneof_group: typing_extensions.Literal["_precision", b"_precision"]
+            ) -> typing_extensions.Literal["precision"] | None: ...
+
         NULL_FIELD_NUMBER: builtins.int
         BINARY_FIELD_NUMBER: builtins.int
         BOOLEAN_FIELD_NUMBER: builtins.int
@@ -675,6 +705,7 @@ class Expression(google.protobuf.message.Message):
         MAP_FIELD_NUMBER: builtins.int
         STRUCT_FIELD_NUMBER: builtins.int
         SPECIALIZED_ARRAY_FIELD_NUMBER: builtins.int
+        TIME_FIELD_NUMBER: builtins.int
         @property
         def null(self) -> pyspark.sql.connect.proto.types_pb2.DataType: ...
         binary: builtins.bytes
@@ -706,6 +737,8 @@ class Expression(google.protobuf.message.Message):
         def struct(self) -> global___Expression.Literal.Struct: ...
         @property
         def specialized_array(self) -> global___Expression.Literal.SpecializedArray: ...
+        @property
+        def time(self) -> global___Expression.Literal.Time: ...
         def __init__(
             self,
             *,
@@ -730,6 +763,7 @@ class Expression(google.protobuf.message.Message):
             map: global___Expression.Literal.Map | None = ...,
             struct: global___Expression.Literal.Struct | None = ...,
             specialized_array: global___Expression.Literal.SpecializedArray | None = ...,
+            time: global___Expression.Literal.Time | None = ...,
         ) -> None: ...
         def HasField(
             self,
@@ -772,6 +806,8 @@ class Expression(google.protobuf.message.Message):
                 b"string",
                 "struct",
                 b"struct",
+                "time",
+                b"time",
                 "timestamp",
                 b"timestamp",
                 "timestamp_ntz",
@@ -821,6 +857,8 @@ class Expression(google.protobuf.message.Message):
                 b"string",
                 "struct",
                 b"struct",
+                "time",
+                b"time",
                 "timestamp",
                 b"timestamp",
                 "timestamp_ntz",
@@ -854,6 +892,7 @@ class Expression(google.protobuf.message.Message):
                 "map",
                 "struct",
                 "specialized_array",
+                "time",
             ]
             | None
         ): ...

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala

@@ -3319,6 +3319,7 @@ class PlanGenerationTestSuite
       fn.lit(java.sql.Date.valueOf("2023-02-23")),
       fn.lit(java.time.Duration.ofSeconds(200L)),
       fn.lit(java.time.Period.ofDays(100)),
+      fn.lit(java.time.LocalTime.of(23, 59, 59, 999999999)),
       fn.lit(new CalendarInterval(2, 20, 100L)))
   }
 
@@ -3389,6 +3390,7 @@ class PlanGenerationTestSuite
       fn.typedLit(java.sql.Date.valueOf("2023-02-23")),
       fn.typedLit(java.time.Duration.ofSeconds(200L)),
       fn.typedLit(java.time.Period.ofDays(100)),
+      fn.typedLit(java.time.LocalTime.of(23, 59, 59, 999999999)),
       fn.typedLit(new CalendarInterval(2, 20, 100L)),
 
       // Handle parameterized scala types e.g.: List, Seq and Map.

sql/connect/common/src/main/protobuf/spark/connect/expressions.proto

@@ -195,6 +195,7 @@ message Expression {
       Struct struct = 24;
 
       SpecializedArray specialized_array = 25;
+      Time time = 26;
     }
 
     message Decimal {
@@ -240,6 +241,12 @@ message Expression {
         Strings strings = 6;
       }
     }
+
+    message Time {
+      int64 nano = 1;
+      // The precision of this time, if omitted, uses the default value of MICROS_PRECISION.
+      optional int32 precision = 2;
+    }
   }
 
   // An unresolved attribute that is not explicitly bound to a specific column, but the column

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala

@@ -96,6 +96,11 @@ object LiteralValueProtoConverter {
       case v: Date => builder.setDate(SparkDateTimeUtils.fromJavaDate(v))
       case v: Duration => builder.setDayTimeInterval(SparkIntervalUtils.durationToMicros(v))
       case v: Period => builder.setYearMonthInterval(SparkIntervalUtils.periodToMonths(v))
+      case v: LocalTime =>
+        builder.setTime(
+          builder.getTimeBuilder
+            .setNano(SparkDateTimeUtils.localTimeToNanos(v))
+            .setPrecision(TimeType.DEFAULT_PRECISION))
       case v: Array[_] => builder.setArray(arrayBuilder(v))
       case v: CalendarInterval =>
         builder.setCalendarInterval(calendarIntervalBuilder(v.months, v.days, v.microseconds))
@@ -181,6 +186,11 @@ object LiteralValueProtoConverter {
         } else {
           builder.setNull(toConnectProtoType(dataType))
         }
+      case (v: LocalTime, timeType: TimeType) =>
+        builder.setTime(
+          builder.getTimeBuilder
+            .setNano(SparkDateTimeUtils.localTimeToNanos(v))
+            .setPrecision(timeType.precision))
       case _ => toLiteralProtoBuilder(literal)
     }
   }

sql/connect/common/src/test/resources/query-tests/explain-results/function_lit.explain

@@ -1,2 +1,2 @@
-Project [id#0L, id#0L, true AS true#0, 68 AS 68#0, 9872 AS 9872#0, -8726532 AS -8726532#0, 7834609328726532 AS 7834609328726532#0L, 2.718281828459045 AS 2.718281828459045#0, -0.8 AS -0.8#0, 89.97620 AS 89.97620#0, 89889.7667231 AS 89889.7667231#0, connect! AS connect!#0, T AS T#0, ABCDEFGHIJ AS ABCDEFGHIJ#0, 0x78797A7B7C7D7E7F808182838485868788898A8B8C8D8E AS X'78797A7B7C7D7E7F808182838485868788898A8B8C8D8E'#0, 0x0806 AS X'0806'#0, [8,6] AS ARRAY(8, 6)#0, null AS NULL#0, 2020-10-10 AS DATE '2020-10-10'#0, 8.997620 AS 8.997620#0, 2023-02-23 04:31:59.808 AS TIMESTAMP '2023-02-23 04:31:59.808'#0, 1969-12-31 16:00:12.345 AS TIMESTAMP '1969-12-31 16:00:12.345'#0, 2023-02-23 20:36:00 AS TIMESTAMP_NTZ '2023-02-23 20:36:00'#0, 2023-02-23 AS DATE '2023-02-23'#0, INTERVAL '0 00:03:20' DAY TO SECOND AS INTERVAL '0 00:03:20' DAY TO SECOND#0, ... 2 more fields]
+Project [id#0L, id#0L, true AS true#0, 68 AS 68#0, 9872 AS 9872#0, -8726532 AS -8726532#0, 7834609328726532 AS 7834609328726532#0L, 2.718281828459045 AS 2.718281828459045#0, -0.8 AS -0.8#0, 89.97620 AS 89.97620#0, 89889.7667231 AS 89889.7667231#0, connect! AS connect!#0, T AS T#0, ABCDEFGHIJ AS ABCDEFGHIJ#0, 0x78797A7B7C7D7E7F808182838485868788898A8B8C8D8E AS X'78797A7B7C7D7E7F808182838485868788898A8B8C8D8E'#0, 0x0806 AS X'0806'#0, [8,6] AS ARRAY(8, 6)#0, null AS NULL#0, 2020-10-10 AS DATE '2020-10-10'#0, 8.997620 AS 8.997620#0, 2023-02-23 04:31:59.808 AS TIMESTAMP '2023-02-23 04:31:59.808'#0, 1969-12-31 16:00:12.345 AS TIMESTAMP '1969-12-31 16:00:12.345'#0, 2023-02-23 20:36:00 AS TIMESTAMP_NTZ '2023-02-23 20:36:00'#0, 2023-02-23 AS DATE '2023-02-23'#0, INTERVAL '0 00:03:20' DAY TO SECOND AS INTERVAL '0 00:03:20' DAY TO SECOND#0, ... 3 more fields]
 +- LocalRelation <empty>, [id#0L, a#0, b#0]

sql/connect/common/src/test/resources/query-tests/explain-results/function_typedLit.explain

@@ -1,2 +1,2 @@
-Project [id#0L, id#0L, 1 AS 1#0, null AS NULL#0, true AS true#0, 68 AS 68#0, 9872 AS 9872#0, -8726532 AS -8726532#0, 7834609328726532 AS 7834609328726532#0L, 2.718281828459045 AS 2.718281828459045#0, -0.8 AS -0.8#0, 89.97620 AS 89.97620#0, 89889.7667231 AS 89889.7667231#0, connect! AS connect!#0, T AS T#0, ABCDEFGHIJ AS ABCDEFGHIJ#0, 0x78797A7B7C7D7E7F808182838485868788898A8B8C8D8E AS X'78797A7B7C7D7E7F808182838485868788898A8B8C8D8E'#0, 0x0806 AS X'0806'#0, [8,6] AS ARRAY(8, 6)#0, null AS NULL#0, 2020-10-10 AS DATE '2020-10-10'#0, 8.997620 AS 8.997620#0, 2023-02-23 04:31:59.808 AS TIMESTAMP '2023-02-23 04:31:59.808'#0, 1969-12-31 16:00:12.345 AS TIMESTAMP '1969-12-31 16:00:12.345'#0, 2023-02-23 20:36:00 AS TIMESTAMP_NTZ '2023-02-23 20:36:00'#0, ... 18 more fields]
+Project [id#0L, id#0L, 1 AS 1#0, null AS NULL#0, true AS true#0, 68 AS 68#0, 9872 AS 9872#0, -8726532 AS -8726532#0, 7834609328726532 AS 7834609328726532#0L, 2.718281828459045 AS 2.718281828459045#0, -0.8 AS -0.8#0, 89.97620 AS 89.97620#0, 89889.7667231 AS 89889.7667231#0, connect! AS connect!#0, T AS T#0, ABCDEFGHIJ AS ABCDEFGHIJ#0, 0x78797A7B7C7D7E7F808182838485868788898A8B8C8D8E AS X'78797A7B7C7D7E7F808182838485868788898A8B8C8D8E'#0, 0x0806 AS X'0806'#0, [8,6] AS ARRAY(8, 6)#0, null AS NULL#0, 2020-10-10 AS DATE '2020-10-10'#0, 8.997620 AS 8.997620#0, 2023-02-23 04:31:59.808 AS TIMESTAMP '2023-02-23 04:31:59.808'#0, 1969-12-31 16:00:12.345 AS TIMESTAMP '1969-12-31 16:00:12.345'#0, 2023-02-23 20:36:00 AS TIMESTAMP_NTZ '2023-02-23 20:36:00'#0, ... 19 more fields]
 +- LocalRelation <empty>, [id#0L, a#0, b#0]

sql/connect/common/src/test/resources/query-tests/queries/function_lit.json

@@ -582,6 +582,30 @@
           }
         }
       }
+    }, {
+      "literal": {
+        "time": {
+          "nano": "86399999999999",
+          "precision": 6
+        }
+      },
+      "common": {
+        "origin": {
+          "jvmOrigin": {
+            "stackTrace": [{
+              "classLoaderName": "app",
+              "declaringClass": "org.apache.spark.sql.functions$",
+              "methodName": "lit",
+              "fileName": "functions.scala"
+            }, {
+              "classLoaderName": "app",
+              "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+              "methodName": "~~trimmed~anonfun~~",
+              "fileName": "PlanGenerationTestSuite.scala"
+            }]
+          }
+        }
+      }
     }, {
       "literal": {
         "calendarInterval": {

sql/connect/common/src/test/resources/query-tests/queries/function_typedLit.json

@@ -627,6 +627,30 @@
           }
         }
       }
+    }, {
+      "literal": {
+        "time": {
+          "nano": "86399999999999",
+          "precision": 6
+        }
+      },
+      "common": {
+        "origin": {
+          "jvmOrigin": {
+            "stackTrace": [{
+              "classLoaderName": "app",
+              "declaringClass": "org.apache.spark.sql.functions$",
+              "methodName": "typedLit",
+              "fileName": "functions.scala"
+            }, {
+              "classLoaderName": "app",
+              "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+              "methodName": "~~trimmed~anonfun~~",
+              "fileName": "PlanGenerationTestSuite.scala"
+            }]
+          }
+        }
+      }
     }, {
       "literal": {
         "calendarInterval": {

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverter.scala

@@ -97,6 +97,13 @@ object LiteralExpressionProtoConverter {
       case proto.Expression.Literal.LiteralTypeCase.DAY_TIME_INTERVAL =>
         expressions.Literal(lit.getDayTimeInterval, DayTimeIntervalType())
 
+      case proto.Expression.Literal.LiteralTypeCase.TIME =>
+        var precision = TimeType.DEFAULT_PRECISION
+        if (lit.getTime.hasPrecision) {
+          precision = lit.getTime.getPrecision
+        }
+        expressions.Literal(lit.getTime.getNano, TimeType(precision))
+
       case proto.Expression.Literal.LiteralTypeCase.ARRAY =>
         expressions.Literal.create(
           LiteralValueProtoConverter.toCatalystArray(lit.getArray),

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala

@@ -38,7 +38,7 @@ import org.apache.spark.sql.connect.common.InvalidPlanInput
 import org.apache.spark.sql.connect.common.LiteralValueProtoConverter.toLiteralProto
 import org.apache.spark.sql.execution.arrow.ArrowConverters
 import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType, TimeType}
 import org.apache.spark.unsafe.types.UTF8String
 
 /**
@@ -961,4 +961,49 @@ class SparkConnectPlannerSuite extends SparkFunSuite with SparkConnectPlanTest {
     assert(plan.aggregateExpressions.forall(aggregateExpression =>
       !aggregateExpression.containsPattern(TreePattern.UNRESOLVED_ORDINAL)))
   }
+
+  test("Time literal") {
+    val project = proto.Project.newBuilder
+      .addExpressions(
+        proto.Expression.newBuilder
+          .setLiteral(proto.Expression.Literal.newBuilder.setTime(
+            proto.Expression.Literal.newBuilder.getTimeBuilder
+              .setNano(86399999999999L)
+              .setPrecision(TimeType.MIN_PRECISION)))
+          .build())
+      .addExpressions(
+        proto.Expression.newBuilder
+          .setLiteral(
+            proto.Expression.Literal.newBuilder.setTime(
+              proto.Expression.Literal.newBuilder.getTimeBuilder
+                .setNano(86399999999999L)
+                .setPrecision(TimeType.MAX_PRECISION)))
+          .build())
+      .addExpressions(
+        proto.Expression.newBuilder
+          .setLiteral(
+            proto.Expression.Literal.newBuilder.setTime(
+              proto.Expression.Literal.newBuilder.getTimeBuilder
+                .setNano(86399999999999L)
+                .setPrecision(TimeType.DEFAULT_PRECISION)))
+          .build())
+      .addExpressions(proto.Expression.newBuilder
+        .setLiteral(proto.Expression.Literal.newBuilder.setTime(
+          proto.Expression.Literal.newBuilder.getTimeBuilder.setNano(86399999999999L)))
+        .build())
+      .build()
+
+    val logical = transform(proto.Relation.newBuilder.setProject(project).build())
+    val df = Dataset.ofRows(spark, logical)
+    assertResult(df.schema.fields(0).dataType)(TimeType(TimeType.MIN_PRECISION))
+    assertResult(df.schema.fields(1).dataType)(TimeType(TimeType.MAX_PRECISION))
+    assertResult(df.schema.fields(2).dataType)(TimeType(TimeType.DEFAULT_PRECISION))
+    assertResult(df.schema.fields(3).dataType)(TimeType(TimeType.DEFAULT_PRECISION))
+    assertResult(df.collect()(0).toString)(
+      InternalRow(
+        "23:59:59.999999999",
+        "23:59:59.999999999",
+        "23:59:59.999999999",
+        "23:59:59.999999999").toString)
+  }
 }