voonhous commented on code in PR #19583:
URL: https://github.com/apache/hudi/pull/19583#discussion_r3765208693
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -448,6 +448,86 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
+ test("Test CDC captures VARIANT values from shredded base files") {
+ // #19578: CDC is the only default-config query path that builds the
internal reader
+ // context without a catalyst schema, and its BASE_FILE_INSERT case
additionally reads
+ // the new base file directly, bypassing the context, so it needs its own
full-variant
+ // rewrite. One row only on purpose: a COW update rewrites the base file
through the
+ // avro merge path, whose shredded-read fix is #19582 (in flight), and a
single row
+ // leaves nothing for that path to carry over.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ // OP_KEY_ONLY reconstructs both images by reading the (shredded) file
slices;
+ // DATA_BEFORE_AFTER (the default) reads the update images from the cdc
log instead.
+ // The insert leg takes BASE_FILE_INSERT in both modes.
+ Seq("OP_KEY_ONLY", "DATA_BEFORE_AFTER").foreach { loggingMode =>
+ 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.table.cdc.enabled' = 'true',
+ | 'hoodie.table.cdc.supplemental.logging.mode' = '$loggingMode',
+ | hoodie.parquet.variant.write.shredding.enabled = 'true',
Review Comment:
Addressed -- the CDC test now sweeps `Seq(true, false)` for shredding
instead of gaining a second copy, with the layout pin flipped per leg.
One nuance: unshredded-through-the-rewrite is already green via the
unshredded clustering test from #19558, which drives the same helpers through
the reader context. What genuinely had no coverage is the hand-rolled
rewrite/restore block this PR added to `CDCFileGroupIterator`, and that is what
the new leg hits.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -448,6 +448,86 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
+ test("Test CDC captures VARIANT values from shredded base files") {
+ // #19578: CDC is the only default-config query path that builds the
internal reader
+ // context without a catalyst schema, and its BASE_FILE_INSERT case
additionally reads
+ // the new base file directly, bypassing the context, so it needs its own
full-variant
+ // rewrite. One row only on purpose: a COW update rewrites the base file
through the
+ // avro merge path, whose shredded-read fix is #19582 (in flight), and a
single row
+ // leaves nothing for that path to carry over.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ // OP_KEY_ONLY reconstructs both images by reading the (shredded) file
slices;
+ // DATA_BEFORE_AFTER (the default) reads the update images from the cdc
log instead.
+ // The insert leg takes BASE_FILE_INSERT in both modes.
+ Seq("OP_KEY_ONLY", "DATA_BEFORE_AFTER").foreach { loggingMode =>
+ 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.table.cdc.enabled' = 'true',
+ | 'hoodie.table.cdc.supplemental.logging.mode' = '$loggingMode',
+ | 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)""")
+
+ // Pin the trigger: the base file the CDC reads must actually be
shredded.
+ val baseFiles = listDataParquetFiles(tablePath)
+ assert(baseFiles.nonEmpty, "Should have a base parquet file after the
insert")
+ baseFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"Base file should carry typed_value. Schema:\n$variantGroup")
+ }
+
+ spark.sql(s"""update $tableName set v =
parse_json('{"key":"value2"}'), ts = 1001 where id = 1""")
Review Comment:
Addressed, and the diagnosis moves a bit. A cdc table is forced onto
`FileGroupReaderBasedMergeHandle`, whose reader context follows the merger's
record type, and this was the one test in the file not wrapped in
`withRecordType` -- so it ran AVRO and hit the #19567 reconstruction gap. What
fails is the CDC *before* image being written with `value = null`, not a
carried-over row, so the single-row argument did not hold either way.
Pinned to `HoodieRecordType.SPARK`, which routes that read through the
context #19558 already fixed. The update leg and both logging modes stay, with
no dependency on #19582 -- verified green on Spark 4.1 (all four legs). For the
record, the half of #19582 that would have mattered here is the reconstruction
detection, not the merge-helper alignment.
--
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]
