You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/09/03 17:49:44 UTC
(spark) branch branch-4.x updated: [SPARK-58876][SQL] Map Oracle DATE and TIMESTAMP to TimestampNTZType
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 f15124783d49 [SPARK-58876][SQL] Map Oracle DATE and TIMESTAMP to TimestampNTZType
f15124783d49 is described below
commit f15124783d4937139f68aa9a4bd32423ca6b01b8
Author: aleksandar-trajkovic-db <al...@databricks.com>
AuthorDate: Thu Sep 3 17:49:21 2026 +0000
[SPARK-58876][SQL] Map Oracle DATE and TIMESTAMP to TimestampNTZType
### What changes were proposed in this pull request?
The Oracle JDBC dialect maps Oracle `DATE` and `TIMESTAMP` to `TimestampNTZType` by default, independent of the `preferTimestampNTZ` read option (`DATE` only when the driver default `oracle.jdbc.mapDateToTimestamp=true` surfaces it as `TIMESTAMP`; otherwise it stays `DateType`). `TIMESTAMP WITH TIME ZONE` and `TIMESTAMP WITH LOCAL TIME ZONE` are unchanged (`TimestampType`). A new `spark.sql.legacy.oracle.timestampNTZMapping.enabled` (default `false`, `SESSION` binding policy) restores [...]
Because these Oracle types are zoneless wall-clock values, the read and write conversions must not route through `java.sql.Timestamp`, which rebases through the JVM default zone. Rather than override the dialect `convert*` methods — which `AggregatedDialect` does not delegate — the dialect stamps field-metadata markers that the framework's getter/setter selection honors:
- `getCatalystType` stamps a read marker; `JdbcUtils.makeGetter` then reads the column via `getObject(LocalDateTime)`, which is zone-independent (and exact for a value that falls in a JVM DST gap).
- A new delegated `JdbcDialect.updateExtraColumnMetaForWrite` hook stamps a write marker; `JdbcUtils.makeSetter` then writes via `setObject(LocalDateTime)`.
- Read and write use distinct marker keys, so a read-origin marker carried on a DataFrame's schema cannot force wall-clock writes on a later write to a different dialect.
- `AggregatedDialect` delegates the new write hook to its members (as it already does for `getCatalystType` / `getJDBCType`), so the mapping stays correct when `OracleDialect` is wrapped.
`compileValue` also gains a `LocalDateTime` case so NTZ filter literals push down as JDBC `{ts ...}` escapes.
### Why are the changes needed?
Oracle `DATE` and `TIMESTAMP` are zoneless wall-clock types. Reading them as `TimestampType` (an instant) rebases the value through the session time zone, so the same stored value reads back differently depending on the reader's zone. `TimestampNTZType` represents these types faithfully and zone-invariantly. Driving the wall-clock read/write through metadata markers (rather than `convert*` overrides) keeps the behavior correct even when `OracleDialect` is wrapped in an `AggregatedDial [...]
### Does this PR introduce _any_ user-facing change?
Yes. Oracle `TIMESTAMP` (and `DATE` under the driver default `oracle.jdbc.mapDateToTimestamp=true`) now reads as `TimestampNTZType` instead of `TimestampType`. Set `spark.sql.legacy.oracle.timestampNTZMapping.enabled=true` to restore the previous behavior. Documented in the SQL migration guide.
### How was this patch tested?
New/updated unit tests in `JDBCSuite`: the `DATE`/`TIMESTAMP` -> `TimestampNTZType` mapping and read-marker stamping (directly and wrapped in `AggregatedDialect`, in both member orders, legacy-gated), the write-marker stamping (likewise), a marked column reading wall-clock regardless of the resolved dialect, the `compileValue` literal, and that the mapping stays microsecond `TimestampNTZType` under the nanosecond-timestamp preview. New/updated `OracleIntegrationSuite` docker tests: NT [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Claude Opus 4.8)
This is my original work and I license it to the project under the project's open source license.
Closes #58203 from aleksandar-trajkovic-db/spark-oracle-ntz-mapping.
Authored-by: aleksandar-trajkovic-db <al...@databricks.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
(cherry picked from commit 59f6ffa4b7eaab8f3339f72db70f11bd378af36d)
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../spark/sql/jdbc/OracleIntegrationSuite.scala | 323 ++++++++++++++++++---
docs/sql-migration-guide.md | 1 +
.../org/apache/spark/sql/internal/SQLConf.scala | 23 ++
.../datasources/jdbc/JDBCValueGetter.scala | 11 +
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 21 +-
.../apache/spark/sql/jdbc/AggregatedDialect.scala | 4 +
.../org/apache/spark/sql/jdbc/JdbcDialects.scala | 9 +
.../org/apache/spark/sql/jdbc/OracleDialect.scala | 20 +-
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 160 +++++++++-
9 files changed, 526 insertions(+), 46 deletions(-)
diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala
index b76307a59d9c..c51736fb28b3 100644
--- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala
+++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala
@@ -19,15 +19,16 @@ package org.apache.spark.sql.jdbc
import java.math.BigDecimal
import java.sql.{Connection, Date, Timestamp}
-import java.time.{Duration, Period}
+import java.time.{Duration, LocalDateTime, Period}
import java.util.{Properties, TimeZone}
import org.apache.spark.sql.{DataFrame, Row, SaveMode}
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.{RowDataSourceScanExec, WholeStageCodegenExec}
import org.apache.spark.sql.execution.datasources.LogicalRelationWithTable
-import org.apache.spark.sql.execution.datasources.jdbc.{JDBCPartition, JDBCRelation}
+import org.apache.spark.sql.execution.datasources.jdbc.{JDBCPartition, JDBCRelation, JdbcUtils}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
@@ -150,6 +151,14 @@ class OracleIntegrationSuite extends SharedJDBCIntegrationSuite
""".stripMargin.replaceAll("\n", " ")).executeUpdate()
conn.commit()
+ // Single-row table with a sub-second (microsecond) TIMESTAMP for NTZ precision tests.
+ conn.prepareStatement("CREATE TABLE datetimeFraction (id NUMBER(10), t TIMESTAMP)")
+ .executeUpdate()
+ conn.prepareStatement(
+ "INSERT INTO datetimeFraction VALUES (1, {ts '1996-01-01 01:23:45.123456'})")
+ .executeUpdate()
+ conn.commit()
+
conn.prepareStatement("CREATE TABLE test_ltz(t TIMESTAMP WITH LOCAL TIME ZONE)")
.executeUpdate()
conn.prepareStatement(
@@ -285,7 +294,7 @@ class OracleIntegrationSuite extends SharedJDBCIntegrationSuite
def checkRow(row: Row): Unit = {
assert(row.getDecimal(0).equals(BigDecimal.valueOf(1)))
assert(row.getDate(1).equals(Date.valueOf("1991-11-09")))
- assert(row.getTimestamp(2).equals(Timestamp.valueOf("1996-01-01 01:23:45")))
+ assert(row.get(2) === LocalDateTime.of(1996, 1, 1, 1, 23, 45))
}
checkRow(sql("SELECT * FROM datetime where id = 1").head())
sql("INSERT INTO TABLE datetime1 SELECT * FROM datetime where id = 1")
@@ -437,7 +446,7 @@ class OracleIntegrationSuite extends SharedJDBCIntegrationSuite
(3, "2018-07-08", "2018-07-08 13:32:01"),
(4, "2018-07-12", "2018-07-12 09:51:15")
).map { case (id, date, timestamp) =>
- Row(BigDecimal.valueOf(id), Date.valueOf(date), Timestamp.valueOf(timestamp))
+ Row(BigDecimal.valueOf(id), Date.valueOf(date), Timestamp.valueOf(timestamp).toLocalDateTime)
}
// DateType partition column
@@ -497,24 +506,28 @@ class OracleIntegrationSuite extends SharedJDBCIntegrationSuite
}
val query = "SELECT id, d, t FROM datetime WHERE id = 1"
- // query option to pass on the query string.
- val df = spark.read.format("jdbc")
- .option("url", jdbcUrl)
- .option("query", query)
- .option("oracle.jdbc.mapDateToTimestamp", "false")
- .load()
- assert(df.collect().toSet === expectedResult)
-
- // query option in the create table path.
- sql(
- s"""
- |CREATE OR REPLACE TEMPORARY VIEW queryOption
- |USING org.apache.spark.sql.jdbc
- |OPTIONS (url '$jdbcUrl',
- | query '$query',
- | oracle.jdbc.mapDateToTimestamp false)
- """.stripMargin.replaceAll("\n", " "))
- assert(sql("select id, d, t from queryOption").collect().toSet == expectedResult)
+ // Keep t (Oracle TIMESTAMP) mapped to TimestampType so the expected Timestamp rows match;
+ // the NTZ mapping is exercised by the dedicated tests below.
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ // query option to pass on the query string.
+ val df = spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("query", query)
+ .option("oracle.jdbc.mapDateToTimestamp", "false")
+ .load()
+ assert(df.collect().toSet === expectedResult)
+
+ // query option in the create table path.
+ sql(
+ s"""
+ |CREATE OR REPLACE TEMPORARY VIEW queryOption
+ |USING org.apache.spark.sql.jdbc
+ |OPTIONS (url '$jdbcUrl',
+ | query '$query',
+ | oracle.jdbc.mapDateToTimestamp false)
+ """.stripMargin.replaceAll("\n", " "))
+ assert(sql("select id, d, t from queryOption").collect().toSet == expectedResult)
+ }
}
test("SPARK-32992: map Oracle's ROWID type to StringType") {
@@ -544,27 +557,255 @@ class OracleIntegrationSuite extends SharedJDBCIntegrationSuite
}
test("SPARK-42627: Support ORACLE TIMESTAMP WITH LOCAL TIME ZONE") {
+ // flag="true" round-trips through a plain Oracle TIMESTAMP column, which now reads as
+ // TimestampNTZType; pin the pre-4.4 read mapping since this test covers only the write mapping.
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ Seq("true", "false").foreach { flag =>
+ withSQLConf((SQLConf.LEGACY_ORACLE_TIMESTAMP_MAPPING_ENABLED.key, flag)) {
+ val df = spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "test_ltz")
+ .load()
+ val row1 = df.collect().head.getTimestamp(0)
+ assert(df.count() === 1)
+ assert(row1 === Timestamp.valueOf("2018-11-17 13:33:33"))
+
+ df.write.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "test_ltz" + flag)
+ .save()
+
+ val df2 = spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "test_ltz" + flag)
+ .load()
+ checkAnswer(df2, Row(row1))
+ }
+ }
+ }
+ }
+
+ test("Oracle DATE and TIMESTAMP read as TimestampNTZType by default; " +
+ "TimestampType under the legacy flag") {
+ // Read the `datetime` table's DATE column D (1991-11-09, no time zone) and TIMESTAMP column T
+ // (1996-01-01 01:23:45) in the driver's default mode (oracle.jdbc.mapDateToTimestamp=true), so
+ // both arrive under JDBC Types.TIMESTAMP.
+ def readDatetime(): DataFrame =
+ spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "datetime")
+ .load()
+
+ // Default: both Oracle DATE and TIMESTAMP -> TimestampNTZType.
+ val df = readDatetime()
+ assert(df.schema("D").dataType === TimestampNTZType)
+ assert(df.schema("T").dataType === TimestampNTZType)
+ val row = df.select("D", "T").collect().head
+ assert(row.get(0) === LocalDateTime.of(1991, 11, 9, 0, 0, 0))
+ assert(row.get(1) === LocalDateTime.of(1996, 1, 1, 1, 23, 45))
+
+ // Sub-second precision: a TIMESTAMP(6) microsecond value round-trips as NTZ without truncation.
+ val fracDf = spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "datetimeFraction")
+ .load()
+ assert(fracDf.schema("T").dataType === TimestampNTZType)
+ assert(fracDf.select("T").collect().head.get(0)
+ === LocalDateTime.of(1996, 1, 1, 1, 23, 45, 123456000))
+
+ // Legacy flag on: both fall back to TimestampType.
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ val legacyDf = readDatetime()
+ assert(legacyDf.schema("D").dataType === TimestampType)
+ assert(legacyDf.schema("T").dataType === TimestampType)
+ val legacyRow = legacyDf.select("D", "T").collect().head
+ assert(legacyRow.getTimestamp(0) === Timestamp.valueOf("1991-11-09 00:00:00"))
+ assert(legacyRow.getTimestamp(1) === Timestamp.valueOf("1996-01-01 01:23:45"))
+ }
+ }
+
+ // Reads the `datetime` table and asserts both columns come back as TimestampNTZType with their
+ // exact zoneless wall-clock values. `readerOptions` lets each test toggle one timezone mechanism.
+ private def assertDatetimeReadsAsNtz(readerOptions: Map[String, String] = Map.empty): Unit = {
+ val df = spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "datetime")
+ .options(readerOptions)
+ .load()
+ assert(df.schema("D").dataType === TimestampNTZType)
+ assert(df.schema("T").dataType === TimestampNTZType)
+ val row = df.select("D", "T").collect().head
+ assert(row.get(0) === LocalDateTime.of(1991, 11, 9, 0, 0, 0))
+ assert(row.get(1) === LocalDateTime.of(1996, 1, 1, 1, 23, 45))
+ }
+
+ // Each test flips one timezone mechanism and asserts the zoneless DATE/TIMESTAMP -> NTZ read is
+ // unchanged. The fixed values are off any DST boundary, so the read is deterministic.
+
+ test("NTZ read is invariant to the JVM default time zone") {
+ Seq(UTC, PST, LA).foreach { zone =>
+ withDefaultTimeZone(zone) {
+ assertDatetimeReadsAsNtz()
+ }
+ }
+ }
+
+ test("NTZ read is invariant to spark.sql.session.timeZone") {
+ Seq("UTC", "America/Los_Angeles", "Asia/Kolkata").foreach { tz =>
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
+ assertDatetimeReadsAsNtz()
+ }
+ }
+ }
+
+ test("NTZ read is invariant to the Oracle session TIME_ZONE (sessionInitStatement)") {
+ Seq("+00:00", "-08:00", "+05:30").foreach { tz =>
+ assertDatetimeReadsAsNtz(
+ Map("sessionInitStatement" -> s"ALTER SESSION SET TIME_ZONE='$tz'"))
+ }
+ }
+
+ test("NTZ read is invariant to oracle.jdbc.timezoneAsRegion") {
Seq("true", "false").foreach { flag =>
- withSQLConf((SQLConf.LEGACY_ORACLE_TIMESTAMP_MAPPING_ENABLED.key, flag)) {
- val df = spark.read.format("jdbc")
- .option("url", jdbcUrl)
- .option("dbtable", "test_ltz")
- .load()
- val row1 = df.collect().head.getTimestamp(0)
- assert(df.count() === 1)
- assert(row1 === Timestamp.valueOf("2018-11-17 13:33:33"))
-
- df.write.format("jdbc")
- .option("url", jdbcUrl)
- .option("dbtable", "test_ltz" + flag)
- .save()
-
- val df2 = spark.read.format("jdbc")
- .option("url", jdbcUrl)
- .option("dbtable", "test_ltz" + flag)
- .load()
- checkAnswer(df2, Row(row1))
+ assertDatetimeReadsAsNtz(Map("oracle.jdbc.timezoneAsRegion" -> flag))
+ }
+ }
+
+ test("NTZ read is invariant to NLS date/timestamp formats") {
+ Seq(
+ "ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS'",
+ "ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF'"
+ ).foreach { initStmt =>
+ assertDatetimeReadsAsNtz(Map("sessionInitStatement" -> initStmt))
+ }
+ }
+
+ // How NTZ-mapped Oracle DATE (D) and TIMESTAMP (T) behave as WHERE predicates, using the 4-row
+ // datetimePartitionTest table. pushDown=false forces Spark-side evaluation (on the materialized
+ // zoneless LocalDateTime) instead of pushing to Oracle, so the two paths can be compared.
+ private def filteredIds(where: String, pushDown: Boolean = true): Set[Long] = {
+ spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "datetimePartitionTest")
+ .option("pushDownPredicate", pushDown.toString)
+ .load()
+ .where(where)
+ .selectExpr("cast(id as long) as id")
+ .collect().map(_.getLong(0)).toSet
+ }
+
+ test("WHERE on NTZ columns filters correctly, with pushdown on and off") {
+ // (predicate, matching ids). DATE has no time component. Each case is checked with pushdown on
+ // (pushed to Oracle) and off (Spark-side eval on the materialized LocalDateTime); both agree.
+ val cases = Seq(
+ "D = TIMESTAMP_NTZ'2018-07-08 00:00:00'" -> Set(3L),
+ // ids 1,2 have D = 2018-07-06 00:00:00 (equal, not greater); ids 3,4 are strictly after.
+ "D > TIMESTAMP_NTZ'2018-07-06 00:00:00'" -> Set(3L, 4L),
+ "T = TIMESTAMP_NTZ'2018-07-06 08:10:08'" -> Set(2L),
+ "T > TIMESTAMP_NTZ'2018-07-06 06:00:00' AND T < TIMESTAMP_NTZ'2018-07-09 00:00:00'" ->
+ Set(2L, 3L),
+ "T > TIMESTAMP_NTZ'2018-07-06 06:00:00'" -> Set(2L, 3L, 4L))
+ cases.foreach { case (where, expected) =>
+ assert(filteredIds(where, pushDown = true) === expected, s"pushdown on: $where")
+ assert(filteredIds(where, pushDown = false) === expected, s"pushdown off: $where")
+ }
+ }
+
+ test("Predicates on NTZ-mapped Oracle columns are pushed down to Oracle") {
+ val df = spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "datetimePartitionTest")
+ .load()
+ .where("D > TIMESTAMP_NTZ'2018-07-06 00:00:00' AND " +
+ "T > TIMESTAMP_NTZ'2018-07-06 06:00:00'")
+ val scan = df.queryExecution.executedPlan.collectFirst {
+ case r: RowDataSourceScanExec => r
+ }.getOrElse(fail("Expected a RowDataSourceScanExec in the plan"))
+ val pushed = scan.metadata.getOrElse("PushedFilters", "")
+ assert(pushed.contains("GreaterThan(D,2018-07-06T00:00)"),
+ s"NTZ DATE predicate was not pushed down; PushedFilters=$pushed")
+ assert(pushed.contains("GreaterThan(T,2018-07-06T06:00)"),
+ s"NTZ TIMESTAMP predicate was not pushed down; PushedFilters=$pushed")
+ assert(df.count() === 2)
+ }
+
+ test("TimestampNTZType round-trips through an Oracle write and read") {
+ // The zoneless NTZ wall-clock (incl. sub-second) must survive a write+read regardless of the
+ // JVM default zone, since getObject/setObject(LocalDateTime) bypass the java.sql.Timestamp zone
+ // bridge. Each (zone, value) case is written and read back and must equal the original.
+ val cases = Seq(
+ (UTC, LocalDateTime.of(1996, 1, 1, 1, 23, 45, 123456000)),
+ (LA, LocalDateTime.of(1996, 1, 1, 1, 23, 45, 123456000)),
+ // 2024-03-10 02:30 has no instant in America/Los_Angeles (spring-forward DST gap); the
+ // java.sql.Timestamp path would shift it an hour, getObject/setObject keep it exact.
+ (LA, LocalDateTime.of(2024, 3, 10, 2, 30, 0)))
+ val schema = StructType(Seq(StructField("T", TimestampNTZType)))
+ cases.foreach { case (zone, ldt) =>
+ withDefaultTimeZone(zone) {
+ spark.createDataFrame(spark.sparkContext.parallelize(Seq(Row(ldt))), schema)
+ .write.format("jdbc").mode(SaveMode.Overwrite)
+ .option("url", jdbcUrl).option("dbtable", "ntz_write_roundtrip").save()
+ val dfRead = spark.read.format("jdbc")
+ .option("url", jdbcUrl).option("dbtable", "ntz_write_roundtrip").load()
+ assert(dfRead.schema.fields.head.dataType === TimestampNTZType, s"zone=$zone ldt=$ldt")
+ assert(dfRead.collect().head.get(0) === ldt, s"zone=$zone ldt=$ldt")
+ }
+ }
+ }
+
+ test("legacy flag writes TimestampNTZ using the base conversion") {
+ // Legacy write uses the base conversion, so in a non-UTC zone the default wall-clock reader
+ // sees base(ldt).toLocalDateTime, not ldt (a wall-clock write would round-trip to ldt here).
+ val ldt = LocalDateTime.of(1996, 1, 1, 1, 23, 45, 123456000)
+ val schema = StructType(Seq(StructField("T", TimestampNTZType)))
+ withDefaultTimeZone(LA) {
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ spark.createDataFrame(spark.sparkContext.parallelize(Seq(Row(ldt))), schema)
+ .write.format("jdbc").mode(SaveMode.Overwrite)
+ .option("url", jdbcUrl).option("dbtable", "ntz_legacy_write").save()
+ }
+ val expected = DateTimeUtils.toJavaTimestampNoRebase(
+ DateTimeUtils.localDateTimeToMicros(ldt)).toLocalDateTime
+ assert(expected != ldt) // guard: LA must shift base vs wall-clock for this to discriminate
+ val df = spark.read.format("jdbc")
+ .option("url", jdbcUrl).option("dbtable", "ntz_legacy_write").load()
+ assert(df.schema.fields.head.dataType === TimestampNTZType)
+ assert(df.collect().head.get(0) === expected)
+ }
+ }
+
+ test("SPARK-58876: an NTZ read marker does not leak into a later write under the legacy flag") {
+ // Read an Oracle TIMESTAMP as NTZ (legacy off) so the read DataFrame's schema carries the read
+ // marker; then flip legacy on and write it back. The write is gated by the WRITE marker (not
+ // stamped under legacy), so despite the stale read marker it takes the legacy base conversion,
+ // not the wall-clock setter -- the read marker cannot force a wall-clock write.
+ val ldt = LocalDateTime.of(1996, 1, 1, 1, 23, 45, 123456000)
+ val schema = StructType(Seq(StructField("T", TimestampNTZType)))
+ withDefaultTimeZone(LA) {
+ // Seed a source table, then read it back as NTZ so the read marker rides on df.schema.
+ spark.createDataFrame(spark.sparkContext.parallelize(Seq(Row(ldt))), schema)
+ .write.format("jdbc").mode(SaveMode.Overwrite)
+ .option("url", jdbcUrl).option("dbtable", "ntz_leak_src").save()
+ val readDf = spark.read.format("jdbc")
+ .option("url", jdbcUrl).option("dbtable", "ntz_leak_src").load()
+ val readField = readDf.schema.fields.head
+ assert(readField.dataType === TimestampNTZType)
+ assert(readField.metadata.contains(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK))
+ assert(readDf.collect().head.get(0) === ldt) // wall-clock read round-trips exactly
+
+ // Flip legacy on and write the read DataFrame (which still carries the read marker).
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ readDf.write.format("jdbc").mode(SaveMode.Overwrite)
+ .option("url", jdbcUrl).option("dbtable", "ntz_leak_dst").save()
}
+ // A wall-clock write would have stored ldt exactly; the legacy base conversion instead shifts
+ // it by the zone offset in LA, so the default wall-clock reader sees base(ldt), not ldt.
+ val expected = DateTimeUtils.toJavaTimestampNoRebase(
+ DateTimeUtils.localDateTimeToMicros(ldt)).toLocalDateTime
+ assert(expected != ldt) // guard: LA must shift base vs wall-clock for this to discriminate
+ val dstDf = spark.read.format("jdbc")
+ .option("url", jdbcUrl).option("dbtable", "ntz_leak_dst").load()
+ assert(dstDf.collect().head.get(0) === expected)
}
}
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index e9d199ec277f..1fe513d5941b 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -25,6 +25,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
- Since Spark 4.4, the optimizer rule `SimplifyCaseConversionExpressions` no longer simplifies mixed case conversions `LOWER(UPPER(str))` to `LOWER(str)` or `UPPER(LOWER(str))` to `UPPER(str)`. This preserves standard Unicode case mapping semantics for non-ASCII characters such as Latin small letter dotless i (`ı` U+0131) and micro sign (`µ` U+00B5).
+- Since Spark 4.4, the Oracle JDBC datasource maps Oracle DATE and TIMESTAMP to TimestampNTZType unconditionally, ignoring the JDBC read option `preferTimestampNTZ` (DATE only when the driver default `oracle.jdbc.mapDateToTimestamp=true` surfaces it as TIMESTAMP; otherwise DATE stays DateType). In 4.3 and earlier these were read as TimestampType, or as TimestampNTZType only when `preferTimestampNTZ=true`. This faithfully represents these zoneless Oracle types. Unaffected: Oracle DATE rea [...]
- Since Spark 4.4, for storage-partitioned joins, `spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be covered by some partition key instead of matching the partition keys positionally. As a result, a join-key column partitioned by more than one transform no longer prevents shuffle elimination, and `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false` w [...]
## Upgrading from Spark SQL 4.2 to 4.3
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index de8faf49b59b..6901b587fccc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -6928,6 +6928,26 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED =
+ buildConf("spark.sql.legacy.oracle.timestampNTZMapping.enabled")
+ .internal()
+ .doc("When true, Oracle TIMESTAMP (and Oracle DATE when the driver default " +
+ "oracle.jdbc.mapDateToTimestamp surfaces it as TIMESTAMP) is read per the JDBC read " +
+ "option preferTimestampNTZ (TimestampType by default), preserving pre-Spark-4.4 " +
+ "behavior. When false (default), it is read as TimestampNTZType, which faithfully " +
+ "represents these zoneless Oracle types. The same flag governs the write path: when " +
+ "false a TimestampNTZType column is written to Oracle zoneless via setObject, and when " +
+ "true via a JVM-default-zone java.sql.Timestamp; these can differ for wall-clocks in a " +
+ "DST gap. The flag is read at schema-resolution and write time, so a JDBC relation " +
+ "whose schema was already resolved (e.g. a cached or metastore-registered table) must " +
+ "be re-resolved for a change to take effect. Oracle DATE read as JDBC DATE " +
+ "(oracle.jdbc.mapDateToTimestamp=false, mapped to DateType), TIMESTAMP WITH TIME ZONE, " +
+ "and TIMESTAMP WITH LOCAL TIME ZONE are unaffected.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
+
val LEGACY_DB2_TIMESTAMP_MAPPING_ENABLED =
buildConf("spark.sql.legacy.db2.numericMapping.enabled")
.internal()
@@ -9120,6 +9140,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def legacyOracleTimestampMappingEnabled: Boolean =
getConf(LEGACY_ORACLE_TIMESTAMP_MAPPING_ENABLED)
+ def legacyOracleTimestampNTZMappingEnabled: Boolean =
+ getConf(LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED)
+
def legacyDB2numericMappingEnabled: Boolean =
getConf(LEGACY_DB2_TIMESTAMP_MAPPING_ENABLED)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala
index 3461bcf493e4..64debbd43a31 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala
@@ -189,6 +189,17 @@ private[jdbc] object JDBCValueGetter {
}
}
+ case object TimestampNTZWallClockGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val localDateTime = rs.getObject(pos + 1, classOf[java.time.LocalDateTime])
+ if (localDateTime != null) {
+ row.setLong(pos, localDateTimeToMicros(localDateTime))
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
// Reads a driver TIMESTAMP into a nanosecond-precision local date-time. Uses
// `getObject(LocalDateTime)` to fetch the stored wall-clock directly (mirroring `TimeGetter`),
// which preserves the full sub-microsecond fraction and, being time-zone independent, avoids the
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 ccec597e6e0b..b513490b8c0b 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
@@ -55,6 +55,11 @@ import org.apache.spark.util.ArrayImplicits._
*/
object JdbcUtils extends Logging with SQLConfHelper {
+ // Marks an NTZ column to read (resp. write) as wall-clock. Distinct keys so a read marker on the
+ // resolved schema can't force wall-clock on a later write to a different dialect.
+ private[sql] val READ_TIMESTAMP_NTZ_WALL_CLOCK = "read_timestamp_ntz_wall_clock"
+ private[sql] val WRITE_TIMESTAMP_NTZ_WALL_CLOCK = "write_timestamp_ntz_wall_clock"
+
/**
* Returns true if the table already exists in the JDBC database.
*/
@@ -471,6 +476,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
case TimestampType => JDBCValueGetter.TimestampGetter(dialect)
case TimestampNTZType if metadata.contains("logical_time_type") =>
JDBCValueGetter.LogicalTimeNTZGetter(dialect)
+ case TimestampNTZType if metadata.contains(READ_TIMESTAMP_NTZ_WALL_CLOCK) =>
+ JDBCValueGetter.TimestampNTZWallClockGetter
case TimestampNTZType => JDBCValueGetter.TimestampNTZGetter(dialect)
case t: TimestampNTZNanosType => JDBCValueGetter.TimestampNTZNanosGetter(t.precision)
case t: TimestampLTZNanosType => JDBCValueGetter.TimestampLTZNanosGetter(dialect, t.precision)
@@ -493,7 +500,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
private def makeSetter(
conn: Connection,
dialect: JdbcDialect,
- dataType: DataType): JDBCValueSetter = dataType match {
+ dataType: DataType,
+ metadata: Metadata): JDBCValueSetter = dataType match {
case IntegerType =>
(stmt: PreparedStatement, row: Row, pos: Int) =>
stmt.setInt(pos + 1, row.getInt(pos))
@@ -539,6 +547,9 @@ object JdbcUtils extends Logging with SQLConfHelper {
stmt.setTimestamp(pos + 1, row.getAs[java.sql.Timestamp](pos))
}
+ case TimestampNTZType if metadata.contains(WRITE_TIMESTAMP_NTZ_WALL_CLOCK) =>
+ (stmt: PreparedStatement, row: Row, pos: Int) =>
+ stmt.setObject(pos + 1, row.getAs[java.time.LocalDateTime](pos))
case TimestampNTZType =>
(stmt: PreparedStatement, row: Row, pos: Int) =>
stmt.setTimestamp(pos + 1,
@@ -710,7 +721,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
conn.setTransactionIsolation(finalIsolationLevel)
}
val stmt = conn.prepareStatement(insertStmt)
- val setters = rddSchema.fields.map(f => makeSetter(conn, dialect, f.dataType))
+ val setters = rddSchema.fields.map(f => makeSetter(conn, dialect, f.dataType, f.metadata))
val nullTypes = rddSchema.fields.map(f => getJdbcType(f.dataType, dialect).jdbcNullType)
val numFields = rddSchema.fields.length
@@ -907,7 +918,11 @@ object JdbcUtils extends Logging with SQLConfHelper {
val url = options.url
val table = options.table
val dialect = JdbcDialects.get(url)
- val rddSchema = df.schema
+ val rddSchema = StructType(df.schema.map { field =>
+ val builder = new MetadataBuilder().withMetadata(field.metadata)
+ dialect.updateExtraColumnMetaForWrite(field.dataType, builder)
+ field.copy(metadata = builder.build())
+ })
val batchSize = options.batchSize
val isolationLevel = options.isolationLevel
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala
index 98486701657c..fe503d97cb60 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala
@@ -46,6 +46,10 @@ private class AggregatedDialect(dialects: List[JdbcDialect])
dialects.flatMap(_.getJDBCType(dt)).headOption
}
+ override def updateExtraColumnMetaForWrite(dt: DataType, metadata: MetadataBuilder): Unit = {
+ dialects.foreach(_.updateExtraColumnMetaForWrite(dt, metadata))
+ }
+
override def quoteIdentifier(colName: String): String = {
dialects.head.quoteIdentifier(colName)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
index b20faf52300e..541a1b0d6dc1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
@@ -953,6 +953,15 @@ abstract class JdbcDialect extends Serializable with Logging {
rsmd: ResultSetMetaData,
columnIdx: Int,
metadata: MetadataBuilder): Unit = {}
+
+ /**
+ * Set extra column metadata for the given column on the write path.
+ *
+ * @param dt The Catalyst type of the column being written.
+ * @param metadata The metadata builder to store the extra column information.
+ */
+ @Since("4.4.0")
+ def updateExtraColumnMetaForWrite(dt: DataType, metadata: MetadataBuilder): Unit = {}
}
/**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
index 46080dd57b1d..856808258d1f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.jdbc
import java.sql.{Date, SQLException, Timestamp, Types}
+import java.time.LocalDateTime
import java.util.Locale
import scala.util.control.NonFatal
@@ -26,7 +27,7 @@ import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.connector.expressions.{Expression, Extract, Literal}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
+import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
import org.apache.spark.sql.jdbc.OracleDialect._
import org.apache.spark.sql.types._
@@ -181,10 +182,26 @@ private case class OracleDialect() extends JdbcDialect with SQLConfHelper with N
case BINARY_DOUBLE => Some(DoubleType) // Value for OracleTypes.BINARY_DOUBLE
case INTERVAL_YM => Some(YearMonthIntervalType())
case INTERVAL_DS => Some(DayTimeIntervalType())
+ case Types.TIMESTAMP if !conf.legacyOracleTimestampNTZMappingEnabled && typeName != null &&
+ typeName.toUpperCase(Locale.ROOT).matches("DATE|TIMESTAMP") =>
+ // Oracle DATE and TIMESTAMP are zoneless; map to NTZ and mark it wall-clock so a later flag
+ // flip can't desync the read. TZ/LTZ variants are handled above.
+ if (md != null) md.putBoolean(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK, value = true)
+ // TODO: map sub-microsecond TIMESTAMP(7-9) to TimestampNTZNanosType when the nanosecond
+ // timestamp preview is enabled, instead of truncating to microsecond TimestampNTZType.
+ Some(TimestampNTZType)
case _ => None
}
}
+ override def updateExtraColumnMetaForWrite(dt: DataType, metadata: MetadataBuilder): Unit = {
+ dt match {
+ case TimestampNTZType if !conf.legacyOracleTimestampNTZMappingEnabled =>
+ metadata.putBoolean(JdbcUtils.WRITE_TIMESTAMP_NTZ_WALL_CLOCK, value = true)
+ case _ =>
+ }
+ }
+
override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
// For more details, please see
// https://docs.oracle.com/cd/E19501-01/819-3659/gcmaz/
@@ -210,6 +227,7 @@ private case class OracleDialect() extends JdbcDialect with SQLConfHelper with N
// Appendix A Reference Information.
case stringValue: String => s"'${escapeSql(stringValue)}'"
case timestampValue: Timestamp => "{ts '" + timestampValue + "'}"
+ case localDateTimeValue: LocalDateTime => "{ts '" + Timestamp.valueOf(localDateTimeValue) + "'}"
case dateValue: Date => "{d '" + dateValue + "'}"
case arrayValue: Array[Any] => arrayValue.map(compileValue).mkString(", ")
case binaryValue: Array[Byte] =>
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 7105208ecfdb..b1114194f9c7 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
@@ -32,11 +32,12 @@ import org.mockito.ArgumentMatchers._
import org.mockito.Mockito._
import org.apache.spark.{SparkException, SparkIllegalArgumentException, SparkSQLException}
+import org.apache.spark.executor.InputMetrics
import org.apache.spark.sql.{AnalysisException, DataFrame, Observation, Row}
import org.apache.spark.sql.catalyst.{analysis, TableIdentifier}
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.ShowCreateTable
-import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, DateTimeTestUtils}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, DateTimeTestUtils, DateTimeUtils}
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression => V2Expression, FieldReference, GeneralScalarExpression, LiteralValue}
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, AlwaysTrue, Predicate}
@@ -1682,6 +1683,163 @@ class JDBCSuite extends SharedSparkSession {
Some(TimestampType))
}
+ test("SPARK-58876: Oracle stamps the NTZ wall-clock write marker, legacy-gated, even wrapped") {
+ val custom = new JdbcDialect {
+ override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle")
+ override def getCatalystType(
+ sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = None
+ }
+ val dialects = Seq[JdbcDialect](
+ OracleDialect(),
+ new AggregatedDialect(List(custom, OracleDialect())),
+ new AggregatedDialect(List(OracleDialect(), custom)))
+ dialects.foreach { dialect =>
+ val ntzMd = new MetadataBuilder()
+ dialect.updateExtraColumnMetaForWrite(TimestampNTZType, ntzMd)
+ assert(ntzMd.build().contains(JdbcUtils.WRITE_TIMESTAMP_NTZ_WALL_CLOCK), s"dialect=$dialect")
+
+ val tsMd = new MetadataBuilder()
+ dialect.updateExtraColumnMetaForWrite(TimestampType, tsMd)
+ assert(!tsMd.build().contains(JdbcUtils.WRITE_TIMESTAMP_NTZ_WALL_CLOCK), s"dialect=$dialect")
+
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ val legacyMd = new MetadataBuilder()
+ dialect.updateExtraColumnMetaForWrite(TimestampNTZType, legacyMd)
+ assert(!legacyMd.build().contains(JdbcUtils.WRITE_TIMESTAMP_NTZ_WALL_CLOCK),
+ s"dialect=$dialect")
+ }
+ }
+ }
+
+ test("SPARK-58876: Oracle DATE/TIMESTAMP map to NTZ, read marker, legacy-gated, even wrapped") {
+ val custom = new JdbcDialect {
+ override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle")
+ override def getCatalystType(
+ sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = None
+ }
+ val dialects = Seq[JdbcDialect](
+ OracleDialect(),
+ new AggregatedDialect(List(custom, OracleDialect())),
+ new AggregatedDialect(List(OracleDialect(), custom)))
+ dialects.foreach { dialect =>
+ Seq("DATE", "TIMESTAMP").foreach { typeName =>
+ val hint = s"dialect=$dialect typeName=$typeName"
+ val md = new MetadataBuilder()
+ assert(dialect.getCatalystType(java.sql.Types.TIMESTAMP, typeName, 0, md) ===
+ Some(TimestampNTZType), hint)
+ assert(md.build().contains(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK), hint)
+
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ val legacyMd = new MetadataBuilder()
+ val legacyType = dialect.getCatalystType(java.sql.Types.TIMESTAMP, typeName, 0, legacyMd)
+ assert(legacyType === None, hint)
+ assert(!legacyMd.build().contains(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK), hint)
+ }
+ }
+ }
+ }
+
+ test("SPARK-58876: only zoneless Oracle temporal types map to TimestampNTZType") {
+ val oracleDialect = OracleDialect()
+ // Zoneless DATE/TIMESTAMP map to NTZ (with the marker); the WITH [LOCAL] TIME ZONE variants
+ // (distinct sqlTypes) must not map to NTZ, and carry no marker.
+ val cases = Seq(
+ (java.sql.Types.TIMESTAMP, "TIMESTAMP", true),
+ (java.sql.Types.TIMESTAMP, "DATE", true),
+ (OracleDialect.TIMESTAMP_TZ, "TIMESTAMP WITH TIME ZONE", false),
+ (OracleDialect.TIMESTAMP_LTZ, "TIMESTAMP WITH LOCAL TIME ZONE", false))
+ cases.foreach { case (sqlType, typeName, isNTZ) =>
+ val md = new MetadataBuilder()
+ val hint = s"sqlType=$sqlType typeName=$typeName"
+ val resolved = oracleDialect.getCatalystType(sqlType, typeName, 0, md)
+ assert((resolved == Some(TimestampNTZType)) === isNTZ, hint)
+ assert(md.build().contains(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK) === isNTZ, hint)
+ }
+ }
+
+ test("SPARK-58876: Oracle NTZ mapping and the preferTimestampNTZ read option") {
+ // preferTimestampNTZ reaches the dialect only through getSchema's isTimestampNTZ argument, so
+ // resolve a mocked Oracle TIMESTAMP column via getSchema under each (prefer, flag) combination.
+ def resolve(preferTimestampNTZ: Boolean): StructField = {
+ val rsmd = mock(classOf[java.sql.ResultSetMetaData])
+ when(rsmd.getColumnCount).thenReturn(1)
+ when(rsmd.getColumnLabel(anyInt())).thenReturn("T")
+ when(rsmd.getColumnType(anyInt())).thenReturn(java.sql.Types.TIMESTAMP)
+ when(rsmd.getColumnTypeName(anyInt())).thenReturn("TIMESTAMP")
+ when(rsmd.getPrecision(anyInt())).thenReturn(0)
+ when(rsmd.getScale(anyInt())).thenReturn(0)
+ when(rsmd.isSigned(anyInt())).thenReturn(false)
+ when(rsmd.isNullable(anyInt())).thenReturn(java.sql.ResultSetMetaData.columnNullable)
+ val rs = mock(classOf[ResultSet])
+ when(rs.getMetaData).thenReturn(rsmd)
+ JdbcUtils.getSchema(mock(classOf[Connection]), rs, OracleDialect(),
+ isTimestampNTZ = preferTimestampNTZ).fields.head
+ }
+
+ // Non-legacy always maps to NTZ (with the read marker) regardless of preferTimestampNTZ; the
+ // legacy flag defers to the shared mapping, which honors preferTimestampNTZ; no marker.
+ for {
+ legacy <- Seq(true, false)
+ prefer <- Seq(true, false)
+ } {
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> legacy.toString) {
+ val field = resolve(prefer)
+ val expectedType = if (legacy && !prefer) TimestampType else TimestampNTZType
+ val hint = s"legacy=$legacy prefer=$prefer"
+ assert(field.dataType === expectedType, hint)
+ assert(field.metadata.contains(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK) === !legacy, hint)
+ }
+ }
+ }
+
+ test("SPARK-58876: a marked NTZ column reads wall-clock regardless of the resolved dialect") {
+ val custom = new JdbcDialect {
+ override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle")
+ override def getCatalystType(
+ sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = None
+ }
+ val schema = new StructType().add("t", TimestampNTZType, nullable = true,
+ new MetadataBuilder().putBoolean(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK, value = true)
+ .build())
+ // Bare Oracle, then AggregatedDialect with Oracle at the tail and at the head, so the getter
+ // selection is exercised regardless of Oracle's position among the members.
+ val dialects = Seq[JdbcDialect](
+ OracleDialect(),
+ new AggregatedDialect(List(custom, OracleDialect())),
+ new AggregatedDialect(List(OracleDialect(), custom)))
+ val ldt = LocalDateTime.of(1991, 11, 9, 0, 0, 0)
+ val expected = DateTimeUtils.localDateTimeToMicros(ldt)
+ dialects.foreach { dialect =>
+ val rs = mock(classOf[ResultSet])
+ when(rs.next()).thenReturn(true, false)
+ when(rs.getObject(1, classOf[java.time.LocalDateTime])).thenReturn(ldt)
+ val rows = JdbcUtils.resultSetToSparkInternalRows(
+ rs, dialect, schema, new InputMetrics).toArray
+ assert(rows.length === 1)
+ assert(rows.head.getLong(0) === expected, s"dialect=$dialect")
+ }
+ }
+
+ test("SPARK-58876: Oracle compileValue renders a LocalDateTime as a JDBC timestamp literal") {
+ // Filters on an NTZ-mapped Oracle column push down a LocalDateTime; it must become a valid
+ // Oracle literal rather than LocalDateTime.toString.
+ val oracleDialect = JdbcDialects.get("jdbc:oracle")
+ assert(oracleDialect.compileValue(LocalDateTime.of(2018, 7, 6, 6, 0, 0)) ===
+ "{ts '2018-07-06 06:00:00.0'}")
+ }
+
+ test("SPARK-58876: Oracle TIMESTAMP stays microsecond TimestampNTZType under the nanos preview") {
+ val oracleDialect = JdbcDialects.get("jdbc:oracle")
+ // Even with the nanosecond timestamp preview enabled, the Oracle mapping is microsecond
+ // TimestampNTZType and does not engage that preview (no nanosecond type, no deferral).
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val md = new MetadataBuilder()
+ .putBoolean("preferTimestampNanos", value = true).putLong("scale", 9)
+ assert(oracleDialect.getCatalystType(java.sql.Types.TIMESTAMP, "TIMESTAMP", 0, md) ===
+ Some(TimestampNTZType))
+ }
+ }
+
test("SPARK-42469: OracleDialect Limit query test") {
// JDBC url is a required option but is not used in this test.
val options = new JDBCOptions(Map("url" -> "jdbc:h2://host:port", "dbtable" -> "test"))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org