This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new f223b52bf70b [SPARK-57460][SQL] Support nanosecond-precision timestamp
types in the JDBC datasource
f223b52bf70b is described below
commit f223b52bf70be21c5719985d2796b8cea996010c
Author: Stevo Mitric <[email protected]>
AuthorDate: Tue Aug 11 10:49:51 2026 +0000
[SPARK-57460][SQL] Support nanosecond-precision timestamp types in the JDBC
datasource
### What changes were proposed in this pull request?
Adds read and write support for the nanosecond-capable timestamp types
`TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` (`p` in 7-9) to the built-in JDBC
datasource, reaching parity with the microsecond `TimestampType` /
`TimestampNTZType`, and removes the SPARK-57166 JDBC write rejection guardrail.
Specifically:
- `JdbcUtils.getCommonJDBCType`: maps the nanos types to SQL `TIMESTAMP(p)`
(dialects may override, e.g. to emit `TIMESTAMP(p) WITH TIME ZONE`).
- `JdbcUtils.getCatalystType` / `getSchema`: a new `preferTimestampNanos`
JDBC option maps a driver `TIMESTAMP` reporting a sub-microsecond scale (7-9)
to a nanos type, gated by the `spark.sql.timestampNanosTypes.enabled` preview
flag. The default microsecond mapping is preserved.
- `JDBCValueGetter`: `TimestampNTZNanosGetter` reads the wall-clock
directly via `getObject(LocalDateTime)` (mirroring `TimeGetter`, avoiding a
zone shift); `TimestampLTZNanosGetter` mirrors the micro `TimestampGetter` and
re-attaches the sub-microsecond digits from `java.sql.Timestamp.getNanos`. Both
floor to the column precision.
- `JdbcUtils.makeSetter`: NTZ writes the `LocalDateTime` wall-clock via
`setObject`; LTZ writes the micro `java.sql.Timestamp` and restores full nanos.
- `CreatableRelationProvider.supportsDataType` (interfaces.scala):
whitelists the nanos types so JDBC write is no longer rejected.
### Why are the changes needed?
JDBC was the last built-in file/serde datasource still rejecting nanosecond
timestamp types (umbrella SPARK-56822). This extends the feature to JDBC.
### Does this PR introduce any user-facing change?
Yes, only for the preview feature gated by
`spark.sql.timestampNanosTypes.enabled` (disabled by default). Nanos timestamp
columns can now be written to JDBC, and a driver `TIMESTAMP` can be read as a
nanos type via `option("preferTimestampNanos", "true")`.
### How was this patch tested?
Added read (NTZ/LTZ, default-micros-preserved) and write round-trip
(precisions 7-9, NTZ and LTZ) tests against H2 in `JDBCSuite` /
`JDBCWriteSuite`, replacing the SPARK-57166 rejection test. All JDBC suites
pass.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude Code (Claude Opus 4.8)
Closes #57698 from stevomitric/stevomitric/spark-57460-nanos.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 0e6fc40542490e18c42069ac0f00458f6a42a387)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../spark/sql/jdbc/v2/MySQLIntegrationSuite.scala | 2 +
.../spark/sql/jdbc/v2/OracleIntegrationSuite.scala | 1 +
.../sql/jdbc/v2/PostgresIntegrationSuite.scala | 1 +
.../org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala | 1 +
docs/sql-data-sources-jdbc.md | 8 +++
.../execution/datasources/jdbc/JDBCOptions.scala | 11 ++++
.../sql/execution/datasources/jdbc/JDBCRDD.scala | 3 +-
.../datasources/jdbc/JDBCValueGetter.scala | 42 +++++++++++++-
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 53 +++++++++++++++--
.../org/apache/spark/sql/sources/interfaces.scala | 4 +-
.../v2/jdbc/JDBCTableCatalogSuite.scala | 1 +
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 67 ++++++++++++++++++++++
.../org/apache/spark/sql/jdbc/JDBCWriteSuite.scala | 58 +++++++++++++------
13 files changed, 223 insertions(+), 29 deletions(-)
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
index fc2cad1b7098..b18e9e4dac82 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
@@ -89,6 +89,7 @@ class MySQLIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JDBCTest
new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean("isSigned", dataType.isInstanceOf[NumericType])
.putString("jdbcClientType", jdbcClientType)
.build()
@@ -343,6 +344,7 @@ class MySQLOverMariaConnectorIntegrationSuite extends
MySQLIntegrationSuite {
new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean("isSigned", true)
.putString("jdbcClientType", jdbcClientType)
.build()
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 f7ba1e1e0dbd..68e74ef731fe 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
@@ -101,6 +101,7 @@ class OracleIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JDBCTes
new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean(
"isSigned",
dataType.isInstanceOf[NumericType] ||
dataType.isInstanceOf[StringType])
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
index d57d3aa5ea03..9b349b0484ff 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
@@ -50,6 +50,7 @@ class PostgresIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JDBCT
new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean("isSigned", dataType.isInstanceOf[NumericType])
.putString("jdbcClientType", jdbcClientType)
.build()
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala
index 6f2a9e97ff00..3b1724877241 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala
@@ -56,6 +56,7 @@ private[v2] trait V2JDBCTest
jdbcClientType: String = "STRING"): Metadata = new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean("isSigned", dataType.isInstanceOf[NumericType])
.putString("jdbcClientType", jdbcClientType)
.build()
diff --git a/docs/sql-data-sources-jdbc.md b/docs/sql-data-sources-jdbc.md
index a205ed1a4e54..9ba3eacef63f 100644
--- a/docs/sql-data-sources-jdbc.md
+++ b/docs/sql-data-sources-jdbc.md
@@ -374,6 +374,14 @@ logging into the data sources.
</td>
<td>read</td>
</tr>
+ <tr>
+ <td><code>preferTimestampNanos</code></td>
+ <td>false</td>
+ <td>
+ When the option is set to <code>true</code>, a driver TIMESTAMP column
that reports a sub-microsecond fractional-second scale (7-9) is inferred as one
of the nanosecond-capable timestamp types (<code>TIMESTAMP_NTZ(p)</code> when
<code>preferTimestampNTZ</code> is also <code>true</code>, otherwise
<code>TIMESTAMP_LTZ(p)</code>). Otherwise such columns keep the historical
microsecond mapping. This option only takes effect when the
<code>spark.sql.timestampNanosTypes.enabled</code> pr [...]
+ </td>
+ <td>read</td>
+ </tr>
<tr>
<td><code>hint</code></td>
<td>(none)</td>
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 7188c2b8b2e8..084b192b5763 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
@@ -255,6 +255,16 @@ class JDBCOptions(
.map(_.toBoolean)
.getOrElse(SQLConf.get.timestampType == TimestampNTZType)
+ // Infers driver TIMESTAMP columns that report a sub-microsecond
fractional-second scale (7-9)
+ // as the nanosecond-capable timestamp types TIMESTAMP_NTZ(p) /
TIMESTAMP_LTZ(p). When disabled
+ // (the default) such columns keep the historical microsecond mapping. Only
takes effect when the
+ // `spark.sql.timestampNanosTypes.enabled` preview flag is on.
+ val preferTimestampNanos =
+ parameters
+ .get(JDBC_PREFER_TIMESTAMP_NANOS)
+ .map(_.toBoolean)
+ .getOrElse(false)
+
val hint = parameters.get(JDBC_HINT_STRING).map(value => {
require(value.matches("(?s)^/\\*\\+ .* \\*/$"),
s"Invalid value `$value` for option `$JDBC_HINT_STRING`." +
@@ -366,5 +376,6 @@ object JDBCOptions {
val JDBC_CONNECTION_PROVIDER = newOption("connectionProvider")
val JDBC_PREPARE_QUERY = newOption("prepareQuery")
val JDBC_PREFER_TIMESTAMP_NTZ = newOption("preferTimestampNTZ")
+ val JDBC_PREFER_TIMESTAMP_NANOS = newOption("preferTimestampNanos")
val JDBC_HINT_STRING = newOption("hint")
}
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 2989c0975143..6636db6e70bb 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
@@ -106,7 +106,8 @@ object JDBCRDD extends Logging {
statement.setQueryTimeout(options.queryTimeout)
Using.resource(statement.executeQuery()) { rs =>
JdbcUtils.getSchema(conn, rs, dialect, alwaysNullable = true,
- isTimestampNTZ = options.preferTimestampNTZ)
+ isTimestampNTZ = options.preferTimestampNTZ,
+ preferTimestampNanos = options.preferTimestampNanos)
}
}
}
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 bd8d04760da6..3461bcf493e4 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
@@ -22,13 +22,13 @@ import java.nio.charset.StandardCharsets
import java.sql.{Date, ResultSet, Time, Timestamp}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_MILLIS
+import
org.apache.spark.sql.catalyst.util.DateTimeConstants.{MICROS_PER_MILLIS,
NANOS_PER_MICROS}
import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.sql.catalyst.util.GenericArrayData
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.jdbc.JdbcDialect
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String}
/**
* A `JDBCValueGetter` is responsible for getting a value from a `ResultSet`
into a field of an
@@ -189,6 +189,44 @@ private[jdbc] object JDBCValueGetter {
}
}
+ // 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
+ // JVM-default-zone shift of the microsecond `getTimestamp` path. The value
is then floored to the
+ // column precision.
+ final case class TimestampNTZNanosGetter(precision: Int) extends
JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val localDateTime = rs.getObject(pos + 1,
classOf[java.time.LocalDateTime])
+ if (localDateTime != null) {
+ val fullNanos =
+ localDateTimeToTimestampNanos(localDateTime,
TimestampNTZNanosType.NANOS_PRECISION)
+ row.update(pos, truncateTimestampNanosToPrecision(fullNanos,
precision))
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
+ // Reads a driver TIMESTAMP into a nanosecond-precision instant. Mirrors the
microsecond
+ // `TimestampGetter` (including the Julian->Gregorian rebase in
`fromJavaTimestamp`), then
+ // re-attaches the sub-microsecond digits from `java.sql.Timestamp.getNanos`
that the micro path
+ // drops, before truncating to the column precision.
+ final case class TimestampLTZNanosGetter(dialect: JdbcDialect, precision:
Int)
+ extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ val converted = dialect.convertJavaTimestampToTimestamp(t)
+ val epochMicros = fromJavaTimestamp(converted)
+ val subMicroNanos = (converted.getNanos % NANOS_PER_MICROS).toShort
+ val fullNanos = TimestampNanosVal.fromParts(epochMicros, subMicroNanos)
+ row.update(pos, truncateTimestampNanosToPrecision(fullNanos,
precision))
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
case object BinaryBitGetter extends JDBCValueGetter {
def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
val bytes = rs.getBytes(pos + 1)
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 ca44e8b710b1..ccec597e6e0b 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
@@ -163,6 +163,12 @@ object JdbcUtils extends Logging with SQLConfHelper {
case TimestampNTZType => Option(JdbcType("TIMESTAMP",
java.sql.Types.TIMESTAMP))
case DateType => Option(JdbcType("DATE", java.sql.Types.DATE))
case t: TimeType => Option(JdbcType(s"TIME(${t.precision})",
java.sql.Types.TIME))
+ // Nanosecond-capable timestamps (precision 7-9) map to SQL
TIMESTAMP(p). Dialects may
+ // override this, e.g. to emit TIMESTAMP(p) WITH TIME ZONE for the LTZ
variant.
+ case t: TimestampNTZNanosType =>
+ Option(JdbcType(s"TIMESTAMP(${t.precision})",
java.sql.Types.TIMESTAMP))
+ case t: TimestampLTZNanosType =>
+ Option(JdbcType(s"TIMESTAMP(${t.precision})",
java.sql.Types.TIMESTAMP))
case t: DecimalType => Option(
JdbcType(s"DECIMAL(${t.precision},${t.scale})",
java.sql.Types.DECIMAL))
case _ => None
@@ -178,6 +184,10 @@ object JdbcUtils extends Logging with SQLConfHelper {
if (isTimestampNTZ) TimestampNTZType else TimestampType
}
+ private def getTimestampNanosType(isTimestampNTZ: Boolean, precision: Int):
DataType = {
+ if (isTimestampNTZ) TimestampNTZNanosType(precision) else
TimestampLTZNanosType(precision)
+ }
+
/**
* Maps a JDBC type to a Catalyst type. This function is called only when
* the JdbcDialect class corresponding to your database driver returns null.
@@ -191,7 +201,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
precision: Int,
scale: Int,
signed: Boolean,
- isTimestampNTZ: Boolean): DataType = sqlType match {
+ isTimestampNTZ: Boolean,
+ preferTimestampNanos: Boolean = false): DataType = sqlType match {
case java.sql.Types.BIGINT => if (signed) LongType else DecimalType(20, 0)
case java.sql.Types.BINARY => BinaryType
case java.sql.Types.BIT => BooleanType // @see JdbcDialect for quirks
@@ -233,7 +244,16 @@ object JdbcUtils extends Logging with SQLConfHelper {
else TimeType.DEFAULT_PRECISION
TimeType(timePrecision)
} else getTimestampType(isTimestampNTZ)
- case java.sql.Types.TIMESTAMP => getTimestampType(isTimestampNTZ)
+ case java.sql.Types.TIMESTAMP =>
+ // When nanosecond timestamps are requested (and the preview feature is
enabled), a driver
+ // TIMESTAMP that reports a sub-microsecond fractional-second scale
(7-9) is mapped to the
+ // nanosecond-capable type. Otherwise the historical microsecond mapping
is preserved.
+ if (preferTimestampNanos &&
+ scale >= TimestampNTZNanosType.MIN_PRECISION &&
+ scale <= TimestampNTZNanosType.MAX_PRECISION &&
+ conf.timestampNanosTypesEnabled) {
+ getTimestampNanosType(isTimestampNTZ, scale)
+ } else getTimestampType(isTimestampNTZ)
case java.sql.Types.TINYINT => IntegerType
case java.sql.Types.VARBINARY => BinaryType
case java.sql.Types.VARCHAR if conf.charVarcharAsString => StringType
@@ -262,7 +282,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
try {
statement.setQueryTimeout(options.queryTimeout)
Some(getSchema(conn, statement.executeQuery(), dialect,
- isTimestampNTZ = options.preferTimestampNTZ))
+ isTimestampNTZ = options.preferTimestampNTZ,
+ preferTimestampNanos = options.preferTimestampNanos))
} catch {
case _: SQLException => None
} finally {
@@ -286,7 +307,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
resultSet: ResultSet,
dialect: JdbcDialect,
alwaysNullable: Boolean = false,
- isTimestampNTZ: Boolean = false): StructType = {
+ isTimestampNTZ: Boolean = false,
+ preferTimestampNanos: Boolean = false): StructType = {
val rsmd = resultSet.getMetaData
val ncols = rsmd.getColumnCount
val fields = new Array[StructField](ncols)
@@ -326,13 +348,15 @@ object JdbcUtils extends Logging with SQLConfHelper {
}
metadata.putBoolean("isSigned", isSigned)
metadata.putBoolean("isTimestampNTZ", isTimestampNTZ)
+ metadata.putBoolean("preferTimestampNanos", preferTimestampNanos)
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,
isTimestampNTZ))
+ getCatalystType(dataType, typeName, fieldSize, fieldScale, isSigned,
isTimestampNTZ,
+ preferTimestampNanos))
fields(i) = StructField(columnName, columnType, nullable,
metadata.build())
i = i + 1
}
@@ -448,6 +472,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
case TimestampNTZType if metadata.contains("logical_time_type") =>
JDBCValueGetter.LogicalTimeNTZGetter(dialect)
case TimestampNTZType => JDBCValueGetter.TimestampNTZGetter(dialect)
+ case t: TimestampNTZNanosType =>
JDBCValueGetter.TimestampNTZNanosGetter(t.precision)
+ case t: TimestampLTZNanosType =>
JDBCValueGetter.TimestampLTZNanosGetter(dialect, t.precision)
case BinaryType if metadata.contains("binarylong") =>
JDBCValueGetter.BinaryBitGetter
case BinaryType => JDBCValueGetter.BytesGetter
case _: YearMonthIntervalType =>
JDBCValueGetter.YearMonthIntervalGetter(dialect)
@@ -518,6 +544,23 @@ object JdbcUtils extends Logging with SQLConfHelper {
stmt.setTimestamp(pos + 1,
dialect.convertTimestampNTZToJavaTimestamp(row.getAs[java.time.LocalDateTime](pos)))
+ // Nanosecond-precision timestamps are always materialized as java.time
values by the Nanos
+ // encoders (independent of the datetimeJava8Api flag).
+ // NTZ is time-zone independent: write the LocalDateTime wall-clock
directly (mirroring the
+ // TimeType setter) so the full sub-microsecond fraction survives without
a zone shift.
+ case _: TimestampNTZNanosType =>
+ (stmt: PreparedStatement, row: Row, pos: Int) =>
+ stmt.setObject(pos + 1, row.getAs[java.time.LocalDateTime](pos))
+
+ // LTZ is an absolute instant: go through the micro java.sql.Timestamp
path (matching the
+ // TimestampType setter), then restore the full nanosecond-of-second the
micro path drops.
+ case _: TimestampLTZNanosType =>
+ (stmt: PreparedStatement, row: Row, pos: Int) =>
+ val instant = row.getAs[Instant](pos)
+ val ts = toJavaTimestamp(instantToMicros(instant))
+ ts.setNanos(instant.getNano)
+ stmt.setTimestamp(pos + 1, ts)
+
case DateType =>
if (conf.datetimeJava8ApiEnabled) {
(stmt: PreparedStatement, row: Row, pos: Int) =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
b/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
index 110889563a2b..82cd74a73618 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
@@ -192,8 +192,8 @@ trait CreatableRelationProvider {
case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
case BinaryType | BooleanType | ByteType | _: CharType | DateType | _:
DecimalType |
DoubleType | FloatType | IntegerType | LongType | NullType |
ObjectType(_) | ShortType |
- _: StringType | _: TimeType | TimestampNTZType | TimestampType |
- _: VarcharType => true
+ _: StringType | _: TimeType | _: TimestampNTZNanosType | _:
TimestampLTZNanosType |
+ TimestampNTZType | TimestampType | _: VarcharType => true
case _ => false
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala
index b3b0eb811ead..1caebf506a3d 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala
@@ -65,6 +65,7 @@ class JDBCTableCatalogSuite extends SharedSparkSession {
jdbcClientType: String): Metadata = new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean("isSigned", dataType.isInstanceOf[NumericType])
.putString("jdbcClientType", jdbcClientType)
.build()
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 0213d18922b7..288871b3b1ce 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
@@ -93,6 +93,7 @@ class JDBCSuite extends SharedSparkSession {
jdbcClientType: String): Metadata = new MetadataBuilder()
.putLong("scale", 0)
.putBoolean("isTimestampNTZ", false)
+ .putBoolean("preferTimestampNanos", false)
.putBoolean("isSigned", dataType.isInstanceOf[NumericType])
.putString("jdbcClientType", jdbcClientType)
.build()
@@ -837,6 +838,72 @@ class JDBCSuite extends SharedSparkSession {
}
}
+ test("SPARK-57460: JDBC TIMESTAMP keeps microsecond mapping by default") {
+ // Without preferTimestampNanos, a driver TIMESTAMP(9) still infers as
microsecond
+ // TimestampType even when the nanos preview feature is enabled.
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val conn = java.sql.DriverManager.getConnection(urlWithUserAndPass)
+ try {
+ conn.createStatement().execute("CREATE TABLE TEST.TS_DEFAULT (t
TIMESTAMP(9))")
+ conn.createStatement().execute(
+ "INSERT INTO TEST.TS_DEFAULT VALUES (TIMESTAMP '2020-02-02
04:13:14.123456789')")
+ val df = spark.read.jdbc(urlWithUserAndPass, "TEST.TS_DEFAULT", new
Properties())
+ assert(df.schema("T").dataType === TimestampType)
+ } finally {
+ conn.createStatement().execute("DROP TABLE IF EXISTS TEST.TS_DEFAULT")
+ conn.close()
+ }
+ }
+ }
+
+ test("SPARK-57460: JDBC TIMESTAMP reads nanosecond LTZ precision when
requested") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val conn = java.sql.DriverManager.getConnection(urlWithUserAndPass)
+ try {
+ conn.createStatement().execute("CREATE TABLE TEST.TS_NANOS_LTZ (t
TIMESTAMP(9))")
+ conn.createStatement().execute(
+ "INSERT INTO TEST.TS_NANOS_LTZ VALUES (TIMESTAMP '2020-02-02
04:13:14.123456789')")
+ val df = spark.read
+ .option("preferTimestampNanos", "true")
+ .jdbc(urlWithUserAndPass, "TEST.TS_NANOS_LTZ", new Properties())
+ assert(df.schema("T").dataType === TimestampLTZNanosType(9))
+ val result = df.collect()
+ // H2 stores TIMESTAMP as local wall-clock; the LTZ read binds it to
the session zone.
+ val expected = java.time.LocalDateTime.of(2020, 2, 2, 4, 13, 14,
123456789)
+ .atZone(java.time.ZoneId.systemDefault()).toInstant
+ assert(result(0).getAs[java.time.Instant](0) === expected)
+ } finally {
+ conn.createStatement().execute("DROP TABLE IF EXISTS
TEST.TS_NANOS_LTZ")
+ conn.close()
+ }
+ }
+ }
+
+ test("SPARK-57460: JDBC TIMESTAMP reads nanosecond NTZ precision when
requested") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val conn = java.sql.DriverManager.getConnection(urlWithUserAndPass)
+ try {
+ conn.createStatement().execute("CREATE TABLE TEST.TS_NANOS_NTZ (t
TIMESTAMP(8))")
+ conn.createStatement().execute(
+ "INSERT INTO TEST.TS_NANOS_NTZ VALUES (TIMESTAMP '2020-02-02
04:13:14.123456789')")
+ val df = spark.read
+ .option("preferTimestampNanos", "true")
+ .option("preferTimestampNTZ", "true")
+ .jdbc(urlWithUserAndPass, "TEST.TS_NANOS_NTZ", new Properties())
+ // Reported scale 8 -> precision 8. H2 rounds the stored value to 8
fractional digits
+ // (.123456789 -> .12345679); Spark's read path only floors to
precision, so the value is
+ // preserved at 8 digits.
+ assert(df.schema("T").dataType === TimestampNTZNanosType(8))
+ val result = df.collect()
+ assert(result(0).getAs[java.time.LocalDateTime](0) ===
+ java.time.LocalDateTime.of(2020, 2, 2, 4, 13, 14, 123456790))
+ } finally {
+ conn.createStatement().execute("DROP TABLE IF EXISTS
TEST.TS_NANOS_NTZ")
+ conn.close()
+ }
+ }
+ }
+
test("test DATE types") {
val rows = spark.read.jdbc(
urlWithUserAndPass, "TEST.TIMETYPES", new Properties()).collect()
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
index bce7a883fa2b..1045a5afa5d7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
@@ -28,15 +28,12 @@ import org.scalatest.BeforeAndAfter
import org.apache.spark.SparkException
import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
-import org.apache.spark.sql.{AnalysisException, Column, DataFrame, Row,
SaveMode}
-import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode}
import org.apache.spark.sql.catalyst.parser.ParseException
-import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.TimestampNanosVal
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils
@@ -701,22 +698,45 @@ class JDBCWriteSuite extends SharedSparkSession with
BeforeAndAfter {
=== java.sql.Timestamp.valueOf("2020-02-02 04:13:14.56789"))
}
- test("SPARK-57166: nanosecond timestamp types are not supported in JDBC
write") {
+ test("SPARK-57460: JDBC nanosecond timestamp write round-trip") {
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
- Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9)).foreach {
nanosType =>
- // The nanos literal is built directly from its internal value to
avoid relying on
- // cast/parser support.
- val nanosLiteral = Literal.create(new TimestampNanosVal(0L,
0.toShort), nanosType)
- val df = spark.range(1).select(Column(nanosLiteral).as("ts"))
- checkErrorMatchPVals(
- exception = intercept[AnalysisException] {
- df.write.jdbc(url, "TEST.NANOSTYPES", new Properties())
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" ->
java.util.regex.Pattern.quote(s""""${nanosType.sql}""""),
- "format" -> ".*"))
+ Seq(7, 8, 9).foreach { precision =>
+ // TIMESTAMP_NTZ(p): write a LocalDateTime, read it back as a nanos
NTZ column.
+ val ntzTable = "TEST.NANOS_NTZ"
+ val ldt = java.time.LocalDateTime.of(2020, 2, 2, 4, 13, 14, 123456789)
+ val ntzSchema = new StructType().add("t",
TimestampNTZNanosType(precision))
+ val ntzDf = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(ldt))), ntzSchema)
+ ntzDf.write.mode(SaveMode.Overwrite).jdbc(url, ntzTable, new
Properties())
+ val ntzReadBack = spark.read
+ .option("preferTimestampNanos", "true")
+ .option("preferTimestampNTZ", "true")
+ .jdbc(url, ntzTable, new Properties())
+ assert(ntzReadBack.schema.fields(0).dataType ===
TimestampNTZNanosType(precision))
+ val expectedNtz = precision match {
+ case 7 => java.time.LocalDateTime.of(2020, 2, 2, 4, 13, 14,
123456700)
+ case 8 => java.time.LocalDateTime.of(2020, 2, 2, 4, 13, 14,
123456780)
+ case 9 => ldt
+ }
+ assert(ntzReadBack.collect()(0).getAs[java.time.LocalDateTime](0) ===
expectedNtz)
+
+ // TIMESTAMP_LTZ(p): write an Instant, read it back as a nanos LTZ
column.
+ val ltzTable = "TEST.NANOS_LTZ"
+ val instant = java.time.Instant.parse("2020-02-02T12:13:14.123456789Z")
+ val ltzSchema = new StructType().add("t",
TimestampLTZNanosType(precision))
+ val ltzDf = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(instant))), ltzSchema)
+ ltzDf.write.mode(SaveMode.Overwrite).jdbc(url, ltzTable, new
Properties())
+ val ltzReadBack = spark.read
+ .option("preferTimestampNanos", "true")
+ .jdbc(url, ltzTable, new Properties())
+ assert(ltzReadBack.schema.fields(0).dataType ===
TimestampLTZNanosType(precision))
+ val expectedLtz = precision match {
+ case 7 => java.time.Instant.parse("2020-02-02T12:13:14.123456700Z")
+ case 8 => java.time.Instant.parse("2020-02-02T12:13:14.123456780Z")
+ case 9 => instant
+ }
+ assert(ltzReadBack.collect()(0).getAs[java.time.Instant](0) ===
expectedLtz)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
