diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SpecializedGettersReader.java b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SpecializedGettersReader.java
index ea0648a6cb90..1fed1da47203 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SpecializedGettersReader.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SpecializedGettersReader.java
@@ -65,6 +65,9 @@ public static Object read(
     if (dataType instanceof TimestampType) {
       return obj.getLong(ordinal);
     }
+    if (dataType instanceof TimeType) {
+      return obj.getLong(ordinal);
+    }
     if (dataType instanceof CalendarIntervalType) {
       return obj.getInterval(ordinal);
     }
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java
index 034894bd8608..a775cad41d3d 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java
@@ -90,7 +90,8 @@ public static int calculateBitSetWidthInBytes(int numFields) {
           FloatType,
           DoubleType,
           DateType,
-          TimestampType
+          TimestampType,
+          TimeType
         })));
   }
 
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/types/DataTypes.java b/sql/catalyst/src/main/java/org/apache/spark/sql/types/DataTypes.java
index d786374f69e2..04911b6a2bb8 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/types/DataTypes.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/types/DataTypes.java
@@ -54,6 +54,11 @@ public class DataTypes {
    */
   public static final DataType TimestampType = TimestampType$.MODULE$;
 
+    /**
+     * Gets the TimeType object.
+     */
+    public static final DataType TimeType = TimeType$.MODULE$;
+
   /**
    * Gets the CalendarIntervalType object.
    */
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala
index 5b17f1d65f1b..9bda36619793 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala
@@ -302,6 +302,13 @@ trait Row extends Serializable {
    */
   def getTimestamp(i: Int): java.sql.Timestamp = getAs[java.sql.Timestamp](i)
 
+  /**
+   * Returns the value at position i of date type as java.sql.Time.
+   *
+   * @throws ClassCastException when data type does not match.
+   */
+  def getTime(i: Int): java.sql.Time = getAs[java.sql.Time](i)
+
   /**
    * Returns the value at position i of date type as java.time.Instant.
    *
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala
index 34d2f45e715e..4f889281f335 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst
 import java.lang.{Iterable => JavaIterable}
 import java.math.{BigDecimal => JavaBigDecimal}
 import java.math.{BigInteger => JavaBigInteger}
-import java.sql.{Date, Timestamp}
+import java.sql.{Date, Time, Timestamp}
 import java.time.{Instant, LocalDate}
 import java.util.{Map => JavaMap}
 import javax.annotation.Nullable
@@ -331,6 +331,16 @@ object CatalystTypeConverters {
       DateTimeUtils.toJavaTimestamp(row.getLong(column))
   }
 
+  private object TimeConverter extends CatalystTypeConverter[Time, Time, Any] {
+    override def toCatalystImpl(scalaValue: Time): Long =
+      DateTimeUtils.fromJavaTime(scalaValue)
+    override def toScala(catalystValue: Any): Time =
+      if (catalystValue == null) null
+      else DateTimeUtils.toJavaTime(catalystValue.asInstanceOf[Long])
+    override def toScalaImpl(row: InternalRow, column: Int): Time =
+      DateTimeUtils.toJavaTime(row.getLong(column))
+  }
+
   private object InstantConverter extends CatalystTypeConverter[Instant, Instant, Any] {
     override def toCatalystImpl(scalaValue: Instant): Long =
       DateTimeUtils.instantToMicros(scalaValue)
@@ -451,6 +461,7 @@ object CatalystTypeConverters {
     case d: Date => DateConverter.toCatalyst(d)
     case ld: LocalDate => LocalDateConverter.toCatalyst(ld)
     case t: Timestamp => TimestampConverter.toCatalyst(t)
+    case ti: Time => TimeConverter.toCatalyst(ti);
     case i: Instant => InstantConverter.toCatalyst(i)
     case d: BigDecimal => new DecimalConverter(DecimalType(d.precision, d.scale)).toCatalyst(d)
     case d: JavaBigDecimal => new DecimalConverter(DecimalType(d.precision, d.scale)).toCatalyst(d)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala
index 701e4e3483c0..afc493f9daf1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala
@@ -118,6 +118,15 @@ object DeserializerBuildHelper {
       returnNullable = false)
   }
 
+  def createDeserializerForSqlTime(path: Expression): Expression = {
+    StaticInvoke(
+      DateTimeUtils.getClass,
+      ObjectType(classOf[java.sql.Time]),
+      "toJavaTime",
+      path :: Nil,
+      returnNullable = false)
+  }
+
   def createDeserializerForJavaBigDecimal(
       path: Expression,
       returnNullable: Boolean): Expression = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/InternalRow.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/InternalRow.scala
index f98b59edd422..f4b34e64b921 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/InternalRow.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/InternalRow.scala
@@ -133,7 +133,7 @@ object InternalRow {
       case ByteType => (input, ordinal) => input.getByte(ordinal)
       case ShortType => (input, ordinal) => input.getShort(ordinal)
       case IntegerType | DateType => (input, ordinal) => input.getInt(ordinal)
-      case LongType | TimestampType => (input, ordinal) => input.getLong(ordinal)
+      case LongType | TimestampType | TimeType => (input, ordinal) => input.getLong(ordinal)
       case FloatType => (input, ordinal) => input.getFloat(ordinal)
       case DoubleType => (input, ordinal) => input.getDouble(ordinal)
       case StringType => (input, ordinal) => input.getUTF8String(ordinal)
@@ -168,7 +168,8 @@ object InternalRow {
     case ByteType => (input, v) => input.setByte(ordinal, v.asInstanceOf[Byte])
     case ShortType => (input, v) => input.setShort(ordinal, v.asInstanceOf[Short])
     case IntegerType | DateType => (input, v) => input.setInt(ordinal, v.asInstanceOf[Int])
-    case LongType | TimestampType => (input, v) => input.setLong(ordinal, v.asInstanceOf[Long])
+    case LongType | TimestampType | TimeType => (input, v) =>
+      input.setLong(ordinal, v.asInstanceOf[Long])
     case FloatType => (input, v) => input.setFloat(ordinal, v.asInstanceOf[Float])
     case DoubleType => (input, v) => input.setDouble(ordinal, v.asInstanceOf[Double])
     case CalendarIntervalType =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
index 2248e2eb0259..bb1c8cf92072 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
@@ -402,6 +402,8 @@ object JavaTypeInference {
 
         case c if c == classOf[java.sql.Timestamp] => createSerializerForSqlTimestamp(inputObject)
 
+        case c if c == classOf[java.sql.Time] => createSerializerForSqlTime(inputObject)
+
         case c if c == classOf[java.time.LocalDate] => createSerializerForJavaLocalDate(inputObject)
 
         case c if c == classOf[java.sql.Date] => createSerializerForSqlDate(inputObject)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala
index 85acaa11230b..80ed3244cdec 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala
@@ -83,6 +83,15 @@ object SerializerBuildHelper {
       returnNullable = false)
   }
 
+  def createSerializerForSqlTime(inputObject: Expression): Expression = {
+    StaticInvoke(
+      DateTimeUtils.getClass,
+      TimeType,
+      "fromJavaTime",
+      inputObject :: Nil,
+      returnNullable = false)
+  }
+
   def createSerializerForJavaLocalDate(inputObject: Expression): Expression = {
     StaticInvoke(
       DateTimeUtils.getClass,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
index 765018f07d87..df6db0000bcf 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
@@ -101,6 +101,13 @@ object RowEncoder {
         createSerializerForSqlTimestamp(inputObject)
       }
 
+    case TimeType =>
+      if (SQLConf.get.datetimeJava8ApiEnabled) {
+        createSerializerForJavaInstant(inputObject)
+      } else {
+        createSerializerForSqlTime(inputObject)
+      }
+
     case DateType =>
       if (SQLConf.get.datetimeJava8ApiEnabled) {
         createSerializerForJavaLocalDate(inputObject)
@@ -220,6 +227,12 @@ object RowEncoder {
       } else {
         ObjectType(classOf[java.sql.Timestamp])
       }
+    case TimeType =>
+      if (SQLConf.get.datetimeJava8ApiEnabled) {
+        ObjectType(classOf[java.time.Instant])
+      } else {
+        ObjectType(classOf[java.sql.Time])
+      }
     case DateType =>
       if (SQLConf.get.datetimeJava8ApiEnabled) {
         ObjectType(classOf[java.time.LocalDate])
@@ -274,6 +287,13 @@ object RowEncoder {
         createDeserializerForSqlTimestamp(input)
       }
 
+    case TimeType =>
+      if (SQLConf.get.datetimeJava8ApiEnabled) {
+        createDeserializerForInstant(input)
+      } else {
+        createDeserializerForSqlTime(input)
+      }
+
     case DateType =>
       if (SQLConf.get.datetimeJava8ApiEnabled) {
         createDeserializerForLocalDate(input)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
index 5576e71b5702..32e2decd51d5 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
@@ -54,6 +54,7 @@ object Cast {
     case (StringType, BooleanType) => true
     case (DateType, BooleanType) => true
     case (TimestampType, BooleanType) => true
+    case (TimeType, BooleanType) => true
     case (_: NumericType, BooleanType) => true
 
     case (StringType, TimestampType) => true
@@ -62,8 +63,14 @@ object Cast {
     case (_: NumericType, TimestampType) =>
       SQLConf.get.getConf(SQLConf.LEGACY_ALLOW_CAST_NUMERIC_TO_TIMESTAMP)
 
+    case (StringType, TimeType) => true
+    case (BooleanType, TimeType) => true
+    case (DateType, TimeType) => true
+    case (_: NumericType, TimeType) => true
+
     case (StringType, DateType) => true
     case (TimestampType, DateType) => true
+    case (TimeType, DateType) => true
 
     case (StringType, CalendarIntervalType) => true
 
@@ -71,6 +78,7 @@ object Cast {
     case (BooleanType, _: NumericType) => true
     case (DateType, _: NumericType) => true
     case (TimestampType, _: NumericType) => true
+    case (TimeType, _: NumericType) => true
     case (_: NumericType, _: NumericType) => true
 
     case (ArrayType(fromType, fn), ArrayType(toType, tn)) =>
@@ -108,10 +116,10 @@ object Cast {
    * * Cast.castToTimestamp
    */
   def needsTimeZone(from: DataType, to: DataType): Boolean = (from, to) match {
-    case (StringType, TimestampType | DateType) => true
-    case (DateType, TimestampType) => true
-    case (TimestampType, StringType) => true
-    case (TimestampType, DateType) => true
+    case (StringType, TimestampType | TimeType | DateType) => true
+    case (DateType, TimestampType | TimeType) => true
+    case (TimestampType | TimeType, StringType) => true
+    case (TimestampType | TimeType, DateType) => true
     case (ArrayType(fromType, _), ArrayType(toType, _)) => needsTimeZone(fromType, toType)
     case (MapType(fromKey, fromValue, _), MapType(toKey, toValue, _)) =>
       needsTimeZone(fromKey, toKey) || needsTimeZone(fromValue, toValue)
@@ -135,6 +143,7 @@ object Cast {
     case (from: DecimalType, to: NumericType) if from.isTighterThan(to) => true
     case (f, t) if legalNumericPrecedence(f, t) => true
     case (DateType, TimestampType) => true
+    case (DateType, TimeType) => true
     case (_: AtomicType, StringType) => true
     case (_: CalendarIntervalType, StringType) => true
     case (NullType, _) => true
@@ -144,6 +153,9 @@ object Cast {
     case (TimestampType, LongType) => true
     case (LongType, TimestampType) => true
 
+    case (TimeType, LongType) => true
+    case (LongType, TimeType) => true
+
     case (ArrayType(fromType, fn), ArrayType(toType, tn)) =>
       resolvableNullability(fn, tn) && canUpCast(fromType, toType)
 
@@ -173,6 +185,8 @@ object Cast {
     case (_: CalendarIntervalType, StringType) => true
     case (DateType, TimestampType) => true
     case (TimestampType, DateType) => true
+    case (DateType, TimeType) => true
+    case (TimeType, DateType) => true
 
     case (ArrayType(fromType, fn), ArrayType(toType, tn)) =>
       resolvableNullability(fn, tn) && canANSIStoreAssign(fromType, toType)
@@ -220,10 +234,10 @@ object Cast {
     case (StringType, _) => true
     case (_, StringType) => false
 
-    case (FloatType | DoubleType, TimestampType) => true
-    case (TimestampType, DateType) => false
+    case (FloatType | DoubleType, TimestampType | TimeType) => true
+    case (TimestampType | TimeType, DateType) => false
     case (_, DateType) => true
-    case (DateType, TimestampType) => false
+    case (DateType, TimestampType | TimeType) => false
     case (DateType, _) => true
     case (_, CalendarIntervalType) => true
 
@@ -302,7 +316,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[CalendarInterval](_, i => UTF8String.fromString(i.toString))
     case BinaryType => buildCast[Array[Byte]](_, UTF8String.fromBytes)
     case DateType => buildCast[Int](_, d => UTF8String.fromString(dateFormatter.format(d)))
-    case TimestampType => buildCast[Long](_,
+    case TimestampType | TimeType => buildCast[Long](_,
       t => UTF8String.fromString(DateTimeUtils.timestampToString(timestampFormatter, t)))
     case ArrayType(et, _) =>
       buildCast[ArrayData](_, array => {
@@ -407,7 +421,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
           null
         }
       })
-    case TimestampType =>
+    case TimestampType | TimeType =>
       buildCast[Long](_, t => t != 0)
     case DateType =>
       // Hive would return null when cast from date to boolean
@@ -455,6 +469,33 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Float](_, f => doubleToTimestamp(f.toDouble))
   }
 
+  // TimeConverter
+  private[this] def castToTime(from: DataType): Any => Any = from match {
+    case StringType =>
+      buildCast[UTF8String](_, utfs => DateTimeUtils.stringToTimestamp(utfs, zoneId).orNull)
+    case BooleanType =>
+      buildCast[Boolean](_, b => if (b) 1L else 0)
+    case LongType =>
+      buildCast[Long](_, l => longToTimestamp(l))
+    case IntegerType =>
+      buildCast[Int](_, i => longToTimestamp(i.toLong))
+    case ShortType =>
+      buildCast[Short](_, s => longToTimestamp(s.toLong))
+    case ByteType =>
+      buildCast[Byte](_, b => longToTimestamp(b.toLong))
+    case DateType =>
+      buildCast[Int](_, d => epochDaysToMicros(d, zoneId))
+    // TimestampWritable.decimalToTimestamp
+    case DecimalType() =>
+      buildCast[Decimal](_, d => decimalToTimestamp(d))
+    // TimestampWritable.doubleToTimestamp
+    case DoubleType =>
+      buildCast[Double](_, d => doubleToTimestamp(d))
+    // TimestampWritable.floatToTimestamp
+    case FloatType =>
+      buildCast[Float](_, f => doubleToTimestamp(f.toDouble))
+  }
+
   private[this] def decimalToTimestamp(d: Decimal): Long = {
     (d.toBigDecimal * MICROS_PER_SECOND).longValue
   }
@@ -477,7 +518,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   private[this] def castToDate(from: DataType): Any => Any = from match {
     case StringType =>
       buildCast[UTF8String](_, s => DateTimeUtils.stringToDate(s, zoneId).orNull)
-    case TimestampType =>
+    case TimestampType | TimeType =>
       // throw valid precision more than seconds, according to Hive.
       // Timestamp.nanos is in 0 to 999,999,999, no more than a second.
       buildCast[Long](_, t => microsToEpochDays(t, zoneId))
@@ -500,7 +541,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => if (b) 1L else 0L)
     case DateType =>
       buildCast[Int](_, d => null)
-    case TimestampType =>
+    case TimestampType | TimeType =>
       buildCast[Long](_, t => timestampToLong(t))
     case x: NumericType if ansiEnabled =>
       b => x.exactNumeric.asInstanceOf[Numeric[Any]].toLong(b)
@@ -519,7 +560,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => if (b) 1 else 0)
     case DateType =>
       buildCast[Int](_, d => null)
-    case TimestampType if ansiEnabled =>
+    case TimestampType | TimeType if ansiEnabled =>
       buildCast[Long](_, t => {
         val longValue = timestampToLong(t)
         if (longValue == longValue.toInt) {
@@ -528,7 +569,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
           throw new ArithmeticException(s"Casting $t to int causes overflow")
         }
       })
-    case TimestampType =>
+    case TimestampType TimeType =>
       buildCast[Long](_, t => timestampToLong(t).toInt)
     case x: NumericType if ansiEnabled =>
       b => x.exactNumeric.asInstanceOf[Numeric[Any]].toInt(b)
@@ -551,7 +592,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => if (b) 1.toShort else 0.toShort)
     case DateType =>
       buildCast[Int](_, d => null)
-    case TimestampType if ansiEnabled =>
+    case TimestampType | TimeType if ansiEnabled =>
       buildCast[Long](_, t => {
         val longValue = timestampToLong(t)
         if (longValue == longValue.toShort) {
@@ -560,7 +601,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
           throw new ArithmeticException(s"Casting $t to short causes overflow")
         }
       })
-    case TimestampType =>
+    case TimestampType | TimeType =>
       buildCast[Long](_, t => timestampToLong(t).toShort)
     case x: NumericType if ansiEnabled =>
       b =>
@@ -594,7 +635,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => if (b) 1.toByte else 0.toByte)
     case DateType =>
       buildCast[Int](_, d => null)
-    case TimestampType if ansiEnabled =>
+    case TimestampType | TimeType if ansiEnabled =>
       buildCast[Long](_, t => {
         val longValue = timestampToLong(t)
         if (longValue == longValue.toByte) {
@@ -603,7 +644,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
           throw new ArithmeticException(s"Casting $t to byte causes overflow")
         }
       })
-    case TimestampType =>
+    case TimestampType | TimeType =>
       buildCast[Long](_, t => timestampToLong(t).toByte)
     case x: NumericType if ansiEnabled =>
       b =>
@@ -671,7 +712,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => toPrecision(if (b) Decimal.ONE else Decimal.ZERO, target))
     case DateType =>
       buildCast[Int](_, d => null) // date can't cast to decimal in Hive
-    case TimestampType =>
+    case TimestampType | TimeType =>
       // Note that we lose precision here.
       buildCast[Long](_, t => changePrecision(Decimal(timestampToDouble(t)), target))
     case dt: DecimalType =>
@@ -705,7 +746,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => if (b) 1d else 0d)
     case DateType =>
       buildCast[Int](_, d => null)
