voonhous commented on code in PR #19583:
URL: https://github.com/apache/hudi/pull/19583#discussion_r3765206680
##########
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:
Addressed, though the fix is smaller than a SQL rewrite: `supportsDataType`
is overridden only in `Spark4DefaultSource`, which is reachable just through
the registered short name, so `format("org.apache.hudi")` was resolving to
`DefaultSource` (`hudi_v1`) and losing the override. Writing with
`format("hudi")` is enough.
The `readStream` keeps the qualified name -- it goes through
`StreamSourceProvider`, which never calls `supportsDataType`. (For what it's
worth, `TestVariantDataType` does not create everything through SQL either; its
DataFrame-writer tests use `format("hudi")`.)
##########
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:
Addressed -- the same guard is now hoisted into a local
`assertLegacyRddPlan` step and asserted in both streams.
##########
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:
Addressed. The second batch is log-only, as described:
`MergeOnReadIncrementalRelationV2` builds its view from
`affectedFilesInCommits` alone and that span covers only the update
deltacommit, so `getLatestMergedFileSlicesBeforeOrOn` has no base file to merge
in.
Fixed the comment and added genuine merged coverage: a second `testStream`
gets its own checkpoint, so it replays from INIT and its first batch spans the
compaction commit and the update together, which is the only way to land a base
+ log slice on this path.
--
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]
