You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/07/02 12:53:54 UTC
(spark) branch master updated: [SPARK-57855][SQL] Refactor JDBC value getters into a sealed type for readability
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 589e0078af2f [SPARK-57855][SQL] Refactor JDBC value getters into a sealed type for readability
589e0078af2f is described below
commit 589e0078af2fcd6bd99c795e43483e2877c21e18
Author: Abdelrahman Gamal <a....@databricks.com>
AuthorDate: Thu Jul 2 20:53:34 2026 +0800
[SPARK-57855][SQL] Refactor JDBC value getters into a sealed type for readability
### What changes were proposed in this pull request?
`JdbcUtils.makeGetter` builds a per-column reader (`JDBCValueGetter`) that reads a column from a `ResultSet`, converts it to Spark's internal representation, and stores it into an `InternalRow`. `JDBCValueGetter` was a private type alias for `(ResultSet, InternalRow, Int) => Unit`, so `makeGetter` returned an anonymous closure per column.
This PR makes the getter an explicit type:
- Adds `JDBCValueGetter.scala` with a `private[jdbc]` sealed trait `JDBCValueGetter` (single `apply(rs, row, pos)` method) and one named case per getter in its companion object — `case object` for stateless getters, `final case class` for parameterized ones — plus the getter-only helpers `nullSafeConvert` / `arrayConverter`.
- `JdbcUtils.makeGetter` becomes a straightforward `DataType` -> getter selection.
Class hierarchy (sealed):
```
JDBCValueGetter
├─ BooleanGetter, DoubleGetter, FloatGetter, IntGetter, LongGetter, ShortGetter,
│ ByteGetter, StringGetter, RowIdGetter, BytesGetter, BinaryLongGetter,
│ BinaryBitGetter, TimeGetter, LogicalTimeGetter, PostgresBitArrayGetter, NullGetter (case object)
└─ DateGetter, DecimalGetter, TimestampGetter, LogicalTimeNTZGetter, TimestampNTZGetter,
YearMonthIntervalGetter, DayTimeIntervalGetter, ArrayGetter (final case class)
```
Each getter's body is transcribed unchanged from the corresponding `match` arm, and the selection order and guards in `makeGetter` are unchanged.
### Why are the changes needed?
Because `JDBCValueGetter` was only a function alias, the getter chosen for a column was not expressed in the type system: `makeGetter` read as a large `match` returning indistinguishable closures, and an individual getter could not be named or referred to. Turning it into a sealed trait with one named case per getter makes the set of getters explicit and self-documenting, and reduces `makeGetter` to a simple selection over the column's `DataType`, separating "which getter to use" from [...]
### Does this PR introduce _any_ user-facing change?
No. This is an internal, behavior-preserving refactor of private (`private[jdbc]`) symbols; the per-column read/convert/store logic is unchanged, and there is no public API or binary-compatibility (MiMa) impact.
### How was this patch tested?
Existing JDBC test suites (`JDBCSuite`, `JDBCV2Suite`, `JDBCWriteSuite`, `JdbcUtilsSuite`, etc.). This is a behavior-preserving refactor with no new behavior, so no new tests were added — the getter logic is a direct transcription of the previous `match` arms.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #56936 from Gamal72/jdbc-value-getter-sealed-type.
Authored-by: Abdelrahman Gamal <a....@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../datasources/jdbc/JDBCValueGetter.scala | 305 +++++++++++++++++++++
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 263 ++----------------
2 files changed, 331 insertions(+), 237 deletions(-)
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
new file mode 100644
index 000000000000..bd8d04760da6
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.jdbc
+
+import java.math.{BigDecimal => JBigDecimal}
+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.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
+
+/**
+ * A `JDBCValueGetter` is responsible for getting a value from a `ResultSet` into a field of an
+ * `InternalRow`. `pos` is the index for the value to be set in the row, and is also used for the
+ * value in the `ResultSet`. `JdbcUtils.makeGetter` selects one getter per column.
+ */
+private[jdbc] sealed trait JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit
+}
+
+private[jdbc] object JDBCValueGetter {
+ case object BooleanGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setBoolean(pos, rs.getBoolean(pos + 1))
+ }
+
+ final case class DateGetter(dialect: JdbcDialect) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ // DateTimeUtils.fromJavaDate does not handle null value, so we need to check it.
+ val dateVal = rs.getDate(pos + 1)
+ if (dateVal != null) {
+ row.setInt(pos, fromJavaDate(dialect.convertJavaDateToDate(dateVal)))
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
+ case object TimeGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val localTime = rs.getObject(pos + 1, classOf[java.time.LocalTime])
+ if (localTime != null) {
+ row.setLong(pos, localTime.toNanoOfDay)
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
+ // When connecting with Oracle DB through JDBC, the precision and scale of BigDecimal
+ // object returned by ResultSet.getBigDecimal is not correctly matched to the table
+ // schema reported by ResultSetMetaData.getPrecision and ResultSetMetaData.getScale.
+ // If inserting values like 19999 into a column with NUMBER(12, 2) type, you get through
+ // a BigDecimal object with scale as 0. But the dataframe schema has correct type as
+ // DecimalType(12, 2). Thus, after saving the dataframe into parquet file and then
+ // retrieve it, you will get wrong result 199.99.
+ // So it is needed to set precision and scale for Decimal based on JDBC metadata.
+ final case class DecimalGetter(precision: Int, scale: Int) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val decimal =
+ nullSafeConvert[JBigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, precision, scale))
+ row.update(pos, decimal)
+ }
+ }
+
+ case object DoubleGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setDouble(pos, rs.getDouble(pos + 1))
+ }
+
+ case object FloatGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setFloat(pos, rs.getFloat(pos + 1))
+ }
+
+ case object IntGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setInt(pos, rs.getInt(pos + 1))
+ }
+
+ case object BinaryLongGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val l = nullSafeConvert[Array[Byte]](rs.getBytes(pos + 1), bytes => {
+ var ans = 0L
+ var j = 0
+ while (j < bytes.length) {
+ ans = 256 * ans + (255 & bytes(j))
+ j = j + 1
+ }
+ ans
+ })
+ row.update(pos, l)
+ }
+ }
+
+ case object LongGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setLong(pos, rs.getLong(pos + 1))
+ }
+
+ case object ShortGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setShort(pos, rs.getShort(pos + 1))
+ }
+
+ case object ByteGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.setByte(pos, rs.getByte(pos + 1))
+ }
+
+ case object RowIdGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val rawRowId = rs.getRowId(pos + 1)
+ if (rawRowId == null) {
+ row.update(pos, null)
+ } else {
+ row.update(pos, UTF8String.fromString(rawRowId.toString))
+ }
+ }
+ }
+
+ case object StringGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
+ row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
+ }
+
+ // SPARK-34357 - sql TIME type represents as zero epoch timestamp.
+ // It is mapped as Spark TimestampType but fixed at 1970-01-01 for day,
+ // time portion is time of day, with no reference to a particular calendar,
+ // time zone or date, with a precision till microseconds.
+ // It stores the number of milliseconds after midnight, 00:00:00.000000
+ case object LogicalTimeGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.update(pos, nullSafeConvert[Time](
+ rs.getTime(pos + 1), t => Math.multiplyExact(t.getTime, MICROS_PER_MILLIS)))
+ }
+
+ final case class TimestampGetter(dialect: JdbcDialect) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, fromJavaTimestamp(dialect.convertJavaTimestampToTimestamp(t)))
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
+ final case class LogicalTimeNTZGetter(dialect: JdbcDialect) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val micros = nullSafeConvert[Time](rs.getTime(pos + 1), t => {
+ val time = dialect.convertJavaTimestampToTimestampNTZ(new Timestamp(t.getTime))
+ localDateTimeToMicros(time)
+ })
+ row.update(pos, micros)
+ }
+ }
+
+ final case class TimestampNTZGetter(dialect: JdbcDialect) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, localDateTimeToMicros(dialect.convertJavaTimestampToTimestampNTZ(t)))
+ } 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)
+ if (bytes != null) {
+ val binary = bytes.flatMap(Integer.toBinaryString(_).getBytes(StandardCharsets.US_ASCII))
+ row.update(pos, binary)
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
+ case object BytesGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.update(pos, rs.getBytes(pos + 1))
+ }
+
+ final case class YearMonthIntervalGetter(dialect: JdbcDialect) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.update(pos,
+ nullSafeConvert(rs.getString(pos + 1), dialect.getYearMonthIntervalAsMonths))
+ }
+
+ final case class DayTimeIntervalGetter(dialect: JdbcDialect) extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ row.update(pos,
+ nullSafeConvert(rs.getString(pos + 1), dialect.getDayTimeIntervalAsMicros))
+ }
+
+ // SPARK-47628: Handle PostgreSQL bit(n>1) array type ahead. As in the pgjdbc driver,
+ // bit(n>1)[] is not distinguishable from bit(1)[], and they are all recognized as boolean[].
+ // This is wrong for bit(n>1)[], so we need to handle it first as byte array.
+ case object PostgresBitArrayGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = {
+ val fieldString = rs.getString(pos + 1)
+ if (fieldString != null) {
+ val strArray = fieldString.substring(1, fieldString.length - 1).split(",")
+ // Charset is picked from the pgjdbc driver for consistency.
+ val bytesArray = strArray.map(_.getBytes(StandardCharsets.US_ASCII))
+ row.update(pos, new GenericArrayData(bytesArray))
+ } else {
+ row.update(pos, null)
+ }
+ }
+ }
+
+ final case class ArrayGetter(arrayType: ArrayType, dialect: JdbcDialect, metadata: Metadata)
+ extends JDBCValueGetter {
+ private def elementConversion(et: DataType): AnyRef => Any = et match {
+ case TimestampType => arrayConverter[Timestamp] {
+ (t: Timestamp) => fromJavaTimestamp(dialect.convertJavaTimestampToTimestamp(t))
+ }
+
+ case TimestampNTZType =>
+ arrayConverter[Timestamp] {
+ (t: Timestamp) => localDateTimeToMicros(dialect.convertJavaTimestampToTimestampNTZ(t))
+ }
+
+ case StringType =>
+ arrayConverter[Object]((obj: Object) => UTF8String.fromString(obj.toString))
+
+ case DateType => arrayConverter[Date] {
+ (d: Date) => fromJavaDate(dialect.convertJavaDateToDate(d))
+ }
+
+ case dt: DecimalType =>
+ arrayConverter[java.math.BigDecimal](d => Decimal(d, dt.precision, dt.scale))
+
+ case LongType if metadata.contains("binarylong") =>
+ throw QueryExecutionErrors.unsupportedArrayElementTypeBasedOnBinaryError(arrayType)
+
+ case ArrayType(et0, _) =>
+ arrayConverter[Array[Any]] {
+ arr => new GenericArrayData(elementConversion(et0)(arr))
+ }
+
+ case IntegerType => arrayConverter[Int]((i: Int) => i)
+ case FloatType => arrayConverter[Float]((f: Float) => f)
+ case DoubleType => arrayConverter[Double]((d: Double) => d)
+ case ShortType => arrayConverter[Short]((s: Short) => s)
+ case BooleanType => arrayConverter[Boolean]((b: Boolean) => b)
+ case LongType => arrayConverter[Long]((l: Long) => l)
+
+ case _ => (array: Object) => array.asInstanceOf[Array[Any]]
+ }
+
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit =
+ try {
+ val array = nullSafeConvert[java.sql.Array](
+ input = rs.getArray(pos + 1),
+ arr => new GenericArrayData(elementConversion(arrayType.elementType)(arr.getArray())))
+ row.update(pos, array)
+ } catch {
+ case _: java.lang.ClassCastException =>
+ throw QueryExecutionErrors.wrongDatatypeInSomeRows(pos, arrayType)
+ }
+ }
+
+ case object NullGetter extends JDBCValueGetter {
+ def apply(rs: ResultSet, row: InternalRow, pos: Int): Unit = row.update(pos, null)
+ }
+
+ private def nullSafeConvert[T](input: T, f: T => Any): Any = {
+ if (input == null) {
+ null
+ } else {
+ f(input)
+ }
+ }
+
+ private def arrayConverter[T](elementConvert: T => Any): Any => Any = (array: Any) => {
+ array.asInstanceOf[Array[T]].map(e => nullSafeConvert(e, elementConvert))
+ }
+}
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 cebc45002d44..ca44e8b710b1 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
@@ -17,9 +17,7 @@
package org.apache.spark.sql.execution.datasources.jdbc
-import java.math.{BigDecimal => JBigDecimal}
-import java.nio.charset.StandardCharsets
-import java.sql.{Connection, Date, JDBCType, PreparedStatement, ResultSet, ResultSetMetaData, SQLException, Time, Timestamp}
+import java.sql.{Connection, JDBCType, PreparedStatement, ResultSet, ResultSetMetaData, SQLException}
import java.time.{Instant, LocalDate}
import java.util
@@ -39,8 +37,7 @@ import org.apache.spark.sql.catalyst.analysis.{DecimalPrecisionTypeCoercion, Res
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.expressions.SpecificInternalRow
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, GenericArrayData}
-import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_MILLIS
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.sql.connector.catalog.{Identifier, TableChange}
import org.apache.spark.sql.connector.catalog.index.{SupportsIndex, TableIndex}
@@ -50,7 +47,6 @@ import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcDialects, JdbcType, NoopDialect}
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.SchemaUtils
-import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.{NextIterator, TaskInterruptListener}
import org.apache.spark.util.ArrayImplicits._
@@ -418,11 +414,6 @@ object JdbcUtils extends Logging with SQLConfHelper {
)
}
- // A `JDBCValueGetter` is responsible for getting a value from `ResultSet` into a field
- // for `MutableRow`. The last argument `Int` means the index for the value to be set in
- // the row and also used for the value in `ResultSet`.
- private type JDBCValueGetter = (ResultSet, InternalRow, Int) => Unit
-
/**
* Creates `JDBCValueGetter`s according to [[StructType]], which can set
* each value from `ResultSet` to each field of [[InternalRow]] correctly.
@@ -438,238 +429,36 @@ object JdbcUtils extends Logging with SQLConfHelper {
dt: DataType,
dialect: JdbcDialect,
metadata: Metadata): JDBCValueGetter = dt match {
- case BooleanType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setBoolean(pos, rs.getBoolean(pos + 1))
-
- case DateType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- // DateTimeUtils.fromJavaDate does not handle null value, so we need to check it.
- val dateVal = rs.getDate(pos + 1)
- if (dateVal != null) {
- row.setInt(pos, fromJavaDate(dialect.convertJavaDateToDate(dateVal)))
- } else {
- row.update(pos, null)
- }
-
- case _: TimeType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val localTime = rs.getObject(pos + 1, classOf[java.time.LocalTime])
- if (localTime != null) {
- row.setLong(pos, localTime.toNanoOfDay)
- } else {
- row.update(pos, null)
- }
-
- // When connecting with Oracle DB through JDBC, the precision and scale of BigDecimal
- // object returned by ResultSet.getBigDecimal is not correctly matched to the table
- // schema reported by ResultSetMetaData.getPrecision and ResultSetMetaData.getScale.
- // If inserting values like 19999 into a column with NUMBER(12, 2) type, you get through
- // a BigDecimal object with scale as 0. But the dataframe schema has correct type as
- // DecimalType(12, 2). Thus, after saving the dataframe into parquet file and then
- // retrieve it, you will get wrong result 199.99.
- // So it is needed to set precision and scale for Decimal based on JDBC metadata.
- case DecimalType.Fixed(p, s) =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val decimal =
- nullSafeConvert[JBigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
- row.update(pos, decimal)
-
- case DoubleType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setDouble(pos, rs.getDouble(pos + 1))
-
- case FloatType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setFloat(pos, rs.getFloat(pos + 1))
-
- case IntegerType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setInt(pos, rs.getInt(pos + 1))
-
- case LongType if metadata.contains("binarylong") =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val l = nullSafeConvert[Array[Byte]](rs.getBytes(pos + 1), bytes => {
- var ans = 0L
- var j = 0
- while (j < bytes.length) {
- ans = 256 * ans + (255 & bytes(j))
- j = j + 1
- }
- ans
- })
- row.update(pos, l)
-
- case LongType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setLong(pos, rs.getLong(pos + 1))
-
- case ShortType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setShort(pos, rs.getShort(pos + 1))
-
- case ByteType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.setByte(pos, rs.getByte(pos + 1))
-
- case StringType if metadata.contains("rowid") =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val rawRowId = rs.getRowId(pos + 1)
- if (rawRowId == null) {
- row.update(pos, null)
- } else {
- row.update(pos, UTF8String.fromString(rawRowId.toString))
- }
-
- case StringType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
- row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
-
- // SPARK-34357 - sql TIME type represents as zero epoch timestamp.
- // It is mapped as Spark TimestampType but fixed at 1970-01-01 for day,
- // time portion is time of day, with no reference to a particular calendar,
- // time zone or date, with a precision till microseconds.
- // It stores the number of milliseconds after midnight, 00:00:00.000000
+ case BooleanType => JDBCValueGetter.BooleanGetter
+ case DateType => JDBCValueGetter.DateGetter(dialect)
+ case _: TimeType => JDBCValueGetter.TimeGetter
+ case DecimalType.Fixed(p, s) => JDBCValueGetter.DecimalGetter(p, s)
+ case DoubleType => JDBCValueGetter.DoubleGetter
+ case FloatType => JDBCValueGetter.FloatGetter
+ case IntegerType => JDBCValueGetter.IntGetter
+ case LongType if metadata.contains("binarylong") => JDBCValueGetter.BinaryLongGetter
+ case LongType => JDBCValueGetter.LongGetter
+ case ShortType => JDBCValueGetter.ShortGetter
+ case ByteType => JDBCValueGetter.ByteGetter
+ case StringType if metadata.contains("rowid") => JDBCValueGetter.RowIdGetter
+ case StringType => JDBCValueGetter.StringGetter
case TimestampType if metadata.contains("logical_time_type") =>
- (rs: ResultSet, row: InternalRow, pos: Int) => {
- row.update(pos, nullSafeConvert[Time](
- rs.getTime(pos + 1), t => Math.multiplyExact(t.getTime, MICROS_PER_MILLIS)))
- }
-
- case TimestampType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val t = rs.getTimestamp(pos + 1)
- if (t != null) {
- row.setLong(pos, fromJavaTimestamp(dialect.convertJavaTimestampToTimestamp(t)))
- } else {
- row.update(pos, null)
- }
-
+ JDBCValueGetter.LogicalTimeGetter
+ case TimestampType => JDBCValueGetter.TimestampGetter(dialect)
case TimestampNTZType if metadata.contains("logical_time_type") =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val micros = nullSafeConvert[Time](rs.getTime(pos + 1), t => {
- val time = dialect.convertJavaTimestampToTimestampNTZ(new Timestamp(t.getTime))
- localDateTimeToMicros(time)
- })
- row.update(pos, micros)
-
- case TimestampNTZType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val t = rs.getTimestamp(pos + 1)
- if (t != null) {
- row.setLong(pos, localDateTimeToMicros(dialect.convertJavaTimestampToTimestampNTZ(t)))
- } else {
- row.update(pos, null)
- }
-
- case BinaryType if metadata.contains("binarylong") =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val bytes = rs.getBytes(pos + 1)
- if (bytes != null) {
- val binary = bytes.flatMap(Integer.toBinaryString(_).getBytes(StandardCharsets.US_ASCII))
- row.update(pos, binary)
- } else {
- row.update(pos, null)
- }
-
- case BinaryType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.update(pos, rs.getBytes(pos + 1))
-
- case _: YearMonthIntervalType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.update(pos,
- nullSafeConvert(rs.getString(pos + 1), dialect.getYearMonthIntervalAsMonths))
-
- case _: DayTimeIntervalType =>
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- row.update(pos,
- nullSafeConvert(rs.getString(pos + 1), dialect.getDayTimeIntervalAsMicros))
-
+ JDBCValueGetter.LogicalTimeNTZGetter(dialect)
+ case TimestampNTZType => JDBCValueGetter.TimestampNTZGetter(dialect)
+ case BinaryType if metadata.contains("binarylong") => JDBCValueGetter.BinaryBitGetter
+ case BinaryType => JDBCValueGetter.BytesGetter
+ case _: YearMonthIntervalType => JDBCValueGetter.YearMonthIntervalGetter(dialect)
+ case _: DayTimeIntervalType => JDBCValueGetter.DayTimeIntervalGetter(dialect)
case _: ArrayType if metadata.contains("pg_bit_array_type") =>
- // SPARK-47628: Handle PostgreSQL bit(n>1) array type ahead. As in the pgjdbc driver,
- // bit(n>1)[] is not distinguishable from bit(1)[], and they are all recognized as boolen[].
- // This is wrong for bit(n>1)[], so we need to handle it first as byte array.
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- val fieldString = rs.getString(pos + 1)
- if (fieldString != null) {
- val strArray = fieldString.substring(1, fieldString.length - 1).split(",")
- // Charset is picked from the pgjdbc driver for consistency.
- val bytesArray = strArray.map(_.getBytes(StandardCharsets.US_ASCII))
- row.update(pos, new GenericArrayData(bytesArray))
- } else {
- row.update(pos, null)
- }
-
- case ArrayType(et, _) =>
- def elementConversion(et: DataType): AnyRef => Any = et match {
- case TimestampType => arrayConverter[Timestamp] {
- (t: Timestamp) => fromJavaTimestamp(dialect.convertJavaTimestampToTimestamp(t))
- }
-
- case TimestampNTZType =>
- arrayConverter[Timestamp] {
- (t: Timestamp) => localDateTimeToMicros(dialect.convertJavaTimestampToTimestampNTZ(t))
- }
-
- case StringType =>
- arrayConverter[Object]((obj: Object) => UTF8String.fromString(obj.toString))
-
- case DateType => arrayConverter[Date] {
- (d: Date) => fromJavaDate(dialect.convertJavaDateToDate(d))
- }
-
- case dt: DecimalType =>
- arrayConverter[java.math.BigDecimal](d => Decimal(d, dt.precision, dt.scale))
-
- case LongType if metadata.contains("binarylong") =>
- throw QueryExecutionErrors.unsupportedArrayElementTypeBasedOnBinaryError(dt)
-
- case ArrayType(et0, _) =>
- arrayConverter[Array[Any]] {
- arr => new GenericArrayData(elementConversion(et0)(arr))
- }
-
- case IntegerType => arrayConverter[Int]((i: Int) => i)
- case FloatType => arrayConverter[Float]((f: Float) => f)
- case DoubleType => arrayConverter[Double]((d: Double) => d)
- case ShortType => arrayConverter[Short]((s: Short) => s)
- case BooleanType => arrayConverter[Boolean]((b: Boolean) => b)
- case LongType => arrayConverter[Long]((l: Long) => l)
-
- case _ => (array: Object) => array.asInstanceOf[Array[Any]]
- }
-
- (rs: ResultSet, row: InternalRow, pos: Int) =>
- try {
- val array = nullSafeConvert[java.sql.Array](
- input = rs.getArray(pos + 1),
- array => new GenericArrayData(elementConversion(et)(array.getArray())))
- row.update(pos, array)
- } catch {
- case e: java.lang.ClassCastException =>
- throw QueryExecutionErrors.wrongDatatypeInSomeRows(pos, dt)
- }
-
- case NullType =>
- (_: ResultSet, row: InternalRow, pos: Int) => row.update(pos, null)
-
+ JDBCValueGetter.PostgresBitArrayGetter
+ case at: ArrayType => JDBCValueGetter.ArrayGetter(at, dialect, metadata)
+ case NullType => JDBCValueGetter.NullGetter
case _ => throw QueryExecutionErrors.unsupportedJdbcTypeError(dt.catalogString)
}
- private def nullSafeConvert[T](input: T, f: T => Any): Any = {
- if (input == null) {
- null
- } else {
- f(input)
- }
- }
-
- private def arrayConverter[T](elementConvert: T => Any): Any => Any = (array: Any) => {
- array.asInstanceOf[Array[T]].map(e => nullSafeConvert(e, elementConvert))
- }
-
// A `JDBCValueSetter` is responsible for setting a value from `Row` into a field for
// `PreparedStatement`. The last argument `Int` means the index for the value to be set
// in the SQL statement and also used for the value in `Row`.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org