Skip to content

[SPARK-58814][SQL] Cover CHAR/VARCHAR round-trips and stop ORC truncating scans #58317

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
srielau wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[SPARK-58814][SQL] Cover CHAR/VARCHAR round-trips and stop ORC truncating scans #58317

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0071ff1
[SPARK-58814][SQL] Cover CHAR/VARCHAR round-trips and stop ORC trunca…
srielau Aug 26, 2026
46e1980
[SPARK-58814][SQL] Fix ORC CHAR/VARCHAR row decoding and legacy enfor…
srielau Aug 29, 2026
7a02fee
[SPARK-58814][SQL] Bind ORC scans to analyzed CHAR/VARCHAR semantics
srielau Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 77 additions & 1 deletion connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ import org.apache.spark.sql.TestingUDT.IntervalData
import org.apache.spark.sql.avro.AvroCompressionCodec._
import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.catalyst.plans.logical.Filter
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeTestUtils}
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC}
import org.apache.spark.sql.execution.{FormattedMode, SparkPlan}
import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition}
Expand Down Expand Up @@ -3724,6 +3724,82 @@ abstract class AvroSuite
}
}

test("SPARK-58814: Avro infers nested CHAR/VARCHAR schema and values") {
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
withTempPath { dir =>
val path = dir.getCanonicalPath
val input = spark.range(1).selectExpr(
"cast('ab' AS CHAR(4)) AS c",
"cast('xy' AS VARCHAR(3)) AS v",
"named_struct('c', cast('z' AS CHAR(2))) AS s",
"array(cast('q' AS VARCHAR(2))) AS a",
"map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
input.write.mode("overwrite").format("avro").save(path)

val readBack = spark.read.format("avro").load(path)
assert(DataType.equalsIgnoreNullability(readBack.schema, input.schema))
checkAnswer(
readBack.selectExpr("concat('<', c, '>')", "v", "concat('<', s.c, '>')"),

@cloud-fan cloud-fan Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Non-blocking (P2): This case creates the array and map, but the projection only materializes c, v, and s.c. Column pruning can skip decoding both collection fields, so a regression in array/map conversion or CHAR-key padding would still pass. Please select a and m as well and assert Seq("q") and Map("k " -> "v").

@srielau srielau Aug 31, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in 7a02fee. The Avro round-trip projection now materializes and asserts both collection fields: Seq("q") for the array and Map("k " -> "v") for the CHAR-keyed map. The targeted Avro V1 and V2 tests pass.

Row("<ab >", "xy", "<z >"))

withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
assert(DataType.equalsIgnoreNullability(
spark.read.format("avro").load(path).schema,
CharVarcharUtils.replaceCharVarcharWithString(input.schema)))
}
withSQLConf(
SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
assert(DataType.equalsIgnoreNullability(
spark.read.format("avro").load(path).schema,
input.schema))
}
}

withTempPath { dir =>
Seq("ab").toDF("c").write.format("avro").save(dir.getCanonicalPath)
val charDf = spark.read.schema("c CHAR(4)").format("avro").load(dir.getCanonicalPath)
checkAnswer(
charDf.selectExpr("concat('<', c, '>')"),
Row("<ab >"))
}
withTempPath { dir =>
Seq("abcdef").toDF("c").write.format("avro").save(dir.getCanonicalPath)
Seq("CHAR", "VARCHAR").foreach { typ =>
checkError(
exception = intercept[SparkRuntimeException] {
spark.read.schema(s"c $typ(4)").format("avro")
.load(dir.getCanonicalPath).collect()
},
condition = "EXCEED_LIMIT_LENGTH",
parameters = Map("limit" -> "4"))
}
}

withTable("avro_char_varchar_assignment") {
sql(
"""CREATE TABLE avro_char_varchar_assignment
|(c CHAR(4), v VARCHAR(4)) USING avro""".stripMargin)
sql("INSERT INTO avro_char_varchar_assignment VALUES ('ab', 'xy')")
assert(spark.table("avro_char_varchar_assignment").schema.map(_.dataType) ===
Seq(CharType(4), VarcharType(4)))
checkAnswer(
sql(
"""SELECT concat('<', c, '>'), v
|FROM avro_char_varchar_assignment""".stripMargin),
Row("<ab >", "xy"))
checkError(
exception = intercept[SparkRuntimeException] {
sql(
"""INSERT INTO avro_char_varchar_assignment
|VALUES ('abcde', 'xy')""".stripMargin).collect()
},
condition = "EXCEED_LIMIT_LENGTH",
parameters = Map("limit" -> "4"))
}
}
}

}

