You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/11/13 23:06:31 UTC
(spark) branch master updated: [SPARK-54209][CONNECT] Support TIMESTAMP type in SparkConnectResultSet
This is an automated email from the ASF dual-hosted git repository.
dongjoon 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 9f1dc1ccf8b7 [SPARK-54209][CONNECT] Support TIMESTAMP type in SparkConnectResultSet
9f1dc1ccf8b7 is described below
commit 9f1dc1ccf8b7af08d056f8f30c8ef0b091550657
Author: vinodkc <vi...@gmail.com>
AuthorDate: Thu Nov 13 15:06:18 2025 -0800
[SPARK-54209][CONNECT] Support TIMESTAMP type in SparkConnectResultSet
### What changes were proposed in this pull request?
Add TIMESTAMP type support to the Spark Connect JDBC client
### Why are the changes needed?
TIMESTAMP is a fundamental SQL type required for JDBC compliance and interoperability.
### Does this PR introduce _any_ user-facing change?
Yes, it's part of a new feature under Spark connect JDBC support.
### How was this patch tested?
Added new UTs in `SparkConnectJdbcDataTypeSuite`
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #53046 from vinodkc/br_SPARK-54209.
Authored-by: vinodkc <vi...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../client/jdbc/SparkConnectResultSet.scala | 41 ++++-
.../connect/client/jdbc/util/JdbcTypeUtils.scala | 13 +-
.../jdbc/SparkConnectJdbcDataTypeSuite.scala | 182 ++++++++++++++++++++-
.../client/jdbc/SparkConnectResultSetSuite.scala | 64 ++++++++
4 files changed, 292 insertions(+), 8 deletions(-)
diff --git a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
index c4af5b6f12f0..8b2d4578d0aa 100644
--- a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
+++ b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
@@ -27,6 +27,7 @@ import java.util.Calendar
import org.apache.spark.sql.Row
import org.apache.spark.sql.connect.client.SparkResult
import org.apache.spark.sql.connect.client.jdbc.util.JdbcErrorUtils
+import org.apache.spark.sql.types.{TimestampNTZType, TimestampType}
class SparkConnectResultSet(
sparkResult: SparkResult[Row],
@@ -162,8 +163,27 @@ class SparkConnectResultSet(
}
}
- override def getTimestamp(columnIndex: Int): Timestamp =
- throw new SQLFeatureNotSupportedException
+ override def getTimestamp(columnIndex: Int): Timestamp = {
+ getColumnValue(columnIndex, null: Timestamp) { idx =>
+ val value = currentRow.get(idx)
+ if (value == null) {
+ null
+ } else {
+ sparkResult.schema.fields(idx).dataType match {
+ case TimestampNTZType =>
+ // TIMESTAMP_NTZ is represented as LocalDateTime
+ Timestamp.valueOf(value.asInstanceOf[java.time.LocalDateTime])
+ case TimestampType =>
+ // TIMESTAMP is represented as java.sql.Timestamp
+ value.asInstanceOf[Timestamp]
+ case other =>
+ throw new SQLException(
+ s"Cannot call getTimestamp() on column of type $other. " +
+ s"Expected TIMESTAMP or TIMESTAMP_NTZ.")
+ }
+ }
+ }
+ }
override def getAsciiStream(columnIndex: Int): InputStream =
throw new SQLFeatureNotSupportedException
@@ -211,7 +231,7 @@ class SparkConnectResultSet(
getTime(findColumn(columnLabel))
override def getTimestamp(columnLabel: String): Timestamp =
- throw new SQLFeatureNotSupportedException
+ getTimestamp(findColumn(columnLabel))
override def getAsciiStream(columnLabel: String): InputStream =
throw new SQLFeatureNotSupportedException
@@ -558,11 +578,20 @@ class SparkConnectResultSet(
override def getTime(columnLabel: String, cal: Calendar): Time =
getTime(findColumn(columnLabel))
- override def getTimestamp(columnIndex: Int, cal: Calendar): Timestamp =
- throw new SQLFeatureNotSupportedException
+ /**
+ * Note: The Calendar parameter is ignored. Spark Connect handles timezone conversions
+ * server-side to avoid client/server timezone inconsistencies.
+ */
+ override def getTimestamp(columnIndex: Int, cal: Calendar): Timestamp = {
+ getTimestamp(columnIndex)
+ }
+ /**
+ * Note: The Calendar parameter is ignored. Spark Connect handles timezone conversions
+ * server-side to avoid client/server timezone inconsistencies.
+ */
override def getTimestamp(columnLabel: String, cal: Calendar): Timestamp =
- throw new SQLFeatureNotSupportedException
+ getTimestamp(findColumn(columnLabel), cal)
override def getURL(columnIndex: Int): URL =
throw new SQLFeatureNotSupportedException
diff --git a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
index b3b28359076e..91de849e46c0 100644
--- a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
+++ b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
@@ -37,6 +37,8 @@ private[jdbc] object JdbcTypeUtils {
case StringType => Types.VARCHAR
case _: DecimalType => Types.DECIMAL
case DateType => Types.DATE
+ case TimestampType => Types.TIMESTAMP
+ case TimestampNTZType => Types.TIMESTAMP
case BinaryType => Types.BINARY
case _: TimeType => Types.TIME
case other =>
@@ -55,6 +57,8 @@ private[jdbc] object JdbcTypeUtils {
case StringType => classOf[String].getName
case _: DecimalType => classOf[JBigDecimal].getName
case DateType => classOf[Date].getName
+ case TimestampType => classOf[Timestamp].getName
+ case TimestampNTZType => classOf[Timestamp].getName
case BinaryType => classOf[Array[Byte]].getName
case _: TimeType => classOf[Time].getName
case other =>
@@ -64,7 +68,8 @@ private[jdbc] object JdbcTypeUtils {
def isSigned(field: StructField): Boolean = field.dataType match {
case ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
_: DecimalType => true
- case NullType | BooleanType | StringType | DateType | BinaryType | _: TimeType => false
+ case NullType | BooleanType | StringType | DateType | BinaryType | _: TimeType |
+ TimestampType | TimestampNTZType => false
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
}
@@ -81,6 +86,8 @@ private[jdbc] object JdbcTypeUtils {
case StringType => 255
case DecimalType.Fixed(p, _) => p
case DateType => 10
+ case TimestampType => 29
+ case TimestampNTZType => 29
case BinaryType => Int.MaxValue
// Returns the Spark SQL TIME type precision, even though java.sql.ResultSet.getTime()
// can only retrieve up to millisecond precision (3) due to java.sql.Time limitations.
@@ -94,6 +101,8 @@ private[jdbc] object JdbcTypeUtils {
def getScale(field: StructField): Int = field.dataType match {
case FloatType => 7
case DoubleType => 15
+ case TimestampType => 6
+ case TimestampNTZType => 6
case NullType | BooleanType | ByteType | ShortType | IntegerType | LongType | StringType |
DateType | BinaryType | _: TimeType => 0
case DecimalType.Fixed(_, s) => s
@@ -111,6 +120,8 @@ private[jdbc] object JdbcTypeUtils {
case StringType =>
getPrecision(field)
case DateType => 10 // length of `YYYY-MM-DD`
+ case TimestampType => 29 // length of `YYYY-MM-DD HH:MM:SS.SSSSSS`
+ case TimestampNTZType => 29 // length of `YYYY-MM-DD HH:MM:SS.SSSSSS`
case BinaryType => Int.MaxValue
case TimeType(precision) if precision > 0 => 8 + 1 + precision // length of `HH:MM:SS.ffffff`
case TimeType(_) => 8 // length of `HH:MM:SS`
diff --git a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
index 370f53be8a2b..79b5eb8a052b 100644
--- a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
+++ b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
@@ -268,7 +268,9 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
("cast(1 AS DECIMAL(10,5))", (rs: ResultSet) => rs.getBigDecimal(999)),
("CAST(X'0A0B0C' AS BINARY)", (rs: ResultSet) => rs.getBytes(999)),
("date '2025-11-15'", (rs: ResultSet) => rs.getBytes(999)),
- ("time '12:34:56.123456'", (rs: ResultSet) => rs.getBytes(999))
+ ("time '12:34:56.123456'", (rs: ResultSet) => rs.getBytes(999)),
+ ("timestamp '2025-11-15 10:30:45.123456'", (rs: ResultSet) => rs.getTimestamp(999)),
+ ("timestamp_ntz '2025-11-15 10:30:45.789012'", (rs: ResultSet) => rs.getTimestamp(999)),
).foreach {
case (query, getter) =>
withExecuteQuery(s"SELECT $query") { rs =>
@@ -567,4 +569,182 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
}
}
}
+
+test("get timestamp type") {
+ withExecuteQuery("SELECT timestamp '2025-11-15 10:30:45.123456'") { rs =>
+ assert(rs.next())
+ val timestamp = rs.getTimestamp(1)
+ assert(timestamp !== null)
+ assert(timestamp === java.sql.Timestamp.valueOf("2025-11-15 10:30:45.123456"))
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnName(1) === "TIMESTAMP '2025-11-15 10:30:45.123456'")
+ assert(metaData.getColumnLabel(1) === "TIMESTAMP '2025-11-15 10:30:45.123456'")
+ assert(metaData.getColumnType(1) === Types.TIMESTAMP)
+ assert(metaData.getColumnTypeName(1) === "TIMESTAMP")
+ assert(metaData.getColumnClassName(1) === "java.sql.Timestamp")
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 29)
+ assert(metaData.getScale(1) === 6)
+ assert(metaData.getColumnDisplaySize(1) === 29)
+ }
+ }
+
+ test("get timestamp type with null") {
+ withExecuteQuery("SELECT cast(null as timestamp)") { rs =>
+ assert(rs.next())
+ assert(rs.getTimestamp(1) === null)
+ assert(rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnName(1) === "CAST(NULL AS TIMESTAMP)")
+ assert(metaData.getColumnLabel(1) === "CAST(NULL AS TIMESTAMP)")
+ assert(metaData.getColumnType(1) === Types.TIMESTAMP)
+ assert(metaData.getColumnTypeName(1) === "TIMESTAMP")
+ assert(metaData.getColumnClassName(1) === "java.sql.Timestamp")
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 29)
+ assert(metaData.getScale(1) === 6)
+ assert(metaData.getColumnDisplaySize(1) === 29)
+ }
+ }
+
+ test("get timestamp type by column label and with calendar") {
+ withExecuteQuery("SELECT timestamp '2025-11-15 10:30:45.987654' as test_timestamp") { rs =>
+ assert(rs.next())
+
+ // Test by column label
+ val timestamp = rs.getTimestamp("test_timestamp")
+ assert(timestamp !== null)
+ assert(timestamp === java.sql.Timestamp.valueOf("2025-11-15 10:30:45.987654"))
+ assert(!rs.wasNull)
+
+ // Test with calendar - should return same value (Calendar is ignored)
+ // Note: Spark Connect handles timezone at server, Calendar param is for API compliance
+ val calUTC = java.util.Calendar.getInstance(java.util.TimeZone.getTimeZone("UTC"))
+ val timestampUTC = rs.getTimestamp(1, calUTC)
+ assert(timestampUTC !== null)
+ assert(timestampUTC.getTime === timestamp.getTime)
+
+ val calPST = java.util.Calendar.getInstance(
+ java.util.TimeZone.getTimeZone("America/Los_Angeles"))
+ val timestampPST = rs.getTimestamp(1, calPST)
+ assert(timestampPST !== null)
+ // Same value regardless of calendar
+ assert(timestampPST.getTime === timestamp.getTime)
+ assert(timestampUTC.getTime === timestampPST.getTime)
+
+ // Test with calendar by label
+ val timestampLabel = rs.getTimestamp("test_timestamp", calUTC)
+ assert(timestampLabel !== null)
+ assert(timestampLabel.getTime === timestamp.getTime)
+
+ // Test with null calendar - returns same value
+ val timestampNullCal = rs.getTimestamp(1, null)
+ assert(timestampNullCal !== null)
+ assert(timestampNullCal.getTime === timestamp.getTime)
+
+ assert(!rs.next())
+ }
+ }
+
+ test("get timestamp type with calendar for null value") {
+ withExecuteQuery("SELECT cast(null as timestamp)") { rs =>
+ assert(rs.next())
+
+ // Calendar parameter should not affect null handling
+ val cal = java.util.Calendar.getInstance(java.util.TimeZone.getTimeZone("UTC"))
+ val timestamp = rs.getTimestamp(1, cal)
+ assert(timestamp === null)
+ assert(rs.wasNull)
+ assert(!rs.next())
+ }
+ }
+
+ test("get timestamp_ntz type") {
+ withExecuteQuery("SELECT timestamp_ntz '2025-11-15 10:30:45.123456'") { rs =>
+ assert(rs.next())
+ val timestamp = rs.getTimestamp(1)
+ assert(timestamp !== null)
+ assert(timestamp === java.sql.Timestamp.valueOf("2025-11-15 10:30:45.123456"))
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnName(1) === "TIMESTAMP_NTZ '2025-11-15 10:30:45.123456'")
+ assert(metaData.getColumnLabel(1) === "TIMESTAMP_NTZ '2025-11-15 10:30:45.123456'")
+ assert(metaData.getColumnType(1) === Types.TIMESTAMP)
+ assert(metaData.getColumnTypeName(1) === "TIMESTAMP_NTZ")
+ assert(metaData.getColumnClassName(1) === "java.sql.Timestamp")
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 29)
+ assert(metaData.getScale(1) === 6)
+ assert(metaData.getColumnDisplaySize(1) === 29)
+ }
+ }
+
+ test("get timestamp_ntz type by label, null, and with calendar") {
+ // Test with non-null value
+ withExecuteQuery("SELECT timestamp_ntz '2025-11-15 14:22:33.789456' as test_ts_ntz") { rs =>
+ assert(rs.next())
+
+ // Test by column label
+ val timestamp = rs.getTimestamp("test_ts_ntz")
+ assert(timestamp !== null)
+ assert(timestamp === java.sql.Timestamp.valueOf("2025-11-15 14:22:33.789456"))
+ assert(!rs.wasNull)
+
+ // Test with calendar - should return same value (Calendar is ignored)
+ val calUTC = java.util.Calendar.getInstance(java.util.TimeZone.getTimeZone("UTC"))
+ val timestampCal = rs.getTimestamp(1, calUTC)
+ assert(timestampCal !== null)
+ assert(timestampCal.getTime === timestamp.getTime)
+
+ assert(!rs.next())
+ }
+
+ // Test with null value
+ withExecuteQuery("SELECT cast(null as timestamp_ntz)") { rs =>
+ assert(rs.next())
+ assert(rs.getTimestamp(1) === null)
+ assert(rs.wasNull)
+ assert(!rs.next())
+ }
+ }
+
+ test("get timestamp types with spark.sql.datetime.java8API.enabled") {
+ withStatement { stmt =>
+ Seq(true, false).foreach { java8APIEnabled =>
+ stmt.execute(s"set spark.sql.datetime.java8API.enabled=$java8APIEnabled")
+
+ Using.resource(stmt.executeQuery(
+ """SELECT
+ | timestamp '2025-11-15 10:30:45.123456' as ts,
+ | timestamp_ntz '2025-11-15 14:22:33.789012' as ts_ntz
+ |""".stripMargin)) { rs =>
+ assert(rs.next())
+
+ // Test TIMESTAMP type
+ val timestamp = rs.getTimestamp(1)
+ assert(timestamp !== null)
+ assert(timestamp === java.sql.Timestamp.valueOf("2025-11-15 10:30:45.123456"))
+ assert(!rs.wasNull)
+
+ // Test TIMESTAMP_NTZ type
+ val timestampNtz = rs.getTimestamp(2)
+ assert(timestampNtz !== null)
+ assert(timestampNtz === java.sql.Timestamp.valueOf("2025-11-15 14:22:33.789012"))
+ assert(!rs.wasNull)
+
+ assert(!rs.next())
+ }
+ }
+ }
+ }
}
diff --git a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala
index ac2866837e93..21b8e261aef4 100644
--- a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala
+++ b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala
@@ -122,4 +122,68 @@ class SparkConnectResultSetSuite extends ConnectFunSuite with RemoteSparkSession
assert(rs.isAfterLast)
}
}
+
+ test("getTimestamp with multiple columns, rows, and types") {
+ withExecuteQuery(
+ """SELECT ts_tz, ts_ntz, id FROM VALUES
+ | (timestamp '2025-01-15 10:30:45.123456', timestamp_ntz '2025-06-20 14:22:33.789012', 1),
+ | (null, timestamp_ntz '2025-03-01 18:30:45.456789', 2),
+ | (timestamp '2025-10-31 23:59:59.999999', null, 3)
+ | AS t(ts_tz, ts_ntz, id)
+ |""".stripMargin) { rs =>
+
+ // Test findColumn
+ assert(rs.findColumn("ts_tz") === 1)
+ assert(rs.findColumn("ts_ntz") === 2)
+ assert(rs.findColumn("id") === 3)
+
+ // Verify metadata
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 3)
+ assert(metaData.getColumnTypeName(1) === "TIMESTAMP")
+ assert(metaData.getColumnTypeName(2) === "TIMESTAMP_NTZ")
+
+ // Row 1: Both timestamps have values
+ assert(rs.next())
+ assert(rs.getRow === 1)
+
+ val ts1 = rs.getTimestamp(1)
+ assert(ts1 !== null)
+ assert(ts1 === java.sql.Timestamp.valueOf("2025-01-15 10:30:45.123456"))
+ assert(!rs.wasNull)
+
+ val tsNtz1 = rs.getTimestamp("ts_ntz")
+ assert(tsNtz1 !== null)
+ assert(tsNtz1 === java.sql.Timestamp.valueOf("2025-06-20 14:22:33.789012"))
+ assert(!rs.wasNull)
+
+ // Row 2: TIMESTAMP is null, TIMESTAMP_NTZ has value
+ assert(rs.next())
+ assert(rs.getRow === 2)
+
+ val ts2 = rs.getTimestamp(1)
+ assert(ts2 === null)
+ assert(rs.wasNull)
+
+ val tsNtz2 = rs.getTimestamp(2)
+ assert(tsNtz2 !== null)
+ assert(tsNtz2 === java.sql.Timestamp.valueOf("2025-03-01 18:30:45.456789"))
+ assert(!rs.wasNull)
+
+ // Row 3: TIMESTAMP has value, TIMESTAMP_NTZ is null
+ assert(rs.next())
+ assert(rs.getRow === 3)
+
+ val ts3 = rs.getTimestamp("ts_tz")
+ assert(ts3 !== null)
+ assert(ts3 === java.sql.Timestamp.valueOf("2025-10-31 23:59:59.999999"))
+ assert(!rs.wasNull)
+
+ val tsNtz3 = rs.getTimestamp(2)
+ assert(tsNtz3 === null)
+ assert(rs.wasNull)
+
+ assert(!rs.next())
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org