wombatu-kun commented on code in PR #19783:
URL: https://github.com/apache/hudi/pull/19783#discussion_r3888224040
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -93,28 +93,41 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
private lazy val allFilters = filters ++ requiredFilters
// For each field of `target`, replace its dataType with the matching
field's projected
- // variant struct from `source` (when present). Non-matching fields pass
through. Why a parallel
- // `sparkRequiredSchema` overlay exists at all is documented on that
constructor parameter.
+ // variant struct from `source` (when present), recursing into struct
members so a variant
+ // reached through a struct path is overlaid too. Fields are matched by name
(findFieldByName);
+ // non-matching fields pass through. The recursion mirrors
PushVariantIntoScan's
+ // VariantInRelation.rewriteType, which rewrites variants at the root of the
relation output
+ // and below STRUCT paths only, so an array element or a map value is never
overlaid here
+ // either (#19775). Why a parallel `sparkRequiredSchema` overlay exists at
all is documented on
+ // that constructor parameter.
private def overlayVariantProjections(target: StructType, source:
StructType): StructType = {
StructType(target.fields.map { f =>
- SparkFileFormatInternalRowReaderContext.findFieldByName(source,
f.name).map(_.dataType) match {
- case Some(projStruct: StructType) if
sparkAdapter.isVariantProjectionStruct(projStruct) =>
+ (f.dataType,
SparkFileFormatInternalRowReaderContext.findFieldByName(source,
f.name).map(_.dataType)) match {
+ case (_, Some(projStruct: StructType)) if
sparkAdapter.isVariantProjectionStruct(projStruct) =>
f.copy(dataType = projStruct)
+ case (targetStruct: StructType, Some(sourceStruct: StructType)) =>
Review Comment:
`SparkSchemaTransformUtils.addMissingFields` has no variant-projection case,
so once any sibling member of the enclosing struct triggers an implicit type
change the nested projection installed here is replaced by the file's
`VariantType` in the read schema. Should it get the same special case
`isDataTypeEqualForPhysicalSchema` already has, or is implicit evolution
alongside a nested variant out of scope for this PR?
##########
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:
A native parquet log file is named
`<fileId>_<token>_<instant>_<v>.log.parquet` with no dot prefix, so this
listing also picks up the two deltacommit logs and "expected a compacted base
file" is satisfied even if inline compaction never ran. Filter with
`FSUtils.isBaseFile` the way `assertNestedBaseLayout` does.
##########
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:
TestStreamingSource's table also carries a top-level `v`, so
`shouldRerouteVariantSplit` is true there and the base-only split never reaches
`requiredSchemaReaderSkipMerging` - that leg is not what the test pins. Drop
the reference here and in the matching "Not swept here" note in
TestVariantShreddingMixedLayouts, or add a nested-only leg that actually takes
the branch?
##########
hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala:
##########
@@ -320,6 +324,89 @@ abstract class BaseSpark4Adapter extends SparkAdapter with
Logging {
if (rewritten) Some(StructType(fields)) else None
}
+ /**
+ * Shared implementation behind [[SparkAdapter#buildVariantProjector]] for
the 4.x adapters
+ * whose planner rewrites variants into projection structs (4.1+).
+ *
+ * Recurses into struct members, mirroring PushVariantIntoScan's
`VariantInRelation.rewriteType`:
+ * a variant is rewritten at the root of the relation output or below a
STRUCT path, while
+ * arrays and maps keep their native VariantType, so nothing under a
collection is projected
+ * here either. Before #19775 this walked top-level fields only, and a
projection struct sitting
+ * one struct member down was left holding a raw variant that the plan then
read as its
+ * projected children.
+ */
+ protected final def buildVariantProjectorForStructPaths(
+ sparkDataSchema: StructType,
+ sparkRequiredSchema: StructType): Option[InternalRow => InternalRow] = {
+ // Quick check: does any required field carry a variant projection struct,
at any depth?
+ if (!sparkRequiredSchema.fields.exists(f =>
containsVariantProjection(f.dataType))) {
+ None
+ } else {
+ // Surface mismatched schemas with both field lists rather than Spark's
bare
+ // IllegalArgumentException from fieldIndex. `path` is the dotted field
path of `name`.
+ def lookupDataField(dataStruct: StructType, requiredStruct: StructType,
+ name: String, path: String): (Int, StructField) = {
+ val idx = dataStruct.getFieldIndex(name).getOrElse(
+ throw new IllegalStateException(
+ s"Required field '$path' is absent from sparkDataSchema; " +
+ s"required=${requiredStruct.fieldNames.mkString("[", ",", "]")},
" +
+ s"data=${dataStruct.fieldNames.mkString("[", ",", "]")}"))
+ (idx, dataStruct.fields(idx))
+ }
+
+ // `ref` reads the data-schema value of type `dataType`; the result has
type `requiredType`.
+ def projectionExpr(ref: Expression, dataType: DataType, requiredType:
DataType,
+ path: String): Expression = requiredType match {
+ case projectedStruct: StructType if
VariantMetadata.isVariantStruct(projectedStruct) =>
+ require(isVariantType(dataType),
+ s"Expected VariantType for field '$path' in data schema, got
$dataType")
+ val childExprs: Seq[Expression] =
projectedStruct.fields.toSeq.flatMap { child =>
+ val vm = VariantMetadata.fromMetadata(child.metadata)
+ val pathLit = Literal(UTF8String.fromString(vm.path), StringType)
+ val variantGet: Expression =
+ VariantGet(ref, pathLit, child.dataType, vm.failOnError,
Option(vm.timeZoneId))
+ Seq(Literal(UTF8String.fromString(child.name), StringType),
variantGet)
+ }
+ CreateNamedStruct(childExprs)
+ case requiredStruct: StructType =>
+ dataType match {
+ // Rebuild the struct member by member only when something below
it is projected;
+ // otherwise the reference is already in the required shape and is
cheaper untouched.
+ case dataStruct: StructType if
containsVariantProjection(requiredStruct) =>
+ val childExprs: Seq[Expression] =
requiredStruct.fields.toSeq.flatMap { rf =>
+ val childPath = s"$path.${rf.name}"
+ val (childIdx, childField) = lookupDataField(dataStruct,
requiredStruct, rf.name, childPath)
+ val childRef = GetStructField(ref, childIdx, Some(rf.name))
+ Seq(Literal(UTF8String.fromString(rf.name), StringType),
+ projectionExpr(childRef, childField.dataType, rf.dataType,
childPath))
+ }
+ val rebuilt = CreateNamedStruct(childExprs)
+ // CreateNamedStruct is never null, so a null struct would come
back as a struct of
+ // nulls without this guard.
+ If(IsNull(ref), Literal(null, rebuilt.dataType), rebuilt)
+ case _ => ref
+ }
+ case _ => ref
+ }
+
+ val exprs: Array[Expression] = sparkRequiredSchema.fields.map { rf =>
+ val (dataIdx, dataField) = lookupDataField(sparkDataSchema,
sparkRequiredSchema, rf.name, rf.name)
+ val ref: Expression = BoundReference(dataIdx, dataField.dataType,
dataField.nullable)
+ projectionExpr(ref, dataField.dataType, rf.dataType, rf.name)
+ }
+
+ val projection = UnsafeProjection.create(exprs.toIndexedSeq,
DataTypeUtils.toAttributes(sparkDataSchema))
+ Some(row => projection(row))
+ }
+ }
+
+ /** True when `dataType` is a variant projection struct or holds one below a
struct path. */
+ private def containsVariantProjection(dataType: DataType): Boolean =
dataType match {
Review Comment:
`containsVariantProjection` is now defined identically here and in
`SparkFileFormatInternalRowReaderContext`. `SparkAdapter` already hosts
`isVariantProjectionStruct`, so a default method there would give both call
sites one definition.
##########
hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala:
##########
@@ -320,6 +324,89 @@ abstract class BaseSpark4Adapter extends SparkAdapter with
Logging {
if (rewritten) Some(StructType(fields)) else None
}
+ /**
+ * Shared implementation behind [[SparkAdapter#buildVariantProjector]] for
the 4.x adapters
+ * whose planner rewrites variants into projection structs (4.1+).
+ *
+ * Recurses into struct members, mirroring PushVariantIntoScan's
`VariantInRelation.rewriteType`:
+ * a variant is rewritten at the root of the relation output or below a
STRUCT path, while
+ * arrays and maps keep their native VariantType, so nothing under a
collection is projected
+ * here either. Before #19775 this walked top-level fields only, and a
projection struct sitting
+ * one struct member down was left holding a raw variant that the plan then
read as its
+ * projected children.
+ */
+ protected final def buildVariantProjectorForStructPaths(
+ sparkDataSchema: StructType,
+ sparkRequiredSchema: StructType): Option[InternalRow => InternalRow] = {
+ // Quick check: does any required field carry a variant projection struct,
at any depth?
+ if (!sparkRequiredSchema.fields.exists(f =>
containsVariantProjection(f.dataType))) {
+ None
+ } else {
+ // Surface mismatched schemas with both field lists rather than Spark's
bare
+ // IllegalArgumentException from fieldIndex. `path` is the dotted field
path of `name`.
+ def lookupDataField(dataStruct: StructType, requiredStruct: StructType,
+ name: String, path: String): (Int, StructField) = {
+ val idx = dataStruct.getFieldIndex(name).getOrElse(
+ throw new IllegalStateException(
+ s"Required field '$path' is absent from sparkDataSchema; " +
+ s"required=${requiredStruct.fieldNames.mkString("[", ",", "]")},
" +
+ s"data=${dataStruct.fieldNames.mkString("[", ",", "]")}"))
+ (idx, dataStruct.fields(idx))
+ }
+
+ // `ref` reads the data-schema value of type `dataType`; the result has
type `requiredType`.
+ def projectionExpr(ref: Expression, dataType: DataType, requiredType:
DataType,
+ path: String): Expression = requiredType match {
+ case projectedStruct: StructType if
VariantMetadata.isVariantStruct(projectedStruct) =>
+ require(isVariantType(dataType),
+ s"Expected VariantType for field '$path' in data schema, got
$dataType")
+ val childExprs: Seq[Expression] =
projectedStruct.fields.toSeq.flatMap { child =>
+ val vm = VariantMetadata.fromMetadata(child.metadata)
+ val pathLit = Literal(UTF8String.fromString(vm.path), StringType)
+ val variantGet: Expression =
+ VariantGet(ref, pathLit, child.dataType, vm.failOnError,
Option(vm.timeZoneId))
+ Seq(Literal(UTF8String.fromString(child.name), StringType),
variantGet)
+ }
+ CreateNamedStruct(childExprs)
Review Comment:
A null variant in an avro log record is projected into a non-null struct of
nulls here, while the parquet paths leave the field null - and
`PushVariantIntoScan` rewrites `IsNull(v)` / `IsNotNull(v)` directly onto that
struct. Wrap this in the same `If(IsNull(...))` guard the struct rebuild below
already carries?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala:
##########
@@ -751,35 +806,519 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
assert(getFieldAsGroup(innerGroup, "typed_value").containsField("k"),
s"[$leg] nested typed_value should carry k:\n$innerGroup")
- // #18605 history: the batch-disabling guards in
HoodieFileGroupReaderBasedFileFormat are
- // top-level-only, so a NESTED variant still reaches the vectorized
reader. The session conf
- // does pick the reader -
HoodieFileGroupReaderBasedFileFormat.supportBatch reads
- // sparkSession.sessionState.conf and
ParquetUtils.isBatchReadSupportedForSchema gates on
- // spark.sql.parquet.enableVectorizedReader, and only afterwards does
- // buildReaderWithPartitionValues write that decision back into the conf
- so sweep it and
- // pin both readers. Every nested-variant read bug so far (HUDI-7190,
HUDI-8803, #18605) is
- // vectorized-only, which leaves the row-based leg as the control.
- Seq("true", "false").foreach { vectorizedReader =>
- withSQLConf("spark.sql.parquet.enableVectorizedReader" ->
vectorizedReader) {
- checkAnswer(s"select id, cast(s.inner as string) from $tableName")(
- Seq(1, """{"k":"x1"}""")
- )
- }
- }
+ // One read, whatever spark.sql.parquet.enableVectorizedReader says:
supportBatch vetoes
+ // batch reads for a variant at any depth before that conf is consulted,
and section F2 pins
+ // that decision directly.
+ checkAnswer(s"select id, cast(s.inner as string) from $tableName")(
+ Seq(1, """{"k":"x1"}""")
+ )
// A plain insert bin-packs into the same file group: the small-file
merge must read the
// nested-shredded base back (nested reconstruction on the AVRO record
type leg).
withWriteLayout(Forced("k string")) {
spark.sql(s"""insert into $tableName values (2, named_struct('inner',
parse_json('{"k":"x2"}')), 1000)""")
}
assertSingleFileGroup(tablePath, leg)
+ // The merge rewrote the whole file group under the INCOMING layout, so
the layout below is
+ // the record writer's, not the row writer's seed. On the AVRO leg that
write goes through
+ // HoodieAvroWriteSupport, whose forced hook now reaches the nested
member; before the #19689
+ // fix it silently rewrote s.inner unshredded and nothing here noticed.
It pins the
+ // small-file MERGE handle's write, while the record-writer test below
pins a fresh INSERT
+ // handle, so the two are not interchangeable.
+ assertVariantLayout(tablePath, shredded = true, leg, column = "s.inner")
checkAnswer(s"select id, cast(s.inner as string) from $tableName order
by id")(
Seq(1, """{"k":"x1"}"""),
Seq(2, """{"k":"x2"}""")
)
}
}
+ test("Both record types force-shred a nested variant through the record
writer") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // A plain insert, so the write goes through the record writers rather
than the row writer: the
+ // AVRO leg is HoodieAvroWriteSupport end to end, the SPARK leg
HoodieRowParquetWriteSupport,
+ // and the #19689 fix is what made their forced hooks agree. The DDL
reaches a variant that is
+ // a record MEMBER at any depth - s.inner, and the inner of the struct
element of items - but
+ // never one that is directly a collection element, which is why arr stays
unshredded on both
+ // paths.
+ withVariantTable("record-writer nested forced shredding", "cow",
+ extraCols = "s struct<inner: variant>, items array<struct<inner:
variant>>, arr array<variant>") {
+ (tableName, tablePath, leg) =>
+ withWriteLayout(Forced("k string")) {
+ spark.sql(
+ s"""insert into $tableName values (1, parse_json('{"k":"top"}'),
+ | named_struct('inner', parse_json('{"k":"nested"}')),
+ | array(named_struct('inner', parse_json('{"k":"element"}'))),
+ | array(parse_json('{"k":"bare"}')), 1000)""".stripMargin)
+ }
+
+ val files = listDataParquetFiles(tablePath)
+ assert(files.size == 1, s"[$leg] expected one base file, got $files")
+ Seq("s.inner", "items.inner").foreach { column =>
+ assertVariantLayout(tablePath, shredded = true, leg, column = column)
+ val group = variantGroupOf(files.head, column)
+ assert(getFieldAsGroup(group, "typed_value").containsField("k"),
+ s"[$leg] typed_value of $column should carry k:\n$group")
+ }
+ assertVariantLayout(tablePath, shredded = false, leg, column = "arr")
+
+ checkAnswer(s"select id, cast(v as string), cast(s.inner as string), " +
+ s"cast(items[0].inner as string), cast(arr[0] as string) from
$tableName")(
+ Seq(1, """{"k":"top"}""", """{"k":"nested"}""", """{"k":"element"}""",
"""{"k":"bare"}""")
+ )
+ }
+ }
+
+ test("MOR merge, compaction and clustering carry a nested-shredded base
through the internal reader") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // The nested variant is the ONLY variant in the table, so no top-level
rewrite carries it (see
+ // withNestedOnlyVariantTable): every read below has to resolve s.inner on
its own - through the
+ // nested projection struct PushVariantIntoScan pushes into the scan for
the user queries, and
+ // natively for the internal reads that have no catalyst schema
(compaction, clustering, the
+ // legacy RDD) - which is what makes these paths say anything about nested
shredding at all.
+ // The projected arm crashed the JVM before #19775 (the projection was
applied at the top level
+ // only, so the merged row still held a raw variant where the plan read a
struct).
+ val mergedRows = Seq(
+ Seq(1, """{"k":"n1"}"""), Seq(2, """{"k":"n2b"}"""), Seq(3,
"""{"k":"n3"}"""))
+ val finalRows = Seq(
+ Seq(1, """{"k":"n1"}"""), Seq(2, """{"k":"n2b"}"""), Seq(3,
"""{"k":"n3c"}"""))
+
+ Seq(true, false).foreach { rowWriter =>
+ // No INMEMORY index: the first insert creates a base file, the updates
go to log files.
+ withNestedOnlyVariantTable(s"mor nested rowWriter=$rowWriter", "mor",
+ props = Seq("hoodie.compact.inline = 'false'"),
+ recordTypes = clusteringRecordTypes(rowWriter)) { (tableName,
tablePath, leg) =>
+ val snapshotQuery = s"select id, cast(s.inner as string) from
$tableName order by id"
+ val readOptimizedQuery = s"select id, cast(s.inner as string) from " +
+ s"hudi_query('$tableName', 'read_optimized') order by id"
+
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName values " +
+ """(1, named_struct('inner', parse_json('{"k":"n1"}')), 1000), """
+
+ """(2, named_struct('inner', parse_json('{"k":"n2"}')), 1000), """
+
+ """(3, named_struct('inner', parse_json('{"k":"n3"}')), 1000)""")
+ }
+ val baseFiles = listDataParquetFiles(tablePath)
+ assert(baseFiles.size == 1, s"[$leg] expected one base file, got
$baseFiles")
+ assertVariantLayout(tablePath, shredded = true, leg, column =
"s.inner")
+ val innerGroup = variantGroupOf(baseFiles.head, "s.inner")
+ assert(getFieldAsGroup(innerGroup, "typed_value").containsField("k"),
+ s"[$leg] nested typed_value should carry k:\n$innerGroup")
+
+ // A nested-shredded native log on top of the nested-shredded base.
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"update $tableName set " +
+ """s = named_struct('inner', parse_json('{"k":"n2b"}')), ts = 1001
where id = 2""")
+ }
+
assert(listDataParquetFiles(tablePath).exists(_.endsWith(".log.parquet")),
+ s"[$leg] the update should have written a native parquet log file")
+
+ // The conf only says what the reader MAY do - supportBatch vetoes
vectorization for a
+ // variant at any depth (pinned in F2) - so the sweep is the control
if that guard is ever
+ // narrowed back to top-level columns.
+ Seq("true", "false").foreach { vectorizedReader =>
+ withSQLConf("spark.sql.parquet.enableVectorizedReader" ->
vectorizedReader) {
+ checkAnswer(snapshotQuery)(mergedRows: _*)
+ }
+ }
+
+ // Not swept here: hoodie.file.group.reader.enabled=false, which no
longer routes a batch
+ // read anywhere (only the streaming sources consult it). The legacy
RDD path over a
+ // nested-shredded base - HoodieMergeOnReadRDDV2, whose
shouldRerouteVariantSplit stays
+ // false without a top-level variant - is pinned by
TestStreamingSource's legacy leg.
+
+ // Read-optimized serves the base file alone: id 2 is still the
pre-update value.
+ checkAnswer(readOptimizedQuery)(
+ Seq(1, """{"k":"n1"}"""), Seq(2, """{"k":"n2"}"""), Seq(3,
"""{"k":"n3"}"""))
+
+ // Compaction merges the nested-shredded log onto the nested-shredded
base and re-derives
+ // the layout from the forced DDL. On the AVRO record type that write
goes through
+ // HoodieAvroWriteSupport, whose nested forced hook is #19689's parity
fix.
+ withWriteLayout(Forced("k string")) {
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 1, leg)
+ assertNestedBaseLayout(tablePath, latestCompletedInstant(tablePath),
shredded = true, leg)
+ checkAnswer(snapshotQuery)(mergedRows: _*)
+
+ // Unshredded round: the update and the compaction both run with
shredding off, so
+ // typed_value has to be stripped at depth on the way out.
+ withWriteLayout(Unshredded) {
+ spark.sql(s"update $tableName set " +
+ """s = named_struct('inner', parse_json('{"k":"n3c"}')), ts = 1002
where id = 3""")
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 2, leg)
+ assertNestedBaseLayout(tablePath, latestCompletedInstant(tablePath),
shredded = false, leg)
+ checkAnswer(snapshotQuery)(finalRows: _*)
+
+ // Clustering re-derives the nested layout from the forced DDL over
that unshredded input:
+ // the row-writer path when rowWriter is true, the record writers
otherwise.
+ withWriteLayout(Forced("k string")) {
+ runClustering(tableName, rowWriter)
+ }
+ val clusteringInstant = completedClusteringInstant(tablePath, leg)
+ assertNestedBaseLayout(tablePath, clusteringInstant, shredded = true,
leg)
+ checkAnswer(snapshotQuery)(finalRows: _*)
+ checkAnswer(readOptimizedQuery)(finalRows: _*)
+ }
+ }
+ }
+
+ test("CDC images carry a nested-shredded variant") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // The nested twin of TestVariantDataType's CDC test. OP_KEY_ONLY
reconstructs both images by
+ // reading the 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, which
reads the new base
+ // file directly rather than through the reader context.
+ Seq("OP_KEY_ONLY", "DATA_BEFORE_AFTER").foreach { loggingMode =>
+ // SPARK pinned: a cdc-enabled table always writes through
FileGroupReaderBasedMergeHandle and
+ // the merger's record type picks that handle's reader context; AVRO
would route the update's
+ // base-file read through HoodieAvroParquetReader, the separate defect
tracked as #19567.
+ withNestedOnlyVariantTable(s"cdc nested $loggingMode", "cow", props =
Seq(
+ "'hoodie.table.cdc.enabled' = 'true'",
+ s"'hoodie.table.cdc.supplemental.logging.mode' = '$loggingMode'",
+ "hoodie.index.type = 'INMEMORY'"),
+ recordTypes = Seq(HoodieRecordType.SPARK)) { (tableName, tablePath,
leg) =>
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName values " +
+ """(1, named_struct('inner', parse_json('{"k":"c1"}')), 1000)""")
+ }
+ assertVariantLayout(tablePath, shredded = true, leg, column =
"s.inner")
+
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"update $tableName set " +
+ """s = named_struct('inner', parse_json('{"k":"c2"}')), ts = 1001
where id = 1""")
+ }
+ // Layout flip: the second update rewrites the file unshredded, so the
images below span
+ // both physical slots.
+ withWriteLayout(Unshredded) {
+ spark.sql(s"update $tableName set " +
+ """s = named_struct('inner', parse_json('{"k":"c3"}')), ts = 1002
where id = 1""")
+ }
+ assertNestedBaseLayout(tablePath, latestCompletedInstant(tablePath),
shredded = false, leg)
+
+ val cdc = spark.sql(s"select op, get_json_object(before,
'$$.s.inner.k') as before_k, " +
+ s"get_json_object(after, '$$.s.inner.k') as after_k " +
+ s"from hudi_table_changes('$tableName', 'cdc', 'earliest')")
+ val insertRows = cdc.where("op = 'i'").collect()
+ assert(insertRows.length == 1, s"[$leg] expected exactly one insert
cdc row")
+ assert(insertRows(0).getString(2) == "c1",
+ s"[$leg] insert after-image lost the nested variant payload:
${insertRows(0)}")
+
+ val updateRows = cdc.where("op = 'u'").orderBy("after_k").collect()
+ assert(updateRows.length == 2, s"[$leg] expected two update cdc rows")
+ assert(updateRows(0).getString(1) == "c1" &&
updateRows(0).getString(2) == "c2",
+ s"[$leg] first update images lost the nested variant payload:
${updateRows(0)}")
+ assert(updateRows(1).getString(1) == "c2" &&
updateRows(1).getString(2) == "c3",
+ s"[$leg] second (layout-flipped) update images lost the nested
variant payload: ${updateRows(1)}")
+ }
+ }
+ }
+
+ test("variant_get projections and filters resolve a nested-shredded
variant") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // The nested twin of the mixed-layout variant_get test above; SPARK
pinned for the same reason.
+ // pushVariantIntoScan is swept because the two arms reach the file
differently: on, Spark
+ // rewrites the s.inner struct path into its own projection struct and
pushes it into the scan;
+ // off, the whole variant is read and variant_get evaluates on top of it.
+ def nestedRowsSql(lo: Int, hi: Int): String =
+ s"""select cast(id as int) as id,
+ | named_struct('inner', parse_json(concat('{"k":"x', id, '"}'))) as s,
+ | 1000L as ts from range($lo, $hi, 1, 1)""".stripMargin
+
+ Seq("true", "false").foreach { pushIntoScan =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
Review Comment:
Nothing in these legs asserts that `PushVariantIntoScan` actually fired, and
both arms expect the same rows, so if the rule ever stops matching
`HoodieFileGroupReaderBasedFileFormat` the projected arm silently becomes a
duplicate of the other one. Worth pinning the plan in the true arm the way
`assertLegacyRddPlan` pins its own path in TestStreamingSource - 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]
