This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new f97481501e56 [SPARK-57380][CONNECT] Support complex types (ARRAY, MAP,
STRUCT) in the Connect JDBC driver
f97481501e56 is described below
commit f97481501e5642817729d792ba6667936d135175
Author: Jiwon Park <[email protected]>
AuthorDate: Mon Jun 22 11:34:01 2026 +0800
[SPARK-57380][CONNECT] Support complex types (ARRAY, MAP, STRUCT) in the
Connect JDBC driver
### What changes were proposed in this pull request?
Support complex types (ARRAY, MAP, STRUCT) in the Spark Connect JDBC
driver, in both metadata (`SparkConnectResultSetMetaData`,
`SparkConnectDatabaseMetaData.getColumns`) and values
(`SparkConnectResultSet`). These previously threw
`SQLFeatureNotSupportedException`.
| | ARRAY | MAP | STRUCT |
|---|---|---|---|
| `getColumnType` | `Types.ARRAY` | `Types.JAVA_OBJECT` | `Types.STRUCT` |
| `getColumnClassName` / `getObject` | `java.sql.Array` | `java.util.Map` |
`java.sql.Struct` |
| `getPrecision` / `getScale` | 0 | 0 | 0 |
| `getColumnDisplaySize` / `getColumns` COLUMN_SIZE | `Int.MaxValue` |
`Int.MaxValue` | `Int.MaxValue` |
`getObject` returns standard JDBC types (nested types converted
recursively); JDBC has no Map type, so MAP returns `java.util.Map`. `getString`
and the complex-type `toString` render JSON (`Row.json` format). MAP uses
`JAVA_OBJECT` since there is no `Types.MAP`; precision/scale 0 follow the JDBC
javadoc for not-applicable types.
### Why are the changes needed?
JDBC tools read `ResultSetMetaData` before rendering rows and fetch complex
columns via `getObject`/`getArray`, so without this a complex-typed column
breaks both query rendering and `getColumns` schema introspection (DataGrip,
DBeaver). Returning standard `java.sql.Array` / `java.sql.Struct` lets tools
render structs as nested fields and arrays in dedicated viewers.
### Does this PR introduce _any_ user-facing change?
Yes, in the unreleased master only: the methods above now return values
instead of throwing.
### How was this patch tested?
New/updated tests in `SparkConnectJdbcDataTypeSuite` and
`SparkConnectDatabaseMetaDataSuite`.
build/sbt 'connect-client-jdbc/testOnly *SparkConnectJdbcDataTypeSuite
*SparkConnectDatabaseMetaDataSuite'
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
Closes #56447 from j1wonpark/SPARK-57380.
Authored-by: Jiwon Park <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../connect/client/jdbc/SparkConnectArray.scala | 66 ++++++++
.../sql/connect/client/jdbc/SparkConnectMap.scala | 41 +++++
.../client/jdbc/SparkConnectResultSet.scala | 28 +++-
.../connect/client/jdbc/SparkConnectStruct.scala | 47 ++++++
.../connect/client/jdbc/util/JdbcTypeUtils.scala | 58 ++++++-
.../jdbc/SparkConnectDatabaseMetaDataSuite.scala | 8 +-
.../jdbc/SparkConnectJdbcDataTypeSuite.scala | 186 ++++++++++++++++++++-
7 files changed, 421 insertions(+), 13 deletions(-)
diff --git
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectArray.scala
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectArray.scala
new file mode 100644
index 000000000000..1e2600f01ca2
--- /dev/null
+++
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectArray.scala
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.client.jdbc
+
+import java.sql.{Array => JdbcArray, ResultSet,
SQLFeatureNotSupportedException}
+import java.util
+
+import org.apache.spark.sql.connect.client.jdbc.util.JdbcTypeUtils
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+/**
+ * A [[java.sql.Array]] backed by an ARRAY value materialized by the Spark
Connect client.
+ * Elements are converted to standard JDBC objects on access (nested
ARRAY/MAP/STRUCT become
+ * [[java.sql.Array]] / [[java.util.Map]] / [[java.sql.Struct]]).
+ */
+private[jdbc] class SparkConnectArray(
+ elements: scala.collection.Seq[Any],
+ elementType: DataType) extends JdbcArray {
+
+ override def getBaseTypeName: String = elementType.sql
+
+ override def getBaseType: Int = JdbcTypeUtils.getColumnType(elementType)
+
+ override def getArray: AnyRef =
+ elements.map(JdbcTypeUtils.toJdbcObject(_, elementType)).toArray
+
+ override def getArray(map: util.Map[String, Class[_]]): AnyRef =
+ throw new SQLFeatureNotSupportedException
+
+ override def getArray(index: Long, count: Int): AnyRef =
+ throw new SQLFeatureNotSupportedException
+
+ override def getArray(index: Long, count: Int, map: util.Map[String,
Class[_]]): AnyRef =
+ throw new SQLFeatureNotSupportedException
+
+ override def getResultSet: ResultSet =
+ throw new SQLFeatureNotSupportedException
+
+ override def getResultSet(map: util.Map[String, Class[_]]): ResultSet =
+ throw new SQLFeatureNotSupportedException
+
+ override def getResultSet(index: Long, count: Int): ResultSet =
+ throw new SQLFeatureNotSupportedException
+
+ override def getResultSet(index: Long, count: Int, map: util.Map[String,
Class[_]]): ResultSet =
+ throw new SQLFeatureNotSupportedException
+
+ override def free(): Unit = ()
+
+ override def toString: String = JdbcTypeUtils.toJson(elements,
ArrayType(elementType))
+}
diff --git
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectMap.scala
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectMap.scala
new file mode 100644
index 000000000000..90854ea2300b
--- /dev/null
+++
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectMap.scala
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.client.jdbc
+
+import org.apache.spark.sql.connect.client.jdbc.util.JdbcTypeUtils
+import org.apache.spark.sql.types.MapType
+
+/**
+ * A [[java.util.Map]] backed by a MAP value materialized by the Spark Connect
client. JDBC has
+ * no dedicated type for maps, so getObject returns a [[java.util.Map]];
entries are converted to
+ * standard JDBC objects (nested ARRAY/MAP/STRUCT become [[java.sql.Array]] /
[[java.util.Map]] /
+ * [[java.sql.Struct]]). toString renders JSON, consistent with
[[SparkConnectArray]] and
+ * [[SparkConnectStruct]].
+ */
+private[jdbc] class SparkConnectMap(
+ entries: scala.collection.Map[Any, Any],
+ mapType: MapType) extends java.util.LinkedHashMap[AnyRef, AnyRef] {
+
+ entries.foreach { case (k, v) =>
+ put(
+ JdbcTypeUtils.toJdbcObject(k, mapType.keyType),
+ JdbcTypeUtils.toJdbcObject(v, mapType.valueType))
+ }
+
+ override def toString: String = JdbcTypeUtils.toJson(entries, mapType)
+}
diff --git
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
index e90f80f783dc..6b02f655f04c 100644
---
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
+++
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
@@ -27,8 +27,8 @@ import java.util.Calendar
import org.apache.spark.sql.Row
import org.apache.spark.sql.connect.client.SparkResult
-import org.apache.spark.sql.connect.client.jdbc.util.JdbcErrorUtils
-import org.apache.spark.sql.types.{TimestampNTZType, TimestampType}
+import org.apache.spark.sql.connect.client.jdbc.util.{JdbcErrorUtils,
JdbcTypeUtils}
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType,
TimestampNTZType, TimestampType}
class SparkConnectResultSet(
sparkResult: SparkResult[Row],
@@ -115,10 +115,15 @@ class SparkConnectResultSet(
override def getString(columnIndex: Int): String = {
getColumnValue(columnIndex, null: String) { idx =>
- currentRow.get(idx) match {
- case bytes: Array[Byte] =>
- new String(bytes, java.nio.charset.StandardCharsets.UTF_8)
- case other => String.valueOf(other)
+ sparkResult.schema(idx).dataType match {
+ case dt @ (_: ArrayType | _: MapType | _: StructType) =>
+ JdbcTypeUtils.toJson(currentRow.get(idx), dt)
+ case _ =>
+ currentRow.get(idx) match {
+ case bytes: Array[Byte] =>
+ new String(bytes, java.nio.charset.StandardCharsets.UTF_8)
+ case other => String.valueOf(other)
+ }
}
}
}
@@ -269,7 +274,7 @@ class SparkConnectResultSet(
override def getObject(columnIndex: Int): AnyRef = {
getColumnValue(columnIndex, null: AnyRef) { idx =>
- currentRow.get(idx).asInstanceOf[AnyRef]
+ JdbcTypeUtils.toJdbcObject(currentRow.get(idx),
sparkResult.schema(idx).dataType)
}
}
@@ -529,7 +534,14 @@ class SparkConnectResultSet(
throw new SQLFeatureNotSupportedException
override def getArray(columnIndex: Int): JdbcArray =
- throw new SQLFeatureNotSupportedException
+ getColumnValue(columnIndex, null: JdbcArray) { idx =>
+ JdbcTypeUtils.toJdbcObject(currentRow.get(idx),
sparkResult.schema(idx).dataType) match {
+ case array: JdbcArray => array
+ case _ =>
+ throw new SQLException(
+ s"The column ${idx + 1} is not an ARRAY type.")
+ }
+ }
override def getObject(columnLabel: String, map: util.Map[String,
Class[_]]): AnyRef =
throw new SQLFeatureNotSupportedException
diff --git
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStruct.scala
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStruct.scala
new file mode 100644
index 000000000000..76411051dbd0
--- /dev/null
+++
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStruct.scala
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.client.jdbc
+
+import java.sql.{SQLFeatureNotSupportedException, Struct}
+import java.util
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connect.client.jdbc.util.JdbcTypeUtils
+import org.apache.spark.sql.types.StructType
+
+/**
+ * A [[java.sql.Struct]] backed by a STRUCT value materialized by the Spark
Connect client.
+ * Attributes are converted to standard JDBC objects on access (nested
ARRAY/MAP/STRUCT become
+ * [[java.sql.Array]] / [[java.util.Map]] / [[java.sql.Struct]]).
+ */
+private[jdbc] class SparkConnectStruct(
+ row: Row,
+ structType: StructType) extends Struct {
+
+ override def getSQLTypeName: String = structType.sql
+
+ override def getAttributes: Array[AnyRef] =
+ structType.fields.zipWithIndex.map { case (field, i) =>
+ JdbcTypeUtils.toJdbcObject(row.get(i), field.dataType)
+ }
+
+ override def getAttributes(map: util.Map[String, Class[_]]): Array[AnyRef] =
+ throw new SQLFeatureNotSupportedException
+
+ override def toString: String = JdbcTypeUtils.toJson(row, structType)
+}
diff --git
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
index 458f94c51f89..48a42cd9ec9c 100644
---
a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
+++
b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala
@@ -21,11 +21,19 @@ import java.lang.{Boolean => JBoolean, Byte => JByte,
Double => JDouble, Float =
import java.math.{BigDecimal => JBigDecimal}
import java.sql.{Array => _, _}
+import org.json4s.JObject
+import org.json4s.jackson.JsonMethods.{compact, render}
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.connect.client.jdbc.{SparkConnectArray,
SparkConnectMap, SparkConnectStruct}
import org.apache.spark.sql.types._
private[jdbc] object JdbcTypeUtils {
- def getColumnType(field: StructField): Int = field.dataType match {
+ def getColumnType(field: StructField): Int = getColumnType(field.dataType)
+
+ def getColumnType(dataType: DataType): Int = dataType match {
case NullType => Types.NULL
case BooleanType => Types.BOOLEAN
case ByteType => Types.TINYINT
@@ -41,6 +49,9 @@ private[jdbc] object JdbcTypeUtils {
case TimestampNTZType => Types.TIMESTAMP
case BinaryType => Types.VARBINARY
case _: TimeType => Types.TIME
+ case _: ArrayType => Types.ARRAY
+ case _: MapType => Types.JAVA_OBJECT
+ case _: StructType => Types.STRUCT
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not
supported yet.")
}
@@ -61,6 +72,9 @@ private[jdbc] object JdbcTypeUtils {
case TimestampNTZType => classOf[Timestamp].getName
case BinaryType => classOf[Array[Byte]].getName
case _: TimeType => classOf[Time].getName
+ case _: ArrayType => classOf[java.sql.Array].getName
+ case _: MapType => classOf[java.util.Map[_, _]].getName
+ case _: StructType => classOf[java.sql.Struct].getName
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not
supported yet.")
}
@@ -69,7 +83,7 @@ private[jdbc] object JdbcTypeUtils {
case ByteType | ShortType | IntegerType | LongType | FloatType |
DoubleType |
_: DecimalType => true
case NullType | BooleanType | StringType | DateType | BinaryType | _:
TimeType |
- TimestampType | TimestampNTZType => false
+ TimestampType | TimestampNTZType | _: ArrayType | _: MapType | _:
StructType => false
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not
supported yet.")
}
@@ -94,6 +108,9 @@ private[jdbc] object JdbcTypeUtils {
// Users can call getObject(index, classOf[LocalTime]) to access full
microsecond
// precision when the source type is TIME(4) or higher.
case TimeType(precision) => precision
+ // ResultSetMetaData.getPrecision: "0 is returned for data types where the
+ // column size is not applicable."
+ case _: ArrayType | _: MapType | _: StructType => 0
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not
supported yet.")
}
@@ -104,7 +121,7 @@ private[jdbc] object JdbcTypeUtils {
case TimestampType => 6
case TimestampNTZType => 6
case NullType | BooleanType | ByteType | ShortType | IntegerType |
LongType | StringType |
- DateType | BinaryType | _: TimeType => 0
+ DateType | BinaryType | _: TimeType | _: ArrayType | _: MapType | _:
StructType => 0
case DecimalType.Fixed(_, s) => s
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not
supported yet.")
@@ -131,6 +148,7 @@ private[jdbc] object JdbcTypeUtils {
case DecimalType.Fixed(p, s) if s == 0 => p + 1
// precision + negative sign + decimal point, like DECIMAL(5,2) = -123.45
case DecimalType.Fixed(p, _) => p + 2
+ case _: ArrayType | _: MapType | _: StructType => Int.MaxValue
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not
supported yet.")
}
@@ -149,4 +167,38 @@ private[jdbc] object JdbcTypeUtils {
case _: NumericType => 10
case _ => null
}
+
+ /**
+ * Converts a value materialized by the Spark Connect client (Scala Seq /
Map / Row for
+ * complex types) into the corresponding standard JDBC object, recursively:
+ * ARRAY -> java.sql.Array, STRUCT -> java.sql.Struct, MAP -> java.util.Map.
Scalar values
+ * are returned as is.
+ */
+ def toJdbcObject(value: Any, dataType: DataType): AnyRef = {
+ if (value == null) {
+ null
+ } else {
+ dataType match {
+ case at: ArrayType =>
+ new SparkConnectArray(value.asInstanceOf[scala.collection.Seq[Any]],
at.elementType)
+ case st: StructType =>
+ new SparkConnectStruct(value.asInstanceOf[Row], st)
+ case mt: MapType =>
+ new SparkConnectMap(value.asInstanceOf[scala.collection.Map[Any,
Any]], mt)
+ case _ => value.asInstanceOf[AnyRef]
+ }
+ }
+ }
+
+ /**
+ * Renders a complex type value as JSON text following the Row.json format,
used as a
+ * stable string representation for getString and the complex type toString
fallbacks.
+ * Row.jsonValue requires a Row with a schema, so wrap the value into a
single-field row
+ * and extract the field back from the rendered JSON object.
+ */
+ def toJson(value: Any, dataType: DataType): String = {
+ val wrapped = new GenericRowWithSchema(
+ Array(value), StructType(Array(StructField("c", dataType))))
+ compact(render(wrapped.jsonValue.asInstanceOf[JObject].obj.head._2))
+ }
}
diff --git
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
index 747d9dc5d0ff..49c0dccc780c 100644
---
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
+++
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
@@ -667,7 +667,10 @@ class SparkConnectDatabaseMetaDataSuite extends
ConnectFunSuite with RemoteSpark
| col_timestamp TIMESTAMP,
| col_timestamp_ntz TIMESTAMP_NTZ,
| col_binary BINARY,
- | col_time TIME)""".stripMargin)
+ | col_time TIME,
+ | col_array ARRAY<INT>,
+ | col_map MAP<STRING, INT>,
+ | col_struct STRUCT<a: INT, b: STRING>)""".stripMargin)
spark.sql(
"""CREATE VIEW IF NOT EXISTS spark_catalog.db1.t1_v AS
@@ -720,6 +723,9 @@ class SparkConnectDatabaseMetaDataSuite extends
ConnectFunSuite with RemoteSpark
("spark_catalog", "db1", "t2", 13, "col_timestamp_ntz"),
("spark_catalog", "db1", "t2", 14, "col_binary"),
("spark_catalog", "db1", "t2", 15, "col_time"),
+ ("spark_catalog", "db1", "t2", 16, "col_array"),
+ ("spark_catalog", "db1", "t2", 17, "col_map"),
+ ("spark_catalog", "db1", "t2", 18, "col_struct"),
("spark_catalog", "db_", "t_", 1, "id"),
("spark_catalog", "db_", "t_", 2, "i_"),
("spark_catalog", "db_2", "t_2", 1, "id"),
diff --git
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
index 99511c56740b..4ad8ab91d4b2 100644
---
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
+++
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala
@@ -17,12 +17,13 @@
package org.apache.spark.sql.connect.client.jdbc
-import java.sql.{ResultSet, SQLException, Types}
+import java.sql.{Array => JdbcArray, ResultSet, SQLException, Struct, Types}
import scala.util.Using
import org.apache.spark.sql.connect.client.jdbc.test.JdbcHelper
import org.apache.spark.sql.connect.test.{ConnectFunSuite, RemoteSparkSession}
+import org.apache.spark.util.SparkClassUtils
class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with
RemoteSparkSession
with JdbcHelper {
@@ -789,4 +790,187 @@ class SparkConnectJdbcDataTypeSuite extends
ConnectFunSuite with RemoteSparkSess
}
}
}
+
+ test("get array type") {
+ withExecuteQuery("SELECT array(1, 2, 3)") { rs =>
+ assert(rs.next())
+ val value = rs.getObject(1)
+ assert(value.isInstanceOf[JdbcArray])
+ val array = value.asInstanceOf[JdbcArray]
+ assert(array.getBaseType === Types.INTEGER)
+ assert(array.getArray.asInstanceOf[Array[_]].length === 3)
+ assert(array.toString === "[1,2,3]")
+ assert(rs.getArray(1).getArray.asInstanceOf[Array[_]].length === 3)
+ assert(rs.getString(1) === "[1,2,3]")
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnName(1) === "array(1, 2, 3)")
+ assert(metaData.getColumnLabel(1) === "array(1, 2, 3)")
+ assert(metaData.getColumnType(1) === Types.ARRAY)
+ assert(metaData.getColumnTypeName(1) === "ARRAY<INT>")
+ assert(metaData.getColumnClassName(1) === "java.sql.Array")
+
assert(SparkClassUtils.classForName(metaData.getColumnClassName(1)).isInstance(value))
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 0)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === Int.MaxValue)
+ }
+ }
+
+ test("get map type") {
+ withExecuteQuery("SELECT map('a', 1)") { rs =>
+ assert(rs.next())
+ val value = rs.getObject(1)
+ assert(value.isInstanceOf[java.util.Map[_, _]])
+ val map = value.asInstanceOf[java.util.Map[String, AnyRef]]
+ assert(map.get("a") === 1)
+ assert(value.toString === """{"a":1}""")
+ assert(rs.getString(1) === """{"a":1}""")
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnName(1) === "map(a, 1)")
+ assert(metaData.getColumnLabel(1) === "map(a, 1)")
+ assert(metaData.getColumnType(1) === Types.JAVA_OBJECT)
+ assert(metaData.getColumnTypeName(1) === "MAP<STRING, INT>")
+ assert(metaData.getColumnClassName(1) === "java.util.Map")
+
assert(SparkClassUtils.classForName(metaData.getColumnClassName(1)).isInstance(value))
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 0)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === Int.MaxValue)
+ }
+ }
+
+ test("get struct type") {
+ withExecuteQuery("SELECT named_struct('a', 1, 'b', 'x')") { rs =>
+ assert(rs.next())
+ val value = rs.getObject(1)
+ assert(value.isInstanceOf[Struct])
+ val struct = value.asInstanceOf[Struct]
+ val attrs = struct.getAttributes
+ assert(attrs(0) === 1)
+ assert(attrs(1) === "x")
+ assert(value.toString === """{"a":1,"b":"x"}""")
+ assert(rs.getString(1) === """{"a":1,"b":"x"}""")
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnName(1) === "named_struct(a, 1, b, x)")
+ assert(metaData.getColumnLabel(1) === "named_struct(a, 1, b, x)")
+ assert(metaData.getColumnType(1) === Types.STRUCT)
+ assert(metaData.getColumnTypeName(1) === "STRUCT<a: INT NOT NULL, b:
STRING NOT NULL>")
+ assert(metaData.getColumnClassName(1) === "java.sql.Struct")
+
assert(SparkClassUtils.classForName(metaData.getColumnClassName(1)).isInstance(value))
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 0)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === Int.MaxValue)
+ }
+ }
+
+ test("get nested array of struct type") {
+ withExecuteQuery("SELECT array(named_struct('a', 1), named_struct('a',
2))") { rs =>
+ assert(rs.next())
+ val value = rs.getObject(1)
+ assert(value.isInstanceOf[JdbcArray])
+ val elems = value.asInstanceOf[JdbcArray].getArray.asInstanceOf[Array[_]]
+ assert(elems.length === 2)
+ assert(elems(0).isInstanceOf[Struct])
+ assert(elems(0).asInstanceOf[Struct].getAttributes()(0) === 1)
+ assert(elems(1).asInstanceOf[Struct].getAttributes()(0) === 2)
+ assert(value.toString === """[{"a":1},{"a":2}]""")
+ assert(rs.getString(1) === """[{"a":1},{"a":2}]""")
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnType(1) === Types.ARRAY)
+ assert(metaData.getColumnTypeName(1) === "ARRAY<STRUCT<a: INT NOT
NULL>>")
+ assert(metaData.getColumnClassName(1) === "java.sql.Array")
+
assert(SparkClassUtils.classForName(metaData.getColumnClassName(1)).isInstance(value))
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 0)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === Int.MaxValue)
+ }
+ }
+
+ test("get nested map of array type") {
+ withExecuteQuery("SELECT map('x', array(1, 2))") { rs =>
+ assert(rs.next())
+ val value = rs.getObject(1)
+ assert(value.isInstanceOf[java.util.Map[_, _]])
+ val map = value.asInstanceOf[java.util.Map[String, AnyRef]]
+ assert(map.get("x").isInstanceOf[JdbcArray])
+
assert(map.get("x").asInstanceOf[JdbcArray].getArray.asInstanceOf[Array[_]].length
=== 2)
+ assert(value.toString === """{"x":[1,2]}""")
+ assert(rs.getString(1) === """{"x":[1,2]}""")
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnType(1) === Types.JAVA_OBJECT)
+ assert(metaData.getColumnTypeName(1) === "MAP<STRING, ARRAY<INT>>")
+ assert(metaData.getColumnClassName(1) === "java.util.Map")
+
assert(SparkClassUtils.classForName(metaData.getColumnClassName(1)).isInstance(value))
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 0)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === Int.MaxValue)
+ }
+ }
+
+ test("get nested struct of map type") {
+ withExecuteQuery("SELECT named_struct('m', map('k', 1))") { rs =>
+ assert(rs.next())
+ val value = rs.getObject(1)
+ assert(value.isInstanceOf[Struct])
+ val attr = value.asInstanceOf[Struct].getAttributes()(0)
+ assert(attr.isInstanceOf[java.util.Map[_, _]])
+ assert(attr.asInstanceOf[java.util.Map[String, AnyRef]].get("k") === 1)
+ assert(value.toString === """{"m":{"k":1}}""")
+ assert(rs.getString(1) === """{"m":{"k":1}}""")
+ assert(!rs.wasNull)
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnCount === 1)
+ assert(metaData.getColumnType(1) === Types.STRUCT)
+ assert(metaData.getColumnTypeName(1) === "STRUCT<m: MAP<STRING, INT> NOT
NULL>")
+ assert(metaData.getColumnClassName(1) === "java.sql.Struct")
+
assert(SparkClassUtils.classForName(metaData.getColumnClassName(1)).isInstance(value))
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 0)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === Int.MaxValue)
+ }
+ }
+
+ test("get complex types with null") {
+ Seq(
+ "cast(null as array<int>)",
+ "cast(null as map<string, int>)",
+ "cast(null as struct<a: int>)").foreach { expr =>
+ withExecuteQuery(s"SELECT $expr") { rs =>
+ assert(rs.next())
+ assert(rs.getObject(1) === null)
+ assert(rs.wasNull)
+ assert(rs.getString(1) === null)
+ assert(rs.wasNull)
+ assert(rs.getArray(1) === null)
+ assert(rs.wasNull)
+ assert(!rs.next())
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
