[WIP][SQL] Incapsulate type operations

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]

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

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

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) =>

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]
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhyTypeOps.scala

@@ -0,0 +1,42 @@
+/*
+ * 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.MutableValue
+import org.apache.spark.sql.types.{DataType, TimeType}
+
+// Base operations over Catalyst's types.
+trait PhyTypeOps extends TypeOps {
+  // Gets the underlying physical type
+  def getPhysicalType: PhysicalDataType
+
+  // Gets the Java class of the physical type
+  def getJavaClass: Class[_]
+
+  // Gets a mutable container for the physical type
+  def getMutableValue: MutableValue
+}
+
+object PhyTypeOps {
+  def supports(dt: DataType): Boolean = dt match {
+    case _: TimeType => true
+    case _ => false
+  }
+
+  def apply(dt: DataType): PhyTypeOps = TypeOps(dt).asInstanceOf[PhyTypeOps]
+}

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,7 @@ sealed abstract class PhysicalDataType {
 
 object PhysicalDataType {
   def apply(dt: DataType): PhysicalDataType = dt match {
+    case _ if PhyTypeOps.supports(dt) => PhyTypeOps(dt).getPhysicalType
     case NullType => PhysicalNullType
     case ByteType => PhysicalByteType
     case ShortType => PhysicalShortType
@@ -54,7 +55,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) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/TimeTypeOps.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.catalyst.expressions.{Literal, MutableLong, MutableValue}
+import org.apache.spark.sql.types.TimeType
+
+case class TimeTypeOps (t: TimeType)
+  extends TypeOps
+  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)
+}

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}")
+  }
+}