[SPARK-58806][SQL][CONNECT] Expose CHAR/VARCHAR metadata and decode result schemas

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()
 

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)

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(

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 = "'"),

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:

sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala

@@ -886,30 +886,35 @@ 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)))
 
         // 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}")

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(