wombatu-kun commented on code in PR #19558:
URL: https://github.com/apache/hudi/pull/19558#discussion_r3748118648
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -159,7 +159,31 @@ 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). Request the
full-variant projection
+ // shape instead and restore native VariantType after the scan.
User-facing reads pass
+ // sparkRequiredSchema and are overlaid above; Spark < 4.1 has no shredded
read support
+ // and the adapter returns None (shredded files cannot be written there
either).
Review Comment:
Spark4_0Adapter inherits generateVariantWriteShreddingSchema and
createShreddedVariantWriter from BaseSpark4Adapter with no version gate, so
Spark 4.0 does write shredded files - it is the read side that is missing
there. Worth rewording the parenthetical.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -159,7 +159,31 @@ 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). Request the
full-variant projection
+ // shape instead and restore native VariantType after the scan.
User-facing reads pass
+ // sparkRequiredSchema and are overlaid above; Spark < 4.1 has no shredded
read support
+ // and the adapter returns None (shredded files cannot be written there
either).
+ val isParquetBaseFile = !isInlineLog && !FSUtils.isLogFile(filePath) &&
+ HoodieFileFormat.fromFileExtension(filePath.getFileExtension) ==
HoodieFileFormat.PARQUET
+ val (readStructTypeForScan, variantOrdinals) =
+ if (sparkRequiredSchema.isEmpty && isParquetBaseFile) {
Review Comment:
HoodieMergeOnReadRDDV2 and CDCFileGroupIterator build this context without
sparkRequiredSchema, so MOR incremental, streaming and CDC queries also take
this rewrite. Is that intended - the impact section reads as internal-only?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -253,6 +253,201 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
+ test("Test COW clustering preserves VARIANT values") {
Review Comment:
Clustering is covered, but the compaction leg is not: the existing MOR
compaction test compacts a log-only slice, so no test reads a shredded parquet
base file through the internal reader. Worth a second compaction round after a
base file exists, or a note that it is follow-up.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -253,6 +253,201 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
+ 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',
Review Comment:
Nothing here asserts the base files actually carry typed_value, so if the
forced shredding schema ever stops taking effect this test silently degrades
into the unshredded twin below and still passes. Worth adding the same
listDataParquetFiles/getFieldAsGroup typed_value check the new bulk_insert test
already 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]
