This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch branch-1.2.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-1.2.x by this push:
new 1768edc35bc5 fix(variant): port #20136 to branch-1.2.x (#20184)
1768edc35bc5 is described below
commit 1768edc35bc565d40286ed69c336068b677003f0
Author: voonhous <[email protected]>
AuthorDate: Fri Oct 2 18:15:29 2026 +0800
fix(variant): port #20136 to branch-1.2.x (#20184)
On Spark 4.1, where spark.sql.variant.pushVariantIntoScan is on by
default, a variant column added by ALTER TABLE ADD COLUMNS without
schema-on-read cannot be read from files written before the DDL:
the pushed-down read fails with INVALID_VARIANT_SHREDDING_SCHEMA on
COW and MOR, base files and log blocks alike. Hudi's readers turn
nested schema pruning off, so the missing column stayed in the
parquet read schema as a plain group, over which Spark builds its
variant converter.
Port of master 827644e8699b. HoodieParquetReadSupport and its unit
test apply as they are. Master's end-to-end test runs over shredded
files, which this branch cannot write, so TestVariantPushVariantIntoScan
gets an unshredded version on its table type x rule x record type
sweep: reads over the old files only, after an insert that carries
v2, after an update of one row per file group (the COW merge handle;
on MOR beside a log block written before the DDL) and, on MOR, after
compaction. The rule-on legs and the unit pin fail without the fix.
---
.../parquet/HoodieParquetReadSupport.scala | 17 +++-
.../parquet/TestHoodieParquetReadSupport.scala | 36 +++++++
.../schema/TestVariantPushVariantIntoScan.scala | 108 +++++++++++++++++++++
3 files changed, 159 insertions(+), 2 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieParquetReadSupport.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieParquetReadSupport.scala
index c63ec0eef9e4..a164d09838a9 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieParquetReadSupport.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieParquetReadSupport.scala
@@ -60,16 +60,29 @@ object HoodieParquetReadSupport {
* schema is trimmed down to the requested fields. This can happen when the
table schema evolves and only a subset of
* the nested fields are required by the query.
*
+ * A top-level field the file does not have at all (a column added by DDL,
or by a later write) is
+ * removed as well. ParquetRowConverter builds its converters per parquet
field and leaves the
+ * catalyst columns it never sees null, which is what Spark's own row-based
reader gets from
+ * intersectParquetGroups when nested schema pruning is on (the file group
reader switches it
+ * off). Kept, the field reaches the converter as a group synthesised from
the catalyst type, and
+ * for a Spark 4.1 variant requested as the PushVariantIntoScan projection
struct that is a plain
+ * group the variant converter rejects (INVALID_VARIANT_SHREDDING_SCHEMA,
#20135).
+ *
+ * This holds whatever enableVectorizedReader the read support was built
with: every reader that
+ * uses it is row-based, HoodieSparkParquetReader included although it
passes true, and the
+ * vectorized readers pin READ_SUPPORT_CLASS to Spark's own
ParquetReadSupport.
+ *
* @param requestedSchema the initial parquet schema requested by Spark
* @param fileSchema the actual parquet schema of the file
- * @return a potentially updated schema with empty struct fields removed
+ * @return a potentially updated schema with missing fields and empty struct
fields removed
*/
def trimParquetSchema(requestedSchema: MessageType, fileSchema:
MessageType): MessageType = {
val trimmedFields = requestedSchema.getFields.asScala.map(field => {
if (fileSchema.containsField(field.getName)) {
trimParquetType(field, fileSchema.asGroupType().getType(field.getName))
} else {
- Some(field)
+ // Not in the file: left out of the read schema, and the row converter
null-fills it.
+ None
}
}).filter(_.isDefined).map(_.get).toArray[Type]
Types.buildMessage().addFields(trimmedFields:
_*).named(requestedSchema.getName)
diff --git
a/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieParquetReadSupport.scala
b/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieParquetReadSupport.scala
index 973454d1b70c..6461c9fe853f 100644
---
a/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieParquetReadSupport.scala
+++
b/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieParquetReadSupport.scala
@@ -113,4 +113,40 @@ class TestHoodieParquetReadSupport {
.named("required")
Assertions.assertEquals(expectedSchema, trimmedSchema)
}
+
+ /**
+ * A top-level field the file does not have at all (a column added by DDL,
say) is dropped from
+ * the read schema; the row converter leaves the catalyst column it never
sees null. The missing
+ * field is shaped like the Spark 4.1 PushVariantIntoScan projection struct
of a variant column,
+ * the case that fails when it is kept (#20135): the synthesised group is
not a variant group,
+ * yet the variant converter is built over it. A request left with no field
at all trims to an
+ * empty message.
+ */
+ @Test
+ def testSchemaTrimmingMissingTopLevelField(): Unit = {
+ val projectionStruct = Types.optionalGroup()
+ .addField(Types.optional(PrimitiveTypeName.BINARY).named("0"))
+ .named("v2")
+ val requiredSchema = Types.buildMessage()
+ .addField(Types.required(PrimitiveTypeName.INT32).named("id"))
+ .addField(Types.optional(PrimitiveTypeName.BINARY).named("v"))
+ .addField(projectionStruct)
+ .named("required")
+ val dataSchema = Types.buildMessage()
+ .addField(Types.required(PrimitiveTypeName.INT32).named("id"))
+ .addField(Types.optional(PrimitiveTypeName.BINARY).named("v"))
+ .named("data")
+
+ val expectedSchema = Types.buildMessage()
+ .addField(Types.required(PrimitiveTypeName.INT32).named("id"))
+ .addField(Types.optional(PrimitiveTypeName.BINARY).named("v"))
+ .named("required")
+ Assertions.assertEquals(expectedSchema,
HoodieParquetReadSupport.trimParquetSchema(requiredSchema, dataSchema))
+
+ // A request whose only field is missing trims to an empty message, and
the reader returns
+ // rows of nulls for it.
+ val onlyMissingField =
Types.buildMessage().addField(projectionStruct).named("required")
+ Assertions.assertEquals(Types.buildMessage().named("required"),
+ HoodieParquetReadSupport.trimParquetSchema(onlyMissingField, dataSchema))
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
index a05a1f844cda..b2d7af1d522a 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
@@ -21,6 +21,7 @@ package org.apache.spark.sql.hudi.dml.schema
import org.apache.hudi.{DataSourceWriteOptions, HoodieSparkUtils,
SparkAdapterSupport}
import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+import org.apache.hudi.common.model.WriteOperationType
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieWriteConfig}
import org.apache.hudi.keygen.NonpartitionedKeyGenerator
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
@@ -30,6 +31,9 @@ import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
import org.apache.spark.sql.types.{DataType, StructType}
+import java.nio.file.{Files, Paths}
+
+import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
import scala.util.{Failure, Success, Try}
@@ -214,6 +218,110 @@ class TestVariantPushVariantIntoScan extends
HoodieSparkSqlTestBase {
}
}
+ test("a second variant column added by DDL reads through old and new files")
{
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // Without schema-on-read, `alter table add columns` writes an empty
ALTER_SCHEMA commit and
+ // touches no data file, so every file written before it lacks v2. Hudi's
readers switch
+ // Spark's nested schema pruning off, so the missing column has to leave
the parquet read schema
+ // through HoodieParquetReadSupport.trimParquetSchema; kept there, it
becomes a plain group over
+ // which Spark builds its variant converter and fails with
INVALID_VARIANT_SHREDDING_SCHEMA
+ // (#20135). The same read sits under the COW merge handle, the native
parquet log reader and
+ // MOR compaction, so the old file group is read through each of them.
+ def dataFiles(tablePath: String): Seq[String] = {
+ val stream = Files.walk(Paths.get(tablePath))
+ try {
+ stream.iterator().asScala.filter(Files.isRegularFile(_))
+ .map(p => Paths.get(tablePath).relativize(p).toString)
+ .filter(p => !p.startsWith(".hoodie") && !p.endsWith(".crc") &&
(p.endsWith(".parquet") || p.contains(".log.")))
+ .toSeq.sorted
+ } finally {
+ stream.close()
+ }
+ }
+ def rowsSql(lo: Int, hi: Int, kPrefix: String, ts: Long, v2Prefix:
Option[String]): String = {
+ val v2Col = v2Prefix.map(p => s""", parse_json(concat('{"k":"$p', id,
'"}')) as v2""").getOrElse("")
+ s"""select cast(id as int) as id, parse_json(concat('{"k":"$kPrefix',
id, '"}')) as v,
+ | ${ts}L as ts$v2Col from range($lo, $hi, 1, 1)""".stripMargin
+ }
+ // ids 0-2 come from the writes before the DDL, ids 3-5 from the insert
after it.
+ def vOf(id: Int): String = if (id < 3) s"a$id" else s"b$id"
+
+ // `v2Of` gives the expected $.k of v2 per id, null where the row has no
v2; `v2FilterIds` names
+ // one row per file or block that holds a v2 value, each matched by its
own filter.
+ def assertReads(tableName: String, leg: String, pushed: Boolean, ids:
Range, v2Of: Int => String,
+ v2FilterIds: Seq[Int] = Seq.empty): Unit = {
+ checkAnswer(s"select id, variant_get(v, '$$.k', 'string'),
variant_get(v2, '$$.k', 'string') " +
+ s"from $tableName order by id")(ids.map(id => Seq(id, vOf(id),
v2Of(id))): _*)
+ checkAnswer(s"select id from $tableName where v2 is null order by id")(
+ ids.filter(v2Of(_) == null).map(Seq(_)): _*)
+ checkAnswer(s"select id from $tableName where variant_get(v, '$$.k',
'string') = 'a1'")(Seq(1))
+ checkAnswer(s"select id from $tableName where variant_get(v2, '$$.k',
'string') = 'none'")()
+ v2FilterIds.foreach { id =>
+ checkAnswer(s"select id from $tableName where variant_get(v2, '$$.k',
'string') = '${v2Of(id)}'")(Seq(id))
+ }
+ checkAnswer(s"select id, cast(v2 as string) from $tableName order by
id")(
+ ids.map(id => Seq(id, Option(v2Of(id)).map(k =>
s"""{"k":"$k"}""").orNull)): _*)
+ Seq("v", "v2").foreach { column =>
+ assertPushed(s"select id, variant_get($column, '$$.k', 'string') from
$tableName", pushed, s"$leg, $column")
+ }
+ }
+
+ sweep("ddl-variant") { (tableName, tablePath, tableType, pushed) =>
+ val leg = s"$tableType, $tableName"
+ createTable(tableName, tablePath, tableType, "v variant")
+ // Seeded values that the update below replaces; on MOR the update is a
log block written
+ // before the DDL, so the old file group reads v2 through a log file
that lacks it too.
+ spark.sql(s"insert into $tableName ${rowsSql(0, 3, "o", 1000L, None)}")
+ spark.sql(s"update $tableName set " + """v =
parse_json(concat('{"k":"a', id, '"}')), ts = 1001""")
+ if (tableType == "mor") {
+ assert(dataFiles(tablePath).count(_.contains(".log.")) == 1,
+ s"[$leg] expected one log file before the DDL:
${dataFiles(tablePath)}")
+ }
+ val filesBeforeDdl = dataFiles(tablePath)
+
+ spark.sql(s"alter table $tableName add columns (v2 variant)")
+ val timeline = createMetaClient(spark, tablePath).getActiveTimeline
+ val alterMetadata = timeline.readCommitMetadata(
+ timeline.getCommitsTimeline.filterCompletedInstants.lastInstant.get)
+ assert(alterMetadata.getOperationType == WriteOperationType.ALTER_SCHEMA,
+ s"[$leg] the DDL commit should be ALTER_SCHEMA, got
${alterMetadata.getOperationType}")
+ assert(dataFiles(tablePath) == filesBeforeDdl, s"[$leg] the DDL must not
add or rewrite data files")
+ assert(spark.table(tableName).schema("v2").dataType.simpleString ==
"variant",
+ s"[$leg] the catalog should report v2 as variant")
+
+ // Old files only: every row reads v2 through a file that lacks the
column.
+ assertReads(tableName, leg, pushed, 0 until 3, _ => null)
+
+ // Rows that carry v2; the small-file limit of 0 keeps them out of the
old file group.
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ spark.sql(s"insert into $tableName ${rowsSql(3, 6, "b", 1000L,
Some("z"))}")
+ }
+ assertReads(tableName, leg, pushed, 0 until 6, id => if (id < 3) null
else s"z$id", v2FilterIds = Seq(3))
+
+ // One row of each file group: COW rewrites both groups through the
merge handle, MOR writes
+ // one log block per group beside the one from before the DDL.
+ spark.sql(s"update $tableName set " +
+ """v2 = parse_json(concat('{"k":"u', id, '"}')), ts = 1002 where id in
(2, 4)""")
+ def updatedV2(id: Int): String = id match {
+ case 2 | 4 => s"u$id"
+ case _ if id < 3 => null
+ case _ => s"z$id"
+ }
+ assertReads(tableName, leg, pushed, 0 until 6, updatedV2, v2FilterIds =
Seq(2, 3, 4))
+
+ if (tableType == "mor") {
+ assert(dataFiles(tablePath).count(_.contains(".log.")) == 3,
+ s"[$leg] expected three log files after the update:
${dataFiles(tablePath)}")
+ val baseFilesBefore =
dataFiles(tablePath).count(_.endsWith(".parquet"))
+ spark.sql(s"call run_compaction(op => 'run', table => '$tableName')")
+ assert(dataFiles(tablePath).count(_.endsWith(".parquet")) ==
baseFilesBefore + 2,
+ s"[$leg] compaction should write a new base file for both file
groups: ${dataFiles(tablePath)}")
+ assertReads(tableName, leg, pushed, 0 until 6, updatedV2, v2FilterIds
= Seq(2, 3, 4))
+ }
+ }
+ }
+
test("bootstrapped tables read the projection through the skeleton and data
file join") {
assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
