urosstan-db commented on code in PR #59671:
URL: https://github.com/apache/spark/pull/59671#discussion_r4387183526
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTable.scala:
##########
@@ -60,13 +60,13 @@ case class JDBCTable(
override def newScanBuilder(options: CaseInsensitiveStringMap):
JDBCScanBuilder = {
val mergedOptions = new JDBCOptions(
- jdbcOptions.parameters.originalMap ++
options.asCaseSensitiveMap().asScala)
Review Comment:
Why do we need this change?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala:
##########
@@ -151,11 +153,28 @@ class JDBCTableCatalog extends TableCatalog
}
}
+ // The per-read value decides the resolved schema, so it must reach
loadTable.
+ override def tableStateOptionKeys(): util.Set[String] =
+ util.Set.of(JDBCOptions.JDBC_ZONELESS_TIMESTAMP_AS_NTZ)
+
+ // Carries the state options down the default dispatch to loadTable(ident),
so subclass
+ // overrides of loadTable(ident) and loadTable(ident, writePrivileges) stay
in the path.
+ private val currentStateOptions =
Review Comment:
This means we need to use same instance of catalog for both loadTable
invocations?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala:
##########
@@ -352,13 +352,17 @@ object JdbcUtils extends Logging with SQLConfHelper {
metadata.putBoolean("isSigned", isSigned)
metadata.putBoolean("isTimestampNTZ", options.preferTimestampNTZ)
metadata.putBoolean("preferTimestampNanos", options.preferTimestampNanos)
+ metadata.putBoolean(
+ JDBCOptions.JDBC_ZONELESS_TIMESTAMP_AS_NTZ,
options.zonelessTimestampAsNTZ)
metadata.putLong("scale", fieldScale)
metadata.putString("jdbcClientType", typeName)
dialect.updateExtraColumnMeta(conn, rsmd, i + 1, metadata)
val columnType =
dialect.getCatalystType(dataType, typeName, fieldSize,
metadata).getOrElse(
getCatalystType(dataType, typeName, fieldSize, fieldScale, isSigned,
options))
+ // Only an input to the dialect's type mapping, so keep it out of the
schema.
+ metadata.remove(JDBCOptions.JDBC_ZONELESS_TIMESTAMP_AS_NTZ)
Review Comment:
It is fine to keep it, since other options are kept as well?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala:
##########
@@ -151,11 +153,28 @@ class JDBCTableCatalog extends TableCatalog
}
}
+ // The per-read value decides the resolved schema, so it must reach
loadTable.
+ override def tableStateOptionKeys(): util.Set[String] =
+ util.Set.of(JDBCOptions.JDBC_ZONELESS_TIMESTAMP_AS_NTZ)
Review Comment:
**B1 (blocker): forward the timestamp preference during table loading.**
With the default Oracle legacy flag (`false`), no catalog-level timestamp
preference, and `spark.sql.timestampType=TIMESTAMP_LTZ`, this combination
resolves `TS` as `TimestampType` instead of the documented `TimestampNTZType`
fallback:
```sql
SELECT TS FROM oracle_catalog.schema_name.T WITH (
'zonelessTimestampAsNTZ' = 'false',
'preferTimestampNTZ' = 'true')
```
`CatalogV2Util` filters the load options to these declared keys, so
`preferTimestampNTZ=true` is lost during schema discovery. `newScanBuilder`
receives it later but keeps the already-resolved schema. The older legacy-mode
limitation predates this PR; the new opt-out exposes it with the legacy flag
disabled.
Could we include the preference in the table-state keys and add a V2 schema
assertion for this combination?
```suggestion
override def tableStateOptionKeys(): util.Set[String] =
util.Set.of(
JDBCOptions.JDBC_ZONELESS_TIMESTAMP_AS_NTZ,
JDBCOptions.JDBC_PREFER_TIMESTAMP_NTZ)
```
##########
connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala:
##########
@@ -629,6 +636,19 @@ class OracleIntegrationSuite extends
SharedJDBCIntegrationSuite
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"))
+
+ withDefaultTimeZone(LA) {
+ val optInDf = readDatetime(
+ Map("zonelessTimestampAsNTZ" -> "true", "timestampNTZAsWallClock" ->
"true"))
+ assert(optInDf.schema("T").dataType === TimestampNTZType)
Review Comment:
**S1: cover the independent read-conversion opt-out.**
The new read checks enable NTZ mapping and wall-clock conversion together.
The mapping opt-out produces `TimestampType`, so it bypasses the NTZ getter;
the explicit false conversion tests exercise writes.
Could we add one read with `timestampNTZAsWallClock=false` while retaining
NTZ mapping? Insert this before the enclosing test enables the legacy flag,
reusing the `datetime` fixture:
```scala
withDefaultTimeZone(LA) {
val baseRead = readDatetime(Map("timestampNTZAsWallClock" -> "false"))
assert(baseRead.schema("T").dataType === TimestampNTZType)
checkAnswer(baseRead.select("T"), Row(LocalDateTime.of(1996, 1, 1, 9, 23,
45)))
}
```
The fixture stores `1996-01-01 01:23:45`; LA is UTC-08:00 on that date, so
the legacy conversion yields `09:23:45`. This checks the independent read
contract without duplicating B1.
##########
docs/sql-data-sources-jdbc.md:
##########
@@ -382,6 +382,22 @@ logging into the data sources.
</td>
<td>read</td>
</tr>
+ <tr>
+ <td><code>zonelessTimestampAsNTZ</code></td>
+ <td><code>true</code> for Oracle, <code>false</code> for other
dialects</td>
+ <td>
+ When the option is set to <code>true</code>, timestamp types without a
time zone are inferred as Spark's TimestampNTZ type regardless of
<code>preferTimestampNTZ</code>. Otherwise, they follow
<code>preferTimestampNTZ</code>. This option is currently only supported by
Oracle. Its default follows the dialect's legacy TimestampNTZ mapping
configuration: it is <code>false</code> when that configuration is enabled.
Depending on the driver, values read as TimestampNTZ may not keep their
wall-clock time unless <code>timestampNTZAsWallClock</code> is also
<code>true</code>. On writes through a JDBC catalog, it also decides the types
of the target table's columns, and so how the written values are cast. For
tables created with <code>CREATE TABLE ... USING jdbc</code>, the schema is
fixed when the table is created, so set this option in the table's
<code>OPTIONS</code>.
+ </td>
+ <td>read/write</td>
+ </tr>
+ <tr>
+ <td><code>timestampNTZAsWallClock</code></td>
+ <td><code>true</code> for Oracle, <code>false</code> for other
dialects</td>
+ <td>
+ When the option is set to <code>true</code>, TimestampNTZ values are
read and written as <code>java.time.LocalDateTime</code>, preserving the
wall-clock value without a time zone conversion. This requires a JDBC driver
that supports <code>java.time.LocalDateTime</code>. Otherwise, they are
converted through <code>java.sql.Timestamp</code> in the JVM default time zone.
Its default follows the dialect's legacy TimestampNTZ mapping configuration: it
is <code>false</code> when that configuration is enabled (only Oracle has one,
so the default is <code>true</code> for Oracle and <code>false</code> for other
dialects).
Review Comment:
**N1 (optional): name the legacy configuration directly.**
The final sentence repeats the adjacent default column. Naming the
configuration preserves the useful exception more directly:
```suggestion
When the option is set to <code>true</code>, TimestampNTZ values are
read and written as <code>java.time.LocalDateTime</code>, preserving the
wall-clock value without a time zone conversion. This requires a JDBC driver
that supports <code>java.time.LocalDateTime</code>. Otherwise, they are
converted through <code>java.sql.Timestamp</code> in the JVM default time zone.
For Oracle, enabling
<code>spark.sql.legacy.oracle.timestampNTZMapping.enabled</code> changes the
default to <code>false</code>.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala:
##########
@@ -151,11 +153,28 @@ class JDBCTableCatalog extends TableCatalog
}
}
+ // The per-read value decides the resolved schema, so it must reach
loadTable.
+ override def tableStateOptionKeys(): util.Set[String] =
+ util.Set.of(JDBCOptions.JDBC_ZONELESS_TIMESTAMP_AS_NTZ)
+
+ // Carries the state options down the default dispatch to loadTable(ident),
so subclass
+ // overrides of loadTable(ident) and loadTable(ident, writePrivileges) stay
in the path.
+ private val currentStateOptions =
+ new DynamicVariable[collection.Map[String, String]](Map.empty)
Review Comment:
**Optional: group the scoped state with the other catalog fields.**
Could we move this declaration and its comment beside the fields at the top
of the class? That keeps the `tableStateOptionKeys` and `loadTable` overrides
together:
```scala
private var dialect: JdbcDialect = _
private var functions: Map[String, UnboundFunction] = _
private val currentStateOptions =
new DynamicVariable[collection.Map[String, String]](Map.empty)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
