[SPARK-54609][SQL] Disable TIME type by default

### What changes were proposed in this pull request?

Introducing a new SQL config for TIME type: `spark.sql.timeType.enabled`.

The default value is `false` and it is enabled only in tests.

### Why are the changes needed?

TIME data type support is not complete, so we need to guard it before it is completed, especially ahead of Spark 4.1 release.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Need to add tests for disabled config.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #53344 from davidm-db/davidm-db/time-config.

Lead-authored-by: David Milicevic <[email protected]>
Co-authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: 2 people

common/utils/src/main/resources/error/error-conditions.json

@@ -7309,6 +7309,12 @@
     ],
     "sqlState" : "0A001"
   },
+  "UNSUPPORTED_TIME_TYPE" : {
+    "message" : [
+      "The data type TIME is not supported."
+    ],
+    "sqlState" : "0A000"
+  },
   "UNSUPPORTED_TYPED_LITERAL" : {
     "message" : [
       "Literals of the type <unsupportedType> are not supported. Supported types are <supportedTypes>."

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala

@@ -25,12 +25,11 @@ import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime, Period
 import java.util.{Map => JavaMap}
 import javax.annotation.Nullable
 
-import scala.language.existentials
-
 import org.apache.spark.SparkIllegalArgumentException
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.types.DayTimeIntervalType._
@@ -79,6 +78,8 @@ object CatalystTypeConverters {
         new GeometryConverter(g)
       case DateType if SQLConf.get.datetimeJava8ApiEnabled => LocalDateConverter
       case DateType => DateConverter
+      case _: TimeType if !SQLConf.get.isTimeTypeEnabled =>
+        QueryCompilationErrors.unsupportedTimeTypeError()
       case _: TimeType => TimeConverter
       case TimestampType if SQLConf.get.datetimeJava8ApiEnabled => InstantConverter
       case TimestampType => TimestampConverter

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.catalyst.util.DateTimeConstants._
 import org.apache.spark.sql.catalyst.util.DateTimeUtils._
 import org.apache.spark.sql.catalyst.util.IntervalUtils.{dayTimeIntervalToByte, dayTimeIntervalToDecimal, dayTimeIntervalToInt, dayTimeIntervalToLong, dayTimeIntervalToShort, yearMonthIntervalToByte, yearMonthIntervalToInt, yearMonthIntervalToShort}
-import org.apache.spark.sql.errors.{QueryErrorsBase, QueryExecutionErrors}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase, QueryExecutionErrors}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.{GeographyVal, UTF8String, VariantVal}
@@ -617,6 +617,12 @@ case class Cast(
   }
 
   override def checkInputDataTypes(): TypeCheckResult = {
+    dataType match {
+      // If the cast is to a TIME type, first check if TIME type is enabled.
+      case _: TimeType if !SQLConf.get.isTimeTypeEnabled =>
+        throw QueryCompilationErrors.unsupportedTimeTypeError()
+      case _ =>
+    }
     val canCast = evalMode match {
       case EvalMode.LEGACY => Cast.canCast(child.dataType, dataType)
       case EvalMode.ANSI => Cast.canAnsiCast(child.dataType, dataType)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/timeExpressions.scala

@@ -34,11 +34,22 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.catalyst.util.TimeFormatter
 import org.apache.spark.sql.catalyst.util.TypeUtils.ordinalNumber
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.types.StringTypeWithCollation
 import org.apache.spark.sql.types.{AbstractDataType, AnyTimeType, ByteType, DataType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, IntegralType, LongType, NumericType, ObjectType, TimeType}
 import org.apache.spark.sql.types.DayTimeIntervalType.{HOUR, SECOND}
 import org.apache.spark.unsafe.types.UTF8String
 
+trait TimeExpression extends Expression {
+  override def checkInputDataTypes(): TypeCheckResult = {
+    if (SQLConf.get.isTimeTypeEnabled) {
+      super.checkInputDataTypes()
+    } else {
+      throw QueryCompilationErrors.unsupportedTimeTypeError()
+    }
+  }
+}
+
 /**
  * Parses a column to a time based on the given format.
  */
@@ -66,7 +77,7 @@ import org.apache.spark.unsafe.types.UTF8String
   since = "4.1.0")
 // scalastyle:on line.size.limit
 case class ToTime(str: Expression, format: Option[Expression])
-  extends RuntimeReplaceable with ExpectsInputTypes {
+  extends RuntimeReplaceable with ExpectsInputTypes with TimeExpression {
 
   def this(str: Expression, format: Expression) = this(str, Option(format))
   def this(str: Expression) = this(str, None)
@@ -202,7 +213,7 @@ object TryToTimeExpressionBuilder extends ExpressionBuilder {
 // scalastyle:on line.size.limit
 case class MinutesOfTime(child: Expression)
   extends RuntimeReplaceable
-    with ExpectsInputTypes {
+    with ExpectsInputTypes with TimeExpression {
 
   override def replacement: Expression = StaticInvoke(
     classOf[DateTimeUtils.type],
@@ -261,7 +272,7 @@ object MinuteExpressionBuilder extends ExpressionBuilder {
 
 case class HoursOfTime(child: Expression)
   extends RuntimeReplaceable
-    with ExpectsInputTypes {
+    with ExpectsInputTypes with TimeExpression {
 
   override def replacement: Expression = StaticInvoke(
     classOf[DateTimeUtils.type],
@@ -318,7 +329,7 @@ object HourExpressionBuilder extends ExpressionBuilder {
 
 case class SecondsOfTimeWithFraction(child: Expression)
   extends RuntimeReplaceable
-  with ExpectsInputTypes {
+  with ExpectsInputTypes with TimeExpression {
   override def replacement: Expression = {
     val precision = child.dataType match {
       case TimeType(p) => p
@@ -344,7 +355,7 @@ case class SecondsOfTimeWithFraction(child: Expression)
 
 case class SecondsOfTime(child: Expression)
   extends RuntimeReplaceable
-    with ExpectsInputTypes {
+    with ExpectsInputTypes with TimeExpression {
 
   override def replacement: Expression = StaticInvoke(
     classOf[DateTimeUtils.type],
@@ -435,7 +446,8 @@ object SecondExpressionBuilder extends ExpressionBuilder {
 case class CurrentTime(
     child: Expression = Literal(TimeType.MICROS_PRECISION),
     timeZoneId: Option[String] = None) extends UnaryExpression
-  with TimeZoneAwareExpression with ImplicitCastInputTypes with CodegenFallback {
+  with TimeZoneAwareExpression with ImplicitCastInputTypes with CodegenFallback
+  with TimeExpression {
 
   def this() = {
     this(Literal(TimeType.MICROS_PRECISION), None)
@@ -547,7 +559,7 @@ case class MakeTime(
     secsAndMicros: Expression)
   extends RuntimeReplaceable
     with ImplicitCastInputTypes
-    with ExpectsInputTypes {
+    with ExpectsInputTypes with TimeExpression {
 
   // Accept `sec` as DecimalType to avoid loosing precision of microseconds while converting
   // it to the fractional part of `sec`. If `sec` is an IntegerType, it can be cast into decimal
@@ -572,7 +584,8 @@ case class MakeTime(
  * Adds day-time interval to time.
  */
 case class TimeAddInterval(time: Expression, interval: Expression)
-  extends BinaryExpression with RuntimeReplaceable with ExpectsInputTypes {
+  extends BinaryExpression with RuntimeReplaceable with ExpectsInputTypes
+  with TimeExpression {
   override def nullIntolerant: Boolean = true
 
   override def left: Expression = time
@@ -613,7 +626,8 @@ case class TimeAddInterval(time: Expression, interval: Expression)
  * Returns a day-time interval between time values.
  */
 case class SubtractTimes(left: Expression, right: Expression)
-  extends BinaryExpression with RuntimeReplaceable with ExpectsInputTypes {
+  extends BinaryExpression with RuntimeReplaceable with ExpectsInputTypes
+  with TimeExpression {
   override def nullIntolerant: Boolean = true
   override def inputTypes: Seq[AbstractDataType] = Seq(AnyTimeType, AnyTimeType)
 
@@ -670,7 +684,8 @@ case class TimeDiff(
     end: Expression)
   extends TernaryExpression
   with RuntimeReplaceable
-  with ImplicitCastInputTypes {
+  with ImplicitCastInputTypes
+  with TimeExpression {
 
   override def first: Expression = unit
   override def second: Expression = start
@@ -725,7 +740,8 @@ case class TimeDiff(
   since = "4.1.0")
 // scalastyle:on line.size.limit
 case class TimeTrunc(unit: Expression, time: Expression)
-  extends BinaryExpression with RuntimeReplaceable with ImplicitCastInputTypes {
+  extends BinaryExpression with RuntimeReplaceable with ImplicitCastInputTypes
+  with TimeExpression {
 
   override def left: Expression = unit
   override def right: Expression = time
@@ -753,7 +769,8 @@ case class TimeTrunc(unit: Expression, time: Expression)
 }
 
 abstract class IntegralToTimeBase
-  extends UnaryExpression with ExpectsInputTypes with CodegenFallback {
+  extends UnaryExpression with ExpectsInputTypes with CodegenFallback
+  with TimeExpression {
   protected def upScaleFactor: Long
 
   override def inputTypes: Seq[AbstractDataType] = Seq(IntegralType)
@@ -772,7 +789,8 @@ abstract class IntegralToTimeBase
   }
 }
 
-abstract class TimeToLongBase extends UnaryExpression with ExpectsInputTypes {
+abstract class TimeToLongBase extends UnaryExpression with ExpectsInputTypes
+  with TimeExpression {
   protected def scaleFactor: Long
 
   override def inputTypes: Seq[AbstractDataType] = Seq(AnyTimeType)
@@ -819,7 +837,8 @@ abstract class TimeToLongBase extends UnaryExpression with ExpectsInputTypes {
   group = "datetime_funcs")
 // scalastyle:on line.size.limit
 case class TimeFromSeconds(child: Expression)
-  extends UnaryExpression with ExpectsInputTypes with CodegenFallback {
+  extends UnaryExpression with ExpectsInputTypes with CodegenFallback
+  with TimeExpression {
   override def inputTypes: Seq[AbstractDataType] = Seq(NumericType)
   override def dataType: DataType = TimeType(TimeType.MICROS_PRECISION)
   override def nullable: Boolean = true

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -4492,4 +4492,10 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
         "colType" -> "metadata",
         "errors" -> errors.mkString("- ", "\n- ", "")))
   }
+
+  def unsupportedTimeTypeError(): Throwable = {
+    new AnalysisException(
+      errorClass = "UNSUPPORTED_TIME_TYPE",
+      messageParameters = Map.empty)
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -6731,6 +6731,13 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val TIME_TYPE_ENABLED =
+    buildConf("spark.sql.timeType.enabled")
+      .doc("When true, the TIME data type is supported.")
+      .version("4.1.0")
+      .booleanConf
+      .createWithDefault(Utils.isTesting)
+
   /**
    * Holds information about keys that have been deprecated.
    *
@@ -7938,6 +7945,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
   def coerceMergeNestedTypes: Boolean =
     getConf(SQLConf.MERGE_INTO_NESTED_TYPE_COERCION_ENABLED)
 
+  def isTimeTypeEnabled: Boolean = getConf(SQLConf.TIME_TYPE_ENABLED)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala

@@ -37,10 +37,11 @@ import org.apache.spark.sql.connect.config.Connect.{CONNECT_GRPC_ARROW_MAX_BATCH
 import org.apache.spark.sql.connect.planner.{InvalidInputErrors, SparkConnectPlanner}
 import org.apache.spark.sql.connect.service.ExecuteHolder
 import org.apache.spark.sql.connect.utils.{MetricGenerator, PipelineAnalysisContextUtils}
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.{DoNotCleanup, LocalTableScanExec, QueryExecution, RemoveShuffleFiles, SkipMigration, SQLExecution}
 import org.apache.spark.sql.execution.arrow.ArrowConverters
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.sql.types.{DataType, StructType, TimeType}
 import org.apache.spark.util.ThreadUtils
 
 /**
@@ -143,6 +144,10 @@ private[execution] class SparkConnectPlanExecution(executeHolder: ExecuteHolder)
         errorClass = "UNSUPPORTED_FEATURE.GEOSPATIAL_DISABLED",
         messageParameters = scala.collection.immutable.Map.empty)
     }
+    val timeTypeEnabled = spark.sessionState.conf.isTimeTypeEnabled
+    if (!timeTypeEnabled && schema.existsRecursively(_.isInstanceOf[TimeType])) {
+      throw QueryCompilationErrors.unsupportedTimeTypeError()
+    }
     val maxRecordsPerBatch = spark.sessionState.conf.arrowMaxRecordsPerBatch
     val timeZoneId = spark.sessionState.conf.sessionLocalTimeZone
     val largeVarTypes = spark.sessionState.conf.arrowUseLargeVarTypes

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala

@@ -93,6 +93,9 @@ object DataSourceUtils extends PredicateHelper {
    * in a driver side.
    */
   def verifySchema(format: FileFormat, schema: StructType, readOnly: Boolean = false): Unit = {
+    if (!SQLConf.get.isTimeTypeEnabled && schema.existsRecursively(_.isInstanceOf[TimeType])) {
+      throw QueryCompilationErrors.unsupportedTimeTypeError()
+    }
     schema.foreach { field =>
       val supported = if (readOnly) {
         format.supportReadDataType(field.dataType)

sql/gen-sql-functions-docs.py

@@ -240,6 +240,8 @@ def generate_functions_examples_html(jvm, jspark, html_output_dir):
     </pre></div>
 
     """
+    print("Enabling TIME data type")
+    jspark.sql("SET spark.sql.timeType.enabled = true")
     print("Running SQL examples to generate formatted output.")
     for key, infos in _list_grouped_function_infos(jvm):
         examples = _make_pretty_examples(jspark, infos)