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/10/12 08:36:09 UTC
(spark) branch branch-4.x updated: [SPARK-59865][SQL] Decide Oracle TimestampNTZ wall-clock conversion from JDBCOptions instead of column metadata
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 caf025de2f8e [SPARK-59865][SQL] Decide Oracle TimestampNTZ wall-clock conversion from JDBCOptions instead of column metadata
caf025de2f8e is described below
commit caf025de2f8e4a35e2f7a2d3f272db5eb8c6efce
Author: aleksandar-trajkovic-db <al...@databricks.com>
AuthorDate: Mon Oct 12 08:35:39 2026 +0000
[SPARK-59865][SQL] Decide Oracle TimestampNTZ wall-clock conversion from JDBCOptions instead of column metadata
### What changes were proposed in this pull request?
SPARK-58876 decides per column, through column-metadata markers, whether Oracle `TimestampNTZType` values are read and written as wall-clock `LocalDateTime`. This PR makes it a per-read/write decision:
- Adds `JdbcDialect.timestampNTZAsWallClock(options: JDBCOptions)`. `OracleDialect` returns true unless the legacy mapping is on, and `AggregatedDialect` returns true if any wrapped dialect does.
- `JDBCOptions` snapshots `spark.sql.legacy.oracle.timestampNTZMapping.enabled` when it is built. `JDBCTableCatalog.loadTable` pins that snapshot into the table options, so V2 scans and writes, which rebuild `JDBCOptions` later in the query, keep it.
- Removes the markers and the unreleased `JdbcDialect.updateExtraColumnMetaForWrite`.
### Why are the changes needed?
The decision depends only on the dialect and the legacy flag, not on the column, so it belongs with the read/write options rather than in each column's metadata. This also drops a dialect hook whose only job was stamping that metadata for writes.
### Does this PR introduce _any_ user-facing change?
Only relative to the unreleased SPARK-58876: Oracle reads apply the wall-clock conversion to every `TimestampNTZType` column, not just the `DATE`/`TIMESTAMP` columns `OracleDialect` maps (e.g. a column that `customSchema` forces to `TIMESTAMP_NTZ`). `JdbcDialect` gains `timestampNTZAsWallClock` and loses `updateExtraColumnMetaForWrite`.
### How was this patch tested?
Updated the SPARK-58876 tests in `JDBCSuite` and `OracleIntegrationSuite`, including reads planned before a legacy-flag flip, and added a `JDBCTableCatalogSuite` test that a V2 scan keeps the flag value pinned by `loadTable`. Ran `JDBCSuite`, `JDBCTableCatalogSuite`, and `OracleIntegrationSuite` (Docker) locally.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Claude Opus 5.5)
This pull request and its description were written by Isaac.
---
This is my original work and I license it to the project under the project's open source license.
Closes #59140 from aleksandar-trajkovic-db/timestampntz-jdbcoption-refactor.
Authored-by: aleksandar-trajkovic-db <al...@databricks.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
(cherry picked from commit a90dc180e04ae8eee7848c9200d11f95a0fb4790)
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../spark/sql/jdbc/OracleIntegrationSuite.scala | 41 ++----
.../spark/sql/jdbc/v2/OracleIntegrationSuite.scala | 16 +++
.../execution/datasources/jdbc/JDBCOptions.scala | 8 ++
.../sql/execution/datasources/jdbc/JDBCRDD.scala | 3 +-
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 44 +++----
.../datasources/v2/jdbc/JDBCTableCatalog.scala | 9 +-
.../apache/spark/sql/jdbc/AggregatedDialect.scala | 4 +-
.../org/apache/spark/sql/jdbc/JdbcDialects.scala | 16 +--
.../org/apache/spark/sql/jdbc/OracleDialect.scala | 16 +--
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 143 ++++++++-------------
10 files changed, 125 insertions(+), 175 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 010b231f727e..d83a77313ddc 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
@@ -28,7 +28,7 @@ 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, JdbcUtils}
+import org.apache.spark.sql.execution.datasources.jdbc.{JDBCPartition, JDBCRelation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
@@ -797,38 +797,19 @@ class OracleIntegrationSuite extends SharedJDBCIntegrationSuite
}
}
- 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)))
+ test("SPARK-58876: an NTZ read planned before a legacy flag flip still reads wall-clock") {
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).
+ val df = withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "false") {
+ spark.read.format("jdbc")
+ .option("url", jdbcUrl)
+ .option("dbtable", "datetime")
+ .load()
+ }
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()
+ 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))
}
- // 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/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
index 68e74ef731fe..d11f9c0e89da 100644
--- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
+++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.jdbc.v2
import java.sql.Connection
+import java.time.LocalDateTime
import java.util.Locale
import scala.util.Using
@@ -25,8 +26,10 @@ import scala.util.Using
import org.apache.spark.{SparkConf, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.catalyst.util.CharVarcharUtils.CHAR_VARCHAR_TYPE_STRING_METADATA_KEY
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.jdbc.OracleDatabaseOnDocker
import org.apache.spark.sql.types._
import org.apache.spark.tags.DockerTest
@@ -184,6 +187,19 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCTes
override def caseConvert(tableName: String): String = tableName.toUpperCase(Locale.ROOT)
+ test("SPARK-58876: a V2 NTZ read planned before a legacy flag flip still reads wall-clock") {
+ val table = s"$catalogName.${namespaceOpt.get}.${caseConvert("datetime")}"
+ withDefaultTimeZone(LA) {
+ val df = withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "false") {
+ sql(s"SELECT time1 FROM $table WHERE name = 'amy'")
+ }
+ assert(df.schema.head.dataType === TimestampNTZType)
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ assert(df.collect().head.get(0) === LocalDateTime.of(2022, 5, 19, 0, 0, 0))
+ }
+ }
+ }
+
test("SPARK-46478: Revert SPARK-43049 to use varchar(255) for string") {
val tableName = catalogName + ".t1"
withTable(tableName) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
index e4bd4df8cf2f..a5c866cbfa38 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
@@ -29,6 +29,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.jdbc.JdbcDialects
import org.apache.spark.sql.types.TimestampNTZType
import org.apache.spark.util.Utils
@@ -265,6 +266,12 @@ class JDBCOptions(
.map(_.toBoolean)
.getOrElse(false)
+ val timestampNTZAsWallClock: Boolean =
+ parameters
+ .get(JDBC_TIMESTAMP_NTZ_WALL_CLOCK)
+ .map(_.toBoolean)
+ .getOrElse(JdbcDialects.get(url).timestampNTZAsWallClock)
+
val hint = parameters.get(JDBC_HINT_STRING).map(value => {
require(value.matches("(?s)^/\\*\\+ .* \\*/$"),
s"Invalid value `$value` for option `$JDBC_HINT_STRING`." +
@@ -379,4 +386,5 @@ object JDBCOptions {
val JDBC_PREFER_TIMESTAMP_NTZ = newOption("preferTimestampNTZ")
val JDBC_PREFER_TIMESTAMP_NANOS = newOption("preferTimestampNanos")
val JDBC_HINT_STRING = newOption("hint")
+ val JDBC_TIMESTAMP_NTZ_WALL_CLOCK = newOption("__timestampNTZAsWallClock")
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
index a681517ccfc0..626ce6ff2437 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
@@ -400,7 +400,8 @@ class JDBCRDD(
dialect,
schema,
inputMetrics,
- Some(fetchAndTransformToInternalRowsMetric))
+ Some(fetchAndTransformToInternalRowsMetric),
+ Some(options))
CompletionIterator[InternalRow, Iterator[InternalRow]](
new InterruptibleIterator(context, rowsIterator), close())
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 0f630b8cc171..652fb134a2e5 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,11 +55,6 @@ 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.
*/
@@ -380,18 +375,21 @@ object JdbcUtils extends Logging with SQLConfHelper {
*/
def resultSetToRows(
resultSet: ResultSet,
- schema: StructType): Iterator[Row] = {
- resultSetToRows(resultSet, schema, NoopDialect)
+ schema: StructType,
+ options: Option[JDBCOptions]): Iterator[Row] = {
+ resultSetToRows(resultSet, schema, NoopDialect, options)
}
def resultSetToRows(
resultSet: ResultSet,
schema: StructType,
- dialect: JdbcDialect): Iterator[Row] = {
+ dialect: JdbcDialect,
+ options: Option[JDBCOptions]): Iterator[Row] = {
val inputMetrics =
Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
val fromRow = ExpressionEncoder(schema).resolveAndBind().createDeserializer()
- val internalRows = resultSetToSparkInternalRows(resultSet, dialect, schema, inputMetrics)
+ val internalRows = resultSetToSparkInternalRows(
+ resultSet, dialect, schema, inputMetrics, options = options)
internalRows.map(fromRow)
}
@@ -400,10 +398,11 @@ object JdbcUtils extends Logging with SQLConfHelper {
dialect: JdbcDialect,
schema: StructType,
inputMetrics: InputMetrics,
- fetchAndTransformToInternalRowsMetric: Option[SQLMetric] = None): Iterator[InternalRow] = {
+ fetchAndTransformToInternalRowsMetric: Option[SQLMetric] = None,
+ options: Option[JDBCOptions]): Iterator[InternalRow] = {
new NextIterator[InternalRow] {
private[this] val rs = resultSet
- private[this] val getters: Array[JDBCValueGetter] = makeGetters(dialect, schema)
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(dialect, schema, options)
private[this] val mutableRow =
new SpecificInternalRow(schema.fields.map(x => x.dataType).toImmutableArraySeq)
@@ -456,15 +455,17 @@ object JdbcUtils extends Logging with SQLConfHelper {
*/
private def makeGetters(
dialect: JdbcDialect,
- schema: StructType): Array[JDBCValueGetter] = {
+ schema: StructType,
+ options: Option[JDBCOptions]): Array[JDBCValueGetter] = {
val replaced = CharVarcharUtils.replaceCharVarcharWithStringInSchema(schema)
- replaced.fields.map(sf => makeGetter(sf.dataType, dialect, sf.metadata))
+ replaced.fields.map(sf => makeGetter(sf.dataType, dialect, sf.metadata, options))
}
private def makeGetter(
dt: DataType,
dialect: JdbcDialect,
- metadata: Metadata): JDBCValueGetter = dt match {
+ metadata: Metadata,
+ options: Option[JDBCOptions]): JDBCValueGetter = dt match {
case BooleanType => JDBCValueGetter.BooleanGetter
case DateType => JDBCValueGetter.DateGetter(dialect)
case _: TimeType => JDBCValueGetter.TimeGetter
@@ -483,7 +484,7 @@ 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) =>
+ case TimestampNTZType if options.exists(_.timestampNTZAsWallClock) =>
JDBCValueGetter.TimestampNTZWallClockGetter
case TimestampNTZType => JDBCValueGetter.TimestampNTZGetter(dialect)
case t: TimestampNTZNanosType => JDBCValueGetter.TimestampNTZNanosGetter(t.precision)
@@ -508,7 +509,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
conn: Connection,
dialect: JdbcDialect,
dataType: DataType,
- metadata: Metadata): JDBCValueSetter = dataType match {
+ options: JDBCOptions): JDBCValueSetter = dataType match {
case IntegerType =>
(stmt: PreparedStatement, row: Row, pos: Int) =>
stmt.setInt(pos + 1, row.getInt(pos))
@@ -554,7 +555,7 @@ 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) =>
+ case TimestampNTZType if options.timestampNTZAsWallClock =>
(stmt: PreparedStatement, row: Row, pos: Int) =>
stmt.setObject(pos + 1, row.getAs[java.time.LocalDateTime](pos))
case TimestampNTZType =>
@@ -728,7 +729,8 @@ 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, f.metadata))
+ val setters =
+ rddSchema.fields.map(f => makeSetter(conn, dialect, f.dataType, options))
val nullTypes = rddSchema.fields.map(f => getJdbcType(f.dataType, dialect).jdbcNullType)
val numFields = rddSchema.fields.length
@@ -925,11 +927,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
val url = options.url
val table = options.table
val dialect = JdbcDialects.get(url)
- 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 rddSchema = df.schema
val batchSize = options.batchSize
val isolationLevel = options.isolationLevel
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala
index ff6d28744364..dc7009d7c6b4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala
@@ -153,8 +153,9 @@ class JDBCTableCatalog extends TableCatalog
override def loadTable(ident: Identifier): Table = {
JdbcUtils.withConnection(options) { conn =>
- val optionsWithTableName = new JDBCOptions(
- options.parameters + (JDBCOptions.JDBC_TABLE_NAME -> getTableName(ident)))
+ val tableOptions = new JDBCOptions(
+ options.parameters + (JDBCOptions.JDBC_TABLE_NAME -> getTableName(ident)) +
+ (JDBCOptions.JDBC_TIMESTAMP_NTZ_WALL_CLOCK -> dialect.timestampNTZAsWallClock.toString))
JdbcUtils.classifyException(
condition = "FAILED_JDBC.LOAD_TABLE",
messageParameters = Map(
@@ -167,9 +168,9 @@ class JDBCTableCatalog extends TableCatalog
val remoteSchemaFetchMetric = JdbcUtils
.createSchemaFetchMetric(SparkSession.active.sparkContext)
val schema = SQLMetrics.withTimingNs(remoteSchemaFetchMetric) {
- JDBCRDD.resolveTable(optionsWithTableName, conn, Some(ident), Some(name()))
+ JDBCRDD.resolveTable(tableOptions, conn, Some(ident), Some(name()))
}
- JDBCTable(ident, schema, optionsWithTableName,
+ JDBCTable(ident, schema, tableOptions,
Map(JDBCRelation.schemaFetchKey -> remoteSchemaFetchMetric))
}
}
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 fe503d97cb60..3b4f2f784989 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,9 +46,7 @@ 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 timestampNTZAsWallClock: Boolean = dialects.exists(_.timestampNTZAsWallClock)
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 541a1b0d6dc1..7d57fa6a5ed6 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
@@ -217,6 +217,13 @@ abstract class JdbcDialect extends Serializable with Logging {
toJavaTimestampNoRebase(micros)
}
+ /**
+ * Whether TimestampNTZType values are read and written as zoneless `LocalDateTime` rather than
+ * through `java.sql.Timestamp`. Snapshotted into `JDBCOptions` when the options are built.
+ */
+ @Since("4.4.0")
+ def timestampNTZAsWallClock: Boolean = false
+
/**
* Returns a factory for creating connections to the given JDBC URL.
* In general, creating a connection has nothing to do with JDBC partition id.
@@ -953,15 +960,6 @@ 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 0dab4c0e9652..55538fa4e3e8 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.sql.Types.TIMESTAMP
import java.time.LocalDateTime
import java.util.Locale
@@ -184,7 +185,7 @@ 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 &&
+ case TIMESTAMP if !conf.legacyOracleTimestampNTZMappingEnabled && typeName != null &&
typeName.toUpperCase(Locale.ROOT).matches("DATE|TIMESTAMP") =>
val metadata = if (md != null) md.build() else Metadata.empty
// Absent scale metadata: Oracle TIMESTAMP defaults to TIMESTAMP(6).
@@ -193,23 +194,12 @@ private case class OracleDialect() extends JdbcDialect with SQLConfHelper with N
metadata.getBoolean("preferTimestampNanos")
val resolved = JdbcUtils.resolveTimestampType(
isTimestampNTZ = true, scale = scale, preferTimestampNanos = preferNanos)
- // Oracle DATE/TIMESTAMP are zoneless; mark the microsecond NTZ wall-clock so a later flag
- // flip can't desync the read. The nanos NTZ getter is wall-clock by construction.
- if (md != null && resolved == TimestampNTZType) {
- md.putBoolean(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK, value = true)
- }
Some(resolved)
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 timestampNTZAsWallClock: Boolean = !conf.legacyOracleTimestampNTZMappingEnabled
override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
// For more details, please see
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 4f731ea84f9c..b5e4b0e29b5e 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
@@ -33,7 +33,7 @@ import org.mockito.Mockito._
import org.apache.spark.{
SparkArithmeticException, SparkException,
- SparkIllegalArgumentException, SparkSQLException
+ SparkIllegalArgumentException, SparkSQLException, TaskContext
}
import org.apache.spark.executor.InputMetrics
import org.apache.spark.sql.{AnalysisException, DataFrame, Observation, Row}
@@ -1745,78 +1745,22 @@ class JDBCSuite extends SharedSparkSession {
}
}
- test("SPARK-58876: Oracle stamps the NTZ wall-clock write marker, legacy-gated, even wrapped") {
+ test("SPARK-58876: Oracle NTZ wall-clock follows the legacy flag, 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")
- }
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "false") {
+ dialects.foreach(d => assert(d.timestampNTZAsWallClock, s"dialect=$d"))
}
- }
-
- 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)
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
+ dialects.foreach(d => assert(!d.timestampNTZAsWallClock, s"dialect=$d"))
}
+ // Dialects that don't opt in keep the java.sql.Timestamp conversion.
+ assert(!NoopDialect.timestampNTZAsWallClock)
}
test("SPARK-58876: Oracle NTZ mapping and the preferTimestampNTZ read option") {
@@ -1838,8 +1782,8 @@ class JDBCSuite extends SharedSparkSession {
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.
+ // Non-legacy always maps to NTZ regardless of preferTimestampNTZ; the legacy flag defers to the
+ // shared mapping, which honors preferTimestampNTZ.
for {
legacy <- Seq(true, false)
prefer <- Seq(true, false)
@@ -1849,36 +1793,57 @@ class JDBCSuite extends SharedSparkSession {
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
+ test("SPARK-58876: Oracle NTZ reads wall-clock via frozen options after a flag flip") {
+ val schema = new StructType().add("t", TimestampNTZType, nullable = true)
+ val options = withSQLConf(
+ SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "false") {
+ new JDBCOptions("jdbc:oracle:thin:@//host:1521/db", "t", Map.empty[String, String])
}
- 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 =>
+ withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") {
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
+ rs, OracleDialect(), schema, new InputMetrics, options = Some(options)).toArray
assert(rows.length === 1)
- assert(rows.head.getLong(0) === expected, s"dialect=$dialect")
+ assert(rows.head.getLong(0) === DateTimeUtils.localDateTimeToMicros(ldt))
+ }
+ }
+
+ test("SPARK-58876: Oracle NTZ writes pick the setter from the snapshotted options") {
+ val ldt = LocalDateTime.of(1991, 11, 9, 0, 0, 0)
+ val schema = new StructType().add("t", TimestampNTZType, nullable = true)
+ for (legacy <- Seq(true, false)) {
+ val options = withSQLConf(
+ SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> legacy.toString) {
+ new JDBCOptions("jdbc:oracle:thin:@//host:1521/db", "t", Map.empty[String, String])
+ }
+ val stmt = mock(classOf[java.sql.PreparedStatement])
+ val conn = mock(classOf[Connection])
+ when(conn.prepareStatement(anyString())).thenReturn(stmt)
+ val dialect = new JdbcDialect {
+ override def canHandle(url: String): Boolean = false
+ override def createConnectionFactory(options: JDBCOptions): Int => Connection = _ => conn
+ }
+ TaskContext.setTaskContext(TaskContext.empty())
+ try {
+ JdbcUtils.savePartition("t", Iterator(Row(ldt)), schema, "INSERT INTO t VALUES (?)",
+ batchSize = 1, dialect, Connection.TRANSACTION_NONE, options)
+ } finally {
+ TaskContext.unset()
+ }
+ if (legacy) {
+ verify(stmt).setTimestamp(1, dialect.convertTimestampNTZToJavaTimestamp(ldt))
+ verify(stmt, never()).setObject(anyInt(), any())
+ } else {
+ verify(stmt).setObject(1, ldt)
+ verify(stmt, never()).setTimestamp(anyInt(), any())
+ }
}
}
@@ -1901,7 +1866,7 @@ class JDBCSuite extends SharedSparkSession {
StructField("va", ArrayType(VarcharType(2)))))
val rows = JdbcUtils.resultSetToSparkInternalRows(
- rs, NoopDialect, schema, new InputMetrics).toArray
+ rs, NoopDialect, schema, new InputMetrics, options = None).toArray
assert(rows.length === 1)
assert(rows.head.getUTF8String(0).toString === "a ")
assert(rows.head.getUTF8String(1).toString === "bb")
@@ -1977,23 +1942,17 @@ class JDBCSuite extends SharedSparkSession {
preferTimestampNanos = preferNanos).fields.head
}
}
- def marked(f: StructField): Boolean =
- f.metadata.contains(JdbcUtils.READ_TIMESTAMP_NTZ_WALL_CLOCK)
// Sub-microsecond scales (7-9) widen to the nanosecond NTZ type only when both the read option
// and the preview are on; every coarser scale and either flag off stays microsecond NTZ.
(TimestampNTZNanosType.MIN_PRECISION to TimestampNTZNanosType.MAX_PRECISION).foreach { s =>
val f = resolve(s, preferNanos = true, nanosEnabled = true)
assert(f.dataType === TimestampNTZNanosType(s), s"scale=$s")
- // The nanos NTZ getter is wall-clock by construction, so the marker must not be stamped.
- assert(!marked(f), s"scale=$s")
}
- // Microsecond NTZ results carry the wall-clock read marker.
Seq(
resolve(6, preferNanos = true, nanosEnabled = true),
resolve(9, preferNanos = false, nanosEnabled = true),
resolve(9, preferNanos = true, nanosEnabled = false)).foreach { f =>
assert(f.dataType === TimestampNTZType)
- assert(marked(f))
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org