Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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

/**
Expand Down Expand Up @@ -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 TIME(6),
// and the sub-microsecond 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)))
Comment on lines +435 to +437

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
}

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")
}
}
}


Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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] = {
Expand Down Expand Up @@ -184,6 +221,17 @@ 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 md0 = md.build()
val precision = if (md0.contains("datetimePrecision")) {
md0.getLong("datetimePrecision").toInt
} else {
TimeType.DEFAULT_PRECISION
}
Some(TimeType(math.min(precision, TimeType.MAX_PRECISION)))
case _ => None
}
}
Expand Down Expand Up @@ -282,6 +330,14 @@ 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 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))
case _ => JdbcUtils.getCommonJDBCType(dt)
}

Expand Down