diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala index 7ed6c7c0ab911..05b96d1eb16aa 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala @@ -19,7 +19,7 @@ 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 @@ -27,7 +27,7 @@ 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") + } + } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala index b301c0c0bd5bc..6a3ff731ee6c3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala +++ b/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) }