You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by JoshRosen <gi...@git.apache.org> on 2016/09/01 00:49:53 UTC
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
GitHub user JoshRosen opened a pull request:
https://github.com/apache/spark/pull/14907
[SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Seq[Row] utility methods
This patch refactors the internals of the JDBC data source in order to allow some of its code to be re-used in an automated comparison testing harness. Here are the key changes:
- Move the JDBC `ResultSetMetadata` to `StructType` conversion logic from `JDBCRDD.resolveTable()` to the `JdbcUtils` object (as a new `getSchema(ResultSet, JdbcDialect)` method), allowing it to be applied on `ResultSet`s that are created elsewhere.
- Move the `ResultSet` to `InternalRow` conversion methods from `JDBCRDD` to `JdbcUtils`:
- It makes sense to move the `JDBCValueGetter` type and `makeGetter` functions here given that their write-path counterparts (`JDBCValueSetter`) are already in `JdbcUtils`.
- Add an internal `resultSetToSparkInternalRows` method which takes a `ResultSet` and schema and returns an `Iterator[InternalRow]`. This effectively extracts the main loop of `JDBCRDD` into its own method.
- Add a public `resultSetToRows` method to `JdbcUtils`, which wraps the minimal machinery around `resultSetToSparkInternalRows` in order to allow it to be called from outside of a Spark job.
- Make `JdbcDialect.get` into a `DeveloperApi` (`JdbcDialect` itself is already a `DeveloperApi`).
Put together, these changes enable the following testing pattern:
```scala
val jdbResultSet: ResultSet = conn.prepareStatement(query).executeQuery()
val resultSchema: StructType = JdbcUtils.getSchema(jdbResultSet, JdbcDialects.get("jdbc:postgresql"))
val jdbcRows: Seq[Row] = JdbcUtils.resultSetToRows(jdbResultSet, schema).toSeq
checkAnswer(sparkResult, jdbcRows) // in a test case
```
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/JoshRosen/spark modularize-jdbc-internals
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/14907.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #14907
----
commit 17d770a85e0921a5bfcbe00aead71cf169f76119
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-08-23T19:02:55Z
Move ResultSet -> Seq[InternalRow] conversion into JdbcUtils
commit 682b5917341a530627a5d873196f6c4a3259a91b
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-08-23T19:07:44Z
Make new method private[spark]
commit ec49accbf2c532766fb66c3b8910fd7c81563839
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-08-23T19:18:03Z
Move getCatalystType to JdbcUtils and add new getSchema() method.
commit 025c9d08d485ebfab0dd23f1ed5065e537ad0437
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-08-23T19:54:54Z
Add public resultSetToRows() method for converting to public rows.
commit 05dfe5276017862dadf8791672de818329d52723
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-08-24T02:25:35Z
Remove InputMetrics from a public API.
commit fca548ae24bbc60e8c04ea4e6756dfb19942fb61
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-08-25T00:49:28Z
Open up JdbcDialects.get as developerapi.
commit 43cbef6b4310dd9af08672bcaa01d8114b1fe5fc
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-09-01T00:32:25Z
Merge remote-tracking branch 'origin/master' into modularize-jdbc-internals
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77384493
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
+ }
+ }
+ }
+ }
+
+ // 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, MutableRow, Int) => Unit
+
+ /**
+ * Creates `JDBCValueGetter`s according to [[StructType]], which can set
+ * each value from `ResultSet` to each field of [[MutableRow]] correctly.
+ */
+ private def makeGetters(schema: StructType): Array[JDBCValueGetter] =
+ schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
+
+ private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
+ case BooleanType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setBoolean(pos, rs.getBoolean(pos + 1))
+
+ case DateType =>
+ (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
+ } 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: MutableRow, pos: Int) =>
+ val decimal =
+ nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
+ row.update(pos, decimal)
+
+ case DoubleType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setDouble(pos, rs.getDouble(pos + 1))
+
+ case FloatType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setFloat(pos, rs.getFloat(pos + 1))
+
+ case IntegerType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setInt(pos, rs.getInt(pos + 1))
+
+ case LongType if metadata.contains("binarylong") =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val bytes = rs.getBytes(pos + 1)
+ var ans = 0L
+ var j = 0
+ while (j < bytes.size) {
+ ans = 256 * ans + (255 & bytes(j))
+ j = j + 1
+ }
+ row.setLong(pos, ans)
+
+ case LongType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setLong(pos, rs.getLong(pos + 1))
+
+ case ShortType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setShort(pos, rs.getShort(pos + 1))
+
+ case StringType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
+ row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
+
+ case TimestampType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
+ } else {
+ row.update(pos, null)
+ }
+
+ case BinaryType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.update(pos, rs.getBytes(pos + 1))
+
+ case ArrayType(et, _) =>
+ val elementConversion = et match {
+ case TimestampType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
+ nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
+ }
+
+ case StringType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.lang.String]]
+ .map(UTF8String.fromString)
+
+ case DateType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Date]].map { date =>
+ nullSafeConvert(date, DateTimeUtils.fromJavaDate)
+ }
+
+ case dt: DecimalType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
+ nullSafeConvert[java.math.BigDecimal](
+ decimal, d => Decimal(d, dt.precision, dt.scale))
+ }
+
+ case LongType if metadata.contains("binarylong") =>
+ throw new IllegalArgumentException(s"Unsupported array element " +
+ s"type ${dt.simpleString} based on binary")
+
+ case ArrayType(_, _) =>
+ throw new IllegalArgumentException("Nested arrays unsupported")
+
+ case _ => (array: Object) => array.asInstanceOf[Array[Any]]
+ }
+
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val array = nullSafeConvert[Object](
+ rs.getArray(pos + 1).getArray,
+ array => new GenericArrayData(elementConversion.apply(array)))
+ row.update(pos, array)
+
+ case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
+ }
+
+ private def nullSafeConvert[T](input: T, f: T => Any): Any = {
--- End diff --
Aha, forgot about using lower type bounds for this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/14907
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77283758
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
+ }
+ }
+ }
+ }
+
+ // 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, MutableRow, Int) => Unit
+
+ /**
+ * Creates `JDBCValueGetter`s according to [[StructType]], which can set
+ * each value from `ResultSet` to each field of [[MutableRow]] correctly.
+ */
+ private def makeGetters(schema: StructType): Array[JDBCValueGetter] =
+ schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
+
+ private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
+ case BooleanType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setBoolean(pos, rs.getBoolean(pos + 1))
+
+ case DateType =>
+ (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
+ } 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: MutableRow, pos: Int) =>
+ val decimal =
+ nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
+ row.update(pos, decimal)
+
+ case DoubleType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setDouble(pos, rs.getDouble(pos + 1))
+
+ case FloatType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setFloat(pos, rs.getFloat(pos + 1))
+
+ case IntegerType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setInt(pos, rs.getInt(pos + 1))
+
+ case LongType if metadata.contains("binarylong") =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val bytes = rs.getBytes(pos + 1)
+ var ans = 0L
+ var j = 0
+ while (j < bytes.size) {
+ ans = 256 * ans + (255 & bytes(j))
+ j = j + 1
+ }
+ row.setLong(pos, ans)
+
+ case LongType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setLong(pos, rs.getLong(pos + 1))
+
+ case ShortType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setShort(pos, rs.getShort(pos + 1))
+
+ case StringType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
+ row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
+
+ case TimestampType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
+ } else {
+ row.update(pos, null)
+ }
+
+ case BinaryType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.update(pos, rs.getBytes(pos + 1))
+
+ case ArrayType(et, _) =>
+ val elementConversion = et match {
+ case TimestampType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
+ nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
+ }
+
+ case StringType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.lang.String]]
+ .map(UTF8String.fromString)
+
+ case DateType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Date]].map { date =>
+ nullSafeConvert(date, DateTimeUtils.fromJavaDate)
+ }
+
+ case dt: DecimalType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
+ nullSafeConvert[java.math.BigDecimal](
+ decimal, d => Decimal(d, dt.precision, dt.scale))
+ }
+
+ case LongType if metadata.contains("binarylong") =>
+ throw new IllegalArgumentException(s"Unsupported array element " +
+ s"type ${dt.simpleString} based on binary")
+
+ case ArrayType(_, _) =>
+ throw new IllegalArgumentException("Nested arrays unsupported")
+
+ case _ => (array: Object) => array.asInstanceOf[Array[Any]]
+ }
+
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val array = nullSafeConvert[Object](
+ rs.getArray(pos + 1).getArray,
+ array => new GenericArrayData(elementConversion.apply(array)))
+ row.update(pos, array)
+
+ case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
+ }
+
+ private def nullSafeConvert[T](input: T, f: T => Any): Any = {
--- End diff --
Actually, I don't think that we can do this because I don't think that there's a way to set an upper type bound to say that `U` must be a nullable object, so you have to do a dangerous `null.asInstanceOf` cast. And since we're not working with primitives here there's no savings on boxing, etc. Therefore, I'd prefer to just leave this unchanged since it's a carryover from the old code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/14907
We exposed `JdbcUtils.resultSetToRows`, but forgot to add a test case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14907
**[Test build #64745 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64745/consoleFull)** for PR 14907 at commit [`43cbef6`](https://github.com/apache/spark/commit/43cbef6b4310dd9af08672bcaa01d8114b1fe5fc).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the issue:
https://github.com/apache/spark/pull/14907
I've merged #14911 so this should now be ready for review.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77099971
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,288 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = rsmd.isSigned(i + 1)
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
--- End diff --
Here, I've used `NextIterator` to simplify the implementation of this code. Take a look at the `NextIterator` source and note the similarities between it and the `hasNext` code in the old `JDBCRDD` iterator.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77377425
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
+ }
+ }
+ }
+ }
+
+ // 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, MutableRow, Int) => Unit
+
+ /**
+ * Creates `JDBCValueGetter`s according to [[StructType]], which can set
+ * each value from `ResultSet` to each field of [[MutableRow]] correctly.
+ */
+ private def makeGetters(schema: StructType): Array[JDBCValueGetter] =
+ schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
+
+ private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
+ case BooleanType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setBoolean(pos, rs.getBoolean(pos + 1))
+
+ case DateType =>
+ (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
+ } 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: MutableRow, pos: Int) =>
+ val decimal =
+ nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
+ row.update(pos, decimal)
+
+ case DoubleType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setDouble(pos, rs.getDouble(pos + 1))
+
+ case FloatType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setFloat(pos, rs.getFloat(pos + 1))
+
+ case IntegerType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setInt(pos, rs.getInt(pos + 1))
+
+ case LongType if metadata.contains("binarylong") =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val bytes = rs.getBytes(pos + 1)
+ var ans = 0L
+ var j = 0
+ while (j < bytes.size) {
+ ans = 256 * ans + (255 & bytes(j))
+ j = j + 1
+ }
+ row.setLong(pos, ans)
+
+ case LongType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setLong(pos, rs.getLong(pos + 1))
+
+ case ShortType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setShort(pos, rs.getShort(pos + 1))
+
+ case StringType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
+ row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
+
+ case TimestampType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
+ } else {
+ row.update(pos, null)
+ }
+
+ case BinaryType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.update(pos, rs.getBytes(pos + 1))
+
+ case ArrayType(et, _) =>
+ val elementConversion = et match {
+ case TimestampType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
+ nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
+ }
+
+ case StringType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.lang.String]]
+ .map(UTF8String.fromString)
+
+ case DateType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Date]].map { date =>
+ nullSafeConvert(date, DateTimeUtils.fromJavaDate)
+ }
+
+ case dt: DecimalType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
+ nullSafeConvert[java.math.BigDecimal](
+ decimal, d => Decimal(d, dt.precision, dt.scale))
+ }
+
+ case LongType if metadata.contains("binarylong") =>
+ throw new IllegalArgumentException(s"Unsupported array element " +
+ s"type ${dt.simpleString} based on binary")
+
+ case ArrayType(_, _) =>
+ throw new IllegalArgumentException("Nested arrays unsupported")
+
+ case _ => (array: Object) => array.asInstanceOf[Array[Any]]
+ }
+
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val array = nullSafeConvert[Object](
+ rs.getArray(pos + 1).getArray,
+ array => new GenericArrayData(elementConversion.apply(array)))
+ row.update(pos, array)
+
+ case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
+ }
+
+ private def nullSafeConvert[T](input: T, f: T => Any): Any = {
--- End diff --
You can use the `U >: Null` bound right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14907
**[Test build #64816 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64816/consoleFull)** for PR 14907 at commit [`f09174b`](https://github.com/apache/spark/commit/f09174b03a27b9fcd8ed6c28608a16a5433a5b19).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14907
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64745/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14907
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on the issue:
https://github.com/apache/spark/pull/14907
LGTM. Merging to master. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by srinathshankar <gi...@git.apache.org>.
Github user srinathshankar commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77277372
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
+ }
+ }
+ }
+ }
+
+ // 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, MutableRow, Int) => Unit
+
+ /**
+ * Creates `JDBCValueGetter`s according to [[StructType]], which can set
+ * each value from `ResultSet` to each field of [[MutableRow]] correctly.
+ */
+ private def makeGetters(schema: StructType): Array[JDBCValueGetter] =
+ schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
+
+ private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
+ case BooleanType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setBoolean(pos, rs.getBoolean(pos + 1))
+
+ case DateType =>
+ (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
+ } 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: MutableRow, pos: Int) =>
+ val decimal =
+ nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
+ row.update(pos, decimal)
+
+ case DoubleType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setDouble(pos, rs.getDouble(pos + 1))
+
+ case FloatType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setFloat(pos, rs.getFloat(pos + 1))
+
+ case IntegerType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setInt(pos, rs.getInt(pos + 1))
+
+ case LongType if metadata.contains("binarylong") =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val bytes = rs.getBytes(pos + 1)
+ var ans = 0L
+ var j = 0
+ while (j < bytes.size) {
+ ans = 256 * ans + (255 & bytes(j))
+ j = j + 1
+ }
+ row.setLong(pos, ans)
+
+ case LongType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setLong(pos, rs.getLong(pos + 1))
+
+ case ShortType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setShort(pos, rs.getShort(pos + 1))
+
+ case StringType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
+ row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
+
+ case TimestampType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
+ } else {
+ row.update(pos, null)
+ }
+
+ case BinaryType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.update(pos, rs.getBytes(pos + 1))
+
+ case ArrayType(et, _) =>
+ val elementConversion = et match {
+ case TimestampType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
+ nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
+ }
+
+ case StringType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.lang.String]]
+ .map(UTF8String.fromString)
+
+ case DateType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Date]].map { date =>
+ nullSafeConvert(date, DateTimeUtils.fromJavaDate)
+ }
+
+ case dt: DecimalType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
+ nullSafeConvert[java.math.BigDecimal](
+ decimal, d => Decimal(d, dt.precision, dt.scale))
+ }
+
+ case LongType if metadata.contains("binarylong") =>
+ throw new IllegalArgumentException(s"Unsupported array element " +
+ s"type ${dt.simpleString} based on binary")
+
+ case ArrayType(_, _) =>
+ throw new IllegalArgumentException("Nested arrays unsupported")
+
+ case _ => (array: Object) => array.asInstanceOf[Array[Any]]
+ }
+
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val array = nullSafeConvert[Object](
+ rs.getArray(pos + 1).getArray,
+ array => new GenericArrayData(elementConversion.apply(array)))
+ row.update(pos, array)
+
+ case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
+ }
+
+ private def nullSafeConvert[T](input: T, f: T => Any): Any = {
--- End diff --
Can this be
private def nullSafeConvert[T, U](input: T, f: T => U): U
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14907
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64816/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14907
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77377079
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,288 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = rsmd.isSigned(i + 1)
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
--- End diff --
That is a valid assumption.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77277642
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
+ }
+ }
+ }
+ }
+
+ // 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, MutableRow, Int) => Unit
+
+ /**
+ * Creates `JDBCValueGetter`s according to [[StructType]], which can set
+ * each value from `ResultSet` to each field of [[MutableRow]] correctly.
+ */
+ private def makeGetters(schema: StructType): Array[JDBCValueGetter] =
+ schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
+
+ private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
+ case BooleanType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setBoolean(pos, rs.getBoolean(pos + 1))
+
+ case DateType =>
+ (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
+ } 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: MutableRow, pos: Int) =>
+ val decimal =
+ nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
+ row.update(pos, decimal)
+
+ case DoubleType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setDouble(pos, rs.getDouble(pos + 1))
+
+ case FloatType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setFloat(pos, rs.getFloat(pos + 1))
+
+ case IntegerType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setInt(pos, rs.getInt(pos + 1))
+
+ case LongType if metadata.contains("binarylong") =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val bytes = rs.getBytes(pos + 1)
+ var ans = 0L
+ var j = 0
+ while (j < bytes.size) {
+ ans = 256 * ans + (255 & bytes(j))
+ j = j + 1
+ }
+ row.setLong(pos, ans)
+
+ case LongType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setLong(pos, rs.getLong(pos + 1))
+
+ case ShortType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.setShort(pos, rs.getShort(pos + 1))
+
+ case StringType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
+ row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
+
+ case TimestampType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val t = rs.getTimestamp(pos + 1)
+ if (t != null) {
+ row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
+ } else {
+ row.update(pos, null)
+ }
+
+ case BinaryType =>
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ row.update(pos, rs.getBytes(pos + 1))
+
+ case ArrayType(et, _) =>
+ val elementConversion = et match {
+ case TimestampType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
+ nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
+ }
+
+ case StringType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.lang.String]]
+ .map(UTF8String.fromString)
+
+ case DateType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.sql.Date]].map { date =>
+ nullSafeConvert(date, DateTimeUtils.fromJavaDate)
+ }
+
+ case dt: DecimalType =>
+ (array: Object) =>
+ array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
+ nullSafeConvert[java.math.BigDecimal](
+ decimal, d => Decimal(d, dt.precision, dt.scale))
+ }
+
+ case LongType if metadata.contains("binarylong") =>
+ throw new IllegalArgumentException(s"Unsupported array element " +
+ s"type ${dt.simpleString} based on binary")
+
+ case ArrayType(_, _) =>
+ throw new IllegalArgumentException("Nested arrays unsupported")
+
+ case _ => (array: Object) => array.asInstanceOf[Array[Any]]
+ }
+
+ (rs: ResultSet, row: MutableRow, pos: Int) =>
+ val array = nullSafeConvert[Object](
+ rs.getArray(pos + 1).getArray,
+ array => new GenericArrayData(elementConversion.apply(array)))
+ row.update(pos, array)
+
+ case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
+ }
+
+ private def nullSafeConvert[T](input: T, f: T => Any): Any = {
--- End diff --
This is carried over from the old code. Good catch.
Changing this signature might actually improve performance in the `ArrayType` case because Scala should be able to statically determine that it can allocate primitive arrays once it knows the return type of `nullSafeConvert`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the issue:
https://github.com/apache/spark/pull/14907
Please merge #14911 ahead of this so that I can bring this up-to-date with that change. Merging in this order reduces the amount of work to backport #14911.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77099804
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala ---
@@ -322,195 +237,15 @@ private[jdbc] class JDBCRDD(
}
}
- // 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, MutableRow, Int) => Unit
-
- /**
- * Creates `JDBCValueGetter`s according to [[StructType]], which can set
- * each value from `ResultSet` to each field of [[MutableRow]] correctly.
- */
- def makeGetters(schema: StructType): Array[JDBCValueGetter] =
- schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
-
- private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
- case BooleanType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setBoolean(pos, rs.getBoolean(pos + 1))
-
- case DateType =>
- (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
- } 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: MutableRow, pos: Int) =>
- val decimal =
- nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
- row.update(pos, decimal)
-
- case DoubleType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setDouble(pos, rs.getDouble(pos + 1))
-
- case FloatType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setFloat(pos, rs.getFloat(pos + 1))
-
- case IntegerType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setInt(pos, rs.getInt(pos + 1))
-
- case LongType if metadata.contains("binarylong") =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val bytes = rs.getBytes(pos + 1)
- var ans = 0L
- var j = 0
- while (j < bytes.size) {
- ans = 256 * ans + (255 & bytes(j))
- j = j + 1
- }
- row.setLong(pos, ans)
-
- case LongType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setLong(pos, rs.getLong(pos + 1))
-
- case ShortType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setShort(pos, rs.getShort(pos + 1))
-
- case StringType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
- row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
-
- case TimestampType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val t = rs.getTimestamp(pos + 1)
- if (t != null) {
- row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
- } else {
- row.update(pos, null)
- }
-
- case BinaryType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.update(pos, rs.getBytes(pos + 1))
-
- case ArrayType(et, _) =>
- val elementConversion = et match {
- case TimestampType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
- nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
- }
-
- case StringType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.lang.String]]
- .map(UTF8String.fromString)
-
- case DateType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.sql.Date]].map { date =>
- nullSafeConvert(date, DateTimeUtils.fromJavaDate)
- }
-
- case dt: DecimalType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
- nullSafeConvert[java.math.BigDecimal](
- decimal, d => Decimal(d, dt.precision, dt.scale))
- }
-
- case LongType if metadata.contains("binarylong") =>
- throw new IllegalArgumentException(s"Unsupported array element " +
- s"type ${dt.simpleString} based on binary")
-
- case ArrayType(_, _) =>
- throw new IllegalArgumentException("Nested arrays unsupported")
-
- case _ => (array: Object) => array.asInstanceOf[Array[Any]]
- }
-
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val array = nullSafeConvert[Object](
- rs.getArray(pos + 1).getArray,
- array => new GenericArrayData(elementConversion.apply(array)))
- row.update(pos, array)
-
- case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
- }
-
/**
* Runs the SQL query against the JDBC driver.
*
*/
- override def compute(thePart: Partition, context: TaskContext): Iterator[InternalRow] =
- new Iterator[InternalRow] {
+ override def compute(thePart: Partition, context: TaskContext): Iterator[InternalRow] = {
var closed = false
- var finished = false
- var gotNext = false
- var nextValue: InternalRow = null
-
- context.addTaskCompletionListener{ context => close() }
- val inputMetrics = context.taskMetrics().inputMetrics
- val part = thePart.asInstanceOf[JDBCPartition]
- val conn = getConnection()
- val dialect = JdbcDialects.get(url)
- import scala.collection.JavaConverters._
- dialect.beforeFetch(conn, properties.asScala.toMap)
-
- // H2's JDBC driver does not support the setSchema() method. We pass a
- // fully-qualified table name in the SELECT statement. I don't know how to
- // talk about a table in a completely portable way.
-
- val myWhereClause = getWhereClause(part)
-
- val sqlText = s"SELECT $columnList FROM $fqTable $myWhereClause"
- val stmt = conn.prepareStatement(sqlText,
- ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)
- val fetchSize = properties.getProperty(JdbcUtils.JDBC_BATCH_FETCH_SIZE, "0").toInt
- require(fetchSize >= 0,
- s"Invalid value `${fetchSize.toString}` for parameter " +
- s"`${JdbcUtils.JDBC_BATCH_FETCH_SIZE}`. The minimum value is 0. When the value is 0, " +
- "the JDBC driver ignores the value and does the estimates.")
- stmt.setFetchSize(fetchSize)
- val rs = stmt.executeQuery()
-
- val getters: Array[JDBCValueGetter] = makeGetters(schema)
- val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
-
- def getNext(): InternalRow = {
- if (rs.next()) {
- inputMetrics.incRecordsRead(1)
- var i = 0
- while (i < getters.length) {
- getters(i).apply(rs, mutableRow, i)
- if (rs.wasNull) mutableRow.setNullAt(i)
- i = i + 1
- }
- mutableRow
- } else {
- finished = true
- null.asInstanceOf[InternalRow]
- }
- }
+ var rs: ResultSet = null
+ var stmt: PreparedStatement = null
+ var conn: Connection = null
def close() {
--- End diff --
Here, `close()` tears down the result set, statement, and connection. While in principle closing the connection should be sufficient to close the statement (which should be sufficient to close the result set), not all JDBC drivers adhere to this contract so we can't simplify this code at all.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77279354
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
--- End diff --
The close is handled automatically by the `NextIterator` superclass; see https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/util/NextIterator.scala#L28
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77099720
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala ---
@@ -322,195 +237,15 @@ private[jdbc] class JDBCRDD(
}
}
- // 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, MutableRow, Int) => Unit
-
- /**
- * Creates `JDBCValueGetter`s according to [[StructType]], which can set
- * each value from `ResultSet` to each field of [[MutableRow]] correctly.
- */
- def makeGetters(schema: StructType): Array[JDBCValueGetter] =
- schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
-
- private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
- case BooleanType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setBoolean(pos, rs.getBoolean(pos + 1))
-
- case DateType =>
- (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
- } 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: MutableRow, pos: Int) =>
- val decimal =
- nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
- row.update(pos, decimal)
-
- case DoubleType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setDouble(pos, rs.getDouble(pos + 1))
-
- case FloatType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setFloat(pos, rs.getFloat(pos + 1))
-
- case IntegerType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setInt(pos, rs.getInt(pos + 1))
-
- case LongType if metadata.contains("binarylong") =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val bytes = rs.getBytes(pos + 1)
- var ans = 0L
- var j = 0
- while (j < bytes.size) {
- ans = 256 * ans + (255 & bytes(j))
- j = j + 1
- }
- row.setLong(pos, ans)
-
- case LongType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setLong(pos, rs.getLong(pos + 1))
-
- case ShortType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setShort(pos, rs.getShort(pos + 1))
-
- case StringType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
- row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
-
- case TimestampType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val t = rs.getTimestamp(pos + 1)
- if (t != null) {
- row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
- } else {
- row.update(pos, null)
- }
-
- case BinaryType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.update(pos, rs.getBytes(pos + 1))
-
- case ArrayType(et, _) =>
- val elementConversion = et match {
- case TimestampType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
- nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
- }
-
- case StringType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.lang.String]]
- .map(UTF8String.fromString)
-
- case DateType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.sql.Date]].map { date =>
- nullSafeConvert(date, DateTimeUtils.fromJavaDate)
- }
-
- case dt: DecimalType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
- nullSafeConvert[java.math.BigDecimal](
- decimal, d => Decimal(d, dt.precision, dt.scale))
- }
-
- case LongType if metadata.contains("binarylong") =>
- throw new IllegalArgumentException(s"Unsupported array element " +
- s"type ${dt.simpleString} based on binary")
-
- case ArrayType(_, _) =>
- throw new IllegalArgumentException("Nested arrays unsupported")
-
- case _ => (array: Object) => array.asInstanceOf[Array[Any]]
- }
-
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val array = nullSafeConvert[Object](
- rs.getArray(pos + 1).getArray,
- array => new GenericArrayData(elementConversion.apply(array)))
- row.update(pos, array)
-
- case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
- }
-
/**
* Runs the SQL query against the JDBC driver.
*
*/
- override def compute(thePart: Partition, context: TaskContext): Iterator[InternalRow] =
- new Iterator[InternalRow] {
--- End diff --
First, note that the old code was mis-idented, so it might have been easy to miss that the entire method body was defining this iterator.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77384218
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala ---
@@ -546,33 +281,33 @@ private[jdbc] class JDBCRDD(
closed = true
}
- override def hasNext: Boolean = {
- if (!finished) {
- if (!gotNext) {
- nextValue = getNext()
- if (finished) {
- close()
- }
- gotNext = true
- }
- }
- !finished
- }
+ context.addTaskCompletionListener{ context => close() }
--- End diff --
Yeah, this is an unfortunate style carryover from the old code :(
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by srinathshankar <gi...@git.apache.org>.
Github user srinathshankar commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77278500
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,297 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = {
+ try {
+ rsmd.isSigned(i + 1)
+ } catch {
+ // Workaround for HIVE-14684:
+ case e: SQLException if
+ e.getMessage == "Method not supported" &&
+ rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
+ }
+ }
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
+ try {
+ rs.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing resultset", e)
+ }
+ }
+
+ override protected def getNext(): InternalRow = {
+ if (rs.next()) {
+ inputMetrics.incRecordsRead(1)
+ var i = 0
+ while (i < getters.length) {
+ getters(i).apply(rs, mutableRow, i)
+ if (rs.wasNull) mutableRow.setNullAt(i)
+ i = i + 1
+ }
+ mutableRow
+ } else {
+ finished = true
+ null.asInstanceOf[InternalRow]
--- End diff --
Why don't we close here ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77099874
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala ---
@@ -546,33 +281,33 @@ private[jdbc] class JDBCRDD(
closed = true
}
- override def hasNext: Boolean = {
- if (!finished) {
- if (!gotNext) {
- nextValue = getNext()
- if (finished) {
- close()
- }
- gotNext = true
- }
- }
- !finished
- }
+ context.addTaskCompletionListener{ context => close() }
--- End diff --
Note that `close()` is idempotent because it may be called twice, at the end of the task and at the end of iteration. In the old code the end-of-iteration case was handled in `hasNext`; here, it's handled in a `CompletionIterator` defined below.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14907
**[Test build #64745 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64745/consoleFull)** for PR 14907 at commit [`43cbef6`](https://github.com/apache/spark/commit/43cbef6b4310dd9af08672bcaa01d8114b1fe5fc).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77099616
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala ---
@@ -322,195 +237,15 @@ private[jdbc] class JDBCRDD(
}
}
- // 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, MutableRow, Int) => Unit
-
- /**
- * Creates `JDBCValueGetter`s according to [[StructType]], which can set
- * each value from `ResultSet` to each field of [[MutableRow]] correctly.
- */
- def makeGetters(schema: StructType): Array[JDBCValueGetter] =
- schema.fields.map(sf => makeGetter(sf.dataType, sf.metadata))
-
- private def makeGetter(dt: DataType, metadata: Metadata): JDBCValueGetter = dt match {
- case BooleanType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setBoolean(pos, rs.getBoolean(pos + 1))
-
- case DateType =>
- (rs: ResultSet, row: MutableRow, 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, DateTimeUtils.fromJavaDate(dateVal))
- } 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: MutableRow, pos: Int) =>
- val decimal =
- nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
- row.update(pos, decimal)
-
- case DoubleType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setDouble(pos, rs.getDouble(pos + 1))
-
- case FloatType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setFloat(pos, rs.getFloat(pos + 1))
-
- case IntegerType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setInt(pos, rs.getInt(pos + 1))
-
- case LongType if metadata.contains("binarylong") =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val bytes = rs.getBytes(pos + 1)
- var ans = 0L
- var j = 0
- while (j < bytes.size) {
- ans = 256 * ans + (255 & bytes(j))
- j = j + 1
- }
- row.setLong(pos, ans)
-
- case LongType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setLong(pos, rs.getLong(pos + 1))
-
- case ShortType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.setShort(pos, rs.getShort(pos + 1))
-
- case StringType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
- row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
-
- case TimestampType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val t = rs.getTimestamp(pos + 1)
- if (t != null) {
- row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
- } else {
- row.update(pos, null)
- }
-
- case BinaryType =>
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- row.update(pos, rs.getBytes(pos + 1))
-
- case ArrayType(et, _) =>
- val elementConversion = et match {
- case TimestampType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
- nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
- }
-
- case StringType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.lang.String]]
- .map(UTF8String.fromString)
-
- case DateType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.sql.Date]].map { date =>
- nullSafeConvert(date, DateTimeUtils.fromJavaDate)
- }
-
- case dt: DecimalType =>
- (array: Object) =>
- array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
- nullSafeConvert[java.math.BigDecimal](
- decimal, d => Decimal(d, dt.precision, dt.scale))
- }
-
- case LongType if metadata.contains("binarylong") =>
- throw new IllegalArgumentException(s"Unsupported array element " +
- s"type ${dt.simpleString} based on binary")
-
- case ArrayType(_, _) =>
- throw new IllegalArgumentException("Nested arrays unsupported")
-
- case _ => (array: Object) => array.asInstanceOf[Array[Any]]
- }
-
- (rs: ResultSet, row: MutableRow, pos: Int) =>
- val array = nullSafeConvert[Object](
- rs.getArray(pos + 1).getArray,
- array => new GenericArrayData(elementConversion.apply(array)))
- row.update(pos, array)
-
- case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
- }
-
/**
* Runs the SQL query against the JDBC driver.
*
*/
- override def compute(thePart: Partition, context: TaskContext): Iterator[InternalRow] =
- new Iterator[InternalRow] {
+ override def compute(thePart: Partition, context: TaskContext): Iterator[InternalRow] = {
--- End diff --
While most of the changes in this block stem from moving the inner loop into a `JdbcUtils` method, there are one or two non-trivial changes that may impact cleanup in error situations. I'll comment on these changes below in order to help walk through them.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSet -> Se...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14907
**[Test build #64816 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64816/consoleFull)** for PR 14907 at commit [`f09174b`](https://github.com/apache/spark/commit/f09174b03a27b9fcd8ed6c28608a16a5433a5b19).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77100063
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala ---
@@ -154,6 +163,288 @@ object JdbcUtils extends Logging {
throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.simpleString}"))
}
+ /**
+ * Maps a JDBC type to a Catalyst type. This function is called only when
+ * the JdbcDialect class corresponding to your database driver returns null.
+ *
+ * @param sqlType - A field of java.sql.Types
+ * @return The Catalyst type corresponding to sqlType.
+ */
+ private def getCatalystType(
+ sqlType: Int,
+ precision: Int,
+ scale: Int,
+ signed: Boolean): DataType = {
+ val answer = sqlType match {
+ // scalastyle:off
+ case java.sql.Types.ARRAY => null
+ 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
+ case java.sql.Types.BLOB => BinaryType
+ case java.sql.Types.BOOLEAN => BooleanType
+ case java.sql.Types.CHAR => StringType
+ case java.sql.Types.CLOB => StringType
+ case java.sql.Types.DATALINK => null
+ case java.sql.Types.DATE => DateType
+ case java.sql.Types.DECIMAL
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.DECIMAL => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.DISTINCT => null
+ case java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.FLOAT => FloatType
+ case java.sql.Types.INTEGER => if (signed) { IntegerType } else { LongType }
+ case java.sql.Types.JAVA_OBJECT => null
+ case java.sql.Types.LONGNVARCHAR => StringType
+ case java.sql.Types.LONGVARBINARY => BinaryType
+ case java.sql.Types.LONGVARCHAR => StringType
+ case java.sql.Types.NCHAR => StringType
+ case java.sql.Types.NCLOB => StringType
+ case java.sql.Types.NULL => null
+ case java.sql.Types.NUMERIC
+ if precision != 0 || scale != 0 => DecimalType.bounded(precision, scale)
+ case java.sql.Types.NUMERIC => DecimalType.SYSTEM_DEFAULT
+ case java.sql.Types.NVARCHAR => StringType
+ case java.sql.Types.OTHER => null
+ case java.sql.Types.REAL => DoubleType
+ case java.sql.Types.REF => StringType
+ case java.sql.Types.ROWID => LongType
+ case java.sql.Types.SMALLINT => IntegerType
+ case java.sql.Types.SQLXML => StringType
+ case java.sql.Types.STRUCT => StringType
+ case java.sql.Types.TIME => TimestampType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.TINYINT => IntegerType
+ case java.sql.Types.VARBINARY => BinaryType
+ case java.sql.Types.VARCHAR => StringType
+ case _ => null
+ // scalastyle:on
+ }
+
+ if (answer == null) throw new SQLException("Unsupported type " + sqlType)
+ answer
+ }
+
+ /**
+ * Takes a [[ResultSet]] and returns its Catalyst schema.
+ *
+ * @return A [[StructType]] giving the Catalyst schema.
+ * @throws SQLException if the schema contains an unsupported type.
+ */
+ def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+ val rsmd = resultSet.getMetaData
+ val ncols = rsmd.getColumnCount
+ val fields = new Array[StructField](ncols)
+ var i = 0
+ while (i < ncols) {
+ val columnName = rsmd.getColumnLabel(i + 1)
+ val dataType = rsmd.getColumnType(i + 1)
+ val typeName = rsmd.getColumnTypeName(i + 1)
+ val fieldSize = rsmd.getPrecision(i + 1)
+ val fieldScale = rsmd.getScale(i + 1)
+ val isSigned = rsmd.isSigned(i + 1)
+ val nullable = rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
+ val metadata = new MetadataBuilder()
+ .putString("name", columnName)
+ .putLong("scale", fieldScale)
+ val columnType =
+ dialect.getCatalystType(dataType, typeName, fieldSize, metadata).getOrElse(
+ getCatalystType(dataType, fieldSize, fieldScale, isSigned))
+ fields(i) = StructField(columnName, columnType, nullable, metadata.build())
+ i = i + 1
+ }
+ new StructType(fields)
+ }
+
+ /**
+ * Convert a [[ResultSet]] into an iterator of Catalyst Rows.
+ */
+ def resultSetToRows(resultSet: ResultSet, schema: StructType): Iterator[Row] = {
+ val inputMetrics =
+ Option(TaskContext.get()).map(_.taskMetrics().inputMetrics).getOrElse(new InputMetrics)
+ val encoder = RowEncoder(schema).resolveAndBind()
+ val internalRows = resultSetToSparkInternalRows(resultSet, schema, inputMetrics)
+ internalRows.map(encoder.fromRow)
+ }
+
+ private[spark] def resultSetToSparkInternalRows(
+ resultSet: ResultSet,
+ schema: StructType,
+ inputMetrics: InputMetrics): Iterator[InternalRow] = {
+ new NextIterator[InternalRow] {
+ private[this] val rs = resultSet
+ private[this] val getters: Array[JDBCValueGetter] = makeGetters(schema)
+ private[this] val mutableRow = new SpecificMutableRow(schema.fields.map(x => x.dataType))
+
+ override protected def close(): Unit = {
--- End diff --
Note that we may now close the `ResultSet` three times in `JDBCRDD`: here, at the `CompletionIterator`, and in the `TaskContext` completion callback. This is fine, though, since the old code already assumed that `rs.close()` was idempotent.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14907: [SPARK-17351] Refactor JDBCRDD to expose ResultSe...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14907#discussion_r77375713
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala ---
@@ -546,33 +281,33 @@ private[jdbc] class JDBCRDD(
closed = true
}
- override def hasNext: Boolean = {
- if (!finished) {
- if (!gotNext) {
- nextValue = getNext()
- if (finished) {
- close()
- }
- gotNext = true
- }
- }
- !finished
- }
+ context.addTaskCompletionListener{ context => close() }
--- End diff --
Nit: space before the bracket.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org