From 3985441f3e0be5ea8449981c080ab093314f42a3 Mon Sep 17 00:00:00 2001 From: shrirangmhalgi Date: Fri, 10 Jul 2026 22:56:33 -0700 Subject: [PATCH 1/3] [SPARK-57555][SQL] Support TIME data type in MySQL JDBC dialect Add TimeType support to MySQLDialect: - getCatalystType: maps Types.TIME to TimeType(precision) when timeType.enabled=true and legacyJdbcTimeMappingEnabled=false. Precision is derived from JDBC metadata (scale parameter). - getJDBCType: maps TimeType(p) to TIME(p) DDL for write path. - Integration tests: read round-trip, write round-trip, and legacy escape hatch verification. --- .../sql/jdbc/MySQLIntegrationSuite.scala | 51 ++++++++++++++++++- .../apache/spark/sql/jdbc/MySQLDialect.scala | 7 +++ 2 files changed, 57 insertions(+), 1 deletion(-) 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..e199eb90a785e 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 @@ -364,6 +364,55 @@ 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") + assert(timeCol.dataType === TimeType(6), + s"Expected TimeType(6) but got ${timeCol.dataType}") + val time3Col = df.schema("t1") + assert(time3Col.dataType === TimeType(3), + s"Expected TimeType(3) but got ${time3Col.dataType}") + checkAnswer(df.select("t", "t1"), Row( + LocalTime.of(13, 31, 24, 123000000), + 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 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..cffe13492faa7 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 @@ -184,6 +184,11 @@ 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 TIME(p) precision is reported via the scale parameter. + val precision = if (size > 0 && size <= TimeType.MAX_PRECISION) size + else TimeType.DEFAULT_PRECISION + Some(TimeType(precision)) case _ => None } } @@ -282,6 +287,8 @@ 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 => + Option(JdbcType(s"TIME(${t.precision})", java.sql.Types.TIME)) case _ => JdbcUtils.getCommonJDBCType(dt) } From 741a8e261953762fe82bb1f95e51a650b358815b Mon Sep 17 00:00:00 2001 From: shrirangmhalgi Date: Fri, 10 Jul 2026 23:02:42 -0700 Subject: [PATCH 2/3] Address review patterns from postgres PR: precision via scale, nanos clamping, proactive fixes - getCatalystType: use md.build().getLong("scale") for fractional-second precision (getScale reports DECIMAL_DIGITS, not display width). Accept scale=0 for bare TIME (TIME(0) is valid). - getJDBCType: use TIME(p) for p<=6, bare TIME for out-of-range precisions (e.g. nanosecond TimeType), matching PostgreSQL pattern. - Fix read test: bare TIME column reports scale=0, truncates fractional seconds (value 13:31:24 not 13:31:24.123). - Add precision preservation round-trip test (TimeType(3) -> TIME(3)) - Add nanosecond TimeType(9) write test (MySQL truncates to micros) - Add comment explaining precision source from JDBC metadata --- .../sql/jdbc/MySQLIntegrationSuite.scala | 44 +++++++++++++++++-- .../apache/spark/sql/jdbc/MySQLDialect.scala | 12 +++-- 2 files changed, 49 insertions(+), 7 deletions(-) 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 e199eb90a785e..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 @@ -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 /** @@ -369,13 +369,15 @@ class MySQLIntegrationSuite extends SharedJDBCIntegrationSuite { withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "true") { val df = spark.read.jdbc(jdbcUrl, "dates", new Properties) val timeCol = df.schema("t") - assert(timeCol.dataType === TimeType(6), - s"Expected TimeType(6) but got ${timeCol.dataType}") + // 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, 123000000), + LocalTime.of(13, 31, 24), LocalTime.of(13, 31, 24, 123000000))) } } @@ -402,6 +404,40 @@ class MySQLIntegrationSuite extends SharedJDBCIntegrationSuite { } } + 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", 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 cffe13492faa7..b81fb4fc09187 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 @@ -185,8 +185,10 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No Some(getTimestampType(md.build())) case Types.TIMESTAMP if !conf.legacyMySqlTimestampNTZMappingEnabled => Some(TimestampType) case Types.TIME if conf.isTimeTypeEnabled && !conf.legacyJdbcTimeMappingEnabled => - // MySQL TIME(p) precision is reported via the scale parameter. - val precision = if (size > 0 && size <= TimeType.MAX_PRECISION) size + // MySQL TIME(p) fractional-second precision is reported via getScale() (DECIMAL_DIGITS). + // Use scale >= 0 to accept TIME(0); default to DEFAULT_PRECISION when not reported. + val scale = md.build().getLong("scale").toInt + val precision = if (scale >= 0 && scale <= TimeType.MAX_PRECISION) scale else TimeType.DEFAULT_PRECISION Some(TimeType(precision)) case _ => None @@ -288,7 +290,11 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No case TimestampNTZType if !conf.legacyMySqlTimestampNTZMappingEnabled => Option(JdbcType("DATETIME", java.sql.Types.TIMESTAMP)) case t: TimeType => - Option(JdbcType(s"TIME(${t.precision})", java.sql.Types.TIME)) + // 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) } From 75cd5c7c8d86f3788c22d22a18ed882204055f6a Mon Sep 17 00:00:00 2001 From: shrirangmhalgi Date: Sat, 11 Jul 2026 07:01:50 -0700 Subject: [PATCH 3/3] Fix MySQL TIME precision: query INFORMATION_SCHEMA for DATETIME_PRECISION MySQL Connector/J Bug #84308: getPrecision()=8 and getScale()=0 for all TIME columns regardless of declared precision. Override updateExtraColumnMeta to query INFORMATION_SCHEMA.COLUMNS.DATETIME_PRECISION and inject the real precision into the metadata, which getCatalystType then reads. --- .../apache/spark/sql/jdbc/MySQLDialect.scala | 50 ++++++++++++++++--- 1 file changed, 43 insertions(+), 7 deletions(-) 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 b81fb4fc09187..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] = { @@ -185,12 +222,11 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No Some(getTimestampType(md.build())) case Types.TIMESTAMP if !conf.legacyMySqlTimestampNTZMappingEnabled => Some(TimestampType) case Types.TIME if conf.isTimeTypeEnabled && !conf.legacyJdbcTimeMappingEnabled => - // MySQL TIME(p) fractional-second precision is reported via getScale() (DECIMAL_DIGITS). - // Use scale >= 0 to accept TIME(0); default to DEFAULT_PRECISION when not reported. - val scale = md.build().getLong("scale").toInt - val precision = if (scale >= 0 && scale <= TimeType.MAX_PRECISION) scale - else TimeType.DEFAULT_PRECISION - Some(TimeType(precision)) + // 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 } }