-    case TimestampType =>
+    case TimestampType | TimeType =>
       buildCast[Long](_, t => timestampToDouble(t))
     case x: NumericType =>
       b => x.numeric.asInstanceOf[Numeric[Any]].toDouble(b)
@@ -730,7 +771,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       buildCast[Boolean](_, b => if (b) 1f else 0f)
     case DateType =>
       buildCast[Int](_, d => null)
-    case TimestampType =>
+    case TimestampType | TimeType =>
       buildCast[Long](_, t => timestampToDouble(t).toFloat)
     case x: NumericType =>
       b => x.numeric.asInstanceOf[Numeric[Any]].toFloat(b)
@@ -798,6 +839,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
         case DateType => castToDate(from)
         case decimal: DecimalType => castToDecimal(from, decimal)
         case TimestampType => castToTimestamp(from)
+        case TimeType => castToTime(from)
         case CalendarIntervalType => castToInterval(from)
         case BooleanType => castToBoolean(from)
         case ByteType => castToByte(from)
@@ -857,6 +899,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
     case DateType => castToDateCode(from, ctx)
     case decimal: DecimalType => castToDecimalCode(from, decimal, ctx)
     case TimestampType => castToTimestampCode(from, ctx)
+    case TimeType => castToTimestampCode(from, ctx)
     case CalendarIntervalType => castToIntervalCode(from)
     case BooleanType => castToBooleanCode(from)
     case ByteType => castToByteCode(from, ctx)
