This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 0919cd405232 [SPARK-58825][SQL] Support TimestampNTZType as a JDBC
partition column
0919cd405232 is described below
commit 0919cd4052329fc1cb8160c5965eefd15ebb1edd
Author: aleksandar-trajkovic-db <[email protected]>
AuthorDate: Fri Aug 21 11:17:08 2026 +0000
[SPARK-58825][SQL] Support TimestampNTZType as a JDBC partition column
### What changes were proposed in this pull request?
`JDBCRelation` restricts the `partitionColumn` option to numeric,
`DateType`, and `TimestampType` columns (SPARK-22814). This extends that set to
include `TimestampNTZType`:
- `verifyAndGetNormalizedPartitionColumn` accepts `TimestampNTZType`.
- Bound parsing uses `stringToTimestampWithoutTimeZone(_, allowTimeZone =
false)`, so `lowerBound`/`upperBound` are parsed as zoneless wall-clock values.
- Bound-to-string conversion formats NTZ micros in UTC, so no
session-time-zone shift is applied to the generated partition predicates.
Behavior for existing numeric/`DateType`/`TimestampType` partition columns
is unchanged.
### Why are the changes needed?
More JDBC sources map zoneless database types (timestamp-without-time-zone)
to `TimestampNTZType`. Such a column is a valid, orderable partition column,
but today it is rejected with `invalidPartitionColumnTypeError`.
### Does this PR introduce _any_ user-facing change?
Yes. A `TimestampNTZType` column can now be used as the JDBC
`partitionColumn`; it was previously rejected with
`invalidPartitionColumnTypeError`. There is no change for existing
numeric/`DateType`/`TimestampType` partition columns. This is a new capability
on `master`, not a change to released behavior.
### How was this patch tested?
New unit tests in `JDBCSuite`:
- `columnPartition supports TimestampNTZType partition column` — bound
normalization (`lowerBound`/`upperBound` → partition predicates) for an NTZ
partition column.
- `columnPartition rejects zoned bounds for a TimestampNTZType partition
column` — negative case, zoned bounds are rejected.
- `support TimestampNTZType partition column end-to-end` — reads a genuine
`TimestampNTZType` column partitioned across bounds and verifies the results.
All pass locally (`build/sbt "sql/testOnly
org.apache.spark.sql.jdbc.JDBCSuite"`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
---
This is my original work and I license it to the project under the
project's open source license.
Closes #58082 from aleksandar-trajkovic-db/spark-jdbc-ntz-partition.
Authored-by: aleksandar-trajkovic-db <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../execution/datasources/jdbc/JDBCRelation.scala | 14 ++-
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 107 +++++++++++++++++++++
2 files changed, 117 insertions(+), 4 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
index 972bb3e35ee6..1c529279744e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
@@ -27,14 +27,14 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession,
SQLContext}
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.util.{DateFormatter, DateTimeUtils,
TimestampFormatter}
-import org.apache.spark.sql.catalyst.util.DateTimeUtils.{getZoneId,
stringToDate, stringToTimestamp}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils.{getZoneId,
stringToDate, stringToTimestamp, stringToTimestampWithoutTimeZone}
import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.jdbc.JdbcDialects
import org.apache.spark.sql.sources._
-import org.apache.spark.sql.types.{DataType, DateType, NumericType,
StructType, TimestampType}
+import org.apache.spark.sql.types.{DataType, DateType, NumericType,
StructType, TimestampNTZType, TimestampType}
import org.apache.spark.unsafe.types.UTF8String
/**
@@ -188,7 +188,7 @@ private[sql] object JDBCRelation extends Logging {
columnName, schema.simpleString(maxNumToStringFields))
}
column.dataType match {
- case _: NumericType | DateType | TimestampType =>
+ case _: NumericType | DateType | TimestampType | TimestampNTZType =>
case _ =>
throw QueryCompilationErrors.invalidPartitionColumnTypeError(column)
}
@@ -209,6 +209,7 @@ private[sql] object JDBCRelation extends Logging {
case _: NumericType => value.toLong
case DateType => parse(stringToDate).toLong
case TimestampType => parse(stringToTimestamp(_, getZoneId(timeZoneId)))
+ case TimestampNTZType => parse(stringToTimestampWithoutTimeZone(_,
allowTimeZone = false))
}
}
@@ -224,12 +225,17 @@ private[sql] object JDBCRelation extends Logging {
val timestampFormatter = TimestampFormatter.getFractionFormatter(
DateTimeUtils.getZoneId(timeZoneId))
timestampFormatter.format(value)
+ case TimestampNTZType =>
+ // NTZ micros are zoneless wall-clock values; format in UTC so no
zone shift is applied.
+ val timestampFormatter = TimestampFormatter.getFractionFormatter(
+ DateTimeUtils.getZoneId("UTC"))
+ timestampFormatter.format(value)
}
s"'$dateTimeStr'"
}
columnType match {
case _: NumericType => value.toString
- case DateType | TimestampType => dateTimeToString()
+ case DateType | TimestampType | TimestampNTZType => dateTimeToString()
}
}
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 16f3f697ff17..6642b527e40e 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
@@ -479,6 +479,67 @@ class JDBCSuite extends SharedSparkSession {
assert(lastPredicate == """"PartitionColumn" >= '2020-08-02'""")
}
+ test("columnPartition supports TimestampNTZType partition column") {
+ val schema = StructType(Seq(
+ StructField("PartitionColumn", TimestampNTZType)
+ ))
+
+ // (lowerBound, upperBound, numPartitions, expected where clauses in
partition order).
+ val cases = Seq(
+ ("2018-07-06 10:00:00", "2018-07-06 16:00:00", "3", Seq(
+ """"PartitionColumn" < '2018-07-06 12:00:00' or "PartitionColumn" is
null""",
+ """"PartitionColumn" >= '2018-07-06 12:00:00' AND """ +
+ """"PartitionColumn" < '2018-07-06 14:00:00'""",
+ """"PartitionColumn" >= '2018-07-06 14:00:00'""")),
+ // Fractional-second bounds parse, and the zoneless midpoint keeps
sub-second precision.
+ ("2018-07-06 10:00:00.100", "2018-07-06 10:00:00.300", "2", Seq(
+ """"PartitionColumn" < '2018-07-06 10:00:00.2' or "PartitionColumn" is
null""",
+ """"PartitionColumn" >= '2018-07-06 10:00:00.2'"""))
+ )
+
+ // NTZ bounds are zoneless, so the generated predicates must be identical
regardless of the
+ // session time zone (unlike TimestampType, which shifts by the zone).
+ Seq("UTC", "America/Los_Angeles", "Asia/Kolkata").foreach { tz =>
+ cases.foreach { case (lowerBound, upperBound, numPartitions, expected) =>
+ val partitions = JDBCRelation.columnPartition(
+ schema,
+ analysis.caseInsensitiveResolution,
+ tz,
+ new JDBCOptions(url, "table", Map(
+ "lowerBound" -> lowerBound,
+ "upperBound" -> upperBound,
+ "numPartitions" -> numPartitions,
+ "partitionColumn" -> "PartitionColumn")))
+
+ val clauses = partitions.map(_.asInstanceOf[JDBCPartition].whereClause)
+ assert(clauses === expected.toArray,
+ s"NTZ partition clauses should be time-zone independent, but
differed for tz=$tz " +
+ s"(bounds $lowerBound..$upperBound)")
+ }
+ }
+ }
+
+ test("columnPartition rejects zoned bounds for a TimestampNTZType partition
column") {
+ val schema = StructType(Seq(
+ StructField("PartitionColumn", TimestampNTZType)
+ ))
+ // allowTimeZone = false: a bound carrying a zone offset is rejected
rather than silently
+ // shifted, so NTZ bounds stay zoneless.
+ val e = intercept[IllegalArgumentException] {
+ JDBCRelation.columnPartition(
+ schema,
+ analysis.caseInsensitiveResolution,
+ "America/Los_Angeles",
+ new JDBCOptions(url, "table", Map(
+ "lowerBound" -> "2018-07-06 10:00:00+05:00",
+ "upperBound" -> "2018-07-06 16:00:00+05:00",
+ "numPartitions" -> "2",
+ "partitionColumn" -> "PartitionColumn")))
+ }
+ assert(e.getMessage.contains("Cannot parse the bound value"))
+ assert(e.getMessage.contains("2018-07-06 10:00:00+05:00"))
+ }
+
test("overflow of partition bound difference does not give negative stride")
{
val df = sql("SELECT * FROM partsoverflow")
checkNumPartitions(df, expectedNumPartitions = 3)
@@ -2487,6 +2548,52 @@ class JDBCSuite extends SharedSparkSession {
checkAnswer(df2, expectedResult)
}
+ test("support TimestampNTZType partition column end-to-end") {
+ val tableName = "timestamp_ntz_partition_table"
+ // Write a genuine TimestampNTZType column through Spark so it round-trips
as a zoneless
+ // wall-clock value (a raw JDBC TIMESTAMP column would pick up a
JVM-time-zone shift on read).
+ val df = Seq(
+ "2018-07-06T05:50:00",
+ "2018-07-06T08:10:08",
+ "2018-07-08T13:32:01",
+ "2018-07-12T09:51:15"
+ ).map(LocalDateTime.parse).toDF("t")
+ df.write.format("jdbc")
+ .mode("overwrite")
+ .option("url", urlWithUserAndPass)
+ .option("dbtable", tableName)
+ .save()
+
+ // Bounds are zoneless, so both the generated predicates and the results
must be identical
+ // regardless of the JVM default time zone.
+ DateTimeTestUtils.outstandingZoneIds.foreach { zoneId =>
+ DateTimeTestUtils.withDefaultTimeZone(zoneId) {
+ val readDf = spark.read.format("jdbc")
+ .option("url", urlWithUserAndPass)
+ .option("dbtable", tableName)
+ .option("preferTimestampNTZ", true)
+ .option("partitionColumn", "t")
+ .option("lowerBound", "2018-07-04 03:30:00")
+ .option("upperBound", "2018-07-27 14:11:05")
+ .option("numPartitions", 2)
+ .load()
+
+ assert(readDf.schema("t").dataType === TimestampNTZType)
+
+ readDf.logicalPlan match {
+ case LogicalRelationWithTable(JDBCRelation(_, parts, _, _), _) =>
+ val whereClauses =
parts.map(_.asInstanceOf[JDBCPartition].whereClause).toSet
+ assert(whereClauses === Set(
+ """"t" < '2018-07-15 20:50:32.5' or "t" is null""",
+ """"t" >= '2018-07-15 20:50:32.5'"""),
+ s"NTZ partition predicates should be time-zone independent, but
differed for " +
+ s"zone=$zoneId")
+ }
+ checkAnswer(readDf, df)
+ }
+ }
+ }
+
test("throws an exception for unsupported partition column types") {
val errMsg = intercept[AnalysisException] {
spark.read.format("jdbc")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
