wombatu-kun commented on code in PR #19583:
URL: https://github.com/apache/hudi/pull/19583#discussion_r3758929100
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -148,12 +148,19 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
}
}
+ // The plain skip-merging reader cannot read a SHREDDED variant base file:
it requests native
+ // VariantType, which clips the shredded group to {metadata, value} and
reads value=null (the
+ // #19556 defect family). Splits with variant columns take the file-group
reader below, whose
+ // reader context requests the full-variant projection shape instead
(#19578).
+ private val requiredSchemaHasVariant: Boolean =
+ requiredSchema.structTypeSchema.fields.exists(f =>
sparkAdapter.isVariantType(f.dataType))
+
override def compute(split: Partition, context: TaskContext):
Iterator[InternalRow] = {
val partition = split.asInstanceOf[HoodieMergeOnReadPartition]
val bytesReadCallback =
HoodieSparkInputMetricsUtils.getFSBytesReadOnThreadCallback()
val iter: Iterator[InternalRow] = partition.split match {
- case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty =>
+ case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty &&
!requiredSchemaHasVariant =>
Review Comment:
requiredSchemaReaderSkipMerging appends partition values from the partition
path when shouldExtractPartitionValuesFromPartitionPath holds, and the
file-group-reader branch has no equivalent, so re-routed base-only splits would
read partition columns as null. The merged branch already has that gap - is
widening it to base-only splits intended here?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala:
##########
@@ -90,6 +92,11 @@ class InternalRowToJsonStringConverter(schema: StructType) {
structMap.toMap
case _ => value // fallback
}
+ case dt if SparkAdapterSupport.sparkAdapter.isVariantType(dt) =>
Review Comment:
This guard resolves SparkAdapterSupport.sparkAdapter for every field that is
not string, array, map or struct, and hudi-spark-common has no version adapter
on its test classpath, so all 11 TestInternalRowToJsonStringConverter cases now
error with `ClassNotFoundException:
org.apache.spark.sql.adapter.Spark4_2Adapter`. Detecting the variant type
without the adapter (its `typeName` is `variant`) both fixes that and lets the
suite cover the new case.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -381,6 +381,69 @@ class TestStreamingSource extends StreamTest {
testLegacyIncrementalStreamSource(MERGE_ON_READ, HoodieTableVersion.EIGHT)
}
+ test("test mor stream source reads shredded variant with legacy file group
reader disabled") {
+ // #19578: with the file group reader disabled, MOR streaming batches
materialize through
+ // HoodieMergeOnReadRDDV2, the only user-facing path reading shredded
variant base files
+ // without a catalyst schema. Covers both RDD branches: the base-only
split (first batch,
+ // right after inline compaction) and the merged split (second batch,
after a
+ // post-compaction update lands in a log file).
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withTempDir { inputDir =>
+ val tablePath =
s"${inputDir.getCanonicalPath}/test_mor_variant_legacy_stream"
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(MERGE_ON_READ)
+ .setTableName(getTableName(tablePath))
+ .setRecordKeyFields("id")
+ .setOrderingFields("ts")
+
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
+
+ // INMEMORY index routes MOR inserts to log files, so the first base
file is the
+ // compaction's SHREDDED one; compact = true trips inline compaction on
that write.
+ def addVariantData(valuesSql: String, compact: Boolean): Unit = {
+ spark.sql(valuesSql).write.format("org.apache.hudi")
+ .options(commonOptions)
+ .option(TBL_NAME.key, getTableName(tablePath))
+ .option(TABLE_TYPE.key, MERGE_ON_READ.name)
+ .option("hoodie.index.type", "INMEMORY")
+ .option("hoodie.parquet.variant.write.shredding.enabled", "true")
+ .option("hoodie.parquet.variant.force.shredding.schema.for.test",
"key string")
+ .option(HoodieCompactionConfig.INLINE_COMPACT.key, compact.toString)
+ .option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key,
"2")
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ }
+
+ addVariantData("""select 1 as id, parse_json('{"key":"v1"}') as v, 1000L
as ts""", compact = false)
+ addVariantData("""select 2 as id, parse_json('{"key":"v2"}') as v, 1000L
as ts""", compact = true)
+
+ val df = spark.readStream
+ .format("org.apache.hudi")
+ // force the legacy (non file-group-reader) incremental relation path
+ .option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key, "false")
+ .load(tablePath)
+ .selectExpr("id", "cast(v as string) as v", "ts")
+
+ testStream(df)(
+ // Base-only split: the compacted shredded base file must round-trip
its variants.
+ AssertOnQuery { q => q.processAllAvailable(); true },
+ CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1\"}", 1000L), Row(2,
"{\"key\":\"v2\"}", 1000L)),
+ lastOnly = true, isSorted = false),
+ StopStream,
+
+ // Merged split: the update lands in a log file on the compacted
slice, so the next
+ // batch merges the shredded base with the log.
Review Comment:
MergeOnReadIncrementalRelationV2 builds its file-system view from
affectedFilesInCommits alone, so the second batch sees only the appended log
file and gets a log-only split rather than base plus log. The merged branch is
not covered here - worth reflecting that in the comment, or covering it another
way.
##########
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:
fullVariantReadSchemaWithOrdinals asks the adapter and never the file, so
every unshredded CDC read on Spark 4.1+ now also takes the rewritten shape plus
a restore projection, and no test covers that. Add an unshredded twin here, the
way the clustering and bulk_insert tests already have one.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -148,12 +148,19 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
}
}
+ // The plain skip-merging reader cannot read a SHREDDED variant base file:
it requests native
+ // VariantType, which clips the shredded group to {metadata, value} and
reads value=null (the
+ // #19556 defect family). Splits with variant columns take the file-group
reader below, whose
+ // reader context requests the full-variant projection shape instead
(#19578).
+ private val requiredSchemaHasVariant: Boolean =
+ requiredSchema.structTypeSchema.fields.exists(f =>
sparkAdapter.isVariantType(f.dataType))
Review Comment:
isVariantType is also true on Spark 4.0, where buildFullVariantReadSchema
returns None, so a base-only split there loses the skip-merging reader for a
file-group read that does not apply the rewrite either. Gate on
`sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined`
so the routing tracks the rewrite that motivates it.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -381,6 +381,69 @@ class TestStreamingSource extends StreamTest {
testLegacyIncrementalStreamSource(MERGE_ON_READ, HoodieTableVersion.EIGHT)
}
+ test("test mor stream source reads shredded variant with legacy file group
reader disabled") {
+ // #19578: with the file group reader disabled, MOR streaming batches
materialize through
+ // HoodieMergeOnReadRDDV2, the only user-facing path reading shredded
variant base files
+ // without a catalyst schema. Covers both RDD branches: the base-only
split (first batch,
+ // right after inline compaction) and the merged split (second batch,
after a
+ // post-compaction update lands in a log file).
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withTempDir { inputDir =>
+ val tablePath =
s"${inputDir.getCanonicalPath}/test_mor_variant_legacy_stream"
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(MERGE_ON_READ)
+ .setTableName(getTableName(tablePath))
+ .setRecordKeyFields("id")
+ .setOrderingFields("ts")
+
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
+
+ // INMEMORY index routes MOR inserts to log files, so the first base
file is the
+ // compaction's SHREDDED one; compact = true trips inline compaction on
that write.
+ def addVariantData(valuesSql: String, compact: Boolean): Unit = {
+ spark.sql(valuesSql).write.format("org.apache.hudi")
+ .options(commonOptions)
+ .option(TBL_NAME.key, getTableName(tablePath))
+ .option(TABLE_TYPE.key, MERGE_ON_READ.name)
+ .option("hoodie.index.type", "INMEMORY")
+ .option("hoodie.parquet.variant.write.shredding.enabled", "true")
+ .option("hoodie.parquet.variant.force.shredding.schema.for.test",
"key string")
+ .option(HoodieCompactionConfig.INLINE_COMPACT.key, compact.toString)
+ .option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key,
"2")
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ }
+
+ addVariantData("""select 1 as id, parse_json('{"key":"v1"}') as v, 1000L
as ts""", compact = false)
+ addVariantData("""select 2 as id, parse_json('{"key":"v2"}') as v, 1000L
as ts""", compact = true)
+
+ val df = spark.readStream
+ .format("org.apache.hudi")
+ // force the legacy (non file-group-reader) incremental relation path
+ .option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key, "false")
+ .load(tablePath)
+ .selectExpr("id", "cast(v as string) as v", "ts")
+
+ testStream(df)(
Review Comment:
testLegacyIncrementalStreamSource asserts the executed plan is a `Scan
ExistingRDD` with no `FileScan`, precisely so a silent fallback to the
file-group reader is caught, and this test has no such guard. Add the same plan
assertion so it cannot go green while covering the already-fixed path.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -381,6 +381,69 @@ class TestStreamingSource extends StreamTest {
testLegacyIncrementalStreamSource(MERGE_ON_READ, HoodieTableVersion.EIGHT)
}
+ test("test mor stream source reads shredded variant with legacy file group
reader disabled") {
+ // #19578: with the file group reader disabled, MOR streaming batches
materialize through
+ // HoodieMergeOnReadRDDV2, the only user-facing path reading shredded
variant base files
+ // without a catalyst schema. Covers both RDD branches: the base-only
split (first batch,
+ // right after inline compaction) and the merged split (second batch,
after a
+ // post-compaction update lands in a log file).
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withTempDir { inputDir =>
+ val tablePath =
s"${inputDir.getCanonicalPath}/test_mor_variant_legacy_stream"
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(MERGE_ON_READ)
+ .setTableName(getTableName(tablePath))
+ .setRecordKeyFields("id")
+ .setOrderingFields("ts")
+
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
+
+ // INMEMORY index routes MOR inserts to log files, so the first base
file is the
+ // compaction's SHREDDED one; compact = true trips inline compaction on
that write.
+ def addVariantData(valuesSql: String, compact: Boolean): Unit = {
+ spark.sql(valuesSql).write.format("org.apache.hudi")
Review Comment:
This write fails in CI with `UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE`: the v1
DefaultSource rejects the `v` VARIANT column, so the test dies before the
streaming read. Create and populate the table through SQL the way
TestVariantDataType does.
##########
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:
This update fails in CI with `Null-value for required field: value` out of
FileGroupReaderBasedMergeHandle.close, so a single-row table does not remove
the #19582 prerequisite. Either land #19582 first or keep this test to the
insert leg.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala:
##########
@@ -90,6 +92,11 @@ class InternalRowToJsonStringConverter(schema: StructType) {
structMap.toMap
case _ => value // fallback
}
+ case dt if SparkAdapterSupport.sparkAdapter.isVariantType(dt) =>
+ // VariantVal.toString renders the variant as JSON; embed it as a
real JSON node so
+ // the image carries the variant's structure. Falling through to the
default would
+ // serialize the VariantVal bean, i.e. its raw value/metadata bytes
as base64.
+ mapper.readTree(value.toString)
Review Comment:
Variant.toJson renders a non-finite double as a bare NaN or Infinity token
and this ObjectMapper rejects those, so a variant column holding one turns a
CDC query that previously returned a base64 image into a hard failure. Worth
falling back to the raw string when readTree throws - follow-up, not a blocker.
--
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]
