This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new a78be288c126 fix(spark): check data schema in Spark 4.0 guard (#20158)
a78be288c126 is described below
commit a78be288c126ae02b6af91b2ac8ed1eca3771a0c
Author: voonhous <[email protected]>
AuthorDate: Thu Oct 1 12:25:10 2026 +0800
fix(spark): check data schema in Spark 4.0 guard (#20158)
Spark 4.0 with spark.sql.variant.pushVariantIntoScan on is unsupported
and the read is meant to fail with an error that names the conf
(#20032, #20033). The guard looked at the required schema of the scan
only. PushVariantIntoScan rewrites every variant column of the
relation, into a placeholder struct<0:boolean> when the query never
reads it, so a query that selects no variant slipped past the guard
and died converting the data schema to a HoodieSchema:
HoodieSchemaException "Illegal initial character: 0".
Validate the data schema instead, which carries the rewrite for every
scan of the table. The Spark 4.0 test expected that query to return
rows, which it never did; it now expects the guard.
CI builds no Spark 4.0 profile. Run locally on Spark 4.0.2: the test
fails before the change and passes with it.
---
.../scala/org/apache/spark/sql/hudi/SparkAdapter.scala | 18 ++++++++++--------
.../parquet/HoodieFileGroupReaderBasedFileFormat.scala | 6 ++++--
.../sql/hudi/dml/schema/TestVariantDataType.scala | 9 ++++++---
.../org/apache/spark/sql/adapter/Spark4_0Adapter.scala | 4 ++--
4 files changed, 22 insertions(+), 15 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
index 2b9725b52049..6949c6fbe11a 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
@@ -528,14 +528,16 @@ trait SparkAdapter extends Serializable {
}
/**
- * Fails the read when `requiredSchema` carries a variant projection struct
this Spark version
- * cannot evaluate. Spark 4.0 rewrites a variant column exactly as 4.1 does
once the conf is on,
- * but its readers do not evaluate the projection and Hudi does not align
log records to it, so
- * a read there would fall through to the schema-change path instead of
failing. The Spark 4.0
- * adapter overrides this to throw; every other version reads the shape
(4.1+) or never sees it
- * (3.x). See https://github.com/apache/hudi/issues/20032.
- */
- def validateVariantProjectionReadable(requiredSchema: StructType): Unit = ()
+ * Fails the read when `schema`, the data schema of a scan, carries a
variant projection struct
+ * this Spark version cannot evaluate. Spark 4.0 rewrites a variant column
exactly as 4.1 does
+ * once the conf is on, but its readers do not evaluate the projection and
Hudi does not align
+ * log records to it, so a read there would fall through to the
schema-change path instead of
+ * failing. The data schema rather than the required one: the rule rewrites
every variant column
+ * of the relation, requested or not. The Spark 4.0 adapter overrides this
to throw; every other
+ * version reads the shape (4.1+) or never sees it (3.x). See
+ * https://github.com/apache/hudi/issues/20032.
+ */
+ def validateVariantProjectionReadable(schema: StructType): Unit = ()
/**
* If `sparkRequiredSchema` contains any Spark 4.1 variant projection struct
(i.e., the
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index 823fda9001f6..241bdff63178 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -292,8 +292,10 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
options: Map[String, String],
hadoopConf: Configuration):
PartitionedFile => Iterator[InternalRow] = {
// Driver side, once per scan: Spark 4.0 cannot read a PushVariantIntoScan
projection struct and
- // has to fail here rather than in the schema-change path (#20032).
- sparkAdapter.validateVariantProjectionReadable(requiredSchema)
+ // has to fail here rather than in the schema-change path (#20032).
Checked on the data schema:
+ // the rule rewrites every variant column of the relation, into a
placeholder struct when the
+ // query never reads it, so a scan that requires no variant still carries
the rewrite there.
+ sparkAdapter.validateVariantProjectionReadable(dataStructType)
val outputSchema = StructType(requiredSchema.fields ++
partitionSchema.fields)
val isCount = requiredSchema.isEmpty && !isMOR && !isIncremental
// Spark planner only adds the user-provided predicates (from `WHERE`
clause or `.filter()`)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index c0b8ad06c3f9..6b0c51243347 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
@@ -258,13 +258,16 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
with VariantShreddingTe
withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "true") {
// An extraction and a whole-variant read are both rewritten into a
projection struct.
+ // So is a query that never reads v: the rule rewrites every variant
column of the
+ // relation, and one no expression asks for becomes a placeholder
struct in the data
+ // schema of the scan. Without the guard that read dies on the
placeholder's field name
+ // ("Illegal initial character: 0") instead of naming the conf.
Seq(s"select id, variant_get(v, '$$.key', 'string') from $tableName",
- s"select id, cast(v as string) from $tableName").foreach { sql =>
+ s"select id, cast(v as string) from $tableName",
+ s"select id, ts from $tableName").foreach { sql =>
checkNestedExceptionContains(() => spark.sql(sql).collect())(
"spark.sql.variant.pushVariantIntoScan")
}
- // A query that does not touch the variant column is not rewritten
and still reads.
- checkAnswer(s"select id, ts from $tableName")(Seq(1, 1000))
}
// Back on the default the same reads work.
checkAnswer(s"select id, cast(v as string) from $tableName")(Seq(1,
"{\"key\":\"v2\"}"))
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
index 14031abf21d6..09f89fcbd685 100644
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
@@ -123,8 +123,8 @@ class Spark4_0Adapter extends BaseSpark4Adapter {
// spark.sql.variant.pushVariantIntoScan is on (off by default there), but
its readers do not
// evaluate the projection and Hudi does not align log records to it, so
fail here instead of
// falling through to the schema-change path (#20032).
- override def validateVariantProjectionReadable(requiredSchema: StructType):
Unit = {
- requiredSchema.fields.find(f =>
containsVariantProjection(f.dataType)).foreach { field =>
+ override def validateVariantProjectionReadable(schema: StructType): Unit = {
+ schema.fields.find(f => containsVariantProjection(f.dataType)).foreach {
field =>
throw new HoodieNotSupportedException(
s"Column '${field.name}' was rewritten by
spark.sql.variant.pushVariantIntoScan into a variant " +
s"projection struct, which Hudi does not support on Spark
${HoodieSparkUtils.getSparkVersion}. " +
