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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
