wombatu-kun commented on code in PR #19582:
URL: https://github.com/apache/hudi/pull/19582#discussion_r3758847845
##########
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:
With `hoodie.spark.sql.insert.into.operation` at its default, the small-file
bin-pack goes through `HoodieConcatHandle` -> `HoodieMergeHelper`, where
`isStrictProjectionOf` fails on RECORD vs VARIANT so `recordSchema` stays the
footer-derived reader schema and the requested column is a plain record. Did
the new test go red before the fix, or does this merge take
`FileGroupReaderBasedMergeHandle` instead?
--
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]
