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 827644e8699b fix(spark): null-fill a DDL-added variant column (#20136)
827644e8699b is described below
commit 827644e8699b61d319ed156c92b9f0c39df5aedf
Author: voonhous <[email protected]>
AuthorDate: Thu Oct 1 12:25:57 2026 +0800
fix(spark): null-fill a DDL-added variant column (#20136)
* fix(spark): null-fill a DDL-added variant column
Hudi's parquet readers force spark.sql.optimizer.nestedSchemaPruning
.enabled off, so Spark's ParquetReadSupport skips intersectParquetGroups
and a requested column the file does not have stays in the parquet read
schema as a group synthesised from its catalyst type. HoodieParquet
ReadSupport.trimParquetSchema already drops such NESTED fields but kept
a missing TOP-LEVEL one. For a plain column that was harmless:
parquet-mr null-fills it. For a Spark 4.1 variant column requested as
the PushVariantIntoScan projection struct, or as the full-variant struct
of the internal reader, the synthesised group is a plain struct, Spark
still routes it to ParquetVariantConverter, and buildVariantSchema
throws INVALID_VARIANT_SHREDDING_SCHEMA. After ALTER TABLE ADD COLUMNS
(v2 variant) that killed every pushed-down read of a pre-DDL file, the
COW update of such a file group through the SPARK-record merge handle,
and MOR compaction of it.
Drop the missing top-level field for the row-based reader, the way the
nested ones already are: ParquetRowConverter builds its converters per
parquet field and leaves the catalyst columns it never sees null, which
is what Spark's own reader gets from intersectParquetGroups. The
vectorized reader keeps the field, as in Spark, since it null-fills and
matches by position; every Hudi row read constructs this read support
with enableVectorizedReader = false.
TestHoodieParquetReadSupport pins both arms of the trim over a
projection-shaped missing field. TestVariantShreddingMixedLayouts gets
the end-to-end case from the #18285 checklist: a shredded base with v
only, the DDL pinned as an ALTER_SCHEMA commit typing v2 as VARIANT,
reads over the old file alone, an insert opening a second file group
with v2 shredded, an update rewriting both groups on COW and landing as
log blocks on MOR, compaction on MOR, all read on both
pushVariantIntoScan arms with the plan pinned per arm; COW with both
record types, MOR with parquet log blocks and with avro data blocks.
Fixes #20135. Part of #18285.
* fix(spark): drop missing column for all readers
HoodieSparkParquetReader, which reads native parquet log files, is
row-based but builds the read support with enableVectorizedReader =
true, so the flag-gated drop kept a missing top-level column there: a
pushed-down read of a parquet log block written before the ADD COLUMNS
still failed with INVALID_VARIANT_SHREDDING_SCHEMA. No vectorized
reader uses HoodieParquetReadSupport (they pin READ_SUPPORT_CLASS to
Spark's ParquetReadSupport), so the flag protected nothing.
trimParquetSchema now drops a missing top-level field unconditionally,
like a missing nested one, and the dropMissingTopLevelFields parameter
is gone. The MOR legs of the DDL test write a log block before the DDL
so the old file group holds a log file without the column too.
* test(variant): tighten the DDL-added column test
- Pin the empty message a request trims to when its only field is
missing from the file.
- Build each leg's table once, with the rule on, and sweep only the
reads over both pushVariantIntoScan arms.
- Drop the AVRO record type from the COW leg: its reads repeat the
SPARK leg and the AVRO merge handle passed before the fix (#20135).
- Compare the DDL instant with the one captured right before the
DDL, so the check can fail on the MOR legs again.
- One v2 filter per file or block that holds a value, plus one that
matches nothing, instead of one per id.
- Remove an always-true assertion and one that the unchanged file
listing already covers.
- Say that the file group reader, not every Hudi reader, switches
nested schema pruning off.
---
.../parquet/HoodieParquetReadSupport.scala | 17 +-
.../parquet/TestHoodieParquetReadSupport.scala | 36 ++++
.../schema/TestVariantShreddingMixedLayouts.scala | 234 ++++++++++++++++++++-
3 files changed, 284 insertions(+), 3 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/TestVariantShreddingMixedLayouts.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
index 671840ca5ce5..544b01db4e25 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
@@ -22,7 +22,7 @@ package org.apache.spark.sql.hudi.dml.schema
import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions,
DefaultSparkRecordMerger, HoodieSchemaConversionUtils, HoodieSparkUtils,
HoodieTableSchema, SparkAdapterSupport}
import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.model.{HoodieAvroRecordMerger, HoodieFileFormat}
+import org.apache.hudi.common.model.{HoodieAvroRecordMerger,
HoodieCommitMetadata, HoodieFileFormat, WriteOperationType}
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaType}
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType
@@ -30,6 +30,7 @@ import org.apache.hudi.config.{HoodieBootstrapConfig,
HoodieWriteConfig}
import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter
import org.apache.hudi.keygen.NonpartitionedKeyGenerator
import org.apache.hudi.testutils.DataSourceTestUtils
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
import org.apache.hadoop.fs.{Path => HadoopPath}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
@@ -2363,6 +2364,237 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
}
}
+ test("A second variant column added by DDL reads through old and new files")
{
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // Without hoodie.schema.on.read.enable, `alter table add columns` takes
the v1 route:
+ // HoodieCatalog.loadTable hands back a V1Table, Spark plans
AlterTableAddColumnsCommand, and
+ // HoodiePostAnalysisRule swaps in AlterHoodieTableAddColumnsCommand,
which writes an EMPTY
+ // ALTER_SCHEMA commit carrying the new schema and touches no data file. A
VARIANT column added
+ // that way is its own case: the table has v2 before any file does, so
every read of an old
+ // file has to yield a null v2 and, with PushVariantIntoScan on, a null
projection struct that
+ // variant_get turns into null rather than an error. 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
(INVALID_VARIANT_SHREDDING_SCHEMA, #20135). The
+ // same base-file read sits under the COW merge handle and under MOR
compaction, and the MOR
+ // legs write a log block before the DDL as well: a native parquet log
without v2 goes through
+ // the same read support from HoodieSparkParquetReader. New files shred v2
under the forced
+ // layout (which applies to every variant column), a COW update rewrites
an old file group with
+ // v2 added, and on MOR a log block carrying v2 merges over a base file
and a log block without
+ // it until compaction folds them together.
+ 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 commit 1 (on MOR through the log block written before
the DDL), ids 3-5
+ // from the insert after the DDL.
+ def vOf(id: Int): String = if (id < 3) s"a$id" else s"b$id"
+
+ // The reads every state is checked with, on both pushVariantIntoScan
arms; `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 non-null v2, each with its own filter, so a
slot that reads back
+ // through the wrong file or block cannot hide behind the others.
+ def assertReads(tableName: String, leg: String, ids: Range, v2Of: Int =>
String,
+ v2FilterIds: Seq[Int] = Seq.empty): Unit = {
+ Seq("true", "false").foreach { pushIntoScan =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ 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))
+ // A v2 filter that matches no row, so a state without any v2 value
filters on v2 too.
+ 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)): _*)
+ // Both arms expect the very same rows; only the plan tells them
apart.
+ val pushed = pushIntoScan.toBoolean
+ val verdict = if (pushed) "should have" else "must not have"
+ Seq("v", "v2").foreach { column =>
+ assert(variantProjectionPushedIntoScan(
+ s"select id, variant_get($column, '$$.k', 'string') from
$tableName") == pushed,
+ s"[$leg, pushVariantIntoScan=$pushIntoScan] PushVariantIntoScan
$verdict rewritten $column " +
+ "into a projection struct")
+ }
+ }
+ }
+ }
+
+ def runLeg(label: String, tableType: String, recordTypes:
Seq[HoodieRecordType], tableProps: Seq[String],
+ expectLogBlockType: Option[HoodieLogBlockType]): Unit = {
+ // The table is built once, with the rule on as it is by default on
Spark 4.1+; only the reads
+ // are swept over both arms (assertReads).
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "true") {
+ withVariantTable(label, tableType, props = tableProps,
+ recordTypes = recordTypes) { (tableName, tablePath, leg) =>
+ val isMor = expectLogBlockType.isDefined
+
+ // Commit 1: v only, shredded. The MOR legs seed values that the log
block below
+ // replaces, so a merged read that skipped that block would show the
stale ones.
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName ${rowsSql(0, 3, if (isMor) "o"
else "a", 1000L, None)}")
+ }
+ assertVariantLayout(tablePath, shredded = true, leg)
+ val commit1 = latestCompletedInstant(tablePath)
+ val firstFileIds = baseLayouts(tablePath).filter(_.instantTime ==
commit1).map(_.fileId).distinct
+ assert(firstFileIds.size == 1, s"[$leg] commit 1 should write one
file group, got $firstFileIds")
+ val firstFileId = firstFileIds.head
+
+ // MOR: a log block written before the DDL, so the old file group
reads v2 through a log
+ // file that lacks the column as well as through a base file that
lacks it.
+ val preDdlLogInstant = if (isMor) {
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"update $tableName set " +
+ """v = parse_json(concat('{"k":"a', id, '"}')), ts = 1001""")
+ }
+ Some(latestCompletedInstant(tablePath))
+ } else {
+ None
+ }
+ val filesBeforeDdl = listDataParquetFiles(tablePath).sorted
+ val instantBeforeDdl = latestCompletedInstant(tablePath)
+
+ // The DDL: an empty ALTER_SCHEMA commit whose metadata carries v2
as a VARIANT.
+ spark.sql(s"alter table $tableName add columns (v2 variant)")
+ val alterInstant = latestCompletedInstant(tablePath)
+ assert(alterInstant != instantBeforeDdl, s"[$leg] the DDL should
have completed its own commit")
+ val metaClient = createMetaClient(spark, tablePath)
+ val alterMetadata = metaClient.getActiveTimeline.readCommitMetadata(
+
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants.lastInstant.get)
+ assert(alterMetadata.getOperationType ==
WriteOperationType.ALTER_SCHEMA,
+ s"[$leg] the DDL commit should be ALTER_SCHEMA, got
${alterMetadata.getOperationType}")
+ assert(listDataParquetFiles(tablePath).sorted == filesBeforeDdl,
+ s"[$leg] the DDL must not add or rewrite data files")
+ val committedV2 =
HoodieSchema.parse(alterMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY))
+ .getField("v2").get.schema().getNonNullType.getType
+ assert(committedV2 == HoodieSchemaType.VARIANT,
+ s"[$leg] the committed table schema should carry v2 as a VARIANT,
got $committedV2")
+ val catalogSchema = spark.table(tableName).schema
+ assert(catalogSchema("v2").dataType.simpleString == "variant",
+ s"[$leg] the catalog should report v2 as variant, got
${catalogSchema("v2").dataType}")
+ val dataColumns =
catalogSchema.fieldNames.filterNot(_.startsWith("_hoodie")).toSeq
+ assert(dataColumns == Seq("id", "v", "ts", "v2"),
+ s"[$leg] the DDL appends v2 after the data columns, got
$dataColumns")
+
+ // Old files only: every row reads v2 through a file that lacks the
column.
+ assertReads(tableName, leg, 0 until 3, _ => null)
+
+ // Commit 3: rows that carry v2. The small-file limit of 0 keeps
them out of the old
+ // file group, so they land in a second file group whose base file
has v2.
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName ${rowsSql(3, 6, "b", 1000L,
Some("z"))}")
+ }
+ }
+ val commit3 = latestCompletedInstant(tablePath)
+ val v2Bases = baseLayouts(tablePath, "v2")
+ assert(v2Bases.map(_.fileId).distinct.size == 2,
+ s"[$leg] expected the old and the new file group, got
${v2Bases.map(_.path)}")
+ val oldBase = v2Bases.filter(_.instantTime == commit1)
+ assert(oldBase.size == 1 && !oldBase.head.hasColumn,
+ s"[$leg] the commit-1 base file must not have v2:
${oldBase.map(_.path)}")
+ val newBase = v2Bases.filter(_.instantTime == commit3)
+ assert(newBase.size == 1 && newBase.head.hasColumn &&
newBase.head.typedFields.contains("k"),
+ s"[$leg] the commit-3 base file should carry v2 shredded on k:
${newBase.map(_.path)}")
+ assertAllShredded(baseLayouts(tablePath), shredded = true, s"$leg v
in both base files")
+
+ // id 3 stands for the one slot that holds a v2 value, the new base
file.
+ assertReads(tableName, leg, 0 until 6, id => if (id < 3) null else
s"z$id", v2FilterIds = Seq(3))
+
+ // Commit 4: an update of v2 on one row of each file group.
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"update $tableName set " +
+ """v2 = parse_json(concat('{"k":"u', id, '"}')), ts = 1002 where
id in (2, 4)""")
+ }
+ val commit4 = latestCompletedInstant(tablePath)
+ expectLogBlockType match {
+ case None =>
+ // COW rewrites both groups; the old group's new version gains
v2 while the
+ // version it replaces (kept until clean) still lacks it.
+ val firstGroup = baseLayouts(tablePath, "v2").filter(_.fileId ==
firstFileId)
+ assert(firstGroup.map(_.instantTime).toSet == Set(commit1,
commit4),
+ s"[$leg] the update should rewrite the commit-1 file group:
${firstGroup.map(_.path)}")
+ firstGroup.foreach { layout =>
+ if (layout.instantTime == commit1) {
+ assert(!layout.hasColumn, s"[$leg] the replaced version must
not have v2: ${layout.path}")
+ } else {
+ assert(layout.hasColumn && layout.typedFields.contains("k"),
+ s"[$leg] the rewritten version should carry v2 shredded on
k: ${layout.path}")
+ }
+ }
+ val rewritten = baseLayouts(tablePath,
"v2").filter(_.instantTime == commit4)
+ assert(rewritten.map(_.fileId).distinct.size == 2,
+ s"[$leg] the update should rewrite both file groups:
${rewritten.map(_.path)}")
+ assertAllShredded(rewritten, shredded = true, s"$leg v2 in the
rewritten base files")
+ assertAllShredded(baseLayouts(tablePath).filter(_.instantTime ==
commit4), shredded = true,
+ s"$leg v in the rewritten base files")
+ case Some(blockType) =>
+ // MOR: the log block from before the DDL plus one per file
group from this update;
+ // the base files stay as they were.
+ val blockTypes = listLogBlockTypes(tablePath)
+ assert(blockTypes.count(_ == blockType) == 3 &&
blockTypes.forall(_ == blockType),
+ s"[$leg] expected exactly three $blockType blocks, found:
$blockTypes")
+ if (blockType == HoodieLogBlockType.PARQUET_DATA_BLOCK) {
+ val (preDdlLogs, updateLogs) = nativeLogLayouts(tablePath,
"v2")
+ .partition(log => preDdlLogInstant.contains(log.instantTime))
+ assert(preDdlLogs.size == 1 && !preDdlLogs.head.hasColumn,
+ s"[$leg] the log file written before the DDL must not have
v2: ${preDdlLogs.map(_.path)}")
+ assert(updateLogs.size == 2 && updateLogs.forall(_.hasColumn),
+ s"[$leg] both log files of the update should carry v2:
${updateLogs.map(_.path)}")
+ assertAllShredded(updateLogs, shredded = true, s"$leg v2 in
the log files of the update")
+ }
+ }
+
+ def updatedV2(id: Int): String = id match {
+ case 2 | 4 => s"u$id"
+ case _ if id < 3 => null
+ case _ => s"z$id"
+ }
+ // One filter per slot: the update of each file group (ids 2 and 4)
and the new base file.
+ assertReads(tableName, leg, 0 until 6, updatedV2, v2FilterIds =
Seq(2, 3, 4))
+
+ if (expectLogBlockType.isDefined) {
+ // Compaction folds each log block into a base file without v2
(old group) or with it.
+ withWriteLayout(Forced("k string")) {
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 1, leg)
+ val compactionInstant = latestCompletedInstant(tablePath)
+ val compacted = baseLayouts(tablePath, "v2").filter(_.instantTime
== compactionInstant)
+ assert(compacted.map(_.fileId).distinct.size == 2 &&
compacted.forall(_.hasColumn),
+ s"[$leg] compaction should write both file groups with v2:
${compacted.map(_.path)}")
+ assertAllShredded(compacted, shredded = true, s"$leg v2 in the
compacted base files")
+ assertAllShredded(baseLayouts(tablePath).filter(_.instantTime ==
compactionInstant),
+ shredded = true, s"$leg v in the compacted base files")
+ assertReads(tableName, leg, 0 until 6, updatedV2, v2FilterIds =
Seq(2, 3, 4))
+ }
+ }
+ }
+ }
+
+ // COW, SPARK records: the update goes through the merge handle, which has
to add a null v2 for
+ // the untouched rows while keeping v shredded.
+ runLeg("cow, second variant by DDL", "cow", Seq(HoodieRecordType.SPARK),
+ Seq.empty, expectLogBlockType = None)
+ // MOR, SPARK records on the current table version: native parquet log
files.
+ runLeg("mor, parquet log blocks", "mor", Seq(HoodieRecordType.SPARK),
+ Seq("hoodie.compact.inline = 'false'"), expectLogBlockType =
Some(HoodieLogBlockType.PARQUET_DATA_BLOCK))
+ // MOR, avro data blocks: table version 9 is the only way to get them; SQL
DML and the DDL both
+ // pick the version and block format up from the tblproperties.
+ runLeg("mor, avro data blocks", "mor", Seq(HoodieRecordType.AVRO),
+ Seq("hoodie.compact.inline = 'false'",
+ "hoodie.write.table.version = '9'",
+ "hoodie.logfile.data.block.format = 'avro'"),
+ expectLogBlockType = Some(HoodieLogBlockType.AVRO_DATA_BLOCK))
+ }
+
//
-----------------------------------------------------------------------------------------------
// H. Robustness
//
-----------------------------------------------------------------------------------------------
