wombatu-kun commented on code in PR #19558:
URL: https://github.com/apache/hudi/pull/19558#discussion_r3755412071
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -159,7 +159,37 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
structType
}
- val (readSchema, readFilters) =
getSchemaAndFiltersForRead(parquetReadStructType, hasRowIndexField)
+ // Internal reads have no catalyst plan, so nothing rewrites VariantType
fields the way
+ // PushVariantIntoScan does for user queries. Requesting native
VariantType against a
+ // SHREDDED parquet base file clips the file group to {metadata, value}
and reads
+ // value=null; write-side callers (compaction, clustering, merge) would
then persist the
+ // nulls, silently losing the variant data (#19556). Query paths that
build this context
+ // without sparkRequiredSchema (MOR incremental relation, streaming, CDC)
hit the same
Review Comment:
The impact section now claims MOR incremental, streaming and CDC reads were
returning null for shredded variants and are fixed by this rewrite, but no test
exercises any of those three paths. Worth one incremental-query round trip on a
shredded table, or is that follow-up?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -250,9 +250,248 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
metaClient.reloadActiveTimeline()
assert(metaClient.getActiveTimeline.getCleanerTimeline.countInstants() >
0,
"Expected at least one .clean instant on the timeline after
compaction")
+
+ // Round 2: the compaction above compacted a log-only slice, so it never
read a
+ // parquet base file. Pin that the compacted base file is shredded, then
drive a
+ // second compaction that reads it through the internal reader
+ // (SparkFileFormatInternalRowReaderContext) and must carry rows 3 and
4, which
+ // exist only in that base file, forward (#19556).
+ val compactedFiles = listDataParquetFiles(tablePath)
+ assert(compactedFiles.nonEmpty, "Should have a compacted base parquet
file")
+ compactedFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"Compacted base file should carry typed_value.
Schema:\n$variantGroup")
+ }
+
+ // The v2-merged deltacommit above was the first after the compaction;
four more
+ // reach max.delta.commits = 5 and trip the second compaction inline.
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v1-r2"}'),
ts = 1003 where id = 1""")
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v2-r2"}'),
ts = 1004 where id = 2""")
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v1-r3"}'),
ts = 1005 where id = 1""")
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v2-r3"}'),
ts = 1006 where id = 2""")
+
+ metaClient.reloadActiveTimeline()
+
assertResult(2)(metaClient.getActiveTimeline.getCommitTimeline.filterCompletedInstants.countInstants)
+
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
+ Seq(1, "{\"key\":\"v1-r3\"}", 1005),
+ Seq(2, "{\"key\":\"v2-r3\"}", 1006),
+ Seq(3, "{\"key\":\"value3\"}", 1000),
+ Seq(4, "{\"key\":\"value4\"}", 1000)
+ )
})
}
+ test("Test COW clustering preserves VARIANT values") {
+ // Same Spark 4.1 gate as the compaction test above: clustering reads the
shredded
+ // base files back through the native reader, which rejects the 3-field
shredded
+ // layout before SPARK-54410 (Spark 4.1+).
+ 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
+ // Clustering rewrites ALL rows of the clustered file groups through the
internal
+ // write-side reader context (SparkReaderContextFactory ->
+ // SparkFileFormatInternalRowReaderContext), the stack whose blob
handling silently
+ // lost bytes in #19232. Nothing pinned its VARIANT behavior: this is
the first
+ // clustering coverage for the type. Shredding is forced so the rewrite
reads and
+ // rewrites the shredded layout, the default in production.
+ 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',
+ | hoodie.parquet.small.file.limit = '0',
+ | hoodie.clustering.inline = 'true',
+ | hoodie.clustering.inline.max.commits = '2'
+ | )
+ """.stripMargin)
+
+ // small.file.limit = 0 keeps the second insert in its own file group.
Otherwise the
+ // second commit bin-packs into the first file group and rewrites it
through the CoW
+ // small-file MERGE (a different internal read stack), conflating that
path's variant
+ // handling with the clustering rewrite this test isolates.
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ // The pre-clustering base file must actually carry the shredded layout;
without this
+ // check the test silently degrades into the unshredded twin below if
the forced
+ // shredding schema ever stops taking effect.
+ val preClusteringFiles = listDataParquetFiles(tablePath)
+ assert(preClusteringFiles.nonEmpty, "Should have at least one data
parquet file before clustering")
+ preClusteringFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"Pre-clustering base file should carry typed_value.
Schema:\n$variantGroup")
+ }
+
+ // Second commit trips inline clustering (max.commits = 2), which
rewrites the rows
+ // of the first commit too.
+ spark.sql(s"insert into $tableName values " +
+ "(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
+ "(4, parse_json('{\"key\":\"value4\"}'), 1000)")
+
+ // getLastClusteringInstant filters by action only, so a
REQUESTED/INFLIGHT instant
+ // satisfies isPresent; isCompleted confirms the rewrite finished.
+ val metaClient = createMetaClient(spark, tablePath)
+ val lastClustering =
metaClient.getActiveTimeline.getLastClusteringInstant
+ assert(lastClustering.isPresent && lastClustering.get.isCompleted,
+ "A COMPLETED clustering (replacecommit) instant must exist after
inline clustering; " +
+ "without a completed rewrite the round-trip below proves nothing")
+
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
+ Seq(1, "{\"key\":\"value1\"}", 1000),
+ Seq(2, "{\"key\":\"value2\"}", 1000),
+ Seq(3, "{\"key\":\"value3\"}", 1000),
+ Seq(4, "{\"key\":\"value4\"}", 1000)
+ )
+
+ // VARIANT must still surface as the native type after the clustering
rewrite.
+ val variantField = spark.table(tableName).schema.find(_.name == "v").get
+ assertResult("variant")(variantField.dataType.typeName)
+ })
+ }
+
+ test("Test COW clustering preserves unshredded VARIANT values") {
+ // Companion to the shredded clustering test above, with shredding
disabled. If this
+ // passes while the shredded one fails, the loss is specific to reading
the shredded
+ // layout inside the clustering rewrite, not to variant clustering in
general.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Variant clustering read-back
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 = 'false',
Review Comment:
Nothing here pins the layout, so if the shredding switch stops being honored
this test silently becomes a copy of the shredded twin and the unshredded leg
of the new rewrite goes uncovered. Worth asserting the pre-clustering base
files do not contain typed_value, mirroring the check the shredded test now
does.
--
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]
