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..34e77f8cd505 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,7 +19,7 @@ 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} @@ -150,6 +150,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 +293,8 @@ 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"))) + // Oracle TIMESTAMP column t maps to TimestampNTZType, materialized as a LocalDateTime. + 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,8 @@ 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)) + // DATE d stays DateType under mapDateToTimestamp=false; Oracle TIMESTAMP t maps to NTZ. + Row(BigDecimal.valueOf(id), Date.valueOf(date), Timestamp.valueOf(timestamp).toLocalDateTime) } // DateType partition column @@ -497,24 +507,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,26 +558,215 @@ 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")) + 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 DATE column: equality (DATE has no time component)") { + assert(filteredIds("D = TIMESTAMP_NTZ'2018-07-08 00:00:00'") === Set(3L)) + } + + test("WHERE on NTZ DATE column: range excludes the equal lower bound") { + // ids 1,2 have D = 2018-07-06 00:00:00 (equal, not greater); ids 3,4 are strictly after. + assert(filteredIds("D > TIMESTAMP_NTZ'2018-07-06 00:00:00'") === Set(3L, 4L)) + } + + test("WHERE on NTZ TIMESTAMP column: equality") { + assert(filteredIds("T = TIMESTAMP_NTZ'2018-07-06 08:10:08'") === Set(2L)) + } + + test("WHERE on NTZ TIMESTAMP column: range") { + assert(filteredIds( + "T > TIMESTAMP_NTZ'2018-07-06 06:00:00' AND T < TIMESTAMP_NTZ'2018-07-09 00:00:00'") + === Set(2L, 3L)) + } + + test("WHERE on NTZ DATE column: pushdown off (Spark-side eval) matches pushdown on") { + val where = "D > TIMESTAMP_NTZ'2018-07-06 00:00:00'" + val off = filteredIds(where, pushDown = false) + assert(off === Set(3L, 4L)) + assert(off === filteredIds(where, pushDown = true)) + } + + test("WHERE on NTZ TIMESTAMP column: pushdown off (Spark-side eval) matches pushdown on") { + val where = "T > TIMESTAMP_NTZ'2018-07-06 06:00:00'" + val off = filteredIds(where, pushDown = false) + assert(off === Set(2L, 3L, 4L)) + assert(off === filteredIds(where, pushDown = true)) + } + + 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) + } - df.write.format("jdbc") + test("TimestampNTZType round-trips through an Oracle write and read") { + val ldt = LocalDateTime.of(1996, 1, 1, 1, 23, 45, 123456000) + val schema = StructType(Seq(StructField("T", TimestampNTZType))) + // Write and read under several JVM zones; the zoneless NTZ wall-clock (incl. microseconds) must + // survive the round-trip regardless of the zone. + Seq(UTC, LA).foreach { zone => + withDefaultTimeZone(zone) { + val dfWrite = spark.createDataFrame( + spark.sparkContext.parallelize(Seq(Row(ldt))), schema) + dfWrite.write.format("jdbc") + .mode(SaveMode.Overwrite) .option("url", jdbcUrl) - .option("dbtable", "test_ltz" + flag) + .option("dbtable", "ntz_write_roundtrip") .save() - val df2 = spark.read.format("jdbc") + val dfRead = spark.read.format("jdbc") .option("url", jdbcUrl) - .option("dbtable", "test_ltz" + flag) + .option("dbtable", "ntz_write_roundtrip") .load() - checkAnswer(df2, Row(row1)) + assert(dfRead.schema.fields.head.dataType === TimestampNTZType) + assert(dfRead.collect().head.get(0) === ldt) } } } diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index c07768cd3c99..24eac903dbfb 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -24,6 +24,7 @@ license: | ## Upgrading from Spark SQL 4.3 to 4.4 +- 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 read with `oracle.jdbc.mapDateToTimestamp=false` (mapped to DateType, as before), and TIMESTAMP WITH TIME ZONE / TIMESTAMP WITH LOCAL TIME ZONE. To restore the previous behavior, set `spark.sql.legacy.oracle.timestampNTZMapping.enabled` to `true`. - 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` when the join keys are a subset of the partition keys. As before, when the partition keys cover only part of the join keys, eliminating the shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`. ## 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 0ed4399361d0..4c32a4ed75ea 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 @@ -6901,6 +6901,21 @@ 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. 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() @@ -9054,6 +9069,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/jdbc/OracleDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala index 46080dd57b1d..5f7b9fad673b 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 @@ -181,10 +182,28 @@ 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 and both surface as Types.TIMESTAMP with typeName + // DATE/TIMESTAMP, so NTZ is faithful; WITH [LOCAL] TIME ZONE hit TIMESTAMP_TZ/LTZ above. + Some(TimestampNTZType) case _ => None } } + // Preserve the zoneless wall-clock: the driver decoded the Timestamp in the JVM zone, and + // toLocalDateTime reads those same fields back rather than rebasing through UTC (mirrors + // PostgresDialect). The legacy flag defers to the base conversion, restoring the pre-4.4 value. + override def convertJavaTimestampToTimestampNTZ(t: Timestamp): LocalDateTime = { + if (conf.legacyOracleTimestampNTZMappingEnabled) super.convertJavaTimestampToTimestampNTZ(t) + else t.toLocalDateTime + } + + override def convertTimestampNTZToJavaTimestamp(ldt: LocalDateTime): Timestamp = { + if (conf.legacyOracleTimestampNTZMappingEnabled) super.convertTimestampNTZToJavaTimestamp(ldt) + else Timestamp.valueOf(ldt) + } + 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 +229,9 @@ 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 + "'}" + // Filters on a TimestampNTZType-mapped column push down a LocalDateTime; render it via the same + // JDBC {ts ...} escape (LocalDateTime.toString would be an invalid literal for the driver). + 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 6642b527e40e..d4bf1e802008 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 @@ -36,7 +36,7 @@ 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} @@ -1634,6 +1634,77 @@ class JDBCSuite extends SharedSparkSession { Some(TimestampType)) } + test("SPARK-58876: Oracle DATE and TIMESTAMP map to TimestampNTZType by default") { + val oracleDialect = JdbcDialects.get("jdbc:oracle") + // Oracle DATE and TIMESTAMP are zoneless and arrive as JDBC Types.TIMESTAMP (bare typeName, + // precision in scale), so by default they map to TimestampNTZType, independent of + // the preferTimestampNTZ read option. + Seq("DATE", "TIMESTAMP").foreach { typeName => + assert(oracleDialect.getCatalystType(java.sql.Types.TIMESTAMP, typeName, 0, null) === + Some(TimestampNTZType), s"typeName=$typeName") + } + // The legacy flag restores the pre-4.4 behavior: defer to the shared default mapping, which + // yields TimestampType (or TimestampNTZType only when preferTimestampNTZ is set). + withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") { + assert(oracleDialect.getCatalystType(java.sql.Types.TIMESTAMP, "TIMESTAMP", 0, null) === None) + } + } + + test("SPARK-58876: Oracle NTZ read preserves the wall-clock value across time zones") { + val oracleDialect = JdbcDialects.get("jdbc:oracle") + // The driver decodes an Oracle DATE/TIMESTAMP into a java.sql.Timestamp using the JVM zone; + // convertJavaTimestampToTimestampNTZ must read those same fields back with no zone shift. + val expected = LocalDateTime.of(1991, 11, 9, 0, 0, 0) + Seq("UTC", "America/Los_Angeles", "Asia/Kolkata").foreach { tz => + DateTimeTestUtils.withDefaultTimeZone(java.time.ZoneId.of(tz)) { + assert(oracleDialect.convertJavaTimestampToTimestampNTZ( + Timestamp.valueOf("1991-11-09 00:00:00")) === expected, s"tz=$tz") + } + } + } + + test("SPARK-58876: Oracle NTZ conversions are gated by the legacy flag") { + val oracleDialect = JdbcDialects.get("jdbc:oracle") + // Under a non-UTC JVM zone the two conversions diverge: by default they read the wall-clock + // fields the driver decoded, while the legacy flag defers to the base UTC-rebased conversion. + // This restores the pre-4.4 value (not just the type) when a column still maps to NTZ via + // preferTimestampNTZ, so the flag is a full behavior restore as documented. + DateTimeTestUtils.withDefaultTimeZone(java.time.ZoneId.of("America/Los_Angeles")) { + val ts = Timestamp.valueOf("1991-11-09 00:00:00") + val ldt = LocalDateTime.of(1991, 11, 9, 0, 0, 0) + assert(oracleDialect.convertJavaTimestampToTimestampNTZ(ts) === ts.toLocalDateTime) + assert(oracleDialect.convertTimestampNTZToJavaTimestamp(ldt) === Timestamp.valueOf(ldt)) + withSQLConf(SQLConf.LEGACY_ORACLE_TIMESTAMP_NTZ_MAPPING_ENABLED.key -> "true") { + assert(oracleDialect.convertJavaTimestampToTimestampNTZ(ts) === + DateTimeUtils.microsToLocalDateTime(DateTimeUtils.fromJavaTimestampNoRebase(ts))) + assert(oracleDialect.convertTimestampNTZToJavaTimestamp(ldt) === + DateTimeUtils.toJavaTimestampNoRebase(DateTimeUtils.localDateTimeToMicros(ldt))) + // The flag genuinely changes the value, not only the type. + assert(oracleDialect.convertJavaTimestampToTimestampNTZ(ts) !== ts.toLocalDateTime) + } + } + } + + 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"))