From eef51432f310addbfd956e848c9cbbbf9534d99f Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sun, 13 Jul 2025 19:30:44 +0200
Subject: [PATCH 1/6] Initial implementation

---
 .../spark/sql/catalyst/types/PhyTypeOps.scala | 33 +++++++++++++++++++
 .../sql/catalyst/types/PhysicalDataType.scala |  5 +--
 .../sql/catalyst/types/TimeTypeOps.scala      | 26 +++++++++++++++
 .../spark/sql/catalyst/types/TypeOps.scala    | 33 +++++++++++++++++++
 4 files changed, 95 insertions(+), 2 deletions(-)
 create mode 100644 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
 create mode 100644 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
 create mode 100644 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TypeOps.scala

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
new file mode 100644
index 0000000000000..c37ec5c48be7f
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.catalyst.types
+
+import org.apache.spark.sql.types.{DataType, TimeType}
+
+// Base operations of Catalyst's types.
+abstract class PhyTypeOps extends TypeOps {
+  // Get the underlying physical type
+  def getPhysicalType: PhysicalDataType
+}
+
+object PhyTypeOps {
+  def supports(dt: DataType): Boolean = dt match {
+    case _: TimeType => true
+    case _ => false
+  }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
index 1084e99731510..2e3121ec15fdb 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
@@ -23,7 +23,7 @@ import scala.reflect.runtime.universe.typeTag
 import org.apache.spark.sql.catalyst.expressions.{Ascending, BoundReference, InterpretedOrdering, SortOrder}
 import org.apache.spark.sql.catalyst.util.{ArrayData, CollationFactory, MapData, SQLOrderingUtil}
 import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteExactNumeric, ByteType, CalendarIntervalType, CharType, DataType, DateType, DayTimeIntervalType, Decimal, DecimalExactNumeric, DecimalType, DoubleExactNumeric, DoubleType, FloatExactNumeric, FloatType, FractionalType, IntegerExactNumeric, IntegerType, IntegralType, LongExactNumeric, LongType, MapType, NullType, NumericType, ShortExactNumeric, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType, TimeType, VarcharType, VariantType, YearMonthIntervalType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteExactNumeric, ByteType, CalendarIntervalType, CharType, DataType, DateType, DayTimeIntervalType, Decimal, DecimalExactNumeric, DecimalType, DoubleExactNumeric, DoubleType, FloatExactNumeric, FloatType, FractionalType, IntegerExactNumeric, IntegerType, IntegralType, LongExactNumeric, LongType, MapType, NullType, NumericType, ShortExactNumeric, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType, VarcharType, VariantType, YearMonthIntervalType}
 import org.apache.spark.unsafe.types.{ByteArray, UTF8String, VariantVal}
 import org.apache.spark.util.ArrayImplicits._
 
