voonhous commented on code in PR #19783:
URL: https://github.com/apache/hudi/pull/19783#discussion_r3893264891
##########
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:
Added `variantProjectionPushedIntoScan(sql)` to
`VariantShreddingTestSupport`: it collects the `FileSourceScanExec` off
`sparkPlan` (AQE hides it on `executedPlan`) and asks whether its
`requiredSchema` carries a projection struct, through the new
`SparkAdapter.containsVariantProjection`. The COW loop and both MOR legs assert
it equals their `pushVariantIntoScan` arm, and the new widening leg asserts it
stays pushed.
##########
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:
Fixed: the projection-struct arm carries the same `If(IsNull(ref), null,
...)` guard as the struct rebuild below it. Pinned twice: the projector unit
test feeds a null variant inside a live struct and expects a null struct with
the sibling intact, and the nested MOR legs now null id 4's variant through the
log and check `where s.inner is null` / `is not null`; the avro-block leg with
the rule on returned no row for `is null` before the fix.
--
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]
