You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/20 14:41:28 UTC
(spark) branch branch-4.x updated: [SPARK-58806][SQL][CONNECT] Expose CHAR/VARCHAR metadata and decode result schemas
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new f48d84f0b8b5 [SPARK-58806][SQL][CONNECT] Expose CHAR/VARCHAR metadata and decode result schemas
f48d84f0b8b5 is described below
commit f48d84f0b8b51894a996204dc69a808dbaf9ea4c
Author: Serge Rielau <se...@rielau.com>
AuthorDate: Thu Aug 20 22:40:58 2026 +0800
[SPARK-58806][SQL][CONNECT] Expose CHAR/VARCHAR metadata and decode result schemas
### What changes were proposed in this pull request?
Covers Connect JDBC mapping ([SPARK-58806](https://issues.apache.org/jira/browse/SPARK-58806)), plus HiveServer2 / JDBC result metadata ([SPARK-58804](https://issues.apache.org/jira/browse/SPARK-58804)) and `DatabaseMetaData.getColumns` `COLUMN_SIZE` ([SPARK-58805](https://issues.apache.org/jira/browse/SPARK-58805)). Parent: [SPARK-58794](https://issues.apache.org/jira/browse/SPARK-58794).
Independent of the remaining CHAR/VARCHAR follow-ups. Rebased onto `master` after #58033; can merge without #58080 / #58087 / #58130. Unique vs master: https://github.com/srielau/spark/compare/master...serge-rielau_data/SPARK-58794-clients
Expose first-class `CHAR(n)` and `VARCHAR(n)` to SQL clients when `spark.sql.charVarchar.standardSemantics.enabled` is on, and let those clients decode result rows that carry the types.
Metadata:
- Spark Connect JDBC maps `CharType` / `VarcharType` to `java.sql.Types.CHAR` / `VARCHAR`, with Java string values and precision / display size `n`.
- Connect `DatabaseMetaData.getColumns` reports `COLUMN_SIZE` as the declared character length `n`. `CHAR_OCTET_LENGTH` is the UTF-8 maximum byte capacity `4 * n` (saturating at `Int.MaxValue`); unbounded STRING stays 0.
- Connect `DatabaseMetaData.getTypeInfo` lists `CHAR` and `VARCHAR` with `CREATE_PARAMS = length`. Unbounded `STRING` remains a `Types.VARCHAR` row without create params.
- HiveServer2 `getColumns` reports `COLUMN_SIZE = n` for `VARCHAR(n)` as well as `CHAR(n)`, and `CHAR_OCTET_LENGTH = 4 * n` for both (null for unbounded STRING).
- HiveServer2 result metadata coverage pins Hive JDBC names (`char` / `varchar`, no length) and precision `n` for CAST results.
Decode (JVM Connect client):
- The Connect client process has no engine `SQLConf`, so `RowEncoder.encoderFor` still follows the local `charVarcharFirstClassTypes` flag and rejects CHAR/VARCHAR when first-class types are off.
- `RowEncoder.encoderForResultSchema` always accepts CHAR/VARCHAR. `SparkResult.createEncoder` uses it only for `UnboundRowEncoder` (engine-produced result schemas). Other encoder bindings keep `encoderFor`.
- Arrow serializer / deserializer treat `CharEncoder` / `VarcharEncoder` like `StringEncoder`, and deserialize via `StringHelper.plainStringType` so STRING Arrow vectors up-cast to CHAR/VARCHAR.
- JDBC CAST `ResultSet.next` / `collect` needs this path; metadata-only mapping is not enough.
Python Connect already maps proto `char` / `var_char` in `pyspark.sql.connect.types`; a mixed classic vs Connect test covers schema and `collect()`.
The existing Connect proto already carries CHAR/VARCHAR and their lengths.
### Why are the changes needed?
Spark Connect JDBC rejected first-class `CharType` and `VarcharType` as unsupported because its metadata mapping only recognized the `StringType` singleton. `getColumns` hardcoded `CHAR_OCTET_LENGTH` to 0, and `getTypeInfo` listed only unbounded STRING. HiveServer2 already identified VARCHAR catalog columns but reported `COLUMN_SIZE` as unknown. Clients therefore cannot reliably describe or size CHAR/VARCHAR columns even though Catalyst retains the type and length.
Separately, a server with standard semantics sends CHAR/VARCHAR in the result schema. The JVM Connect client then failed at row decode (`UNSUPPORTED_DATA_TYPE_FOR_ENCODER`) because `RowEncoder.encoderFor` read the client's default conf (`charVarcharFirstClassTypes = false`). JDBC CAST collect and Spark Connect `SparkResult` both hit that path.
### Does this PR introduce _any_ user-facing change?
Yes, when first-class CHAR/VARCHAR types are present: JDBC and HiveServer2 metadata now report the corresponding JDBC type and declared character length instead of rejecting the type or reporting an unknown size. `CHAR_OCTET_LENGTH` is the UTF-8 maximum byte capacity (`4 * n`), not the character length. Connect clients can also collect CHAR/VARCHAR result columns instead of failing to decode the schema.
### How was this patch tested?
- `RowEncoderSuite`: `encoderFor` still raises `UNSUPPORTED_DATA_TYPE_FOR_ENCODER` for CHAR/VARCHAR when both first-class flags are off; `encoderForResultSchema` accepts the same schema.
- `ArrowEncoderSuite`: CHAR/VARCHAR Arrow round-trip (top-level, nested struct, array).
- `SparkConnectJdbcDataTypeSuite`: CAST result metadata and table `getColumns` (`CHAR(4)` / `VARCHAR(6)`, `COLUMN_SIZE`, `CHAR_OCTET_LENGTH` = 16 / 24); CAST collect after decode.
- `SparkConnectDatabaseMetaDataSuite`: `getTypeInfo` rows for CHAR / VARCHAR.
- `SparkMetadataOperationSuite`: HiveServer2 `getColumns` / CAST result metadata (`CHAR_OCTET_LENGTH` = `4 * n`).
- `pyspark.sql.tests.connect.test_connect_basic.SparkConnectBasicTests.test_char_varchar_result_schema`: classic vs Connect schema and `collect()`.
Local compile of `sql-api`, `connect-client-jdbc`, and `hive-thriftserver` passed. HiveServer2 / full Connect JDBC in this environment remain blocked (`CheckReturnValue` while compiling `connect-common`; FIPS Python multiprocessing). Please treat the suites above as the merge gate.
### Was this patch authored or co-authored using generative AI tooling?
Yes (Cursor).
Closes #58132 from srielau/serge-rielau_data/SPARK-58794-clients.
Authored-by: Serge Rielau <se...@rielau.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 6af294ef12f13ed5c6cb0f207e4b29d8d407e337)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/tests/connect/test_connect_basic.py | 15 ++++++
.../spark/sql/catalyst/encoders/RowEncoder.scala | 35 +++++++++----
.../sql/catalyst/encoders/RowEncoderSuite.scala | 22 +++++++-
.../client/jdbc/SparkConnectDatabaseMetaData.scala | 9 ++--
.../connect/client/jdbc/util/JdbcTypeUtils.scala | 28 ++++++++--
.../jdbc/SparkConnectDatabaseMetaDataSuite.scala | 19 +++++--
.../jdbc/SparkConnectJdbcDataTypeSuite.scala | 60 ++++++++++++++++++++++
.../connect/client/arrow/ArrowEncoderSuite.scala | 26 +++++++++-
.../spark/sql/connect/client/SparkResult.scala | 2 +-
.../connect/client/arrow/ArrowDeserializer.scala | 12 +++--
.../sql/connect/client/arrow/ArrowSerializer.scala | 4 +-
.../thriftserver/SparkGetColumnsOperation.scala | 24 +++++++--
.../thriftserver/SparkMetadataOperationSuite.scala | 31 ++++++++++-
13 files changed, 253 insertions(+), 34 deletions(-)
diff --git a/python/pyspark/sql/tests/connect/test_connect_basic.py b/python/pyspark/sql/tests/connect/test_connect_basic.py
index b8d258267d26..80e5cdfc57e3 100755
--- a/python/pyspark/sql/tests/connect/test_connect_basic.py
+++ b/python/pyspark/sql/tests/connect/test_connect_basic.py
@@ -34,6 +34,8 @@ from pyspark.sql.types import (
IntegerType,
MapType,
ArrayType,
+ CharType,
+ VarcharType,
Row,
)
from pyspark.testing import assertDataFrameEqual
@@ -487,6 +489,19 @@ class SparkConnectBasicTests(SparkConnectSQLTestCase):
)
self._check_print_schema(query)
+ def test_char_varchar_result_schema(self):
+ # SPARK-58794: Python Connect maps first-class CHAR/VARCHAR the same as classic.
+ query = "SELECT CAST('ab' AS CHAR(4)) AS c, CAST('cd' AS VARCHAR(6)) AS v"
+ conf = {"spark.sql.charVarchar.standardSemantics.enabled": "true"}
+ with self.both_conf(conf):
+ classic_df = self.spark.sql(query)
+ connect_df = self.connect.sql(query)
+ self.assertEqual(classic_df.schema, connect_df.schema)
+ self.assertEqual(classic_df.schema["c"].dataType, CharType(4))
+ self.assertEqual(classic_df.schema["v"].dataType, VarcharType(6))
+ self.assertEqual(classic_df.collect(), connect_df.collect())
+ self.assertEqual(connect_df.collect(), [Row(c="ab ", v="cd")])
+
def test_to(self):
# SPARK-41464: test DataFrame.to()
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
index 51157b534490..7627998f5a6d 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
@@ -72,14 +72,31 @@ object RowEncoder extends DataTypeErrorsBase {
encoderForDataType(schema, lenient).asInstanceOf[AgnosticEncoder[Row]]
}
+ /**
+ * Builds an encoder for a schema that the engine produced, such as the result schema of a Spark
+ * Connect query. Whether CHAR/VARCHAR are first-class types is decided by the session that
+ * produced the schema, so they are always accepted here. A client cannot read that session's
+ * configuration, and refusing the type would make the result undecodable.
+ */
+ private[sql] def encoderForResultSchema(schema: StructType): AgnosticEncoder[Row] =
+ encoderForDataType(schema, lenient = false, charVarcharFirstClassTypes = true)
+ .asInstanceOf[AgnosticEncoder[Row]]
+
private[sql] def encoderForDataType(dataType: DataType, lenient: Boolean): AgnosticEncoder[_] =
+ encoderForDataType(dataType, lenient, SqlApiConf.get.charVarcharFirstClassTypes)
+
+ private def encoderForDataType(
+ dataType: DataType,
+ lenient: Boolean,
+ charVarcharFirstClassTypes: Boolean): AgnosticEncoder[_] =
TypeApiOps(dataType)
.map(_.getEncoder)
- .getOrElse(encoderForDataTypeDefault(dataType, lenient))
+ .getOrElse(encoderForDataTypeDefault(dataType, lenient, charVarcharFirstClassTypes))
private def encoderForDataTypeDefault(
dataType: DataType,
- lenient: Boolean): AgnosticEncoder[_] =
+ lenient: Boolean,
+ charVarcharFirstClassTypes: Boolean): AgnosticEncoder[_] =
dataType match {
case NullType => NullEncoder
case BooleanType => BoxedBooleanEncoder
@@ -91,9 +108,9 @@ object RowEncoder extends DataTypeErrorsBase {
case DoubleType => BoxedDoubleEncoder
case dt: DecimalType => JavaDecimalEncoder(dt, lenientSerialization = true)
case BinaryType => BinaryEncoder
- case c: CharType if SqlApiConf.get.charVarcharFirstClassTypes =>
+ case c: CharType if charVarcharFirstClassTypes =>
CharEncoder(c)
- case v: VarcharType if SqlApiConf.get.charVarcharFirstClassTypes =>
+ case v: VarcharType if charVarcharFirstClassTypes =>
VarcharEncoder(v)
case s: StringType if StringHelper.isPlainString(s) => StringEncoder
case TimestampType if SqlApiConf.get.datetimeJava8ApiEnabled => InstantEncoder(lenient)
@@ -107,25 +124,25 @@ object RowEncoder extends DataTypeErrorsBase {
case _: VariantType => VariantEncoder
case p: PythonUserDefinedType =>
// TODO check if this works.
- encoderForDataType(p.sqlType, lenient)
+ encoderForDataType(p.sqlType, lenient, charVarcharFirstClassTypes)
case udt: UserDefinedType[_] => UDTEncoder(udt, udt.getClass)
case ArrayType(elementType, containsNull) =>
IterableEncoder(
classTag[mutable.ArraySeq[_]],
- encoderForDataType(elementType, lenient),
+ encoderForDataType(elementType, lenient, charVarcharFirstClassTypes),
containsNull,
lenientSerialization = true)
case MapType(keyType, valueType, valueContainsNull) =>
MapEncoder(
classTag[scala.collection.Map[_, _]],
- encoderForDataType(keyType, lenient),
- encoderForDataType(valueType, lenient),
+ encoderForDataType(keyType, lenient, charVarcharFirstClassTypes),
+ encoderForDataType(valueType, lenient, charVarcharFirstClassTypes),
valueContainsNull)
case StructType(fields) =>
AgnosticRowEncoder(fields.map { field =>
EncoderField(
field.name,
- encoderForDataType(field.dataType, lenient),
+ encoderForDataType(field.dataType, lenient, charVarcharFirstClassTypes),
field.nullable,
field.metadata)
}.toImmutableArraySeq)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
index b7a052d5fc4e..2a2889e441c7 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
@@ -21,7 +21,7 @@ import scala.collection.mutable
import scala.util.Random
import org.apache.spark.{SparkException, SparkRuntimeException}
-import org.apache.spark.sql.{RandomDataGenerator, Row}
+import org.apache.spark.sql.{AnalysisException, RandomDataGenerator, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.plans.CodegenInterpretedPlanTest
import org.apache.spark.sql.catalyst.util.{ArrayData, DateTimeUtils, GenericArrayData, IntervalUtils}
@@ -590,6 +590,26 @@ class RowEncoderSuite extends CodegenInterpretedPlanTest {
val row = encoder.createSerializer()(data)
}
+ test("SPARK-58794: encoderFor rejects CHAR/VARCHAR when first-class types are off") {
+ Seq(CharType(4), VarcharType(6)).foreach { dt =>
+ val schema = new StructType().add("c", dt)
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ RowEncoder.encoderFor(schema)
+ },
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_ENCODER",
+ sqlState = "0A000",
+ parameters = Map("dataType" -> s"\"${dt.sql}\"")
+ )
+ // Engine-produced result schemas still decode on the Connect client.
+ RowEncoder.encoderForResultSchema(schema)
+ }
+ }
+ }
+
test("do not allow serializing too long strings into char/varchar") {
Seq(CharType(5), VarcharType(5)).foreach { typ =>
Seq(
diff --git a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaData.scala b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaData.scala
index 823984f666bc..b49894e92809 100644
--- a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaData.scala
+++ b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaData.scala
@@ -540,7 +540,7 @@ class SparkConnectDatabaseMetaData(conn: SparkConnectConnection) extends Databas
if (field.nullable) columnNullable else columnNoNulls, // NULLABLE
field.getComment().orNull, // REMARKS
field.getCurrentDefaultValue().orNull, // COLUMN_DEF
- 0, // CHAR_OCTET_LENGTH
+ JdbcTypeUtils.getCharOctetLength(field), // CHAR_OCTET_LENGTH
i + 1, // ORDINAL_POSITION
if (field.nullable) "YES" else "NO", // IS_NULLABLE
"", // IS_AUTOINCREMENT
@@ -683,7 +683,7 @@ class SparkConnectDatabaseMetaData(conn: SparkConnectConnection) extends Databas
"SQL_DATA_TYPE",
"SQL_DATETIME_SUB",
"NUM_PREC_RADIX")
- .orderBy("DATA_TYPE")
+ .orderBy("DATA_TYPE", "TYPE_NAME")
new SparkConnectResultSet(df.collectResult())
}
@@ -944,7 +944,8 @@ object SparkConnectDatabaseMetaData {
numPrecRadix)
// Static JDBC type metadata for the Spark SQL atomic types, mirroring the
- // JdbcTypeUtils type-code/precision mapping. Only STRING is case-sensitive.
+ // JdbcTypeUtils type-code/precision mapping. CHAR, VARCHAR, and STRING are
+ // case-sensitive.
// TIMESTAMP_NTZ is omitted because it maps to the same JDBC type code
// (Types.TIMESTAMP) as TIMESTAMP, so the TIMESTAMP row already covers it.
// TIME is omitted for now because its maximum PRECISION/scale representation
@@ -958,6 +959,8 @@ object SparkConnectDatabaseMetaData {
typeRow("FLOAT", Types.FLOAT, 7, null, null, false, 0, 0, 10),
typeRow("DOUBLE", Types.DOUBLE, 15, null, null, false, 0, 0, 10),
typeRow("DECIMAL", Types.DECIMAL, 38, null, "precision,scale", false, 0, 38, 10),
+ typeRow("CHAR", Types.CHAR, Int.MaxValue, "'", "length", true, 0, 0, null),
+ typeRow("VARCHAR", Types.VARCHAR, Int.MaxValue, "'", "length", true, 0, 0, null),
typeRow("STRING", Types.VARCHAR, Int.MaxValue, "'", null, true, 0, 0, null),
typeRow("BINARY", Types.VARBINARY, Int.MaxValue, "X'", null, false, 0, 0, null,
literalSuffix = "'"),
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 48a42cd9ec9c..45bb8fcb43d8 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
@@ -42,6 +42,8 @@ private[jdbc] object JdbcTypeUtils {
case LongType => Types.BIGINT
case FloatType => Types.FLOAT
case DoubleType => Types.DOUBLE
+ case _: CharType => Types.CHAR
+ case _: VarcharType => Types.VARCHAR
case StringType => Types.VARCHAR
case _: DecimalType => Types.DECIMAL
case DateType => Types.DATE
@@ -65,7 +67,7 @@ private[jdbc] object JdbcTypeUtils {
case LongType => classOf[JLong].getName
case FloatType => classOf[JFloat].getName
case DoubleType => classOf[JDouble].getName
- case StringType => classOf[String].getName
+ case _: StringType => classOf[String].getName
case _: DecimalType => classOf[JBigDecimal].getName
case DateType => classOf[Date].getName
case TimestampType => classOf[Timestamp].getName
@@ -82,7 +84,7 @@ private[jdbc] object JdbcTypeUtils {
def isSigned(field: StructField): Boolean = field.dataType match {
case ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
_: DecimalType => true
- case NullType | BooleanType | StringType | DateType | BinaryType | _: TimeType |
+ case NullType | BooleanType | _: StringType | DateType | BinaryType | _: TimeType |
TimestampType | TimestampNTZType | _: ArrayType | _: MapType | _: StructType => false
case other =>
throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
@@ -97,6 +99,8 @@ private[jdbc] object JdbcTypeUtils {
case LongType => 19
case FloatType => 7
case DoubleType => 15
+ case c: CharType => c.length
+ case v: VarcharType => v.length
case StringType => Int.MaxValue
case DecimalType.Fixed(p, _) => p
case DateType => 10
@@ -120,7 +124,7 @@ private[jdbc] object JdbcTypeUtils {
case DoubleType => 15
case TimestampType => 6
case TimestampNTZType => 6
- case NullType | BooleanType | ByteType | ShortType | IntegerType | LongType | StringType |
+ case NullType | BooleanType | ByteType | ShortType | IntegerType | LongType | _: StringType |
DateType | BinaryType | _: TimeType | _: ArrayType | _: MapType | _: StructType => 0
case DecimalType.Fixed(_, s) => s
case other =>
@@ -134,7 +138,7 @@ private[jdbc] object JdbcTypeUtils {
getPrecision(field) + 1 // may have leading negative sign
case FloatType => 14
case DoubleType => 24
- case StringType =>
+ case _: StringType =>
getPrecision(field)
case DateType => 10 // length of `YYYY-MM-DD`
case TimestampType => 29 // length of `YYYY-MM-DD HH:MM:SS.SSSSSS`
@@ -168,6 +172,22 @@ private[jdbc] object JdbcTypeUtils {
case _ => null
}
+ /**
+ * JDBC `CHAR_OCTET_LENGTH` is a byte capacity. Spark CHAR/VARCHAR lengths are in
+ * characters, so report `4 * n` (UTF-8 maximum bytes per character), saturating at
+ * `Int.MaxValue`. Unbounded STRING and non-character types keep 0 (not applicable).
+ */
+ def getCharOctetLength(field: StructField): Int = field.dataType match {
+ case c: CharType => maxUtf8OctetLength(c.length)
+ case v: VarcharType => maxUtf8OctetLength(v.length)
+ case _ => 0
+ }
+
+ private def maxUtf8OctetLength(numChars: Int): Int = {
+ val maxChars = Int.MaxValue / 4
+ if (numChars > maxChars) Int.MaxValue else numChars * 4
+ }
+
/**
* Converts a value materialized by the Spark Connect client (Scala Seq / Map / Row for
* complex types) into the corresponding standard JDBC object, recursively:
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 17f3d77b77c5..de5099714b87 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
@@ -886,17 +886,19 @@ class SparkConnectDatabaseMetaDataSuite extends ConnectFunSuite with RemoteSpark
Option(rs.getObject("NUM_PREC_RADIX")).map(_.asInstanceOf[Integer].toInt))
}.toSeq
- // results are ordered by DATA_TYPE
+ // results are ordered by DATA_TYPE, then TYPE_NAME
assert(types.map(t => (t.name, t.dataType)) === Seq(
("TINYINT", Types.TINYINT),
("BIGINT", Types.BIGINT),
("BINARY", Types.VARBINARY),
+ ("CHAR", Types.CHAR),
("DECIMAL", Types.DECIMAL),
("INT", Types.INTEGER),
("SMALLINT", Types.SMALLINT),
("FLOAT", Types.FLOAT),
("DOUBLE", Types.DOUBLE),
("STRING", Types.VARCHAR),
+ ("VARCHAR", Types.VARCHAR),
("BOOLEAN", Types.BOOLEAN),
("DATE", Types.DATE),
("TIMESTAMP", Types.TIMESTAMP)))
@@ -904,12 +906,15 @@ class SparkConnectDatabaseMetaDataSuite extends ConnectFunSuite with RemoteSpark
// every type is nullable and searchable
assert(types.forall(_.nullable == DatabaseMetaData.typeNullable))
assert(types.forall(_.searchable == DatabaseMetaData.typeSearchable))
- // only STRING is case-sensitive
- assert(types.filter(_.caseSensitive).map(_.name) === Seq("STRING"))
+ // CHAR, VARCHAR and STRING are case-sensitive
+ assert(types.filter(_.caseSensitive).map(_.name) ===
+ Seq("CHAR", "STRING", "VARCHAR"))
// string-like types are quoted with a single quote on both sides, except BINARY,
// whose literals use the hex syntax X'...'. Numeric types carry no literal quote.
val quoted = Map(
+ "CHAR" -> ("'", "'"),
+ "VARCHAR" -> ("'", "'"),
"STRING" -> ("'", "'"),
"DATE" -> ("'", "'"),
"TIMESTAMP" -> ("'", "'"),
@@ -923,14 +928,18 @@ class SparkConnectDatabaseMetaDataSuite extends ConnectFunSuite with RemoteSpark
// PRECISION mirrors JdbcTypeUtils.getPrecision for every type
val precisions = Map(
"BOOLEAN" -> 1, "TINYINT" -> 3, "SMALLINT" -> 5, "INT" -> 10, "BIGINT" -> 19,
- "FLOAT" -> 7, "DOUBLE" -> 15, "DECIMAL" -> 38, "STRING" -> Int.MaxValue,
+ "FLOAT" -> 7, "DOUBLE" -> 15, "DECIMAL" -> 38,
+ "CHAR" -> Int.MaxValue, "VARCHAR" -> Int.MaxValue, "STRING" -> Int.MaxValue,
"BINARY" -> Int.MaxValue, "DATE" -> 10, "TIMESTAMP" -> 29)
types.foreach { t =>
assert(t.precision === precisions(t.name), s"unexpected PRECISION for ${t.name}")
}
// CREATE_PARAMS is set only for the parameterized types
- val createParams = Map("DECIMAL" -> "precision,scale")
+ val createParams = Map(
+ "DECIMAL" -> "precision,scale",
+ "CHAR" -> "length",
+ "VARCHAR" -> "length")
types.foreach { t =>
assert(t.createParams === createParams.getOrElse(t.name, null),
s"unexpected CREATE_PARAMS for ${t.name}")
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 4ad8ab91d4b2..59507c816b30 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
@@ -223,6 +223,66 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
}
}
+ test("SPARK-58794: get char and varchar types") {
+ withStatement { statement =>
+ statement.execute("SET spark.sql.charVarchar.standardSemantics.enabled=true")
+ withExecuteQuery(statement,
+ "SELECT CAST('ab' AS CHAR(4)) AS c, CAST('cd' AS VARCHAR(6)) AS v") { rs =>
+ assert(rs.next())
+ assert(rs.getString(1) === "ab ")
+ assert(rs.getString(2) === "cd")
+ assert(!rs.next())
+
+ val metaData = rs.getMetaData
+ assert(metaData.getColumnType(1) === Types.CHAR)
+ assert(metaData.getColumnTypeName(1) === "CHAR(4)")
+ assert(metaData.getColumnClassName(1) === "java.lang.String")
+ assert(metaData.isSigned(1) === false)
+ assert(metaData.getPrecision(1) === 4)
+ assert(metaData.getScale(1) === 0)
+ assert(metaData.getColumnDisplaySize(1) === 4)
+
+ assert(metaData.getColumnType(2) === Types.VARCHAR)
+ assert(metaData.getColumnTypeName(2) === "VARCHAR(6)")
+ assert(metaData.getColumnClassName(2) === "java.lang.String")
+ assert(metaData.isSigned(2) === false)
+ assert(metaData.getPrecision(2) === 6)
+ assert(metaData.getScale(2) === 0)
+ assert(metaData.getColumnDisplaySize(2) === 6)
+ }
+ }
+ }
+
+ test("SPARK-58794: getColumns for CHAR/VARCHAR table") {
+ val table = "char_varchar_jdbc_cols"
+ withStatement { statement =>
+ statement.execute("SET spark.sql.charVarchar.standardSemantics.enabled=true")
+ statement.execute(s"DROP TABLE IF EXISTS $table")
+ statement.execute(
+ s"CREATE TABLE $table (c CHAR(4), v VARCHAR(6)) USING parquet")
+ try {
+ Using.resource(
+ statement.getConnection.getMetaData.getColumns(null, null, table, null)) { rs =>
+ assert(rs.next())
+ assert(rs.getString("COLUMN_NAME") === "c")
+ assert(rs.getInt("DATA_TYPE") === Types.CHAR)
+ assert(rs.getString("TYPE_NAME") === "CHAR(4)")
+ assert(rs.getInt("COLUMN_SIZE") === 4)
+ assert(rs.getInt("CHAR_OCTET_LENGTH") === 16)
+ assert(rs.next())
+ assert(rs.getString("COLUMN_NAME") === "v")
+ assert(rs.getInt("DATA_TYPE") === Types.VARCHAR)
+ assert(rs.getString("TYPE_NAME") === "VARCHAR(6)")
+ assert(rs.getInt("COLUMN_SIZE") === 6)
+ assert(rs.getInt("CHAR_OCTET_LENGTH") === 24)
+ assert(!rs.next())
+ }
+ } finally {
+ statement.execute(s"DROP TABLE IF EXISTS $table")
+ }
+ }
+ }
+
test("get decimal type") {
withStatement { stmt =>
Seq(
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
index efb6a7a2304f..9963ec5620ff 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
@@ -40,7 +40,7 @@ import org.apache.spark.sql.{Encoders, Row}
import org.apache.spark.sql.catalyst.{DefinedByConstructorParams, JavaTypeInference, ScalaReflection}
import org.apache.spark.sql.catalyst.encoders.{AgnosticEncoder, Codec, OuterScopes}
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{agnosticEncoderFor, BinaryEncoder, BoxedBooleanEncoder, BoxedByteEncoder, BoxedDoubleEncoder, BoxedFloatEncoder, BoxedIntEncoder, BoxedLongEncoder, BoxedShortEncoder, CalendarIntervalEncoder, DateEncoder, DayTimeIntervalEncoder, EncoderField, InstantEncoder, IterableEncoder, JavaDecimalEncoder, LocalDateEncoder, LocalDateTimeEncoder, NullEncoder, PrimitiveBooleanEncoder, PrimitiveByteEncoder, PrimitiveDoubleEncoder, Primitiv [...]
-import org.apache.spark.sql.catalyst.encoders.RowEncoder.{encoderFor => toRowEncoder}
+import org.apache.spark.sql.catalyst.encoders.RowEncoder.{encoderFor => toRowEncoder, encoderForResultSchema => toResultRowEncoder}
import org.apache.spark.sql.catalyst.util.{DateFormatter, TimestampFormatter}
import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_SECOND
import org.apache.spark.sql.catalyst.util.IntervalStringStyles.ANSI_STYLE
@@ -48,7 +48,7 @@ import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
import org.apache.spark.sql.catalyst.util.SparkIntervalUtils._
import org.apache.spark.sql.connect.client.arrow.FooEnum.FooEnum
import org.apache.spark.sql.connect.test.ConnectFunSuite
-import org.apache.spark.sql.types.{ArrayType, DataType, DayTimeIntervalType, Decimal, DecimalType, Geography, Geometry, IntegerType, Metadata, SQLUserDefinedType, StringType, StructType, UserDefinedType, YearMonthIntervalType}
+import org.apache.spark.sql.types.{ArrayType, CharType, DataType, DayTimeIntervalType, Decimal, DecimalType, Geography, Geometry, IntegerType, Metadata, SQLUserDefinedType, StringType, StructType, UserDefinedType, VarcharType, YearMonthIntervalType}
import org.apache.spark.sql.util.CloseableIterator
import org.apache.spark.unsafe.types.VariantVal
import org.apache.spark.util.{MaybeNull, SparkStringUtils}
@@ -309,6 +309,28 @@ class ArrowEncoderSuite extends ConnectFunSuite {
}
}
+ test("SPARK-58794: char/varchar round trip") {
+ // The client cannot see the server's charVarchar configuration, so a result schema carrying
+ // CHAR/VARCHAR must be decodable regardless of the local one. Values are padded and length
+ // checked by the server, so the client passes them through unchanged.
+ val encoder = toResultRowEncoder(
+ new StructType()
+ .add("c", CharType(4))
+ .add("v", VarcharType(6))
+ .add("s", new StructType().add("c", CharType(4)))
+ .add("a", ArrayType(VarcharType(6))))
+ roundTripAndCheckIdentical(encoder) { () =>
+ val maybeNull = MaybeNull(7)
+ Iterator.tabulate(101) { i =>
+ Row(
+ maybeNull("ab "),
+ maybeNull("cd"),
+ maybeNull(Row("ef ")),
+ maybeNull(mutable.ArraySeq.make[String](Array("gh"))))
+ }
+ }
+ }
+
test("single batch") {
val inspector = new CountingBatchInspector
roundTripAndCheckIdentical(singleIntEncoder, inspectBatch = inspector) { () =>
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala
index 375d18514fbf..dd6e0ec47469 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala
@@ -101,7 +101,7 @@ private[sql] class SparkResult[T](
case UnboundRowEncoder =>
// Replace the row encoder with the encoder inferred from the schema.
RowEncoder
- .encoderFor(dataType.asInstanceOf[StructType])
+ .encoderForResultSchema(dataType.asInstanceOf[StructType])
.asInstanceOf[AgnosticEncoder[E]]
case ProductEncoder(clsTag, fields, outer) if ProductEncoder.isTuple(clsTag) =>
// Recursively continue updating the tuple product encoder
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
index fe0ecace05cc..4441d85b7629 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
@@ -40,7 +40,7 @@ import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders._
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.connect.common.types.ops.ConnectTypeOps
import org.apache.spark.sql.errors.{CompilationErrors, ExecutionErrors}
-import org.apache.spark.sql.types.Decimal
+import org.apache.spark.sql.types.{Decimal, StringHelper}
import org.apache.spark.sql.util.{CloseableIterator, ConcatenatingArrowStreamReader, MessageIterator}
import org.apache.spark.unsafe.types.VariantVal
@@ -135,8 +135,14 @@ object ArrowDeserializers {
new Deserializer[Any] {
def get(i: Int): Any = null
}
- case (StringEncoder, v: FieldVector) =>
- new LeafFieldDeserializer[String](encoder, v, timeZoneId) {
+ // CHAR/VARCHAR travel as plain Arrow string vectors; the length is part of the type, not of
+ // the encoding, and the values arrive already padded and length checked by the server. Read
+ // them against the unconstrained string type, since narrowing STRING to CHAR(n)/VARCHAR(n)
+ // is not an up-cast and the reader would reject the vector.
+ case (StringEncoder | _: CharEncoder | _: VarcharEncoder, v: FieldVector) =>
+ val stringReader =
+ ArrowVectorReader(StringHelper.plainStringType(encoder.dataType), v, timeZoneId)
+ new LeafFieldDeserializer[String](stringReader) {
override def value(i: Int): String = reader.getString(i)
}
case (JavaEnumEncoder(tag), v: FieldVector) =>
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala
index 02918e4eb140..10375fa32c87 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala
@@ -281,11 +281,11 @@ object ArrowSerializer {
new FieldSerializer[Unit, NullVector](v) {
override def set(index: Int, value: Unit): Unit = vector.setNull(index)
}
- case (StringEncoder, v: VarCharVector) =>
+ case (StringEncoder | _: CharEncoder | _: VarcharEncoder, v: VarCharVector) =>
new FieldSerializer[String, VarCharVector](v) {
override def set(index: Int, value: String): Unit = setString(v, index, value)
}
- case (StringEncoder, v: LargeVarCharVector) =>
+ case (StringEncoder | _: CharEncoder | _: VarcharEncoder, v: LargeVarCharVector) =>
new FieldSerializer[String, LargeVarCharVector](v) {
override def set(index: Int, value: String): Unit = setString(v, index, value)
}
diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
index b429086db22f..9dff17b80376 100644
--- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
+++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
@@ -126,16 +126,18 @@ private[hive] class SparkGetColumnsOperation(
}
/**
- * For boolean, numeric and datetime types, it returns the default size of its catalyst type
+ * For boolean, numeric and datetime types, this method returns the input type's default size.
+ * For CHAR(n) and VARCHAR(n), it returns the declared character length n.
* For struct type, when its elements are fixed-size, the summation of all element sizes will be
* returned.
- * For array, map, string, and binaries, the column size is variable, return null as unknown.
+ * For array, map, unbounded string, and binaries, the column size is variable; return null.
*/
private def getColumnSize(typ: DataType): Option[Int] = typ match {
case dt @ (BooleanType | _: NumericType | DateType | TimestampType | TimestampNTZType |
CalendarIntervalType | NullType | _: AnsiIntervalType) =>
Some(dt.defaultSize)
case c: CharType => Some(c.length)
+ case v: VarcharType => Some(v.length)
case StructType(fields) =>
val sizeArr = fields.map(f => getColumnSize(f.dataType))
if (sizeArr.contains(None)) {
@@ -146,6 +148,22 @@ private[hive] class SparkGetColumnsOperation(
case other => None
}
+ /**
+ * JDBC CHAR_OCTET_LENGTH is a byte capacity. Spark CHAR/VARCHAR lengths are in
+ * characters, so report 4 * n (UTF-8 maximum bytes per character), saturating at
+ * Int.MaxValue. Unbounded STRING and non-character types stay null (not applicable).
+ */
+ private def getCharOctetLength(typ: DataType): Option[Int] = typ match {
+ case c: CharType => Some(maxUtf8OctetLength(c.length))
+ case v: VarcharType => Some(maxUtf8OctetLength(v.length))
+ case _ => None
+ }
+
+ private def maxUtf8OctetLength(numChars: Int): Int = {
+ val maxChars = Int.MaxValue / 4
+ if (numChars > maxChars) Int.MaxValue else numChars * 4
+ }
+
/**
* The number of fractional digits for this type.
* Null is returned for data types where this is not applicable.
@@ -222,7 +240,7 @@ private[hive] class SparkGetColumnsOperation(
null, // COLUMN_DEF
null, // SQL_DATA_TYPE
null, // SQL_DATETIME_SUB
- null, // CHAR_OCTET_LENGTH
+ getCharOctetLength(column.dataType).map(_.asInstanceOf[AnyRef]).orNull,
ordinal.asInstanceOf[AnyRef], // ORDINAL_POSITION, 1-based
(if (column.nullable) "YES" else "NO"), // IS_NULLABLE
null, // SCOPE_CATALOG
diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
index b3194cfdef3f..af0e115b2492 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
@@ -316,6 +316,7 @@ class SparkMetadataOperationSuite extends HiveThriftServer2TestBase {
|using parquet""".stripMargin
withJdbcStatement(tableName) { statement =>
+ statement.execute(s"SET ${SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key}=true")
statement.execute(ddl)
val databaseMetaData = statement.getConnection.getMetaData
@@ -338,10 +339,21 @@ class SparkMetadataOperationSuite extends HiveThriftServer2TestBase {
val colSize = rowSet.getInt("COLUMN_SIZE")
schema(pos).dataType match {
- case StringType | BinaryType | _: ArrayType | _: MapType | _: VarcharType =>
+ case StringType | BinaryType | _: ArrayType | _: MapType =>
assert(colSize === 0)
+ case c: CharType => assert(colSize === c.length)
+ case v: VarcharType => assert(colSize === v.length)
case o => assert(colSize === o.defaultSize)
}
+ if (schema(pos).name == "c17") assert(colSize === 255)
+ if (schema(pos).name == "c18") assert(colSize === 1024)
+
+ val octetLength = rowSet.getInt("CHAR_OCTET_LENGTH")
+ schema(pos).dataType match {
+ case c: CharType => assert(octetLength === c.length * 4)
+ case v: VarcharType => assert(octetLength === v.length * 4)
+ case _ => assert(octetLength === 0) // JDBC getInt on SQL NULL
+ }
assert(rowSet.getInt("BUFFER_LENGTH") === 0) // not used
val decimalDigits = rowSet.getInt("DECIMAL_DIGITS")
@@ -372,6 +384,23 @@ class SparkMetadataOperationSuite extends HiveThriftServer2TestBase {
}
}
+ test("SPARK-58794: result metadata preserves CHAR and VARCHAR") {
+ withJdbcStatement() { statement =>
+ statement.execute(s"SET ${SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key}=true")
+ val resultSet = statement.executeQuery(
+ "SELECT CAST('ab' AS CHAR(4)) AS c, CAST('cd' AS VARCHAR(6)) AS v")
+ assert(resultSet.next())
+
+ val metadata = resultSet.getMetaData
+ assert(metadata.getColumnType(1) === java.sql.Types.CHAR)
+ assert(metadata.getColumnTypeName(1) === "char")
+ assert(metadata.getPrecision(1) === 4)
+ assert(metadata.getColumnType(2) === java.sql.Types.VARCHAR)
+ assert(metadata.getColumnTypeName(2) === "varchar")
+ assert(metadata.getPrecision(2) === 6)
+ }
+ }
+
test("get columns operation should handle interval column properly") {
val viewName = "view_interval"
val ddl = s"CREATE GLOBAL TEMP VIEW $viewName as select interval 1 day as i"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org