srielau commented on code in PR #58087:
URL: https://github.com/apache/spark/pull/58087#discussion_r3823955176
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -369,6 +376,16 @@ object SchemaConverters extends Logging {
case FloatType => builder.floatType()
case DoubleType => builder.doubleType()
+ // CharType/VarcharType are not equal to the StringType singleton; stamp
the catalyst
+ // type so file-schema inference restores the length constraint.
+ case c: CharType =>
+ val stringSchema = builder.stringType()
+ stringSchema.addProp(CATALYST_TYPE_PROP_NAME, c.typeName)
+ stringSchema
+ case v: VarcharType =>
+ val stringSchema = builder.stringType()
+ stringSchema.addProp(CATALYST_TYPE_PROP_NAME, v.typeName)
+ stringSchema
Review Comment:
Fixed: Avro map keys now match any `StringType` subclass. CHAR/VARCHAR keys
are stamped on the map schema as `spark.sql.catalyst.mapKey.type` so
`toAvroType` / `toAvroTypeWithDefaults` / `toSqlType` round-trip them.
Serializer/deserializer use the same subclass match. Covered in
`BasicCharVarcharTestSuite` (schema conversion plus DataFileWriter ser/de; the
avro data source is not on the sql/core classpath).
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -114,7 +114,14 @@ object SchemaConverters extends Logging {
}
SchemaType(catalystType, nullable = false)
}
- case STRING => SchemaType(StringType, nullable = false)
+ case STRING =>
+ val catalystTypeAttrValue = avroSchema.getProp(CATALYST_TYPE_PROP_NAME)
+ val catalystType = if (catalystTypeAttrValue == null) {
+ StringType
+ } else {
+ CatalystSqlParser.parseDataType(catalystTypeAttrValue)
+ }
+ SchemaType(catalystType, nullable = false)
Review Comment:
Fixed: STRING `spark.sql.catalyst.type` is parsed and required to be a
`StringType` subtype, otherwise `IncompatibleSchemaException`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:
##########
@@ -469,6 +469,16 @@ object OrcUtils extends Logging {
val typeDesc = new TypeDescription(ops.orcCategory)
typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, dt.typeName)
Some(typeDesc)
+ // CharType/VarcharType extend StringType; match them first so the
catalyst attribute
+ // records char(n)/varchar(n) instead of plain string.
+ case c: CharType =>
+ val typeDesc = new TypeDescription(TypeDescription.Category.STRING)
+ typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, c.typeName)
+ Some(typeDesc)
+ case v: VarcharType =>
+ val typeDesc = new TypeDescription(TypeDescription.Category.STRING)
+ typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, v.typeName)
Review Comment:
Documented next to the match: write stays on ORC `STRING` plus the catalyst
attribute so native ORC maxLength does not fight Spark store assignment.
Hive-written native CHAR still round-trips on read via `toCatalystSchema`.
Stamping now goes through `CharVarcharUtils.charVarcharTypeName`.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -1341,6 +1396,295 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
}
}
+ test("SPARK-58794: language surfaces keep CHAR/VARCHAR under
standardSemantics") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // CTAS / CREATE VIEW inherit projected CHAR/VARCHAR (D13).
+ withTable("std_src", "std_ctas") {
+ sql("CREATE TABLE std_src (c CHAR(5), v VARCHAR(5)) USING parquet")
+ sql("INSERT INTO std_src VALUES ('ab', 'ab')")
+ sql("CREATE TABLE std_ctas USING parquet AS SELECT c, v FROM std_src")
+ assert(spark.table("std_ctas").schema.map(_.dataType) ===
+ Seq(CharType(5), VarcharType(5)))
+ checkAnswer(
+ sql("SELECT concat('<', c, '>'), concat('<', v, '>') FROM std_ctas"),
+ Row("<ab >", "<ab>"))
+ }
+ withTable("std_view_src") {
+ withView("std_cv_view") {
+ sql("CREATE TABLE std_view_src (c CHAR(4)) USING parquet")
+ sql("INSERT INTO std_view_src VALUES ('xy')")
+ sql("CREATE VIEW std_cv_view AS SELECT c FROM std_view_src")
+ assert(spark.table("std_cv_view").schema.head.dataType ===
CharType(4))
+ checkAnswer(
+ sql("SELECT concat('<', c, '>') FROM std_cv_view"), Row("<xy >"))
+ }
+ }
+
+ // ALTER COLUMN equal-length CHAR/VARCHAR remains supported with
first-class types.
+ // VARCHAR widen / CHAR->VARCHAR are allowed by CheckAnalysis on V2
tables
+ // (see DSV2CharVarcharDDLTestSuite); V1 file-source ALTER only evolves
collation
+ // (same StringConstraint), so length changes stay rejected there.
+ withTable("std_alter") {
+ sql("CREATE TABLE std_alter (c CHAR(4), v VARCHAR(4)) USING parquet")
+ sql("ALTER TABLE std_alter CHANGE COLUMN c TYPE CHAR(4)")
+ sql("ALTER TABLE std_alter CHANGE COLUMN v TYPE VARCHAR(4)")
+ assert(spark.table("std_alter").schema.map(_.dataType) ===
+ Seq(CharType(4), VarcharType(4)))
+ intercept[AnalysisException] {
+ sql("ALTER TABLE std_alter CHANGE COLUMN c TYPE CHAR(5)")
+ }
+ intercept[AnalysisException] {
+ sql("ALTER TABLE std_alter CHANGE COLUMN v TYPE VARCHAR(5)")
+ }
+ }
+
+ // Session variables: DECLARE / SET keep the type and apply CAST
assignment.
+ sql("DECLARE OR REPLACE VARIABLE std_char_var CHAR(4)")
+ sql("DECLARE OR REPLACE VARIABLE std_varchar_var VARCHAR(4)")
+ try {
+ sql("SET VARIABLE std_char_var = 'ab'")
+ val charVarDf = sql("SELECT std_char_var AS c")
+ assert(charVarDf.schema.head.dataType === CharType(4))
+ checkAnswer(sql("SELECT concat('<', std_char_var, '>')"), Row("<ab
>"))
+ // Oversize by trailing blanks only is trimmed to fit CHAR(n).
+ sql("SET VARIABLE std_char_var = 'abcd '")
+ checkAnswer(sql("SELECT concat('<', std_char_var, '>')"),
Row("<abcd>"))
+ intercept[SparkRuntimeException] {
+ sql("SET VARIABLE std_char_var = 'abcde'").collect()
+ }
+
+ sql("SET VARIABLE std_varchar_var = 'ab'")
+ val varcharVarDf = sql("SELECT std_varchar_var AS v")
+ assert(varcharVarDf.schema.head.dataType === VarcharType(4))
+ checkAnswer(sql("SELECT concat('<', std_varchar_var, '>')"),
Row("<ab>"))
+ // Oversize by trailing blanks only is trimmed to fit.
+ sql("SET VARIABLE std_varchar_var = 'abcd '")
+ checkAnswer(sql("SELECT std_varchar_var"), Row("abcd"))
+ intercept[SparkRuntimeException] {
+ sql("SET VARIABLE std_varchar_var = 'abcde'").collect()
+ }
+ } finally {
+ sql("DROP TEMPORARY VARIABLE IF EXISTS std_char_var")
+ sql("DROP TEMPORARY VARIABLE IF EXISTS std_varchar_var")
+ }
+
+ // SQL scripting local variables keep CHAR/VARCHAR inside a compound
statement.
+ val localVarScript =
+ """
+ |BEGIN
+ | DECLARE c CHAR(4);
+ | DECLARE v VARCHAR(4);
+ | SET c = 'ab';
+ | SET v = 'cd';
+ | SELECT typeof(c), concat('<', c, '>'), typeof(v), concat('<', v,
'>');
+ |END
+ |""".stripMargin
+ val localVarDf = sql(localVarScript)
+ assert(localVarDf.schema.map(_.dataType) ===
+ Seq(StringType, StringType, StringType, StringType))
+ checkAnswer(localVarDf, Row("char(4)", "<ab >", "varchar(4)", "<cd>"))
+ // Trailing-blank trim on local SET into CHAR/VARCHAR.
+ checkAnswer(
+ sql(
+ """
+ |BEGIN
+ | DECLARE c CHAR(4);
+ | DECLARE v VARCHAR(4);
+ | SET c = 'abcd ';
+ | SET v = 'abcd ';
+ | SELECT concat('<', c, '>'), v;
+ |END
+ |""".stripMargin),
+ Row("<abcd>", "abcd"))
+ intercept[SparkRuntimeException] {
+ sql(
+ """
+ |BEGIN
+ | DECLARE c CHAR(4);
+ | SET c = 'abcde';
+ |END
+ |""".stripMargin).collect()
+ }
+ intercept[SparkRuntimeException] {
+ sql(
+ """
+ |BEGIN
+ | DECLARE v VARCHAR(4);
+ | SET v = 'abcde';
+ |END
+ |""".stripMargin).collect()
+ }
+
+ // Cursor FETCH INTO CHAR/VARCHAR locals applies store assignment (pad /
length).
+ withSQLConf(SQLConf.SQL_SCRIPTING_CURSOR_ENABLED.key -> "true") {
+ val cursorScript =
+ """
+ |BEGIN
+ | DECLARE fetched_c CHAR(4);
+ | DECLARE fetched_v VARCHAR(4);
+ | DECLARE cur CURSOR FOR
+ | SELECT cast('ab' AS CHAR(4)) AS c, cast('cd' AS VARCHAR(4))
AS v;
+ | OPEN cur;
+ | FETCH cur INTO fetched_c, fetched_v;
+ | SELECT typeof(fetched_c), concat('<', fetched_c, '>'),
+ | typeof(fetched_v), concat('<', fetched_v, '>');
+ | CLOSE cur;
+ |END
+ |""".stripMargin
+ checkAnswer(
+ sql(cursorScript),
+ Row("char(4)", "<ab >", "varchar(4)", "<cd>"))
+
+ // FETCH plain STRING into CHAR pads via assignment cast.
+ checkAnswer(
+ sql(
+ """
+ |BEGIN
+ | DECLARE fetched CHAR(4);
+ | DECLARE cur CURSOR FOR SELECT 'ab' AS c;
+ | OPEN cur;
+ | FETCH cur INTO fetched;
+ | SELECT typeof(fetched), concat('<', fetched, '>');
+ | CLOSE cur;
+ |END
+ |""".stripMargin),
+ Row("char(4)", "<ab >"))
+
+ // FETCH into a wider CHAR pads; trailing blanks trim into a shorter
target.
+ checkAnswer(
+ sql(
+ """
+ |BEGIN
+ | DECLARE fetched CHAR(5);
+ | DECLARE cur CURSOR FOR SELECT cast('xy' AS CHAR(2)) AS c;
+ | OPEN cur;
+ | FETCH cur INTO fetched;
+ | SELECT concat('<', fetched, '>');
+ | CLOSE cur;
+ |END
+ |""".stripMargin),
+ Row("<xy >"))
+ checkAnswer(
+ sql(
+ """
+ |BEGIN
+ | DECLARE fetched_c CHAR(4);
+ | DECLARE fetched_v VARCHAR(4);
+ | DECLARE cur CURSOR FOR
+ | SELECT cast('abcd ' AS CHAR(5)) AS c, cast('abcd ' AS
VARCHAR(5)) AS v;
+ | OPEN cur;
+ | FETCH cur INTO fetched_c, fetched_v;
+ | SELECT concat('<', fetched_c, '>'), fetched_v;
+ | CLOSE cur;
+ |END
+ |""".stripMargin),
+ Row("<abcd>", "abcd"))
+ intercept[SparkRuntimeException] {
+ sql(
+ """
+ |BEGIN
+ | DECLARE fetched VARCHAR(2);
+ | DECLARE cur CURSOR FOR SELECT cast('abcd' AS VARCHAR(4)) AS v;
+ | OPEN cur;
+ | FETCH cur INTO fetched;
+ | CLOSE cur;
+ |END
+ |""".stripMargin).collect()
+ }
+ }
+
+ // SQL FUNCTION params/RETURNS apply store assignment (pad, blank-trim,
overflow).
+ sql("CREATE OR REPLACE TEMPORARY FUNCTION std_char_fn() RETURNS CHAR(3)
RETURN 'a'")
+ sql(
+ """CREATE OR REPLACE TEMPORARY FUNCTION std_varchar_ret()
+ |RETURNS VARCHAR(3) RETURN 'ab'""".stripMargin)
+ sql(
+ """CREATE OR REPLACE TEMPORARY FUNCTION std_char_param(x CHAR(3))
+ |RETURNS CHAR(3) RETURN x""".stripMargin)
+ sql(
+ """CREATE OR REPLACE TEMPORARY FUNCTION std_varchar_param(x VARCHAR(3))
+ |RETURNS VARCHAR(3) RETURN x""".stripMargin)
+ try {
+ val fnDf = sql("SELECT std_char_fn() AS c")
+ assert(fnDf.schema.head.dataType === CharType(3))
+ checkAnswer(sql("SELECT concat('<', std_char_fn(), '>')"), Row("<a
>"))
+ val retVarcharDf = sql("SELECT std_varchar_ret() AS v")
+ assert(retVarcharDf.schema.head.dataType === VarcharType(3))
+ checkAnswer(retVarcharDf, Row("ab"))
+
+ // STRING -> CHAR(n) param: pad; trailing blanks trim; non-blank
overflow errors.
+ val charParamDf = sql("SELECT std_char_param('a') AS c")
+ assert(charParamDf.schema.head.dataType === CharType(3))
+ checkAnswer(sql("SELECT concat('<', std_char_param('a'), '>')"),
Row("<a >"))
+ checkAnswer(
+ sql("SELECT concat('<', std_char_param('abc '), '>')"),
+ Row("<abc>"))
+ intercept[SparkRuntimeException] {
+ sql("SELECT std_char_param('abcd')").collect()
+ }
+
+ val paramDf = sql("SELECT std_varchar_param('ab') AS v")
+ assert(paramDf.schema.head.dataType === VarcharType(3))
+ checkAnswer(paramDf, Row("ab"))
+ checkAnswer(sql("SELECT std_varchar_param('abc ')"), Row("abc"))
+ intercept[SparkRuntimeException] {
+ sql("SELECT std_varchar_param('abcd')").collect()
+ }
+ } finally {
+ sql("DROP TEMPORARY FUNCTION IF EXISTS std_char_fn")
+ sql("DROP TEMPORARY FUNCTION IF EXISTS std_varchar_ret")
+ sql("DROP TEMPORARY FUNCTION IF EXISTS std_char_param")
+ sql("DROP TEMPORARY FUNCTION IF EXISTS std_varchar_param")
+ }
+
+ // ORC catalog tables stamp the catalyst type so typeof survives
write/read.
+ withTable("std_orc") {
+ sql("CREATE TABLE std_orc (c CHAR(5), v VARCHAR(5)) USING orc")
+ sql("INSERT INTO std_orc VALUES ('ab', 'cd')")
+ assert(spark.table("std_orc").schema.map(_.dataType) ===
+ Seq(CharType(5), VarcharType(5)))
+ checkAnswer(
+ sql("SELECT concat('<', c, '>'), concat('<', v, '>') FROM std_orc"),
+ Row("<ab >", "<cd>"))
+ }
+
+ // File-only ORC inference recovers the catalyst type stamped on write.
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(1).selectExpr("cast('ab' AS CHAR(4)) AS c")
+ .write.mode("overwrite").orc(path)
+ val orcDf = spark.read.orc(path)
+ assert(orcDf.schema.head.dataType === CharType(4))
+ checkAnswer(orcDf.selectExpr("concat('<', c, '>')"), Row("<ab >"))
+ }
+
+ // Avro schema conversion round-trips CHAR/VARCHAR via the catalyst type
property.
+ // (Full Avro data-source E2E lives in the avro module.)
+ Seq(CharType(5), VarcharType(7)).foreach { dt =>
+ val avro = org.apache.spark.sql.avro.SchemaConverters.toAvroType(dt,
nullable = false)
+ val back =
org.apache.spark.sql.avro.SchemaConverters.toSqlType(avro).dataType
+ assert(back === dt, s"Avro round-trip lost $dt, got $back")
+ }
Review Comment:
Added nested struct + CHAR map-key conversion, invalid stamp, Avro
serializer/deserializer file round-trip, and ORC cross-flag cases
(`standardSemantics` on then off, first-class types off,
`preserveCharVarcharTypeInfo` only). Spark's avro data source is
connector-only, so the Avro file path uses `DataFileWriter` against the
sql/core ser/de.
##########
sql/core/src/test/resources/sql-tests/inputs/charvarchar-standard-semantics.sql:
##########
@@ -193,4 +217,18 @@ INSERT INTO char_varchar_std VALUES ('ab', 'ab');
SELECT typeof(c), typeof(v) FROM char_varchar_std;
SELECT concat('[', c, ']'), concat('[', v, ']') FROM char_varchar_std;
SELECT length(c), length(v) FROM char_varchar_std;
+
+-- Language surfaces: CTAS / VIEW inherit CHAR/VARCHAR; ORC catalog round-trip
+CREATE TABLE char_varchar_std_ctas USING parquet AS SELECT c, v FROM
char_varchar_std;
+SELECT typeof(c), typeof(v) FROM char_varchar_std_ctas;
+CREATE VIEW char_varchar_std_view AS SELECT c FROM char_varchar_std;
Review Comment:
Added `CREATE VIEW ... AS SELECT v` and `WITH t AS (SELECT c, v ...) SELECT
typeof(c), typeof(v)` to `charvarchar-standard-semantics.sql`, plus the same
CTE/VARCHAR view coverage in the Scala suite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
