diff --git a/python/pyspark/sql/tests/connect/test_connect_basic.py b/python/pyspark/sql/tests/connect/test_connect_basic.py index b8d258267d267..80e5cdfc57e38 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 @@ IntegerType, MapType, ArrayType, + CharType, + VarcharType, Row, ) from pyspark.testing import assertDataFrameEqual @@ -487,6 +489,19 @@ def test_schema(self): ) 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 51157b5344905..7627998f5a6df 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 b7a052d5fc4eb..2a2889e441c77 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 823984f666bcd..b49894e92809b 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 48a42cd9ec9cb..45bb8fcb43d82 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 17f3d77b77c5d..de5099714b87e 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 4ad8ab91d4b2f..59507c816b306 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 efb6a7a2304f0..9963ec5620ff2 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, PrimitiveFloatEncoder, PrimitiveIntEncoder, PrimitiveLongEncoder, PrimitiveShortEncoder, RowEncoder, ScalaDecimalEncoder, StringEncoder, TimestampEncoder, TransformingEncoder, UDTEncoder, YearMonthIntervalEncoder} -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 375d18514fbf3..dd6e0ec474697 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 fe0ecace05cc0..4441d85b76292 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 02918e4eb1403..10375fa32c875 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 b429086db22f2..9dff17b803764 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 b3194cfdef3f6..af0e115b24929 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"