[SPARK-57555][FOLLOWUP][SQL] Add TimeType support to MySQLDialect

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala

@@ -19,15 +19,15 @@ package org.apache.spark.sql.jdbc
 
 import java.math.BigDecimal
 import java.sql.{Connection, Date, Timestamp}
-import java.time.LocalDateTime
+import java.time.{LocalDateTime, LocalTime}
 import java.util.Properties
 
 import scala.util.Using
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.ShortType
+import org.apache.spark.sql.types.{ShortType, TimeType}
 import org.apache.spark.tags.DockerTest
 
 /**
@@ -364,6 +364,91 @@ class MySQLIntegrationSuite extends SharedJDBCIntegrationSuite {
       .load()
     checkAnswer(df, Row("brown     ", "fox"))
   }
+
+  test("SPARK-57555: MySQL TIME type read as TimeType") {
+    withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "true") {
+      val df = spark.read.jdbc(jdbcUrl, "dates", new Properties)
+      val timeCol = df.schema("t")
+      // MySQL bare TIME (no precision) reports scale=0 via JDBC metadata
+      assert(timeCol.dataType === TimeType(0),
+        s"Expected TimeType(0) for bare TIME but got ${timeCol.dataType}")
+      val time3Col = df.schema("t1")
+      assert(time3Col.dataType === TimeType(3),
+        s"Expected TimeType(3) but got ${time3Col.dataType}")
+      // Bare TIME(0) truncates fractional seconds; TIME(3) preserves milliseconds
+      checkAnswer(df.select("t", "t1"), Row(
+        LocalTime.of(13, 31, 24),
+        LocalTime.of(13, 31, 24, 123000000)))
+    }
+  }
+
+  test("SPARK-57555: MySQL TIME type write round-trip") {
+    withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "true") {
+      val data = Seq(
+        Row(LocalTime.of(8, 30, 0), LocalTime.of(17, 22, 31, 123456000))
+      )
+      val schema = new org.apache.spark.sql.types.StructType()
+        .add("t", TimeType(0))
+        .add("t_micro", TimeType(6))
+      val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+      df.write.mode("overwrite")
+        .option("createTableColumnTypes", "t TIME(0), t_micro TIME(6)")
+        .jdbc(jdbcUrl, "time_write_test", new Properties)
+
+      val result = spark.read.jdbc(jdbcUrl, "time_write_test", new Properties)
+      assert(result.schema("t").dataType === TimeType(0))
+      assert(result.schema("t_micro").dataType === TimeType(6))
+      checkAnswer(result, Row(
+        LocalTime.of(8, 30, 0),
+        LocalTime.of(17, 22, 31, 123456000)))
+    }
+  }
+
+  test("SPARK-57555: MySQL TIME precision preservation round-trip") {
+    // Verifies TimeType(3) -> TIME(3) -> read back as TimeType(3)
+    withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "true") {
+      val data = Seq(Row(LocalTime.of(10, 15, 30, 500000000)))
+      val schema = new org.apache.spark.sql.types.StructType()
+        .add("t3", TimeType(3))
+      val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+      df.write.mode("overwrite").jdbc(jdbcUrl, "time_precision_test", new Properties)
+
+      val result = spark.read.jdbc(jdbcUrl, "time_precision_test", new Properties)
+      assert(result.schema("t3").dataType === TimeType(3),
+        "Precision should be preserved on round-trip")
+      checkAnswer(result, Row(LocalTime.of(10, 15, 30, 500000000)))
+    }
+  }
+
+  test("SPARK-57555: MySQL TIME nanosecond write truncates to microseconds") {
+    // MySQL TIME max precision is 6 (microseconds). TimeType(9) writes as bare TIME (= TIME(6)),
+    // and the nanosecond portion is truncated on read-back.
+    withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "true") {
+      val data = Seq(Row(LocalTime.of(12, 0, 0, 123456789)))
+      val schema = new org.apache.spark.sql.types.StructType()
+        .add("t_nanos", TimeType(9))
+      val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+      df.write.mode("overwrite").jdbc(jdbcUrl, "time_nanos_test", new Properties)
+
+      val result = spark.read.jdbc(jdbcUrl, "time_nanos_test", new Properties)
+      // Read back as TimeType(6) since MySQL stores at most microseconds
+      assert(result.schema("t_nanos").dataType === TimeType(6))
+      // Nanoseconds truncated to microseconds: 123456789 -> 123456000
+      checkAnswer(result, Row(LocalTime.of(12, 0, 0, 123456000)))
+    }
+  }
+
+  test("SPARK-57555: MySQL TIME type legacy mapping with escape hatch") {
+    withSQLConf(
+      SQLConf.TIME_TYPE_ENABLED.key -> "true",
+      SQLConf.LEGACY_JDBC_TIME_MAPPING_ENABLED.key -> "true") {
+      val df = spark.read.jdbc(jdbcUrl, "dates", new Properties)
+      // With escape hatch, TIME columns should remain as Timestamp (legacy behavior)
+      val timeCol = df.schema("t")
+      assert(!timeCol.dataType.isInstanceOf[TimeType],
+        s"With legacyJdbcTimeMappingEnabled=true, TIME should NOT be TimeType")
+    }
+  }
 }
 
 

sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala

@@ -17,11 +17,12 @@
 
 package org.apache.spark.sql.jdbc
 
-import java.sql.{Connection, SQLException, Types}
+import java.sql.{Connection, ResultSetMetaData, SQLException, Types}
 import java.util
 import java.util.Locale
 
 import scala.collection.mutable.ArrayBuilder
+import scala.util.Using
 import scala.util.control.NonFatal
 
 import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
@@ -131,6 +132,42 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No
     }
   }
 
+  override def updateExtraColumnMeta(
+      conn: Connection,
+      rsmd: ResultSetMetaData,
+      columnIdx: Int,
+      metadata: MetadataBuilder): Unit = {
+    // MySQL Connector/J Bug #84308: getPrecision()=8 and getScale()=0 for all TIME columns.
+    // Query INFORMATION_SCHEMA for actual DATETIME_PRECISION.
+    if (rsmd.getColumnType(columnIdx) == Types.TIME) {
+      val tableName = rsmd.getTableName(columnIdx)
+      val columnName = rsmd.getColumnName(columnIdx)
+      if (tableName.nonEmpty && columnName.nonEmpty) {
+        val query =
+          s"SELECT DATETIME_PRECISION FROM INFORMATION_SCHEMA.COLUMNS " +
+            s"WHERE TABLE_NAME = '$tableName' AND COLUMN_NAME = '$columnName' " +
+            s"AND TABLE_SCHEMA = DATABASE()"
+        try {
+          Using.resource(conn.createStatement()) { stmt =>
+            Using.resource(stmt.executeQuery(query)) { rs =>
+              if (rs.next()) {
+                metadata.putLong("datetimePrecision", rs.getLong(1))
+              } else {
+                metadata.putLong("datetimePrecision", TimeType.DEFAULT_PRECISION)
+              }
+            }
+          }
+        } catch {
+          case NonFatal(_) =>
+            metadata.putLong("datetimePrecision", TimeType.DEFAULT_PRECISION)
+        }
+      } else {
+        // For computed columns or expressions without table context, use default
+        metadata.putLong("datetimePrecision", TimeType.DEFAULT_PRECISION)
+      }
+    }
+  }
+
   override def getCatalystType(
       sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = {
     def getCatalystTypeForBitArray: Option[DataType] = {
@@ -184,6 +221,12 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No
         // scalastyle:on line.size.limit
         Some(getTimestampType(md.build()))
       case Types.TIMESTAMP if !conf.legacyMySqlTimestampNTZMappingEnabled => Some(TimestampType)
+      case Types.TIME if conf.isTimeTypeEnabled && !conf.legacyJdbcTimeMappingEnabled =>
+        // MySQL Connector/J Bug #84308: COLUMN_SIZE=8 and DECIMAL_DIGITS=0 for all TIME columns
+        // regardless of declared precision. The actual precision is injected by
+        // updateExtraColumnMeta via INFORMATION_SCHEMA query.
+        val precision = md.build().getLong("datetimePrecision").toInt
+        Some(TimeType(math.min(precision, TimeType.MAX_PRECISION)))
       case _ => None
     }
   }
@@ -282,6 +325,12 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No
     // scalastyle:on line.size.limit
     case TimestampNTZType if !conf.legacyMySqlTimestampNTZMappingEnabled =>
       Option(JdbcType("DATETIME", java.sql.Types.TIMESTAMP))
+    case t: TimeType =>
+      // MySQL supports TIME(p) for p in [0,6]. Use the declared precision when valid;
+      // fall back to bare TIME (= TIME(6)) for out-of-range precisions (e.g. nanosecond).
+      val p = t.precision
+      if (p >= 0 && p <= 6) Option(JdbcType(s"TIME($p)", java.sql.Types.TIME))
+      else Option(JdbcType("TIME", java.sql.Types.TIME))
     case _ => JdbcUtils.getCommonJDBCType(dt)
   }