You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "MaxGekk (via GitHub)" <gi...@apache.org> on 2026/07/14 07:49:25 UTC
Re: [PR] [SPARK-57555][FOLLOWUP][SQL] Add TimeType support to MySQLDialect [spark]
MaxGekk commented on code in PR #57198:
URL: https://github.com/apache/spark/pull/57198#discussion_r3576998999
##########
sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala:
##########
@@ -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))
Review Comment:
This out-of-range fallback loses **all** fractional seconds, not just the sub-microsecond part. `TimeType.MAX_PRECISION` is 9, so `TimeType(7|8|9)` reaches the `else` and emits bare `TIME` — which in MySQL is `TIME(0)` (whole seconds), as this PR's own read test confirms (inserting `13:31:24.123` into a bare `TIME` column reads back `13:31:24`). So `12:00:00.123456789` is stored as `12:00:00`, not truncated to microseconds. MySQL's max fractional precision is 6, so the fallback should emit `TIME(6)`:
```suggestion
// MySQL supports TIME(p) for p in [0,6]. Use the declared precision when valid;
// fall back to TIME(6) (MySQL's max fractional precision) for out-of-range
// precisions (e.g. nanosecond), which truncates to microseconds rather than
// to whole seconds (bare TIME = TIME(0) in MySQL).
val p = t.precision
if (p >= 0 && p <= 6) Option(JdbcType(s"TIME($p)", java.sql.Types.TIME))
else Option(JdbcType("TIME(6)", java.sql.Types.TIME))
```
##########
connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala:
##########
@@ -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.
Review Comment:
The premise here is wrong in the same way as the production comment: bare MySQL `TIME` is `TIME(0)`, not `TIME(6)`. With the `getJDBCType` fix (emit `TIME(6)` for out-of-range precision), the accurate statement is:
```suggestion
// MySQL TIME max precision is 6 (microseconds). TimeType(9) writes as TIME(6),
// and the sub-microsecond portion is truncated on read-back.
```
##########
connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala:
##########
@@ -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)))
Review Comment:
These assertions are correct — `TimeType(6)` with microseconds preserved is what a nanosecond write *should* round-trip to. But they don't hold against the current code: `getJDBCType` emits bare `TIME` (= MySQL `TIME(0)`) for `TimeType(9)`, so the column stores whole seconds and this reads back as `TimeType(0)` / `12:00:00`. On a real MySQL server both assertions would fail. This wasn't caught because the suite is `@DockerTest` and isn't part of standard PR CI (the description only reports the JDBCSuite unit run). Once the `getJDBCType` fix above lands, these assertions become correct as written — just fix the comment premise below.
##########
sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala:
##########
@@ -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
Review Comment:
`MetadataBuilder.getLong` throws `NoSuchElementException` if `datetimePrecision` is absent. It's safe today — `JdbcUtils.getSchema` always calls `updateExtraColumnMeta` before `getCatalystType`, and every branch there sets the key for a TIME column — but coupling this read to an unrelated method having run first is fragile. Consider guarding it:
```suggestion
val md0 = md.build()
val precision = if (md0.contains("datetimePrecision")) {
md0.getLong("datetimePrecision").toInt
} else {
TimeType.DEFAULT_PRECISION
}
```
(Non-blocking.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org