@@ -35,6 +35,8 @@ sealed abstract class PhysicalDataType {
 
 object PhysicalDataType {
   def apply(dt: DataType): PhysicalDataType = dt match {
+    case _ if PhyTypeOps.supports(dt) =>
+      TypeOps(dt).asInstanceOf[PhyTypeOps].getPhysicalType
     case NullType => PhysicalNullType
     case ByteType => PhysicalByteType
     case ShortType => PhysicalShortType
@@ -54,7 +56,6 @@ object PhysicalDataType {
     case DayTimeIntervalType(_, _) => PhysicalLongType
     case YearMonthIntervalType(_, _) => PhysicalIntegerType
     case DateType => PhysicalIntegerType
-    case _: TimeType => PhysicalLongType
     case ArrayType(elementType, containsNull) => PhysicalArrayType(elementType, containsNull)
     case StructType(fields) => PhysicalStructType(fields)
     case MapType(keyType, valueType, valueContainsNull) =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
new file mode 100644
index 0000000000000..3e21cecf38c3b
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
@@ -0,0 +1,26 @@
+/*
+ * 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.catalyst.types
+
+import org.apache.spark.sql.types.TimeType
+
+case class TimeTypeOps (t: TimeType)
+  extends PhyTypeOps {
+
+  override def getPhysicalType: PhysicalDataType = PhysicalLongType
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TypeOps.scala
new file mode 100644
index 0000000000000..d5043cfcc02df
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TypeOps.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.catalyst.types
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.types.{DataType, TimeType}
+
+// Operations over a data type
+trait TypeOps
+
+// The factory of type operation objects.
+object TypeOps {
+  def apply(dt: DataType): TypeOps = dt match {
+    case tt: TimeType => TimeTypeOps(tt)
+    case other => throw SparkException.internalError(
+      s"Cannot create an operation object of the type ${other.sql}")
+  }
+}

From 37e1751dee1aee288ab29287296ffaf38cc3181f Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sun, 13 Jul 2025 19:49:45 +0200
Subject: [PATCH 2/6] Get Java classes

---
 .../apache/spark/sql/catalyst/encoders/EncoderUtils.scala  | 6 +++---
 .../org/apache/spark/sql/catalyst/types/PhyTypeOps.scala   | 7 ++++++-
 .../apache/spark/sql/catalyst/types/PhysicalDataType.scala | 3 +--
 .../org/apache/spark/sql/catalyst/types/TimeTypeOps.scala  | 4 +++-
 4 files changed, 13 insertions(+), 7 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/EncoderUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/EncoderUtils.scala
index 0de459e1196d6..013e559c7a54d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/EncoderUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/EncoderUtils.scala
@@ -22,9 +22,9 @@ import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{BinaryEncoder, CalendarIntervalEncoder, NullEncoder, PrimitiveBooleanEncoder, PrimitiveByteEncoder, PrimitiveDoubleEncoder, PrimitiveFloatEncoder, PrimitiveIntEncoder, PrimitiveLongEncoder, PrimitiveShortEncoder, SparkDecimalEncoder, VariantEncoder}
 import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.types.{PhysicalBinaryType, PhysicalIntegerType, PhysicalLongType}
+import org.apache.spark.sql.catalyst.types.{PhysicalBinaryType, PhysicalIntegerType, PhysicalLongType, PhyTypeOps}
 import org.apache.spark.sql.catalyst.util.{ArrayData, MapData}
-import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, CalendarIntervalType, DataType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, ObjectType, ShortType, StringType, StructType, TimestampNTZType, TimestampType, TimeType, UserDefinedType, VariantType, YearMonthIntervalType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, CalendarIntervalType, DataType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, ObjectType, ShortType, StringType, StructType, TimestampNTZType, TimestampType, UserDefinedType, VariantType, YearMonthIntervalType}
 import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String, VariantVal}
 
 /**
@@ -99,10 +99,10 @@ object EncoderUtils {
 
   def dataTypeJavaClass(dt: DataType): Class[_] = {
     dt match {
+      case _ if PhyTypeOps.supports(dt) => PhyTypeOps(dt).getJavaClass
       case _: DecimalType => classOf[Decimal]
       case _: DayTimeIntervalType => classOf[PhysicalLongType.InternalType]
       case _: YearMonthIntervalType => classOf[PhysicalIntegerType.InternalType]
-      case _: TimeType => classOf[PhysicalLongType.InternalType]
       case _: StringType => classOf[UTF8String]
       case _: StructType => classOf[InternalRow]
       case _: ArrayType => classOf[ArrayData]
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
index c37ec5c48be7f..eae279053ab4d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
@@ -20,9 +20,12 @@ package org.apache.spark.sql.catalyst.types
 import org.apache.spark.sql.types.{DataType, TimeType}
 
 // Base operations of Catalyst's types.
-abstract class PhyTypeOps extends TypeOps {
+trait PhyTypeOps extends TypeOps {
   // Get the underlying physical type
   def getPhysicalType: PhysicalDataType
+
+  // Get the Java class of the physical type
+  def getJavaClass: Class[_]
 }
 
 object PhyTypeOps {
@@ -30,4 +33,6 @@ object PhyTypeOps {
     case _: TimeType => true
     case _ => false
   }
+
+  def apply(dt: DataType): PhyTypeOps = TypeOps(dt).asInstanceOf[PhyTypeOps]
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
index 2e3121ec15fdb..aebe9d0374d8e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
@@ -35,8 +35,7 @@ sealed abstract class PhysicalDataType {
 
 object PhysicalDataType {
   def apply(dt: DataType): PhysicalDataType = dt match {
-    case _ if PhyTypeOps.supports(dt) =>
-      TypeOps(dt).asInstanceOf[PhyTypeOps].getPhysicalType
+    case _ if PhyTypeOps.supports(dt) => PhyTypeOps(dt).getPhysicalType
     case NullType => PhysicalNullType
     case ByteType => PhysicalByteType
     case ShortType => PhysicalShortType
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
index 3e21cecf38c3b..d057e716c5c33 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
@@ -20,7 +20,9 @@ package org.apache.spark.sql.catalyst.types
 import org.apache.spark.sql.types.TimeType
 
 case class TimeTypeOps (t: TimeType)
-  extends PhyTypeOps {
+  extends TypeOps
+  with PhyTypeOps {
 
   override def getPhysicalType: PhysicalDataType = PhysicalLongType
+  override def getJavaClass: Class[_] = classOf[PhysicalLongType.InternalType]
 }

From 579e32e09e428021b14113fa85f982262f52eb3e Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sun, 13 Jul 2025 19:59:42 +0200
Subject: [PATCH 3/6] Modify CodeGenerator

---
 .../spark/sql/catalyst/expressions/codegen/CodeGenerator.scala | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
index 6b2c696fb9933..055496c904827 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
@@ -1985,11 +1985,12 @@ object CodeGenerator extends Logging {
 
   @tailrec
   def javaClass(dt: DataType): Class[_] = dt match {
+    case _ if PhyTypeOps.supports(dt) => PhyTypeOps(dt).getJavaClass
     case BooleanType => java.lang.Boolean.TYPE
     case ByteType => java.lang.Byte.TYPE
     case ShortType => java.lang.Short.TYPE
     case IntegerType | DateType | _: YearMonthIntervalType => java.lang.Integer.TYPE
-    case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType | _: TimeType =>
+    case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
       java.lang.Long.TYPE
     case FloatType => java.lang.Float.TYPE
     case DoubleType => java.lang.Double.TYPE

From 4215f9209dae1ef2ced58857ec4dab19ecf8581f Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Mon, 14 Jul 2025 15:45:41 +0200
Subject: [PATCH 4/6] Handle mutable values

---
 .../sql/catalyst/expressions/SpecificInternalRow.scala    | 4 +++-
 .../org/apache/spark/sql/catalyst/types/PhyTypeOps.scala  | 8 ++++++--
 .../org/apache/spark/sql/catalyst/types/TimeTypeOps.scala | 2 ++
 3 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificInternalRow.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificInternalRow.scala
index 1f755df0516ff..13c6f4bb4f7dc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificInternalRow.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificInternalRow.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions
 
 import scala.annotation.tailrec
 
+import org.apache.spark.sql.catalyst.types.PhyTypeOps
 import org.apache.spark.sql.types._
 
 /**
@@ -196,10 +197,11 @@ final class SpecificInternalRow(val values: Array[MutableValue]) extends BaseGen
 
   @tailrec
   private[this] def dataTypeToMutableValue(dataType: DataType): MutableValue = dataType match {
+    case _ if PhyTypeOps.supports(dataType) => PhyTypeOps(dataType).getMutableValue
     // We use INT for DATE and YearMonthIntervalType internally
     case IntegerType | DateType | _: YearMonthIntervalType => new MutableInt
     // We use Long for Timestamp, Timestamp without time zone and DayTimeInterval internally
-    case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType | _: TimeType =>
+    case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
       new MutableLong
     case FloatType => new MutableFloat
     case DoubleType => new MutableDouble
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
index eae279053ab4d..a9d06ea705798 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
@@ -17,15 +17,19 @@
 
 package org.apache.spark.sql.catalyst.types
 
+import org.apache.spark.sql.catalyst.expressions.MutableValue
 import org.apache.spark.sql.types.{DataType, TimeType}
 
 // Base operations of Catalyst's types.
 trait PhyTypeOps extends TypeOps {
-  // Get the underlying physical type
+  // Gets the underlying physical type
   def getPhysicalType: PhysicalDataType
 
-  // Get the Java class of the physical type
+  // Gets the Java class of the physical type
   def getJavaClass: Class[_]
+
+  // Gets a mutable container for the physical type
+  def getMutableValue: MutableValue
 }
 
 object PhyTypeOps {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
index d057e716c5c33..04526d1ecd0ab 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.types
 
+import org.apache.spark.sql.catalyst.expressions.{MutableLong, MutableValue}
 import org.apache.spark.sql.types.TimeType
 
 case class TimeTypeOps (t: TimeType)
@@ -25,4 +26,5 @@ case class TimeTypeOps (t: TimeType)
 
   override def getPhysicalType: PhysicalDataType = PhysicalLongType
   override def getJavaClass: Class[_] = classOf[PhysicalLongType.InternalType]
+  override def getMutableValue: MutableValue = new MutableLong
 }

From b672cc2b3bbb02a1a0e71cfe165734fd5079751e Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Mon, 14 Jul 2025 16:03:44 +0200
Subject: [PATCH 5/6] Add LiteralTypeOps

---
 .../sql/catalyst/expressions/literals.scala   |  2 +-
 .../sql/catalyst/types/LiteralTypeOps.scala   | 36 +++++++++++++++++++
 .../spark/sql/catalyst/types/PhyTypeOps.scala |  2 +-
 .../sql/catalyst/types/TimeTypeOps.scala      |  7 ++--
 4 files changed, 43 insertions(+), 4 deletions(-)
 create mode 100644 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
index 925735654b73e..a83dd5c1d132e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
@@ -188,6 +188,7 @@ object Literal {
    * Create a literal with default value for given DataType
    */
   def default(dataType: DataType): Literal = dataType match {
+    case _ if LiteralTypeOps.supports(dataType) => LiteralTypeOps(dataType).getDefaultLiteral
     case NullType => create(null, NullType)
     case BooleanType => Literal(false)
     case ByteType => Literal(0.toByte)
@@ -200,7 +201,6 @@ object Literal {
     case DateType => create(0, DateType)
     case TimestampType => create(0L, TimestampType)
     case TimestampNTZType => create(0L, TimestampNTZType)
-    case t: TimeType => create(0L, t)
     case it: DayTimeIntervalType => create(0L, it)
     case it: YearMonthIntervalType => create(0, it)
     case CharType(length) =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala
new file mode 100644
index 0000000000000..ee5c2602024ac
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala
@@ -0,0 +1,36 @@
+/*
+ * 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.catalyst.types
+
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.types.{DataType, TimeType}
+
+// Literal operations over Catalyst's types
+trait LiteralTypeOps {
+  // Gets a literal with default value of the type
+  def getDefaultLiteral: Literal
+}
+
+object LiteralTypeOps {
+  def supports(dt: DataType): Boolean = dt match {
+    case _: TimeType => true
+    case _ => false
+  }
+
+  def apply(dt: DataType): LiteralTypeOps = TypeOps(dt).asInstanceOf[LiteralTypeOps]
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
index a9d06ea705798..7b7fdf8581c1f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.types
 import org.apache.spark.sql.catalyst.expressions.MutableValue
 import org.apache.spark.sql.types.{DataType, TimeType}
 
-// Base operations of Catalyst's types.
+// Base operations over Catalyst's types.
 trait PhyTypeOps extends TypeOps {
   // Gets the underlying physical type
   def getPhysicalType: PhysicalDataType
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
index 04526d1ecd0ab..2eeec759bbc16 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.scala
@@ -17,14 +17,17 @@
 
 package org.apache.spark.sql.catalyst.types
 
-import org.apache.spark.sql.catalyst.expressions.{MutableLong, MutableValue}
+import org.apache.spark.sql.catalyst.expressions.{Literal, MutableLong, MutableValue}
 import org.apache.spark.sql.types.TimeType
 
 case class TimeTypeOps (t: TimeType)
   extends TypeOps
-  with PhyTypeOps {
+  with PhyTypeOps
+  with LiteralTypeOps {
 
   override def getPhysicalType: PhysicalDataType = PhysicalLongType
   override def getJavaClass: Class[_] = classOf[PhysicalLongType.InternalType]
   override def getMutableValue: MutableValue = new MutableLong
+
+  override def getDefaultLiteral: Literal = Literal.create(0L, t)
 }

From 0f3bc557bb8982f4959b1d7a213f8408623a12e6 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Thu, 17 Jul 2025 11:47:27 +0200
Subject: [PATCH 6/6] Improve type checks

---
 .../apache/spark/sql/catalyst/types/LiteralTypeOps.scala  | 8 ++++----
 .../org/apache/spark/sql/catalyst/types/PhyTypeOps.scala  | 8 ++++----
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala
index ee5c2602024ac..d94c8a7d076a2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/LiteralTypeOps.scala
@@ -27,10 +27,10 @@ trait LiteralTypeOps {
 }
 
 object LiteralTypeOps {
-  def supports(dt: DataType): Boolean = dt match {
-    case _: TimeType => true
-    case _ => false
-  }
+  private val supportedDataTypes: Set[DataType] =
+    Set(TimeType.MIN_PRECISION to TimeType.MAX_PRECISION map TimeType.apply: _*)
+
+  def supports(dt: DataType): Boolean = supportedDataTypes.contains(dt)
 
   def apply(dt: DataType): LiteralTypeOps = TypeOps(dt).asInstanceOf[LiteralTypeOps]
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
index 7b7fdf8581c1f..5749fc81b9cd4 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala
@@ -33,10 +33,10 @@ trait PhyTypeOps extends TypeOps {
 }
 
 object PhyTypeOps {
-  def supports(dt: DataType): Boolean = dt match {
-    case _: TimeType => true
-    case _ => false
-  }
+  private val supportedDataTypes: Set[DataType] =
+    Set(TimeType.MIN_PRECISION to TimeType.MAX_PRECISION map TimeType.apply: _*)
+
+  def supports(dt: DataType): Boolean = supportedDataTypes.contains(dt)
 
   def apply(dt: DataType): PhyTypeOps = TypeOps(dt).asInstanceOf[PhyTypeOps]
 }