@@ -1030,7 +1073,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
           ctx.addReferenceObj("dateFormatter", dateFormatter),
           dateFormatter.getClass)
         (c, evPrim, evNull) => code"""$evPrim = UTF8String.fromString(${df}.format($c));"""
-      case TimestampType =>
+      case TimestampType | TimeType =>
         val tf = JavaCode.global(
           ctx.addReferenceObj("timestampFormatter", timestampFormatter),
           timestampFormatter.getClass)
@@ -1116,7 +1159,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
             $evNull = true;
           }
          """
-      case TimestampType =>
+      case TimestampType | TimeType =>
         val zid = getZoneId()
         (c, evPrim, evNull) =>
           code"""$evPrim =
@@ -1183,7 +1226,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       case DateType =>
         // date can't cast to decimal in Hive
         (c, evPrim, evNull) => code"$evNull = true;"
-      case TimestampType =>
+      case TimestampType | TimeType =>
         // Note that we lose precision here.
         (c, evPrim, evNull) =>
           code"""
@@ -1305,7 +1348,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
             $evNull = true;
           }
         """
-    case TimestampType =>
+    case TimestampType | TimeType =>
       (c, evPrim, evNull) => code"$evPrim = $c != 0;"
     case DateType =>
       // Hive would return null when cast from date to boolean
