voonhous commented on code in PR #19582:
URL: https://github.com/apache/hudi/pull/19582#discussion_r3763774931
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -448,6 +449,74 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
+ test("Test COW small-file merge preserves shredded VARIANT values") {
+ // The #19567 repro: the second insert bin-packs into the existing small
file group, and the
+ // small-file MERGE rewrites the old base file through the AVRO read path
+ // (HoodieAvroParquetReader + HoodieVariantReconstruction), not the Spark
reader context the
+ // clustering tests above exercise. Before the fix, the footer-derived
file schema lost the
+ // variant logical type, reconstruction never engaged, and rows 1-2 came
back null after the
+ // second commit. Unlike those tests, small.file.limit stays at its
default on purpose: the
+ // bin-pack is the trigger.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v variant,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.parquet.variant.write.shredding.enabled = 'true',
+ | hoodie.parquet.variant.force.shredding.schema.for.test = 'key
string',
+ | hoodie.index.type = 'INMEMORY'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ // Pin the trigger: the first base file must be shredded, or this
degenerates into a
+ // plain unshredded merge.
+ val firstFiles = listDataParquetFiles(tablePath)
+ assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
+ firstFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"First base file should carry typed_value. Schema:\n$variantGroup")
+ }
+
+ spark.sql(s"insert into $tableName values " +
Review Comment:
Addressed. It does go through `HoodieConcatHandle` -> `HoodieMergeHelper`,
and as you traced, `recordSchema` stayed footer-derived there, so
reconstruction never engaged on that path. Pushed a follow-up commit:
`HoodieMergeHelper` now aligns shredded variant columns to their requested form
(`VariantSchemaUtils.alignShreddedVariants`) before the strict-projection
check, so the merge reads the base file at the writer schema and reconstruction
engages; the lossy rewrite no longer runs. Also pinned the test to the AVRO
record type since the record type picks the reader the merge uses. Verified the
test is red without the `HoodieMergeHelper` change and green with it under the
spark4.1 profile.
--
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]
