This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5d3b74cd802a [SPARK-57555][SQL] Support TIME data type in built-in
JDBC data source
5d3b74cd802a is described below
commit 5d3b74cd802a2eecfb67be93221972327a0f2f0c
Author: Shrirang Mhalgi <[email protected]>
AuthorDate: Thu Jun 25 08:17:10 2026 +0200
[SPARK-57555][SQL] Support TIME data type in built-in JDBC data source
### What changes were proposed in this pull request?
Add native `TimeType` support to the built-in JDBC data source.
`getCatalystType` maps `java.sql.Types.TIME` to `TimeType` (gated behind
`spark.sql.timeType.enabled`), `getCommonJDBCType` maps `TimeType` to SQL
`TIME`, and value read / write uses `java.time.LocalTime`. Per-dialect
overrides are left for follow-up sub-tasks. A DB2 dialect override is included
in this PR emitting bare `TIME` as DB2 does not support `TIME(p)` syntax.
### Why are the changes needed?
JDBC currently maps `java.sql.Types.TIME` to `TimestampType`, losing
time-only semantics. With TimeType now supported in Spark, the JDBC source
should read and write TIME columns natively.
### Does this PR introduce any user-facing change?
Yes. When `spark.sql.timeType.enabled` is `true`, JDBC TIME columns are
inferred as `TimeType` with driver-reported precision, and `TimeType` columns
can be written as SQL TIME. When disabled (default), behavior is unchanged.
### How was this patch tested?
- Read, write round-trip, and sub-second precision tests against H2
- Legacy back-compat tests preserved with `timeType.enabled=false`
- All 6 TIME-related tests pass
### Was this patch authored or co-authored using generative AI tooling?
Co-Authored using Claude Opus 4.6
Closes #56653 from shrirangmhalgi/SPARK-57555-jdbc-time-type.
Authored-by: Shrirang Mhalgi <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 141b23a3a9ef8e45126c1ba712b8f9a906c639ab)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../spark/sql/jdbc/DB2IntegrationSuite.scala | 24 +--
.../sql/jdbc/MsSqlServerIntegrationSuite.scala | 3 +-
.../spark/sql/jdbc/MySQLIntegrationSuite.scala | 38 ++--
.../spark/sql/jdbc/PostgresIntegrationSuite.scala | 196 +++++++++++----------
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 23 ++-
.../org/apache/spark/sql/jdbc/DB2Dialect.scala | 1 +
.../org/apache/spark/sql/sources/interfaces.scala | 3 +-
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 92 ++++++++--
8 files changed, 233 insertions(+), 147 deletions(-)
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/DB2IntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/DB2IntegrationSuite.scala
index 6a489ffb2d42..10402b99e232 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/DB2IntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/DB2IntegrationSuite.scala
@@ -131,17 +131,19 @@ class DB2IntegrationSuite extends
SharedJDBCIntegrationSuite {
test("Date types") {
withDefaultTimeZone(UTC) {
- val df = spark.read.jdbc(jdbcUrl, "dates", new Properties)
- val rows = df.collect()
- assert(rows.length == 1)
- val types = rows(0).toSeq.map(x => x.getClass.toString)
- assert(types.length == 3)
- assert(types(0).equals("class java.sql.Date"))
- assert(types(1).equals("class java.sql.Timestamp"))
- assert(types(2).equals("class java.sql.Timestamp"))
- assert(rows(0).getAs[Date](0).equals(Date.valueOf("1991-11-09")))
- assert(rows(0).getAs[Timestamp](1).equals(Timestamp.valueOf("1970-01-01
13:31:24")))
- assert(rows(0).getAs[Timestamp](2).equals(Timestamp.valueOf("2009-02-13
23:31:30")))
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "false") {
+ val df = spark.read.jdbc(jdbcUrl, "dates", new Properties)
+ val rows = df.collect()
+ assert(rows.length == 1)
+ val types = rows(0).toSeq.map(x => x.getClass.toString)
+ assert(types.length == 3)
+ assert(types(0).equals("class java.sql.Date"))
+ assert(types(1).equals("class java.sql.Timestamp"))
+ assert(types(2).equals("class java.sql.Timestamp"))
+ assert(rows(0).getAs[Date](0).equals(Date.valueOf("1991-11-09")))
+
assert(rows(0).getAs[Timestamp](1).equals(Timestamp.valueOf("1970-01-01
13:31:24")))
+
assert(rows(0).getAs[Timestamp](2).equals(Timestamp.valueOf("2009-02-13
23:31:30")))
+ }
}
}
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MsSqlServerIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MsSqlServerIntegrationSuite.scala
index 0950fd7330c5..858d89f53387 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MsSqlServerIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MsSqlServerIntegrationSuite.scala
@@ -226,7 +226,8 @@ class MsSqlServerIntegrationSuite extends
SharedJDBCIntegrationSuite {
Seq(true, false).foreach { ntz =>
Seq(true, false).foreach { legacy =>
withSQLConf(
- SQLConf.LEGACY_MSSQLSERVER_DATETIMEOFFSET_MAPPING_ENABLED.key ->
legacy.toString) {
+ SQLConf.LEGACY_MSSQLSERVER_DATETIMEOFFSET_MAPPING_ENABLED.key ->
legacy.toString,
+ SQLConf.TIME_TYPE_ENABLED.key -> "false") {
val df = spark.read
.option("preferTimestampNTZ", ntz)
.jdbc(jdbcUrl, "dates", new Properties)
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 345b66e07ae0..7ed6c7c0ab91 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
@@ -184,14 +184,16 @@ class MySQLIntegrationSuite extends
SharedJDBCIntegrationSuite {
test("Date types") {
withDefaultTimeZone(UTC) {
- val df = spark.read.jdbc(jdbcUrl, "dates", new Properties)
- checkAnswer(df, Row(
- Date.valueOf("1991-11-09"),
- Timestamp.valueOf("1970-01-01 13:31:24"),
- Timestamp.valueOf("1996-01-01 01:23:45"),
- Timestamp.valueOf("2009-02-13 23:31:30"),
- Date.valueOf("2001-01-01"),
- Timestamp.valueOf("1970-01-01 13:31:24.123")))
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "false") {
+ val df = spark.read.jdbc(jdbcUrl, "dates", new Properties)
+ checkAnswer(df, Row(
+ Date.valueOf("1991-11-09"),
+ Timestamp.valueOf("1970-01-01 13:31:24"),
+ Timestamp.valueOf("1996-01-01 01:23:45"),
+ Timestamp.valueOf("2009-02-13 23:31:30"),
+ Date.valueOf("2001-01-01"),
+ Timestamp.valueOf("1970-01-01 13:31:24.123")))
+ }
}
val df = spark.read.format("jdbc")
.option("url", jdbcUrl)
@@ -203,15 +205,17 @@ class MySQLIntegrationSuite extends
SharedJDBCIntegrationSuite {
test("SPARK-47406: MySQL datetime types with preferTimestampNTZ") {
withDefaultTimeZone(UTC) {
- val df = spark.read.option("preferTimestampNTZ", true)
- .jdbc(jdbcUrl, "dates", new Properties)
- checkAnswer(df, Row(
- Date.valueOf("1991-11-09"),
- LocalDateTime.of(1970, 1, 1, 13, 31, 24),
- LocalDateTime.of(1996, 1, 1, 1, 23, 45),
- Timestamp.valueOf("2009-02-13 23:31:30"),
- Date.valueOf("2001-01-01"),
- LocalDateTime.of(1970, 1, 1, 13, 31, 24, 123000000)))
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "false") {
+ val df = spark.read.option("preferTimestampNTZ", true)
+ .jdbc(jdbcUrl, "dates", new Properties)
+ checkAnswer(df, Row(
+ Date.valueOf("1991-11-09"),
+ LocalDateTime.of(1970, 1, 1, 13, 31, 24),
+ LocalDateTime.of(1996, 1, 1, 1, 23, 45),
+ Timestamp.valueOf("2009-02-13 23:31:30"),
+ Date.valueOf("2001-01-01"),
+ LocalDateTime.of(1970, 1, 1, 13, 31, 24, 123000000)))
+ }
}
}
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala
index 9e08cc976bb5..78cd846d718c 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala
@@ -186,103 +186,105 @@ class PostgresIntegrationSuite extends
SharedJDBCIntegrationSuite {
}
test("Type mapping for various types") {
- val df = spark.read.jdbc(jdbcUrl, "bar", new Properties)
- val rows = df.collect().sortBy(_.toString())
- assert(rows.length == 2)
- // Test the types, and values using the first row.
- val types = rows(0).toSeq.map(x => x.getClass)
- assert(types.length == 42)
- assert(classOf[String].isAssignableFrom(types(0)))
- assert(classOf[java.lang.Integer].isAssignableFrom(types(1)))
- assert(classOf[java.lang.Double].isAssignableFrom(types(2)))
- assert(classOf[java.lang.Long].isAssignableFrom(types(3)))
- assert(classOf[java.lang.Boolean].isAssignableFrom(types(4)))
- assert(classOf[Array[Byte]].isAssignableFrom(types(5)))
- assert(classOf[Array[Byte]].isAssignableFrom(types(6)))
- assert(classOf[java.lang.Boolean].isAssignableFrom(types(7)))
- assert(classOf[String].isAssignableFrom(types(8)))
- assert(classOf[String].isAssignableFrom(types(9)))
- assert(classOf[scala.collection.Seq[Int]].isAssignableFrom(types(10)))
- assert(classOf[scala.collection.Seq[String]].isAssignableFrom(types(11)))
- assert(classOf[scala.collection.Seq[Double]].isAssignableFrom(types(12)))
-
assert(classOf[scala.collection.Seq[BigDecimal]].isAssignableFrom(types(13)))
- assert(classOf[String].isAssignableFrom(types(14)))
- assert(classOf[java.lang.Float].isAssignableFrom(types(15)))
- assert(classOf[java.lang.Short].isAssignableFrom(types(16)))
-
assert(classOf[scala.collection.Seq[BigDecimal]].isAssignableFrom(types(17)))
- assert(classOf[String].isAssignableFrom(types(18)))
- assert(classOf[String].isAssignableFrom(types(19)))
- assert(classOf[String].isAssignableFrom(types(20)))
- assert(classOf[String].isAssignableFrom(types(21)))
- assert(classOf[String].isAssignableFrom(types(22)))
- assert(classOf[String].isAssignableFrom(types(23)))
- assert(classOf[String].isAssignableFrom(types(24)))
- assert(classOf[String].isAssignableFrom(types(25)))
- assert(classOf[String].isAssignableFrom(types(26)))
- assert(classOf[String].isAssignableFrom(types(27)))
- assert(classOf[String].isAssignableFrom(types(28)))
- assert(classOf[Date].isAssignableFrom(types(29)))
- assert(classOf[String].isAssignableFrom(types(30)))
- assert(classOf[String].isAssignableFrom(types(31)))
- assert(classOf[String].isAssignableFrom(types(32)))
- assert(classOf[JBigDecimal].isAssignableFrom(types(33)))
- assert(classOf[String].isAssignableFrom(types(34)))
- assert(classOf[java.lang.Float].isAssignableFrom(types(35)))
- assert(classOf[java.sql.Timestamp].isAssignableFrom(types(36)))
- assert(classOf[java.sql.Timestamp].isAssignableFrom(types(37)))
- assert(classOf[String].isAssignableFrom(types(38)))
- assert(classOf[String].isAssignableFrom(types(39)))
- assert(classOf[String].isAssignableFrom(types(40)))
- assert(classOf[String].isAssignableFrom(types(41)))
- assert(rows(0).getString(0).equals("hello"))
- assert(rows(0).getInt(1) == 42)
- assert(rows(0).getDouble(2) == 1.25)
- assert(rows(0).getLong(3) == 123456789012345L)
- assert(!rows(0).getBoolean(4))
- // BIT(10)'s come back as ASCII strings of ten ASCII 0's and 1's...
- assert(java.util.Arrays.equals(rows(0).getAs[Array[Byte]](5),
- Array[Byte](49, 48, 48, 48, 49, 48, 48, 49, 48, 49)))
- assert(java.util.Arrays.equals(rows(0).getAs[Array[Byte]](6),
- Array[Byte](0xDE.toByte, 0xAD.toByte, 0xBE.toByte, 0xEF.toByte)))
- assert(rows(0).getBoolean(7))
- assert(rows(0).getString(8) == "172.16.0.42")
- assert(rows(0).getString(9) == "192.168.0.0/16")
- assert(rows(0).getSeq(10) == Seq(1, 2))
- assert(rows(0).getSeq(11) == Seq("a", null, "b"))
- assert(rows(0).getSeq(12).toSeq == Seq(0.11f, 0.22f))
- assert(rows(0).getSeq(13) == Seq("0.11",
"0.22").map(BigDecimal(_).bigDecimal))
- assert(rows(0).getString(14) == "d1")
- assert(rows(0).getFloat(15) == 1.01f)
- assert(rows(0).getShort(16) == 1)
- assert(rows(0).getSeq(17) ==
- Seq("111.222200000000000000",
"333.444400000000000000").map(BigDecimal(_).bigDecimal))
- assert(rows(0).getString(18) == "101010")
- assert(rows(0).getString(19) == "(800,600)")
- assert(rows(0).getString(20) ==
"{-4.359313077939234,-1,159.9516512549538}")
- assert(rows(0).getString(21) == "[(80.12,131.24),(201.5,503.33)]")
- assert(rows(0).getString(22) == "(20.21,11.23),(19.84,2.1)")
- assert(rows(0).getString(23) == "((10.2,30.4),(50.6,70.8),(90.1,11.3))")
- assert(rows(0).getString(24) ==
"((100.3,40.2),(20.198,83.1),(500.821,311.38))")
- assert(rows(0).getString(25) == "<(500,200),100>")
- assert(rows(0).getString(26) == "16/B374D848")
- assert(rows(0).getString(27) == "ab")
- assert(rows(0).getString(28) == "efg")
- assert(rows(0).getDate(29) == new
SimpleDateFormat("yyyy-MM-dd").parse("2021-02-02"))
- assert(rows(0).getString(30) == "00:01:00")
- assert(rows(0).getString(31) == "00:11:22:33:44:55")
- assert(rows(0).getString(32) == "00:11:22:33:44:55:66:77")
- assert(rows(0).getDecimal(33) == new JBigDecimal("12.3456"))
- assert(rows(0).getString(34) == "10:20:10,14,15")
- assert(rows(0).getFloat(35) == 1E+37F)
- assert(rows(0).getTimestamp(36) == Timestamp.valueOf("1970-01-01
17:22:31.123"))
- assert(rows(0).getTimestamp(37) == Timestamp.valueOf("2016-08-12
10:22:31.949271"))
- assert(rows(0).getString(38) == "'cat':AB & 'dog':CD")
- assert(rows(0).getString(39) == "'and' 'cat' 'dog' 'fox'")
- assert(rows(0).getString(40) == "10:20:10,14,15")
- assert(rows(0).getString(41) == "<key>id</key><value>10</value>")
-
- // Test reading null values using the second row.
- assert(0.until(16).forall(rows(1).isNullAt(_)))
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "false") {
+ val df = spark.read.jdbc(jdbcUrl, "bar", new Properties)
+ val rows = df.collect().sortBy(_.toString())
+ assert(rows.length == 2)
+ // Test the types, and values using the first row.
+ val types = rows(0).toSeq.map(x => x.getClass)
+ assert(types.length == 42)
+ assert(classOf[String].isAssignableFrom(types(0)))
+ assert(classOf[java.lang.Integer].isAssignableFrom(types(1)))
+ assert(classOf[java.lang.Double].isAssignableFrom(types(2)))
+ assert(classOf[java.lang.Long].isAssignableFrom(types(3)))
+ assert(classOf[java.lang.Boolean].isAssignableFrom(types(4)))
+ assert(classOf[Array[Byte]].isAssignableFrom(types(5)))
+ assert(classOf[Array[Byte]].isAssignableFrom(types(6)))
+ assert(classOf[java.lang.Boolean].isAssignableFrom(types(7)))
+ assert(classOf[String].isAssignableFrom(types(8)))
+ assert(classOf[String].isAssignableFrom(types(9)))
+ assert(classOf[scala.collection.Seq[Int]].isAssignableFrom(types(10)))
+ assert(classOf[scala.collection.Seq[String]].isAssignableFrom(types(11)))
+ assert(classOf[scala.collection.Seq[Double]].isAssignableFrom(types(12)))
+
assert(classOf[scala.collection.Seq[BigDecimal]].isAssignableFrom(types(13)))
+ assert(classOf[String].isAssignableFrom(types(14)))
+ assert(classOf[java.lang.Float].isAssignableFrom(types(15)))
+ assert(classOf[java.lang.Short].isAssignableFrom(types(16)))
+
assert(classOf[scala.collection.Seq[BigDecimal]].isAssignableFrom(types(17)))
+ assert(classOf[String].isAssignableFrom(types(18)))
+ assert(classOf[String].isAssignableFrom(types(19)))
+ assert(classOf[String].isAssignableFrom(types(20)))
+ assert(classOf[String].isAssignableFrom(types(21)))
+ assert(classOf[String].isAssignableFrom(types(22)))
+ assert(classOf[String].isAssignableFrom(types(23)))
+ assert(classOf[String].isAssignableFrom(types(24)))
+ assert(classOf[String].isAssignableFrom(types(25)))
+ assert(classOf[String].isAssignableFrom(types(26)))
+ assert(classOf[String].isAssignableFrom(types(27)))
+ assert(classOf[String].isAssignableFrom(types(28)))
+ assert(classOf[Date].isAssignableFrom(types(29)))
+ assert(classOf[String].isAssignableFrom(types(30)))
+ assert(classOf[String].isAssignableFrom(types(31)))
+ assert(classOf[String].isAssignableFrom(types(32)))
+ assert(classOf[JBigDecimal].isAssignableFrom(types(33)))
+ assert(classOf[String].isAssignableFrom(types(34)))
+ assert(classOf[java.lang.Float].isAssignableFrom(types(35)))
+ assert(classOf[java.sql.Timestamp].isAssignableFrom(types(36)))
+ assert(classOf[java.sql.Timestamp].isAssignableFrom(types(37)))
+ assert(classOf[String].isAssignableFrom(types(38)))
+ assert(classOf[String].isAssignableFrom(types(39)))
+ assert(classOf[String].isAssignableFrom(types(40)))
+ assert(classOf[String].isAssignableFrom(types(41)))
+ assert(rows(0).getString(0).equals("hello"))
+ assert(rows(0).getInt(1) == 42)
+ assert(rows(0).getDouble(2) == 1.25)
+ assert(rows(0).getLong(3) == 123456789012345L)
+ assert(!rows(0).getBoolean(4))
+ // BIT(10)'s come back as ASCII strings of ten ASCII 0's and 1's...
+ assert(java.util.Arrays.equals(rows(0).getAs[Array[Byte]](5),
+ Array[Byte](49, 48, 48, 48, 49, 48, 48, 49, 48, 49)))
+ assert(java.util.Arrays.equals(rows(0).getAs[Array[Byte]](6),
+ Array[Byte](0xDE.toByte, 0xAD.toByte, 0xBE.toByte, 0xEF.toByte)))
+ assert(rows(0).getBoolean(7))
+ assert(rows(0).getString(8) == "172.16.0.42")
+ assert(rows(0).getString(9) == "192.168.0.0/16")
+ assert(rows(0).getSeq(10) == Seq(1, 2))
+ assert(rows(0).getSeq(11) == Seq("a", null, "b"))
+ assert(rows(0).getSeq(12).toSeq == Seq(0.11f, 0.22f))
+ assert(rows(0).getSeq(13) == Seq("0.11",
"0.22").map(BigDecimal(_).bigDecimal))
+ assert(rows(0).getString(14) == "d1")
+ assert(rows(0).getFloat(15) == 1.01f)
+ assert(rows(0).getShort(16) == 1)
+ assert(rows(0).getSeq(17) ==
+ Seq("111.222200000000000000",
"333.444400000000000000").map(BigDecimal(_).bigDecimal))
+ assert(rows(0).getString(18) == "101010")
+ assert(rows(0).getString(19) == "(800,600)")
+ assert(rows(0).getString(20) ==
"{-4.359313077939234,-1,159.9516512549538}")
+ assert(rows(0).getString(21) == "[(80.12,131.24),(201.5,503.33)]")
+ assert(rows(0).getString(22) == "(20.21,11.23),(19.84,2.1)")
+ assert(rows(0).getString(23) == "((10.2,30.4),(50.6,70.8),(90.1,11.3))")
+ assert(rows(0).getString(24) ==
"((100.3,40.2),(20.198,83.1),(500.821,311.38))")
+ assert(rows(0).getString(25) == "<(500,200),100>")
+ assert(rows(0).getString(26) == "16/B374D848")
+ assert(rows(0).getString(27) == "ab")
+ assert(rows(0).getString(28) == "efg")
+ assert(rows(0).getDate(29) == new
SimpleDateFormat("yyyy-MM-dd").parse("2021-02-02"))
+ assert(rows(0).getString(30) == "00:01:00")
+ assert(rows(0).getString(31) == "00:11:22:33:44:55")
+ assert(rows(0).getString(32) == "00:11:22:33:44:55:66:77")
+ assert(rows(0).getDecimal(33) == new JBigDecimal("12.3456"))
+ assert(rows(0).getString(34) == "10:20:10,14,15")
+ assert(rows(0).getFloat(35) == 1E+37F)
+ assert(rows(0).getTimestamp(36) == Timestamp.valueOf("1970-01-01
17:22:31.123"))
+ assert(rows(0).getTimestamp(37) == Timestamp.valueOf("2016-08-12
10:22:31.949271"))
+ assert(rows(0).getString(38) == "'cat':AB & 'dog':CD")
+ assert(rows(0).getString(39) == "'and' 'cat' 'dog' 'fox'")
+ assert(rows(0).getString(40) == "10:20:10,14,15")
+ assert(rows(0).getString(41) == "<key>id</key><value>10</value>")
+
+ // Test reading null values using the second row.
+ assert(0.until(16).forall(rows(1).isNullAt(_)))
+ }
}
test("Basic write test") {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
index d2d8f7416fe8..1ad38e771208 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
@@ -164,6 +164,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
// Note that some dialects override this setting, e.g. as SQL Server.
case TimestampNTZType => Option(JdbcType("TIMESTAMP",
java.sql.Types.TIMESTAMP))
case DateType => Option(JdbcType("DATE", java.sql.Types.DATE))
+ case t: TimeType => Option(JdbcType(s"TIME(${t.precision})",
java.sql.Types.TIME))
case t: DecimalType => Option(
JdbcType(s"DECIMAL(${t.precision},${t.scale})",
java.sql.Types.DECIMAL))
case _ => None
@@ -227,7 +228,13 @@ object JdbcUtils extends Logging with SQLConfHelper {
case java.sql.Types.SMALLINT => IntegerType
case java.sql.Types.SQLXML => StringType
case java.sql.Types.STRUCT => StringType
- case java.sql.Types.TIME => getTimestampType(isTimestampNTZ)
+ case java.sql.Types.TIME =>
+ if (conf.isTimeTypeEnabled) {
+ // Use reported scale (fractional digits) as precision; TIME(0) is
valid
+ val timePrecision = if (scale >= 0 && scale <= TimeType.MAX_PRECISION)
scale
+ else TimeType.DEFAULT_PRECISION
+ TimeType(timePrecision)
+ } else getTimestampType(isTimestampNTZ)
case java.sql.Types.TIMESTAMP => getTimestampType(isTimestampNTZ)
case java.sql.Types.TINYINT => IntegerType
case java.sql.Types.VARBINARY => BinaryType
@@ -443,6 +450,15 @@ object JdbcUtils extends Logging with SQLConfHelper {
row.update(pos, null)
}
+ case _: TimeType =>
+ (rs: ResultSet, row: InternalRow, pos: Int) =>
+ val localTime = rs.getObject(pos + 1, classOf[java.time.LocalTime])
+ if (localTime != null) {
+ row.setLong(pos, localTime.toNanoOfDay)
+ } else {
+ row.update(pos, null)
+ }
+
// When connecting with Oracle DB through JDBC, the precision and scale of
BigDecimal
// object returned by ResultSet.getBigDecimal is not correctly matched to
the table
// schema reported by ResultSetMetaData.getPrecision and
ResultSetMetaData.getScale.
@@ -720,6 +736,11 @@ object JdbcUtils extends Logging with SQLConfHelper {
stmt.setDate(pos + 1, row.getAs[java.sql.Date](pos))
}
+ case _: TimeType =>
+ (stmt: PreparedStatement, row: Row, pos: Int) =>
+ val localTime = row.getAs[java.time.LocalTime](pos)
+ stmt.setObject(pos + 1, localTime)
+
case t: DecimalType =>
(stmt: PreparedStatement, row: Row, pos: Int) =>
stmt.setBigDecimal(pos + 1, row.getDecimal(pos))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
index 4f70f9a65ae4..71753c1155ce 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
@@ -117,6 +117,7 @@ private case class DB2Dialect() extends JdbcDialect with
SQLConfHelper with NoLe
Option(JdbcType("CHAR(1)", java.sql.Types.CHAR))
case BooleanType => Option(JdbcType("BOOLEAN", java.sql.Types.BOOLEAN))
case ShortType | ByteType => Some(JdbcType("SMALLINT",
java.sql.Types.SMALLINT))
+ case _: TimeType => Some(JdbcType("TIME", java.sql.Types.TIME))
case _ => None
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
b/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
index c221e59d7f92..110889563a2b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
@@ -192,7 +192,8 @@ trait CreatableRelationProvider {
case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
case BinaryType | BooleanType | ByteType | _: CharType | DateType | _:
DecimalType |
DoubleType | FloatType | IntegerType | LongType | NullType |
ObjectType(_) | ShortType |
- _: StringType | TimestampNTZType | TimestampType | _: VarcharType
=> true
+ _: StringType | _: TimeType | TimestampNTZType | TimestampType |
+ _: VarcharType => true
case _ => false
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index e18e84d2a506..4e75e1807cf4 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -714,12 +714,11 @@ class JDBCSuite extends SharedSparkSession {
}
test("H2 time types") {
+ // With timeType.enabled=true (default in tests), TIME columns use TimeType
val rows = sql("SELECT * FROM timetypes").collect()
+ assert(rows(0).getAs[java.time.LocalTime](0) ===
java.time.LocalTime.of(12, 34, 56))
+ // DATE and TIMESTAMP columns unchanged
val cal = new GregorianCalendar(java.util.Locale.ROOT)
- cal.setTime(rows(0).getAs[java.sql.Timestamp](0))
- assert(cal.get(Calendar.HOUR_OF_DAY) === 12)
- assert(cal.get(Calendar.MINUTE) === 34)
- assert(cal.get(Calendar.SECOND) === 56)
cal.setTime(rows(0).getAs[java.sql.Timestamp](1))
assert(cal.get(Calendar.YEAR) === 1996)
assert(cal.get(Calendar.MONTH) === 0)
@@ -736,24 +735,79 @@ class JDBCSuite extends SharedSparkSession {
}
test("SPARK-34357: test TIME types") {
- val rows = spark.read.jdbc(
- urlWithUserAndPass, "TEST.TIMETYPES", new Properties()).collect()
- val cachedRows = spark.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES", new
Properties())
- .cache().collect()
- val expectedTimeAtEpoch = java.sql.Timestamp.valueOf("1970-01-01
12:34:56.0")
- assert(rows(0).getAs[java.sql.Timestamp](0) === expectedTimeAtEpoch)
- assert(rows(1).getAs[java.sql.Timestamp](0) === expectedTimeAtEpoch)
- assert(cachedRows(0).getAs[java.sql.Timestamp](0) === expectedTimeAtEpoch)
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "false") {
+ val rows = spark.read.jdbc(
+ urlWithUserAndPass, "TEST.TIMETYPES", new Properties()).collect()
+ val cachedRows = spark.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES",
new Properties())
+ .cache().collect()
+ val expectedTimeAtEpoch = java.sql.Timestamp.valueOf("1970-01-01
12:34:56.0")
+ assert(rows(0).getAs[java.sql.Timestamp](0) === expectedTimeAtEpoch)
+ assert(rows(1).getAs[java.sql.Timestamp](0) === expectedTimeAtEpoch)
+ assert(cachedRows(0).getAs[java.sql.Timestamp](0) ===
expectedTimeAtEpoch)
+ }
}
test("SPARK-47396: TIME WITHOUT TIME ZONE preferTimestampNTZ") {
- spark.catalog.clearCache()
- val df = spark.read.format("jdbc")
- .option("preferTimestampNTZ", true)
- .option("url", urlWithUserAndPass)
- .option("query", "SELECT A FROM TEST.TIMETYPES limit 1")
- .load()
- assert(df.head().get(0).isInstanceOf[LocalDateTime])
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "false") {
+ spark.catalog.clearCache()
+ val df = spark.read.format("jdbc")
+ .option("preferTimestampNTZ", true)
+ .option("url", urlWithUserAndPass)
+ .option("query", "SELECT A FROM TEST.TIMETYPES limit 1")
+ .load()
+ assert(df.head().get(0).isInstanceOf[LocalDateTime])
+ }
+ }
+
+ test("SPARK-57555: JDBC TIME maps to TimeType when timeType.enabled") {
+ val df = spark.read.jdbc(
+ urlWithUserAndPass, "TEST.TIMETYPES", new Properties())
+ // With timeType.enabled=true (default in tests), TIME column maps to
TimeType
+ assert(df.schema("A").dataType.isInstanceOf[TimeType])
+ val rows = df.collect()
+ assert(rows(0).getAs[java.time.LocalTime](0) ===
java.time.LocalTime.of(12, 34, 56))
+ }
+
+ test("SPARK-57555: JDBC TIME write round-trip") {
+ val url = urlWithUserAndPass
+ val tableName = "TEST.TIME_ROUNDTRIP"
+ val time1 = java.time.LocalTime.of(9, 30, 0)
+ val time2 = java.time.LocalTime.of(23, 59, 59, 123456000)
+ val schema = new StructType().add("t",
TimeType(TimeType.DEFAULT_PRECISION))
+ val rows = Seq(
+ Row(time1),
+ Row(time2)
+ )
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(rows),
schema)
+ df.write.jdbc(url, tableName, new Properties())
+ try {
+ val readBack = spark.read.jdbc(url, tableName, new Properties())
+ assert(readBack.schema.fields(0).dataType.isInstanceOf[TimeType])
+ val result = readBack.orderBy(readBack.columns(0)).collect()
+ assert(result(0).getAs[java.time.LocalTime](0) === time1)
+ assert(result(1).getAs[java.time.LocalTime](0) === time2)
+ } finally {
+ val conn = java.sql.DriverManager.getConnection(url)
+ conn.createStatement().execute(s"DROP TABLE IF EXISTS $tableName")
+ conn.close()
+ }
+ }
+
+ test("SPARK-57555: JDBC TIME preserves sub-second precision") {
+ val conn = java.sql.DriverManager.getConnection(urlWithUserAndPass)
+ try {
+ conn.createStatement().execute(
+ "CREATE TABLE TEST.TIME_PRECISION (t TIME(6))")
+ conn.createStatement().execute(
+ "INSERT INTO TEST.TIME_PRECISION VALUES (TIME '14:30:45.123456')")
+ val df = spark.read.jdbc(urlWithUserAndPass, "TEST.TIME_PRECISION", new
Properties())
+ val result = df.collect()
+ assert(result(0).getAs[java.time.LocalTime](0) ===
+ java.time.LocalTime.of(14, 30, 45, 123456000))
+ } finally {
+ conn.createStatement().execute("DROP TABLE IF EXISTS
TEST.TIME_PRECISION")
+ conn.close()
+ }
}
test("test DATE types") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