@@ -1411,7 +1454,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       (c, evPrim, evNull) => code"$evPrim = $c ? (byte) 1 : (byte) 0;"
     case DateType =>
       (c, evPrim, evNull) => code"$evNull = true;"
-    case TimestampType => castTimestampToIntegralTypeCode(ctx, "byte")
+    case TimestampType | TimeType => castTimestampToIntegralTypeCode(ctx, "byte")
     case DecimalType() => castDecimalToIntegralTypeCode(ctx, "byte")
     case _: ShortType | _: IntegerType | _: LongType if ansiEnabled =>
       castIntegralTypeToIntegralTypeExactCode("byte")
@@ -1444,7 +1487,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       (c, evPrim, evNull) => code"$evPrim = $c ? (short) 1 : (short) 0;"
     case DateType =>
       (c, evPrim, evNull) => code"$evNull = true;"
-    case TimestampType => castTimestampToIntegralTypeCode(ctx, "short")
+    case TimestampType | TimeType => castTimestampToIntegralTypeCode(ctx, "short")
     case DecimalType() => castDecimalToIntegralTypeCode(ctx, "short")
     case _: IntegerType | _: LongType if ansiEnabled =>
       castIntegralTypeToIntegralTypeExactCode("short")
@@ -1475,7 +1518,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       (c, evPrim, evNull) => code"$evPrim = $c ? 1 : 0;"
     case DateType =>
       (c, evPrim, evNull) => code"$evNull = true;"
-    case TimestampType => castTimestampToIntegralTypeCode(ctx, "int")
+    case TimestampType | TimeType => castTimestampToIntegralTypeCode(ctx, "int")
     case DecimalType() => castDecimalToIntegralTypeCode(ctx, "int")
     case _: LongType if ansiEnabled => castIntegralTypeToIntegralTypeExactCode("int")
     case _: FloatType if ansiEnabled =>
@@ -1505,7 +1548,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       (c, evPrim, evNull) => code"$evPrim = $c ? 1L : 0L;"
     case DateType =>
       (c, evPrim, evNull) => code"$evNull = true;"
-    case TimestampType =>
+    case TimestampType | TimeType =>
       (c, evPrim, evNull) => code"$evPrim = (long) ${timestampToLongCode(c)};"
     case DecimalType() => castDecimalToIntegralTypeCode(ctx, "long")
     case _: FloatType if ansiEnabled =>
@@ -1543,7 +1586,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
         (c, evPrim, evNull) => code"$evPrim = $c ? 1.0f : 0.0f;"
       case DateType =>
         (c, evPrim, evNull) => code"$evNull = true;"
-      case TimestampType =>
+      case TimestampType | TimeType =>
         (c, evPrim, evNull) => code"$evPrim = (float) (${timestampToDoubleCode(c)});"
       case DecimalType() =>
         (c, evPrim, evNull) => code"$evPrim = $c.toFloat();"
@@ -1579,7 +1622,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
         (c, evPrim, evNull) => code"$evPrim = $c ? 1.0d : 0.0d;"
       case DateType =>
         (c, evPrim, evNull) => code"$evNull = true;"
