voonhous commented on code in PR #19783:
URL: https://github.com/apache/hudi/pull/19783#discussion_r3893261852
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -149,13 +149,14 @@ 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). Such splits take the file-group reader below,
whose reader context
- // requests the full-variant projection shape instead (#19578). Keyed off
the adapter building
- // that shape rather than the mere presence of a variant column: it is None
below Spark 4.1,
- // where the file-group reader would read the same nulls, so re-routing
there would cost the
- // fast path for nothing.
+ // A split whose required schema has a top-level variant column takes the
file-group reader
+ // below, whose reader context requests the full-variant projection shape
for parquet base
+ // files (#19578), so a SHREDDED base file is read on this legacy path
through the same
+ // contract as everywhere else. The skip-merging reader's native VariantType
request is
+ // reconstructed by the Spark 4.1+ row reader as well (pinned by
TestStreamingSource), so this
Review Comment:
Added the nested-only leg rather than dropping the claim.
`TestStreamingSource` now runs the legacy shredded-variant stream twice, with
and without the top-level `v`. Without it `shouldRerouteVariantSplit` is false,
so the base-only split goes through `requiredSchemaReaderSkipMerging` and
Spark's own row reader rebuilds `s.inner` from the shredded group; the footer
pin asserts `s.inner` is shredded and that the file carries no `v`. This
comment and the "Not swept here" note in `TestVariantShreddingMixedLayouts` now
name that leg.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -441,29 +446,52 @@ class TestStreamingSource extends StreamTest {
true
}
- 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)
+ addVariantData("""select 1 as id, parse_json('{"key":"v1"}') as v,
+ | named_struct('inner', parse_json('{"key":"n1"}')) as
s, 1000L as ts""".stripMargin, compact = false)
+ addVariantData("""select 2 as id, parse_json('{"key":"v2"}') as v,
+ | named_struct('inner', parse_json('{"key":"n2"}')) as
s, 1000L as ts""".stripMargin, compact = true)
+ // Pin that the compacted base file is shredded at both depths: without
it the streams below
+ // would pass just the same over an unshredded base and pin nothing
about shredded reads.
+ val conf = spark.sessionState.newHadoopConf()
+ val baseFiles = new Path(tablePath).getFileSystem(conf).listStatus(new
Path(tablePath))
Review Comment:
Done: the listing filters with `FSUtils.isBaseFile`, and the message says
log files are excluded.
--
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]
