srielau commented on code in PR #58087:
URL: https://github.com/apache/spark/pull/58087#discussion_r3823793937
##########
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:
This STRING branch now restores CHAR/VARCHAR from `spark.sql.catalyst.type`.
The INT interval path already parses the property unchecked; for STRING it is
worth requiring a `StringType` subtype and failing with
`IncompatibleSchemaException` if the property is e.g. `int`. Otherwise
inference succeeds and the deserializer fails later with a worse message.
##########
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:
Leaf CHAR/VARCHAR are handled here, but `MapType(StringType, ...)` below
still uses the `StringType` singleton. A `MAP<CHAR(n), T>` (or VARCHAR key)
will miss this arm and hit `Unexpected type`.
That is the same subclass trap this PR is fixing. `AvroSerializer` has the
same `kt == StringType` guard. `toAvroTypeWithDefaults` is not dead: RocksDB
state encoding goes through it, so CHAR map keys in streaming state would fail
the same way.
Please match `StringType` subclasses for map keys (or reject them with an
explicit incompatible-schema error), and cover it in the round-trip test.
##########
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:
Matching CHAR/VARCHAR before `StringType` is correct. Emitting ORC `STRING`
plus `spark.sql.catalyst.type` is consistent with TimeType/interval stamping
and keeps length enforcement in Spark assignment rather than ORC `maxLength`.
`toCatalystSchema` already notes that native ORC char/varchar can round-trip
on read. Please document here why write stays on `Category.STRING`
(Hive-written CHAR vs Spark-written STRING+attr), so the next change does not
"fix" this into native CHAR and change truncation/padding relative to store
assignment.
##########
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:
VIEW covers CHAR only; CTAS covers both. The PR description also claims CTE
inheritance. A `WITH t AS (SELECT c, v FROM char_varchar_std) SELECT typeof(c),
typeof(v) FROM t` plus `CREATE VIEW ... AS SELECT v` would pin those without
another Scala test.
ORC here is catalog-backed; the file-only inference path lives only in
`CharVarcharTestSuite`.
##########
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:
This only exercises `toAvroType` / `toSqlType`. The PR also changes
`AvroSerializer` / `AvroDeserializer`, and write uses `AvroUtils.prepareWrite`
-> `toAvroType`.
Please add a real Avro write/read (file-only is enough) that asserts
inferred types and values, including CHAR padding. Nested struct is cheap
coverage of the recursive converter; map keys with CHAR would catch the
remaining `StringType` singleton match.
Also add a cross-flag case: write with `standardSemantics` on, read with it
off (and the reverse stamp check when first-class types are off).
--
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]