-      case TimestampType =>
+      case TimestampType | TimeType =>
         (c, evPrim, evNull) => code"$evPrim = ${timestampToDoubleCode(c)};"
       case DecimalType() =>
         (c, evPrim, evNull) => code"$evPrim = $c.toDouble();"
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala
index 39a16e917c4a..3e030af32592 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala
@@ -146,7 +146,7 @@ object InterpretedUnsafeProjection {
       case IntegerType | DateType =>
         (v, i) => writer.write(i, v.getInt(i))
 
-      case LongType | TimestampType =>
+      case LongType | TimestampType | TimeType =>
         (v, i) => writer.write(i, v.getLong(i))
 
       case FloatType =>
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 817dd948f1a6..6b8705f6ec74 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
@@ -41,7 +41,6 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData, MapData}
 import org.apache.spark.sql.catalyst.util.DateTimeConstants.NANOS_PER_MILLIS
-import org.apache.spark.sql.catalyst.util.DateTimeUtils._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.Platform
@@ -1788,7 +1787,7 @@ object CodeGenerator extends Logging {
     case ByteType => JAVA_BYTE
     case ShortType => JAVA_SHORT
     case IntegerType | DateType => JAVA_INT
-    case LongType | TimestampType => JAVA_LONG
+    case LongType | TimestampType | TimeType => JAVA_LONG
     case FloatType => JAVA_FLOAT
     case DoubleType => JAVA_DOUBLE
     case _: DecimalType => "Decimal"
@@ -1809,7 +1808,7 @@ object CodeGenerator extends Logging {
     case ByteType => java.lang.Byte.TYPE
     case ShortType => java.lang.Short.TYPE
     case IntegerType | DateType => java.lang.Integer.TYPE
-    case LongType | TimestampType => java.lang.Long.TYPE
+    case LongType | TimestampType | TimeType => java.lang.Long.TYPE
     case FloatType => java.lang.Float.TYPE
     case DoubleType => java.lang.Double.TYPE
     case _: DecimalType => classOf[Decimal]
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 213a58a3244e..3a46761e4677 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
@@ -26,7 +26,7 @@ import java.lang.{Long => JavaLong}
 import java.lang.{Short => JavaShort}
 import java.math.{BigDecimal => JavaBigDecimal}
 import java.nio.charset.StandardCharsets
-import java.sql.{Date, Timestamp}
+import java.sql.{Date, Time, Timestamp}
 import java.time.{Instant, LocalDate}
 import java.util
 import java.util.Objects
@@ -72,6 +72,7 @@ object Literal {
     case d: Decimal => Literal(d, DecimalType(Math.max(d.precision, d.scale), d.scale))
     case i: Instant => Literal(instantToMicros(i), TimestampType)
     case t: Timestamp => Literal(DateTimeUtils.fromJavaTimestamp(t), TimestampType)
+    case t: Time => Literal(DateTimeUtils.fromJavaTime(t), TimeType)
     case ld: LocalDate => Literal(ld.toEpochDay.toInt, DateType)
     case d: Date => Literal(DateTimeUtils.fromJavaDate(d), DateType)
     case a: Array[Byte] => Literal(a, BinaryType)
@@ -163,6 +164,7 @@ object Literal {
     case dt: DecimalType => Literal(Decimal(0, dt.precision, dt.scale))
     case DateType => create(0, DateType)
     case TimestampType => create(0L, TimestampType)
+    case TimeType => create(0L, TimeType)
     case StringType => Literal("")
     case BinaryType => Literal("".getBytes(StandardCharsets.UTF_8))
     case CalendarIntervalType => Literal(new CalendarInterval(0, 0, 0))
@@ -182,7 +184,7 @@ object Literal {
       case ByteType => v.isInstanceOf[Byte]
       case ShortType => v.isInstanceOf[Short]
       case IntegerType | DateType => v.isInstanceOf[Int]
-      case LongType | TimestampType => v.isInstanceOf[Long]
+      case LongType | TimestampType | TimeType => v.isInstanceOf[Long]
       case FloatType => v.isInstanceOf[Float]
       case DoubleType => v.isInstanceOf[Double]
       case _: DecimalType => v.isInstanceOf[Decimal]
@@ -369,7 +371,7 @@ case class Literal (value: Any, dataType: DataType) extends LeafExpression {
           }
         case ByteType | ShortType =>
           ExprCode.forNonNullValue(JavaCode.expression(s"($javaType)$value", dataType))
-        case TimestampType | LongType =>
+        case TimestampType | TimeType | LongType =>
           toExprCode(s"${value}L")
         case _ =>
           val constRef = ctx.addReferenceObj("literal", value, javaType)
@@ -411,6 +413,10 @@ case class Literal (value: Any, dataType: DataType) extends LeafExpression {
       val formatter = TimestampFormatter.getFractionFormatter(
         DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
       s"TIMESTAMP '${formatter.format(v)}'"
+    case (v: Long, TimeType) =>
+      val formatter = TimestampFormatter.getFractionFormatter(
+        DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
+      s"TIME '${formatter.format(v)}'"
     case (i: CalendarInterval, CalendarIntervalType) =>
       s"INTERVAL '${i.toString}'"
     case (v: Array[Byte], BinaryType) => s"X'${DatatypeConverter.printHexBinary(v)}'"
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index 03571a740df3..5cf2ba1ab9b7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -2202,6 +2202,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
       case ("double", Nil) => DoubleType
       case ("date", Nil) => DateType
       case ("timestamp", Nil) => TimestampType
+      case ("time", Nil) => TimestampType
       case ("string", Nil) => StringType
       case ("character" | "char", length :: Nil) => CharType(length.getText.toInt)
       case ("varchar", length :: Nil) => VarcharType(length.getText.toInt)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
index 41a271b95e83..11b5fc0d4fbd 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.util
 
 import java.nio.charset.StandardCharsets
-import java.sql.{Date, Timestamp}
+import java.sql.{Date, Time, Timestamp}
 import java.time._
 import java.time.temporal.{ChronoField, ChronoUnit, IsoFields}
 import java.util.{Locale, TimeZone}
@@ -159,6 +159,13 @@ object DateTimeUtils {
     ts
   }
 
+  /**
+   * Returns a java.sql.Time from number of micros since epoch.
+   */
+  def toJavaTime(us: SQLTimestamp): Time = {
+    new Time(us)
+  }
+
   /**
    * Converts an instance of `java.sql.Timestamp` to the number of microseconds since
    * 1970-01-01T00:00:00.000000Z. It extracts date-time fields from the input, builds
@@ -183,6 +190,13 @@ object DateTimeUtils {
     rebaseJulianToGregorianMicros(micros)
   }
 
+  /**
+   * Returns the number of micros since epoch from java.sql.Time.
+   */
+  def fromJavaTime(t: Time): SQLTimestamp = {
+    TimeToMicros(t)
+  }
+
   /**
    * Returns the number of microseconds since epoch from Julian day
    * and nanoseconds in a day
@@ -414,6 +428,11 @@ object DateTimeUtils {
     result
   }
 
+  def TimeToMicros(t: Time): Long = {
+    val result = Math.multiplyExact(t.getTime, MICROS_PER_MILLIS)
+    result
+  }
+
   def microsToInstant(us: Long): Instant = {
     val secs = Math.floorDiv(us, MICROS_PER_SECOND)
     // Unfolded Math.floorMod(us, MICROS_PER_SECOND) to reuse the result of
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala
index fe8d7efc9dc1..808d5c0750ae 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala
@@ -135,7 +135,7 @@ object DataType {
   def fromJson(json: String): DataType = parseDataType(parse(json))
 
   private val nonDecimalNameToType = {
-    Seq(NullType, DateType, TimestampType, BinaryType, IntegerType, BooleanType, LongType,
+    Seq(NullType, DateType, TimestampType, TimeType, BinaryType, IntegerType, BooleanType, LongType,
       DoubleType, FloatType, ShortType, ByteType, StringType, CalendarIntervalType)
       .map(t => t.typeName -> t).toMap
   }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/TimeType.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/TimeType.scala
new file mode 100644
index 000000000000..ff1d10aa2ac8
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/TimeType.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.types
+
+import scala.math.Ordering
+import scala.reflect.runtime.universe.typeTag
+
+import org.apache.spark.annotation.Stable
+
+/**
+ * The timestamp type represents a time instant in microsecond precision.
+ * Valid range is [0001-01-01T00:00:00.000000Z, 9999-12-31T23:59:59.999999Z] where
+ * the left/right-bound is a date and time of the proleptic Gregorian
+ * calendar in UTC+00:00.
+ *
+ * Please use the singleton `DataTypes.TimestampType` to refer the type.
+ * @since 3.1.0
+ */
+@Stable
+class TimeType private() extends AtomicType {
+  /**
+   * Internally, a timestamp is stored as the number of microseconds from
+   * the epoch of 1970-01-01T00:00:00.000000Z (UTC+00:00)
+   */
+  private[sql] type InternalType = Long
+
+  @transient private[sql] lazy val tag = typeTag[InternalType]
+
+  private[sql] val ordering = implicitly[Ordering[InternalType]]
+
+  /**
+   * The default size of a value of the TimeType is 8 bytes.
+   */
+  override def defaultSize: Int = 8
+
+  private[spark] override def asNullable: TimeType = this
+}
+
+/**
+ * The companion case object and its class is separated so the companion object also subclasses
+ * the TimestampType class. Otherwise, the companion object would be of type "TimestampType$"
+ * in byte code. Defined with a private constructor so the companion object is the only possible
+ * instantiation.
+ *
+ * @since 3.1.0
+ */
+@Stable
+case object TimeType extends TimeType
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
index fd24f058f357..de319633a092 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
@@ -96,20 +96,22 @@ class RowEncoderSuite extends CodegenInterpretedPlanTest {
 
   encodeDecodeTest(
     new StructType()
-      .add("null", NullType)
-      .add("boolean", BooleanType)
-      .add("byte", ByteType)
-      .add("short", ShortType)
-      .add("int", IntegerType)
-      .add("long", LongType)
-      .add("float", FloatType)
-      .add("double", DoubleType)
-      .add("decimal", DecimalType.SYSTEM_DEFAULT)
-      .add("string", StringType)
-      .add("binary", BinaryType)
-      .add("date", DateType)
-      .add("timestamp", TimestampType)
-      .add("udt", new ExamplePointUDT))
+//      .add("null", NullType)
+//      .add("boolean", BooleanType)
+//      .add("byte", ByteType)
+//      .add("short", ShortType)
+//      .add("int", IntegerType)
+//      .add("long", LongType)
+//      .add("float", FloatType)
+//      .add("double", DoubleType)
+//      .add("decimal", DecimalType.SYSTEM_DEFAULT)
+//      .add("string", StringType)
+//      .add("binary", BinaryType)
+//      .add("date", DateType)
+//      .add("timestamp", TimestampType)
+      .add("time", TimeType)
+//      .add("udt", new ExamplePointUDT)
+            )
 
   encodeDecodeTest(
     new StructType()
@@ -308,6 +310,13 @@ class RowEncoderSuite extends CodegenInterpretedPlanTest {
       toRow(encoder, Row(Array("a")))
     }
     assert(e4.getMessage.contains("java.lang.String is not a valid external type"))
+
+    val e5 = intercept[RuntimeException] {
+      val schema = new StructType().add("a", ArrayType(TimeType))
+      val encoder = RowEncoder(schema)
+      encoder.toRow(Row(Array("a")))
+    }
+    assert(e5.getMessage.contains("java.lang.String is not a valid external type"))
   }
 
   test("SPARK-25791: Datatype of serializers should be accessible") {
@@ -330,6 +339,18 @@ class RowEncoderSuite extends CodegenInterpretedPlanTest {
     }
   }
 
+  test("encoding/decoding TimeType to/from java.time.Instant") {
+    withSQLConf(SQLConf.DATETIME_JAVA8API_ENABLED.key -> "true") {
+      val schema = new StructType().add("t", TimeType)
+      val encoder = RowEncoder(schema).resolveAndBind()
+      val instant = java.time.Instant.parse("2019-02-26T16:56:00Z")
+      val row = encoder.toRow(Row(instant))
+      assert(row.getLong(0) === DateTimeUtils.instantToMicros(instant))
+      val readback = encoder.fromRow(row)
+      assert(readback.get(0) === instant)
+    }
+  }
+
   test("encoding/decoding DateType to/from java.time.LocalDate") {
     withSQLConf(SQLConf.DATETIME_JAVA8API_ENABLED.key -> "true") {
       val schema = new StructType().add("d", DateType)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuite.scala
index 35b401798013..91308f33464d 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuite.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
-import java.sql.{Date, Timestamp}
+import java.sql.{Date, Time, Timestamp}
 import java.util.{Calendar, TimeZone}
 
 import scala.collection.parallel.immutable.ParVector
@@ -70,6 +70,7 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     checkNullCast(StringType, BooleanType)
     checkNullCast(DateType, BooleanType)
     checkNullCast(TimestampType, BooleanType)
+    checkNullCast(TimeType, BooleanType)
     numericTypes.foreach(dt => checkNullCast(dt, BooleanType))
 
     checkNullCast(StringType, TimestampType)
@@ -77,14 +78,21 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     checkNullCast(DateType, TimestampType)
     numericTypes.foreach(dt => checkNullCast(dt, TimestampType))
 
+    checkNullCast(StringType, TimeType)
+    checkNullCast(BooleanType, TimeType)
+    checkNullCast(DateType, TimeType)
+    numericTypes.foreach(dt => checkNullCast(dt, TimeType))
+
     checkNullCast(StringType, DateType)
     checkNullCast(TimestampType, DateType)
+    checkNullCast(TimeType, DateType)
 
     checkNullCast(StringType, CalendarIntervalType)
     numericTypes.foreach(dt => checkNullCast(StringType, dt))
     numericTypes.foreach(dt => checkNullCast(BooleanType, dt))
     numericTypes.foreach(dt => checkNullCast(DateType, dt))
     numericTypes.foreach(dt => checkNullCast(TimestampType, dt))
+    numericTypes.foreach(dt => checkNullCast(TimeType, dt))
     for (from <- numericTypes; to <- numericTypes) checkNullCast(from, to)
   }
 
@@ -251,6 +259,7 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     assert(cast("abcdef", BinaryType).nullable === false)
     assert(cast("abcdef", BooleanType).nullable)
     assert(cast("abcdef", TimestampType).nullable)
+    assert(cast("abcdef", TimeType).nullable)
     assert(cast("abcdef", LongType).nullable)
     assert(cast("abcdef", IntegerType).nullable)
     assert(cast("abcdef", ShortType).nullable)
@@ -267,7 +276,10 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     val zts = sd + " 00:00:00"
     val sts = sd + " 00:00:02"
     val nts = sts + ".1"
-    val ts = withDefaultTimeZone(UTC)(Timestamp.valueOf(nts))
+    val zt = "00:00:00"
+    val t = "00:00:02"
+    val ts = withDefaultTimeZone(TimeZoneGMT)(Timestamp.valueOf(nts))
+    val ts2 = withDefaultTimeZone(TimeZoneGMT)(Time.valueOf(t))
 
     for (tz <- ALL_TIMEZONES) {
       val timeZoneId = Option(tz.getId)
@@ -295,6 +307,9 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(
       cast(cast(ts, StringType, UTC_OPT), TimestampType, UTC_OPT),
       DateTimeUtils.fromJavaTimestamp(ts))
+    checkEvaluation(
+          cast(cast(ts2, StringType, gmtId), TimeType, gmtId),
+          DateTimeUtils.fromJavaTime(ts2))
 
     // all convert to string type to check
     checkEvaluation(
@@ -303,6 +318,10 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(
       cast(cast(cast(ts, DateType, UTC_OPT), TimestampType, UTC_OPT), StringType, UTC_OPT),
       zts)
+    checkEvaluation(cast(cast(cast(nts, TimeType, gmtId), DateType, gmtId), StringType), sd)
+    checkEvaluation(
+      cast(cast(cast(ts2, DateType, gmtId), TimeType, gmtId), StringType, gmtId),
+      zts)
 
     checkEvaluation(cast(cast("abdef", BinaryType), StringType), "abdef")
 
@@ -324,6 +343,10 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         5.toShort)
     }
 
+      checkEvaluation(cast(cast(cast(cast(cast(cast("5", DecimalType.SYSTEM_DEFAULT),
+        ByteType), TimeType), LongType), StringType), ShortType),
+        5.toShort)
+
     checkEvaluation(cast("23", DoubleType), 23d)
     checkEvaluation(cast("23", IntegerType), 23)
     checkEvaluation(cast("23", FloatType), 23f)
@@ -583,6 +606,16 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(cast(inp, targetSchema), expected)
   }
 
+  test("cast struct with a time field") {
+    val originalSchema = new StructType().add("tsField", TimeType, nullable = false)
+    // nine out of ten times I'm casting a struct, it's to normalize its fields nullability
+    val targetSchema = new StructType().add("tsField", TimeType, nullable = true)
+
+    val inp = Literal.create(InternalRow(0L), originalSchema)
+    val expected = InternalRow(0L)
+    checkEvaluation(cast(inp, targetSchema), expected)
+  }
+
   test("complex casting") {
     val complex = Literal.create(
       Row(
@@ -878,6 +911,8 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
       checkExceptionInExpression[ArithmeticException](cast(Decimal(value.toString), dt), "overflow")
       checkExceptionInExpression[ArithmeticException](
         cast(Literal(value * MICROS_PER_SECOND, TimestampType), dt), "overflow")
+      checkExceptionInExpression[ArithmeticException](
+        cast(Literal(value * MICROS_PER_SECOND, TimeType), dt), "overflow")
       checkExceptionInExpression[ArithmeticException](
         cast(Literal(value * 1.5f, FloatType), dt), "overflow")
       checkExceptionInExpression[ArithmeticException](
@@ -904,6 +939,8 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         checkExceptionInExpression[ArithmeticException](cast(value, ByteType), "overflow")
         checkExceptionInExpression[ArithmeticException](
           cast(Literal(value * MICROS_PER_SECOND, TimestampType), ByteType), "overflow")
+        checkExceptionInExpression[ArithmeticException](
+          cast(Literal(value * MICROS_PER_SECOND, TimeType), ByteType), "overflow")
         checkExceptionInExpression[ArithmeticException](
           cast(Literal(value.toFloat, FloatType), ByteType), "overflow")
         checkExceptionInExpression[ArithmeticException](
@@ -915,6 +952,7 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         checkEvaluation(cast(value.toString, ByteType), value)
         checkEvaluation(cast(Decimal(value.toString), ByteType), value)
         checkEvaluation(cast(Literal(value * MICROS_PER_SECOND, TimestampType), ByteType), value)
+        checkEvaluation(cast(Literal(value * MICROS_PER_SECOND, TimeType), ByteType), value)
         checkEvaluation(cast(Literal(value.toInt, DateType), ByteType), null)
         checkEvaluation(cast(Literal(value.toFloat, FloatType), ByteType), value)
         checkEvaluation(cast(Literal(value.toDouble, DoubleType), ByteType), value)
@@ -929,6 +967,8 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         checkExceptionInExpression[ArithmeticException](cast(value, ShortType), "overflow")
         checkExceptionInExpression[ArithmeticException](
           cast(Literal(value * MICROS_PER_SECOND, TimestampType), ShortType), "overflow")
+        checkExceptionInExpression[ArithmeticException](
+          cast(Literal(value * MICROS_PER_SECOND, TimeType), ShortType), "overflow")
         checkExceptionInExpression[ArithmeticException](
           cast(Literal(value.toFloat, FloatType), ShortType), "overflow")
         checkExceptionInExpression[ArithmeticException](
@@ -940,6 +980,7 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         checkEvaluation(cast(value.toString, ShortType), value)
         checkEvaluation(cast(Decimal(value.toString), ShortType), value)
         checkEvaluation(cast(Literal(value * MICROS_PER_SECOND, TimestampType), ShortType), value)
+        checkEvaluation(cast(Literal(value * MICROS_PER_SECOND, TimeType), ShortType), value)
         checkEvaluation(cast(Literal(value.toInt, DateType), ShortType), null)
         checkEvaluation(cast(Literal(value.toFloat, FloatType), ShortType), value)
         checkEvaluation(cast(Literal(value.toDouble, DoubleType), ShortType), value)
@@ -957,6 +998,7 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         checkEvaluation(cast(value.toString, IntegerType), value)
         checkEvaluation(cast(Decimal(value.toString), IntegerType), value)
         checkEvaluation(cast(Literal(value * MICROS_PER_SECOND, TimestampType), IntegerType), value)
+        checkEvaluation(cast(Literal(value * MICROS_PER_SECOND, TimeType), IntegerType), value)
         checkEvaluation(cast(Literal(value * 1.0, DoubleType), IntegerType), value)
       }
       checkEvaluation(cast(Int.MaxValue + 0.9D, IntegerType), Int.MaxValue)
@@ -974,6 +1016,8 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
         checkEvaluation(cast(Decimal(value.toString), LongType), value)
         checkEvaluation(cast(Literal(value, TimestampType), LongType),
           Math.floorDiv(value, MICROS_PER_SECOND))
+        checkEvaluation(cast(Literal(value, TimeType), LongType),
+          Math.floorDiv(value, MICROS_PER_SECOND))
       }
       checkEvaluation(cast(Long.MaxValue + 0.9F, LongType), Long.MaxValue)
       checkEvaluation(cast(Long.MinValue - 0.9F, LongType), Long.MinValue)
@@ -1042,6 +1086,9 @@ class CastSuite extends CastSuiteBase {
       checkEvaluation(cast(cast(-1200, TimestampType), LongType), -1200.toLong)
     }
 
+    checkEvaluation(cast(cast(1000, TimeType), LongType), 1000.toLong)
+    checkEvaluation(cast(cast(-1200, TimeType), LongType), -1200.toLong)
+
     checkEvaluation(cast(123, DecimalType.USER_DEFAULT), Decimal(123))
     checkEvaluation(cast(123, DecimalType(3, 0)), Decimal(123))
     checkEvaluation(cast(123, DecimalType(3, 1)), null)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/HiveResult.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/HiveResult.scala
index 4d388e40fb8b..2c54448eac80 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/HiveResult.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/HiveResult.scala
@@ -18,6 +18,8 @@
 package org.apache.spark.sql.execution
 
 import java.nio.charset.StandardCharsets
+import java.sql.{Date, Time, Timestamp}
+import java.time.{Instant, LocalDate}
 import java.sql.{Date, Timestamp}
 import java.time.{Instant, LocalDate, ZoneOffset}
 
@@ -101,6 +103,8 @@ object HiveResult {
     case (ld: LocalDate, DateType) => formatters.date.format(ld)
     case (t: Timestamp, TimestampType) => formatters.timestamp.format(t)
     case (i: Instant, TimestampType) => formatters.timestamp.format(i)
+    case (t: Time, TimeType) => formatters.timestamp.format(t)
+    case (i: Instant, TimeType) => formatters.timestamp.format(i)
     case (bin: Array[Byte], BinaryType) => new String(bin, StandardCharsets.UTF_8)
     case (decimal: java.math.BigDecimal, DecimalType()) => decimal.toPlainString
     case (n, _: NumericType) => n.toString
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/HiveResultSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/HiveResultSuite.scala
index a49beda2186b..0e61e5fb011d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/HiveResultSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/HiveResultSuite.scala
@@ -56,6 +56,20 @@ class HiveResultSuite extends SharedSparkSession {
     assert(result2 == timestamps.map(x => s"[$x]"))
   }
 
+  test("time formatting in hive result") {
+    val time = Seq(
+      "2018-12-28 01:02:03",
+      "1582-10-13 01:02:03",
+      "1582-10-14 01:02:03",
+      "1582-10-15 01:02:03")
+    val df = time.toDF("a").selectExpr("cast(a as time) as b")
+    val result = HiveResult.hiveResultString(df)
+    assert(result == time)
+    val df2 = df.selectExpr("array(b)")
+    val result2 = HiveResult.hiveResultString(df2)
+    assert(result2 == time.map(x => s"[$x]"))
+  }
+
   test("toHiveString correctly handles UDTs") {
     val point = new ExamplePoint(50.0, 50.0)
     val tpe = new ExamplePointUDT()