class AvroV1Suite extends AvroSuite {
Expand Down
3 changes: 3 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -437,6 +437,9 @@ object OrcUtils extends Logging {
s"array<${getOrcSchemaString(a.elementType)}>"
case m: MapType =>
s"map<${getOrcSchemaString(m.keyType)},${getOrcSchemaString(m.valueType)}>"
// Keep Spark responsible for CHAR/VARCHAR assignment and scan checks. Native ORC
// CHAR/VARCHAR would truncate or pad before Spark can validate the original value.
case _: CharType | _: VarcharType => StringType.catalogString
Comment thread
srielau marked this conversation as resolved.
Outdated
case _: DayTimeIntervalType | _: TimestampNTZType => LongType.catalogString
case _: YearMonthIntervalType => IntegerType.catalogString
// Framework types (TimeType, nanosecond timestamps) supply their own ORC schema string.
Expand Down
125 changes: 84 additions & 41 deletions sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1697,33 +1697,94 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
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>"))
test("SPARK-58814: major formats preserve CHAR/VARCHAR schemas and values") {
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
Seq("parquet", "orc").foreach { format =>
Seq("v1" -> format, "v2" -> "").foreach { case (sourceVersion, useV1List) =>
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
withTempPath { dir =>
val path = dir.getCanonicalPath
val input = spark.range(1).selectExpr(
"cast('ab' AS CHAR(4)) AS c",
"cast('xy' AS VARCHAR(3)) AS v",
"named_struct('c', cast('z' AS CHAR(2))) AS s",
"array(cast('q' AS VARCHAR(2))) AS a",
"map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
input.write.mode("overwrite").format(format).save(path)

val readBack = spark.read.format(format).load(path)
assert(DataType.equalsIgnoreNullability(readBack.schema, input.schema),
s"$format $sourceVersion lost CHAR/VARCHAR schema")
checkAnswer(
readBack.selectExpr("concat('<', c, '>')", "v", "concat('<', s.c, '>')"),
Comment thread
srielau marked this conversation as resolved.
Outdated
Row("<ab >", "xy", "<z >"))

withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
val readOff = spark.read.format(format).load(path)
assert(DataType.equalsIgnoreNullability(
readOff.schema,
CharVarcharUtils.replaceCharVarcharWithString(input.schema)))
}
withSQLConf(
SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
assert(DataType.equalsIgnoreNullability(
spark.read.format(format).load(path).schema,
input.schema))
}
}
}
}
}

// 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 >"))
// Reading with first-class types off replaces CHAR with STRING even if the
// file was stamped under standardSemantics.
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
val readOff = spark.read.orc(path)
assert(readOff.schema.head.dataType === StringType)
Seq("parquet", "orc", "csv").foreach { format =>
Seq("v1" -> format, "v2" -> "").foreach { case (sourceVersion, useV1List) =>
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
withTempPath { dir =>
Seq("ab").toDF("c").write.format(format).save(dir.getCanonicalPath)
val charDf = spark.read.schema("c CHAR(4)").format(format)
.load(dir.getCanonicalPath)
checkAnswer(charDf.selectExpr("concat('<', c, '>')"), Row("<ab >"))
}
withTempPath { dir =>
Seq("abcdef").toDF("c").write.format(format).save(dir.getCanonicalPath)
Seq("CHAR", "VARCHAR").foreach { typ =>
withClue(s"$format $sourceVersion $typ: ") {
val readDf = spark.read.schema(s"c $typ(4)").format(format)
.load(dir.getCanonicalPath)
checkError(
exception = intercept[SparkRuntimeException] {
readDf.collect()
},
condition = "EXCEED_LIMIT_LENGTH",
parameters = Map("limit" -> "4"))
}
}
}

val table = s"std_${format}_${sourceVersion}_assignment"
withTable(table) {
sql(s"CREATE TABLE $table (c CHAR(4), v VARCHAR(4)) USING $format")
sql(s"INSERT INTO $table VALUES ('ab', 'xy')")
assert(spark.table(table).schema.map(_.dataType) ===
Seq(CharType(4), VarcharType(4)))
checkAnswer(
sql(s"SELECT concat('<', c, '>'), v FROM $table"),
Row("<ab >", "xy"))
checkError(
exception = intercept[SparkRuntimeException] {
sql(s"INSERT INTO $table VALUES ('abcde', 'xy')").collect()
},
condition = "EXCEED_LIMIT_LENGTH",
parameters = Map("limit" -> "4"))
}
}
}
}

// First-class types off: CAST CHAR is STRING before the writer, so ORC does not stamp CHAR.
withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
withTempPath { dir =>
Expand All @@ -1733,17 +1794,6 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
assert(spark.read.orc(path).schema.head.dataType === StringType)
}
}
// preserveCharVarcharTypeInfo also keeps first-class types, so write still stamps CHAR.
withSQLConf(
SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
withTempPath { dir =>
val path = dir.getCanonicalPath
spark.range(1).selectExpr("cast('ab' AS CHAR(4)) AS c")
.write.mode("overwrite").orc(path)
assert(spark.read.orc(path).schema.head.dataType === CharType(4))
}
}
// ORC stamps collated unbounded STRING as plain "string"; the inferred type is the
// same as Avro, which omits the catalyst property on unbounded STRING.
withTempPath { dir =>
Expand Down Expand Up @@ -1825,21 +1875,14 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
}
}

// JSON / CSV keep a user-specified CHAR/VARCHAR schema under the flag.
// JSON has no embedded schema, so a user-specified schema supplies the logical type.
withTempPath { dir =>
val path = dir.getCanonicalPath
spark.range(1).selectExpr("cast(id AS STRING) AS c").write.mode("overwrite")
.json(s"$path/json")
val jsonDf = spark.read.schema("c CHAR(5)").json(s"$path/json")
assert(jsonDf.schema.head.dataType === CharType(5))
checkAnswer(jsonDf.selectExpr("concat('<', c, '>')"), Row("<0 >"))

spark.range(1).selectExpr("cast(id AS STRING) AS c").write.mode("overwrite")
.option("header", "true").csv(s"$path/csv")
val csvDf = spark.read.schema("c VARCHAR(5)").option("header", "true")
.csv(s"$path/csv")
assert(csvDf.schema.head.dataType === VarcharType(5))
checkAnswer(csvDf, Row("0"))
}
}
}
Expand Down