This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c59987a024cd fix(variant): project PushVariantIntoScan struct paths in
the ... (#19783)
c59987a024cd is described below
commit c59987a024cd021cbb9aa34bdc74d341ea834b4d
Author: voonhous <[email protected]>
AuthorDate: Tue Sep 1 18:59:39 2026 +0800
fix(variant): project PushVariantIntoScan struct paths in the ... (#19783)
Spark 4.1+ rewrites a variant reached through a struct path into a
projection struct nested in the scan schema, e.g. s: struct<inner:
struct<"0": string>>. SparkFileFormatInternalRowReaderContext overlaid
and projected that struct for top-level fields only, so on a MOR table
with log files the merged row still held a raw VariantVal at s.inner
while the plan read it as the projected struct: SIGBUS, InternalError,
OOM or silent nulls, on any log format and record type. COW was fine.
- SparkFileFormatInternalRowReaderContext: the projection overlay and
its detection recurse into struct members, mirroring Spark's
VariantInRelation.rewriteType; arrays and maps stay native.
- BaseSpark4Adapter: buildVariantProjector hoisted from the identical
4.1/4.2 copies and made recursive. A null variant in an avro log
record projects to a NULL struct, not a struct of nulls, since
PushVariantIntoScan rewrites IsNull/IsNotNull onto that struct.
- SparkSchemaTransformUtils.addMissingFields: a projection struct
requested over a file VariantType stays the reader type when a
sibling member has an implicit type change; it used to be folded
back to VariantType, and the type-change Cast then cast the variant
value to the projected struct.
- SparkAdapter.containsVariantProjection: one definition for the
reader context and the adapter.
- The deferral comments claiming a native VariantType request clips a
shredded file to {metadata, value} now state the verified mechanism:
the Spark 4.1+ parquet reader reconstructs a shredded variant at any
depth. The full-variant rewrite stays top-level as a contract.
Tests, over tables whose only variant is the nested one.
TestVariantShreddingMixedLayouts covers MOR merge, read-optimized,
compaction and clustering on both writer paths, CDC images, nested
variant_get and cast on COW and on MOR over native parquet logs and
avro blocks with pushVariantIntoScan on and off (each arm pins whether
the scan carries the projection struct), a null variant through the
log, implicit widening of a sibling, and an array<variant> element
shredded through a declared write schema.
TestBaseSpark4AdapterVariantMethods covers the projector and the
implicit-change reconciliation as units. TestStreamingSource runs the
legacy-RDD leg with and without a top-level variant beside the nested
one.
Closes #19775. Part of #18937.
---
.../SparkFileFormatInternalRowReaderContext.scala | 63 +--
.../datasources/SparkSchemaTransformUtils.scala | 13 +
.../org/apache/spark/sql/hudi/SparkAdapter.scala | 42 +-
.../org/apache/hudi/HoodieMergeOnReadRDDV2.scala | 19 +-
.../org/apache/hudi/cdc/CDCFileGroupIterator.scala | 12 +-
.../hudi/functional/TestStreamingSource.scala | 115 ++++-
.../TestBaseSpark4AdapterVariantMethods.scala | 109 ++++-
.../sql/hudi/dml/schema/TestVariantDataType.scala | 31 +-
.../schema/TestVariantShreddingMixedLayouts.scala | 530 ++++++++++++++++++++-
.../dml/schema/VariantShreddingTestSupport.scala | 121 ++++-
.../spark/sql/adapter/BaseSpark4Adapter.scala | 94 +++-
.../apache/spark/sql/adapter/Spark4_1Adapter.scala | 48 +-
.../apache/spark/sql/adapter/Spark4_2Adapter.scala | 48 +-
13 files changed, 1032 insertions(+), 213 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
index 1bd90bd86c61..8e42dd9831fc 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
@@ -93,13 +93,20 @@ 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)) =>
+ f.copy(dataType = overlayVariantProjections(targetStruct,
sourceStruct))
case _ => f
}
})
@@ -111,10 +118,8 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
// VariantType, so a row already rewritten into the projected struct shape
would be mis-decoded.
// Single source of truth for both reader paths (parquet native projection +
avro rewrite).
private def shouldProjectVariants(): Boolean = {
- val hasVariantProjection =
sparkRequiredSchema.exists(_.fields.exists(_.dataType match {
- case st: StructType => sparkAdapter.isVariantProjectionStruct(st)
- case _ => false
- }))
+ val hasVariantProjection =
+ sparkRequiredSchema.exists(_.fields.exists(f =>
sparkAdapter.containsVariantProjection(f.dataType)))
// getRecordMerger() is a Lombok getter over a field initialized to null
(not Option.empty());
// it stays null until HoodieReaderContext.initRecordMerger runs
(HoodieFileGroupReader calls it
// from its constructor), so the null guard is required.
@@ -187,22 +192,25 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
}
// Internal reads have no catalyst plan, so nothing rewrites VariantType
fields the way
- // PushVariantIntoScan does for user queries. Requesting native
VariantType against a
- // SHREDDED parquet base file clips the file group to {metadata, value}
and reads
- // value=null; write-side callers (compaction, clustering, merge) would
then persist the
- // nulls, silently losing the variant data (#19556). Query paths that
build this context
- // without sparkRequiredSchema hit the same null reads and take the same
rewrite: CDC
- // (CDCFileGroupIterator) under default configs, plus the legacy RDD paths
(streaming
- // source, MergeOnRead relations) only when
hoodie.file.group.reader.enabled=false;
- // batch incremental always scans through the file format with a catalyst
schema.
- // Request the full-variant projection shape
- // instead and restore native VariantType after the scan. User-facing
reads pass
- // sparkRequiredSchema and are overlaid above. Only top-level variant
fields are
- // rewritten: no production write path shreds a nested variant today (a
nested shredded
- // write schema needs the test-only force config until shredding-schema
inference
- // lands), so the nested leg is deferred until such files can exist. Spark
< 4.1 cannot
- // reconstruct shredded values on read (SPARK-54410) and the adapter
returns None;
- // Spark 4.0 does write shredded files, so its pre-existing read-side gap
stays as is.
+ // PushVariantIntoScan does for user queries. For parquet base files,
request the same
+ // whole-variant shape that rewrite would (one child "0" at path "$") for
each top-level
+ // variant column and restore native VariantType after the scan, so
write-side callers
+ // (compaction, clustering, merge) and the query paths that build this
context without
+ // sparkRequiredSchema (CDC under default configs; the streaming source
only with
+ // hoodie.file.group.reader.enabled=false) read a SHREDDED file through
the contract
+ // PushVariantIntoScan established (#19556, #19578). It is a contract, not
a workaround for
+ // the reader: on Spark 4.1+ a native VariantType request is reconstructed
from a shredded
+ // group at any depth (ParquetReadSupport.clipParquetType passes the leaf
group through,
+ // ParquetToSparkSchemaConverter carries the catalyst target into struct
members, list
+ // elements and map values, and ParquetRowConverter assembles it), which
is why nested
+ // variants are read natively here and are not rewritten (#19775). What
does read a
+ // shredded group as value=null is a request in the physical {metadata,
value} shape,
+ // which only the schema-on-read internal-schema branch produces; the
projection shape is
+ // what lets ParquetSchemaEvolutionUtils.validateNoShreddedVariants name
this route there
+ // instead of failing inside the read. User-facing reads pass
sparkRequiredSchema and are
+ // overlaid above. Spark < 4.1 cannot reconstruct shredded values on read
(SPARK-54410)
+ // and the adapter returns None; Spark 4.0 does write shredded files, so
its pre-existing
+ // read-side gap stays as is.
val isParquetBaseFile = !isInlineLog && !FSUtils.isLogFile(filePath) &&
HoodieFileFormat.fromFileExtension(filePath.getFileExtension) ==
HoodieFileFormat.PARQUET
val (readStructTypeForScan, variantOrdinals) =
@@ -509,9 +517,10 @@ object SparkFileFormatInternalRowReaderContext {
/**
* Rewrites top-level VariantType fields of `structType` into the
full-variant projection
* shape (see SparkAdapter.buildFullVariantReadSchema) and returns it with
the ordinals of
- * the rewritten fields, or None when nothing rewrites (no variant fields,
or no shredded
- * read support on this Spark version). Shared by this context and direct
base-file reads
- * that bypass it (CDCFileGroupIterator's BASE_FILE_INSERT case).
+ * the rewritten fields, or None when nothing rewrites (no top-level variant
field, or no
+ * shredded read support on this Spark version). Variants nested in structs,
arrays and maps
+ * stay native and are reconstructed by the reader as they are. Shared by
this context and
+ * direct base-file reads that bypass it (CDCFileGroupIterator's
BASE_FILE_INSERT case).
*/
private[hudi] def fullVariantReadSchemaWithOrdinals(structType: StructType):
Option[(StructType, Set[Int])] = {
SparkAdapterSupport.sparkAdapter.buildFullVariantReadSchema(structType).map {
rewritten =>
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
index c2676c710b9b..3150216c5e1e 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
@@ -426,6 +426,13 @@ object SparkSchemaTransformUtils {
case (ArrayType(rt, _), ArrayType(ft, _)) =>
ArrayType(addMissingFields(rt, ft))
case (MapType(requiredKey, requiredValue, _), MapType(fileKey, fileValue,
_)) =>
MapType(addMissingFields(requiredKey, fileKey),
addMissingFields(requiredValue, fileValue))
+ // A Spark 4.1 variant projection struct requested over a file VariantType
is the pair
+ // isDataTypeEqual declares equal through the adapter, and the reader has
to be handed the
+ // projection struct, not the file's VariantType: parquet-mr decodes into
the projected shape
+ // natively, while the type-change Cast applied afterwards would cast the
variant VALUE to that
+ // struct. Without this arm a type change on a SIBLING member of the
enclosing struct rewrote
+ // the projection back to VariantType (#19775).
+ case (requiredStruct: StructType, _) if
isVariantProjectionOverVariant(requiredStruct, fileType) => requiredStruct
case (StructType(requiredFields), StructType(fileFields)) =>
val fileFieldMap = fileFields.map(f => f.name -> f).toMap
StructType(requiredFields.map(f => {
@@ -436,4 +443,10 @@ object SparkSchemaTransformUtils {
}))
case _ => fileType
}
+
+ // Same Try as in isDataTypeEqual: the adapter module may be absent from the
classpath
+ // (hudi-spark-client tests), in which case there is no projection struct to
recognise anyway.
+ private def isVariantProjectionOverVariant(requiredType: StructType,
fileType: DataType): Boolean =
+ Try(HoodieSparkUtils.sparkAdapter.isVariantProjectionStruct(requiredType)
+ &&
HoodieSparkUtils.sparkAdapter.isVariantType(fileType)).getOrElse(false)
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
index 03c266c96666..be3966634165 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
@@ -514,11 +514,27 @@ trait SparkAdapter extends Serializable {
def isVariantProjectionStruct(structType: StructType): Boolean = false
/**
- * If `sparkRequiredSchema` contains any field that's a Spark 4.1 variant
projection struct
- * (i.e., the same-named field in `sparkDataSchema` is `VariantType`),
returns a row
- * transformer that takes an InternalRow in the data-schema shape (with full
variants) and
- * produces an InternalRow in the required-schema shape (with each variant
column projected
- * to its requested struct via VariantGet).
+ * True when `dataType` is a variant projection struct or holds one below a
struct path: the two
+ * places PushVariantIntoScan puts them (the root of the relation output and
STRUCT members), so
+ * an array element or a map value never matches. The reader context's
schema overlay and the
+ * adapter's row projector both key off this, and they have to agree on it.
+ */
+ def containsVariantProjection(dataType: DataType): Boolean = dataType match {
+ case st: StructType =>
+ isVariantProjectionStruct(st) || st.fields.exists(f =>
containsVariantProjection(f.dataType))
+ case _ => false
+ }
+
+ /**
+ * If `sparkRequiredSchema` contains any Spark 4.1 variant projection struct
(i.e., the
+ * same-named field in `sparkDataSchema` is `VariantType`), returns a row
transformer that
+ * takes an InternalRow in the data-schema shape (with full variants) and
produces an
+ * InternalRow in the required-schema shape (with each variant projected to
its requested
+ * struct via VariantGet).
+ *
+ * Projection structs are looked for at the root and below any STRUCT path,
the same places
+ * PushVariantIntoScan puts them; a variant that is an array element or a
map value keeps its
+ * native VariantType on both sides and is passed through (#19775).
*
* Used on the MOR log-file path: log records carry the full variant on
disk, but the merger
* expects rows aligned to the post-PushVariantIntoScan required schema.
Returns None when
@@ -530,14 +546,14 @@ trait SparkAdapter extends Serializable {
/**
* Rewrites each top-level VariantType field of `schema` into the
full-variant projection
* struct that PushVariantIntoScan would request for whole-variant access: a
single child
- * field "0" of VariantType carrying `VariantMetadata` for path "$".
Requesting that shape
- * makes the parquet reader reconstruct shredded variants by field name;
requesting native
- * VariantType instead clips a shredded file group down to {metadata, value}
and reads
- * value=null (#19556).
- *
- * Used by internal (non-catalyst) reads of parquet base files, which have no
- * PushVariantIntoScan to do this for them. The caller restores the native
VariantType
- * shape by projecting child 0 of each rewritten field.
+ * field "0" of VariantType carrying `VariantMetadata` for path "$".
Internal (non-catalyst)
+ * reads of parquet base files request that shape so they read a shredded
file through the
+ * same contract as a rewritten user query (#19556); the caller restores
native VariantType
+ * by projecting child 0 of each rewritten field. The shape is a contract,
not a reader
+ * requirement: on Spark 4.1+ the parquet reader reconstructs a shredded
variant for a native
+ * VariantType request too, at any depth, which is why only top-level fields
are rewritten
+ * and variants nested in structs, arrays and maps are read natively
(#19775). A request in
+ * the physical {metadata, value} struct shape is what reads a shredded
group as value=null.
*
* Returns None when the schema has no top-level VariantType field or the
Spark version has
* no shredded-read support (Spark 3.x / 4.0).
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
index 267b4e4389a6..9f619e158ab4 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
@@ -149,13 +149,18 @@ 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. Without a top-level variant the base-only
split stays on
+ // requiredSchemaReaderSkipMerging, whose native VariantType request the
Spark 4.1+ parquet
+ // reader reconstructs at any depth - the vectorized one at stock settings,
since the legacy
+ // file format inherits ParquetFileFormat.supportBatch and VariantType is
atomic; the variant
+ // veto lives in HoodieFileGroupReaderBasedFileFormat only (pinned by
TestStreamingSource's
+ // nested-only legacy leg), so this is about one contract, not a null read
(#19775). Keyed off
+ // the adapter building that shape
+ // rather than the mere presence of a variant column: it is None below Spark
4.1, where
+ // re-routing would cost the fast path for nothing.
private val shouldRerouteVariantSplit: Boolean =
sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala
index b845dbe38d35..a9e8091e9aa1 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala
@@ -378,12 +378,12 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
val pf = sparkPartitionedFileUtils.createPartitionedFile(
InternalRow.empty, absCDCPath, 0, fileStatus.getLength)
- // This read bypasses SparkFileFormatInternalRowReaderContext, so it
needs the same
- // full-variant treatment that context applies: requesting native
VariantType against
- // a SHREDDED base file clips the shredded group to {metadata,
value} and reads
- // value=null, which would surface as null variants in the insert
after-images
- // (#19556 family, #19578). The restore projection reuses one output
buffer, hence
- // the copy before buffering.
+ // This read bypasses SparkFileFormatInternalRowReaderContext, so it
makes the same
+ // whole-variant request that context makes for top-level variant
columns of a parquet
+ // base file and restores native VariantType afterwards (#19578); a
variant nested in a
+ // struct, array or map stays native and the reader reconstructs it,
shredded or not
+ // (#19775). The restore projection reuses one output buffer, hence
the copy before
+ // buffering.
val baseRows = SparkFileFormatInternalRowReaderContext
.fullVariantReadSchemaWithOrdinals(originTableSchema.structTypeSchema) match {
case Some((rewritten, ordinals)) =>
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala
index 3c943037f2e9..4fa1113b0261 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala
@@ -21,6 +21,7 @@ import org.apache.hudi.{DataSourceReadOptions,
HoodieSparkUtils}
import org.apache.hudi.DataSourceReadOptions.{START_OFFSET,
STREAMING_READ_TABLE_VERSION}
import org.apache.hudi.DataSourceWriteOptions.{ORDERING_FIELDS,
RECORDKEY_FIELD, TABLE_TYPE}
import org.apache.hudi.common.config.HoodieReaderConfig
+import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.common.model.HoodieTableType.{COPY_ON_WRITE,
MERGE_ON_READ}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion}
@@ -30,6 +31,9 @@ import
org.apache.hudi.config.HoodieWriteConfig.{DELETE_PARALLELISM_VALUE, INSER
import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.util.JavaConversions
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.util.HadoopInputFile
import org.apache.spark.sql.{DataFrame, Row, SaveMode}
import org.apache.spark.sql.streaming.StreamTest
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
@@ -381,16 +385,27 @@ 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. The first stream covers the base-only split
(first batch, the
- // branch this fix re-routes) and the log-only split (second batch); the
second stream covers
- // the merged base + log split.
+ /**
+ * #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. The first stream covers the base-only split
(first batch) and the
+ * log-only split (second batch); the second stream covers the merged base +
log split.
+ *
+ * The table carries a variant one struct member down, force-shredded
exactly like a top-level
+ * one. With a top-level `v` beside it the base-only split is re-routed to
the file group reader
+ * (HoodieMergeOnReadRDDV2.shouldRerouteVariantSplit), while the nested-only
leg keeps that split
+ * on requiredSchemaReaderSkipMerging - Spark's own parquet reader with a
native VariantType
+ * request one struct member down, which the Spark 4.1+ parquet reader
reconstructs out of the
+ * shredded group; the vectorized one at stock settings, as the legacy file
format inherits
+ * ParquetFileFormat.supportBatch and VariantType is atomic (#19775). That
second leg is the one
+ * HoodieMergeOnReadRDDV2's comment relies on.
+ */
+ private def testLegacyShreddedVariantStream(withTopLevelVariant: Boolean):
Unit = {
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"
+ val suffix = if (withTopLevelVariant) "" else "_nested_only"
+ val tablePath =
s"${inputDir.getCanonicalPath}/test_mor_variant_legacy_stream$suffix"
HoodieTableMetaClient.newTableBuilder()
.setTableType(MERGE_ON_READ)
.setTableName(getTableName(tablePath))
@@ -398,6 +413,20 @@ class TestStreamingSource extends StreamTest {
.setOrderingFields("ts")
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
+ // Whether the top-level `v` column exists at all is decided here and
nowhere else: on the
+ // nested-only leg the table is (id, s, ts) and every write, projection
and expected row
+ // below drops it.
+ def rowSql(id: Int, topLevelKey: String, nestedKey: String, ts: Long):
String = {
+ val topLevelCol = if (withTopLevelVariant)
s"""parse_json('{"key":"$topLevelKey"}') as v, """ else ""
+ s"""select $id as id, ${topLevelCol}named_struct('inner', """ +
+ s"""parse_json('{"key":"$nestedKey"}')) as s, ${ts}L as ts"""
+ }
+
+ def expectedRow(id: Int, topLevelKey: String, nestedKey: String, ts:
Long): Row = {
+ val topLevel = if (withTopLevelVariant)
Seq(s"""{"key":"$topLevelKey"}""") else Seq.empty
+ Row((Seq[Any](id) ++ topLevel ++ Seq(s"""{"key":"$nestedKey"}""",
ts)): _*)
+ }
+
// 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 = {
@@ -421,12 +450,15 @@ class TestStreamingSource extends StreamTest {
// The read keeps the fully qualified name: it goes through
StreamSourceProvider, which
// never calls supportsDataType.
- def variantStreamDf(): DataFrame = 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")
+ def variantStreamDf(): DataFrame = {
+ val topLevel = if (withTopLevelVariant) Seq("cast(v as string) as v")
else Seq.empty
+ 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((Seq("id") ++ topLevel ++ Seq("cast(s.inner as string)
as inner", "ts")): _*)
+ }
// The legacy branch of getBatch materializes the micro batch from an
RDD via
// internalCreateDataFrame, so the physical plan is a "Scan
ExistingRDD"; this fails if the
@@ -441,16 +473,46 @@ 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(rowSql(1, "v1", "n1", 1000L), compact = false)
+ addVariantData(rowSql(2, "v2", "n2", 1000L), compact = true)
+ // Pin that the compacted base file is shredded at every depth the leg
carries: without it
+ // the streams below would pass just the same over an unshredded base
and pin nothing about
+ // shredded reads. The listing goes through FSUtils.isBaseFile rather
than a ".parquet"
+ // suffix test, which a native parquet log file
(<fileId>_<token>_<instant>_<v>.log.parquet)
+ // also matches - so the assertion would hold even if inline compaction
never ran.
+ val conf = spark.sessionState.newHadoopConf()
+ val baseFiles = new Path(tablePath).getFileSystem(conf).listStatus(new
Path(tablePath))
+ .map(_.getPath).filter(p => FSUtils.isBaseFile(p.getName))
+ assertTrue(baseFiles.nonEmpty,
+ "expected a compacted BASE file under " + tablePath + " (log files
excluded)")
+ baseFiles.foreach { file =>
+ val reader = ParquetFileReader.open(HadoopInputFile.fromPath(file,
conf))
+ val footer = try reader.getFooter.getFileMetaData.getSchema finally
reader.close()
+ // getFieldIndex + getType(int): the String overload of getType is
ambiguous from Scala.
+ val s = footer.getType(footer.getFieldIndex("s")).asGroupType()
+ val inner = s.getType(s.getFieldIndex("inner")).asGroupType()
+ assertTrue(inner.containsField("typed_value"), "s.inner must be
shredded in " + file + ":\n" + footer)
+ if (withTopLevelVariant) {
+ val v = footer.getType(footer.getFieldIndex("v")).asGroupType()
+ assertTrue(v.containsField("typed_value"), "v must be shredded in "
+ file + ":\n" + footer)
+ } else {
+ // The absence of `v` is what keeps shouldRerouteVariantSplit false
on this leg.
+ assertTrue(!footer.containsField("v"),
+ "the nested-only leg must not carry a top-level v in " + file +
":\n" + footer)
+ }
+ }
testStream(variantStreamDf())(
// Base-only split: this batch spans both deltacommits and the
compaction commit, whose
- // affected files resolve to the compacted shredded base file with no
log on top. This is
- // the branch the fix re-routes to the file group reader.
+ // affected files resolve to the compacted shredded base file with no
log on top. With a
+ // top-level `v` this is the split shouldRerouteVariantSplit sends to
the file group
+ // reader; without one it stays on requiredSchemaReaderSkipMerging,
the leg that pins the
+ // nested-shredded base read by Spark's own parquet reader.
AssertOnQuery { q => q.processAllAvailable(); true },
assertLegacyRddPlan,
- CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1\"}", 1000L), Row(2,
"{\"key\":\"v2\"}", 1000L)),
+ CheckAnswerRows(Seq(
+ expectedRow(1, "v1", "n1", 1000L),
+ expectedRow(2, "v2", "n2", 1000L)),
lastOnly = true, isSorted = false),
StopStream,
@@ -458,12 +520,13 @@ class TestStreamingSource extends StreamTest {
// span's affected files alone, and this span covers only the update
deltacommit, so the
// slice is the appended log file with no base file.
AssertOnQuery { _ =>
- addVariantData("""select 1 as id, parse_json('{"key":"v1-updated"}')
as v, 1001L as ts""", compact = false)
+ addVariantData(rowSql(1, "v1-updated", "n1-updated", 1001L), compact
= false)
true
},
StartStream(),
AssertOnQuery { q => q.processAllAvailable(); true },
- CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1-updated\"}", 1001L)),
lastOnly = true, isSorted = false)
+ CheckAnswerRows(Seq(expectedRow(1, "v1-updated", "n1-updated", 1001L)),
+ lastOnly = true, isSorted = false)
)
// Merged split: a fresh testStream over a fresh streaming DataFrame
gets its own
@@ -474,12 +537,22 @@ class TestStreamingSource extends StreamTest {
testStream(variantStreamDf())(
AssertOnQuery { q => q.processAllAvailable(); true },
assertLegacyRddPlan,
- CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1-updated\"}", 1001L), Row(2,
"{\"key\":\"v2\"}", 1000L)),
+ CheckAnswerRows(Seq(
+ expectedRow(1, "v1-updated", "n1-updated", 1001L),
+ expectedRow(2, "v2", "n2", 1000L)),
lastOnly = true, isSorted = false)
)
}
}
+ test("test mor stream source reads shredded variant with legacy file group
reader disabled") {
+ testLegacyShreddedVariantStream(withTopLevelVariant = true)
+ }
+
+ test("test mor stream source reads a nested-only shredded variant with
legacy file group reader disabled") {
+ testLegacyShreddedVariantStream(withTopLevelVariant = false)
+ }
+
private def testCheckpointTranslation(tableName: String,
tableType: HoodieTableType,
writeTableVersion: HoodieTableVersion,
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
index 8f14dd1ceed8..b9a7dbe07500 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
@@ -23,7 +23,10 @@ import org.apache.hudi.common.schema.{HoodieSchema,
HoodieSchemaType}
import org.apache.parquet.schema.PrimitiveType
import org.apache.parquet.schema.Type.Repetition
-import org.apache.spark.sql.types.{BinaryType, IntegerType, StringType,
StructField, StructType}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.SparkSchemaTransformUtils
+import org.apache.spark.sql.types.{BinaryType, IntegerType, LongType,
Metadata, MetadataBuilder, StringType, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertThrows, assertTrue}
import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.Test
@@ -259,4 +262,108 @@ class TestBaseSpark4AdapterVariantMethods extends
SparkAdapterSupport {
val result = sparkAdapter.isDataTypeEqualForPhysicalSchema(invalidStruct,
variantType)
assertTrue(result.isEmpty, "Should return None when reversed struct
doesn't match variant physical schema")
}
+
+ /**
+ * The per-child metadata PushVariantIntoScan attaches to a projection
struct. Built key by key
+ * rather than through Spark 4.1's `VariantMetadata`, which does not exist
on the Spark 3.x
+ * profiles this suite also compiles for; the test pins the shape against
the adapter before it
+ * relies on it.
+ */
+ private def variantProjectionMetadata(path: String): Metadata =
+ new MetadataBuilder().putMetadata("__VARIANT_METADATA_KEY",
+ new MetadataBuilder()
+ .putString("path", path)
+ .putBoolean("failOnError", true)
+ .putString("timeZoneId", "UTC")
+ .build()).build()
+
+ /** A `VariantVal` holding `json`, reached by reflection for the same
cross-version reason. */
+ private def parseJson(json: String): Any = {
+ val cls =
Class.forName("org.apache.spark.sql.catalyst.expressions.variant.VariantExpressionEvalUtils$")
+ val module = cls.getField("MODULE$").get(null)
+ // parseJson(UTF8String, allowDuplicateKeys, ...): the trailing flags
gained an entry across
+ // 4.x, so every one of them is passed as its default false.
+ val method = cls.getMethods.find(m => m.getName == "parseJson"
+ && m.getParameterTypes.headOption.contains(classOf[UTF8String])).get
+ val args: Seq[AnyRef] = UTF8String.fromString(json) +:
+ Seq.fill(method.getParameterCount - 1)(Boolean.box(false))
+ method.invoke(module, args: _*)
+ }
+
+ @Test
+ def testBuildVariantProjectorRecursesIntoStructPaths(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_1, "Variant projection structs only
exist on Spark 4.1+")
+ val variantType = sparkAdapter.getVariantDataType.get
+ val projectionStruct = StructType(Seq(
+ StructField("0", StringType, metadata =
variantProjectionMetadata("$.k"))))
+ assertTrue(sparkAdapter.isVariantProjectionStruct(projectionStruct),
+ "the hand-built metadata must be what Spark recognizes as a projection
struct")
+
+ // `cast(s.inner as string)` on `s struct<inner: variant, other: int>`:
PushVariantIntoScan
+ // rewrites the variant one struct member down, leaving the struct itself
in place.
+ val dataSchema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", variantType),
+ StructField("other", IntegerType))))))
+ val requiredSchema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", projectionStruct),
+ StructField("other", IntegerType))))))
+
+ val projector = sparkAdapter.buildVariantProjector(dataSchema,
requiredSchema)
+ assertTrue(projector.isDefined, "a projection below a struct path must
build a projector")
+ val project = projector.get
+
+ val row = project(InternalRow(1, InternalRow(parseJson("""{"k":"n1"}"""),
7)))
+ assertEquals(1, row.getInt(0))
+ val struct = row.getStruct(1, 2)
+ assertEquals("n1", struct.getStruct(0, 1).getUTF8String(0).toString,
+ "the nested variant must be projected to its requested extraction")
+ assertEquals(7, struct.getInt(1), "the sibling of the projected variant
must survive")
+
+ // CreateNamedStruct is never null on its own, so a null struct has to be
preserved explicitly.
+ assertTrue(project(InternalRow(2, null)).isNullAt(1), "a null struct must
stay null")
+
+ // A null variant inside a live struct: its projection struct must be
NULL, not a struct of
+ // nulls, because PushVariantIntoScan rewrites `s.inner is null` onto that
struct.
+ val nullVariant = project(InternalRow(3, InternalRow(null, 9)))
+ val liveStruct = nullVariant.getStruct(1, 2)
+ assertTrue(liveStruct.isNullAt(0), "a null variant must project to a null
struct")
+ assertEquals(9, liveStruct.getInt(1), "the sibling of a null variant must
survive")
+
+ assertTrue(sparkAdapter.buildVariantProjector(dataSchema,
dataSchema).isEmpty,
+ "a required schema with no projection struct anywhere must not build a
projector")
+ }
+
+ @Test
+ def testImplicitSchemaChangeKeepsNestedVariantProjection(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_1, "Variant projection structs only
exist on Spark 4.1+")
+ val variantType = sparkAdapter.getVariantDataType.get
+ val projectionStruct = StructType(Seq(
+ StructField("0", StringType, metadata =
variantProjectionMetadata("$.k"))))
+
+ // The file wrote s.n as int and the table has since widened it to long,
so `s` is an implicit
+ // type change on this file and every member of `s` is reconciled by
addMissingFields --
+ // including the projected variant next to the widened column.
+ val fileStruct = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", variantType),
+ StructField("n", IntegerType))))))
+ val requiredSchema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", projectionStruct),
+ StructField("n", LongType))))))
+
+ val (typeChanges, readerSchema) =
SparkSchemaTransformUtils.buildImplicitSchemaChangeInfo(fileStruct,
requiredSchema)
+ val readerS = readerSchema("s").dataType.asInstanceOf[StructType]
+ assertEquals(projectionStruct, readerS("inner").dataType,
+ "the reader must be handed the projection struct, not the file's
VariantType")
+ assertEquals(IntegerType, readerS("n").dataType, "the widened sibling is
still read at its file type")
+ assertTrue(typeChanges.containsKey(1), "s is an implicit type change")
+ assertEquals(readerS, typeChanges.get(1).getRight, "the recorded reader
type is the reconciled struct")
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index 7bd872880a52..7644c232d61b 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
@@ -22,13 +22,11 @@ package org.apache.spark.sql.hudi.dml.schema
import org.apache.hudi.{DataSourceReadOptions, HoodieSparkUtils}
import org.apache.hudi.common.avro.VariantShreddingRuntime
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.model.HoodieLogFile
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.model.WriteOperationType
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.HoodieSchemaException
import org.apache.hudi.common.table.TableSchemaResolver
-import org.apache.hudi.common.table.log.HoodieLogFormat
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType
import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.common.util.StringUtils
@@ -42,11 +40,9 @@ import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
import org.apache.spark.sql.hudi.command.CreateHoodieTableCommand
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
-import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.{getLastCommitMetadata,
getMetaClientAndFileSystemView}
+import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType, LongType,
MapType, MetadataBuilder, StringType, StructField, StructType}
-import scala.collection.JavaConverters._
-
class TestVariantDataType extends HoodieSparkSqlTestBase with
VariantShreddingTestSupport {
test(s"Test Table with Variant Data Type") {
@@ -1436,29 +1432,4 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
with VariantShreddingTe
}
}
- /**
- * Block types of every log block in the table, read from the log files
themselves. Tests that pin
- * a log format assert on this rather than on file names: native logs carry
a .log.parquet suffix,
- * but an inline log file is named the same whether its data blocks are avro
or parquet.
- */
- private def listLogBlockTypes(tablePath: String): Seq[HoodieLogBlockType] = {
- val (metaClient, fsView) = getMetaClientAndFileSystemView(tablePath)
- val schema = new TableSchemaResolver(metaClient).getTableSchema
- val logFiles = fsView.getAllFileSlices("").iterator().asScala
- .flatMap(slice =>
HoodieTestUtils.getLogFileListFromFileSlice(slice).asScala).toSeq
- assert(logFiles.nonEmpty, "expected at least one log file")
- logFiles.flatMap { path =>
- val reader = HoodieLogFormat.newReader(metaClient, new
HoodieLogFile(path), schema)
- try {
- val types = scala.collection.mutable.ArrayBuffer[HoodieLogBlockType]()
- while (reader.hasNext) {
- types += reader.next().getBlockType
- }
- types.toSeq
- } finally {
- reader.close()
- }
- }
- }
-
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
index 4535a04e6261..345e82d44242 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
@@ -20,16 +20,23 @@
package org.apache.spark.sql.hudi.dml.schema
import org.apache.hudi.{HoodieSchemaConversionUtils, HoodieSparkUtils,
HoodieTableSchema, SparkAdapterSupport}
+import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.HoodieFileFormat
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaType}
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType
import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter
import org.apache.hudi.testutils.DataSourceTestUtils
+import org.apache.hadoop.fs.{Path => HadoopPath}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+import org.apache.spark.sql.SaveMode
import
org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
-import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+import org.apache.spark.sql.types.{IntegerType, LongType, StructField,
StructType}
+
+import scala.collection.JavaConverters._
/**
* Mixed-layout variant shredding matrix: files with DIFFERENT typed_value
layouts in one table,
@@ -84,6 +91,52 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
s"options => 'hoodie.datasource.write.row.writer.enable=$rowWriter')")
}
+ /**
+ * Pins the layout of a NESTED variant column (a dotted path) in the base
files written by one
+ * instant. [[variantFileLayouts]] and its derivatives resolve their column
as a TOP-LEVEL footer
+ * field, so the nested path is walked here with [[variantGroupOf]] instead.
+ */
+ private def assertNestedBaseLayout(tablePath: String, instant: String,
shredded: Boolean,
+ leg: String, column: String = "s.inner"):
Unit = {
+ val files = listDataParquetFiles(tablePath).filter { file =>
+ val name = new HadoopPath(file).getName
+ FSUtils.isBaseFile(name) && FSUtils.getCommitTime(name) == instant
+ }
+ assert(files.nonEmpty, s"[$leg] expected at least one base file written by
instant $instant")
+ files.foreach { file =>
+ val group = variantGroupOf(file, column)
+ val expected = if (shredded) "shredded" else "unshredded"
+ assert(group.containsField("typed_value") == shredded,
+ s"[$leg] base file of instant $instant should be $expected at $column:
$file\n$group")
+ }
+ }
+
+ /**
+ * The declared write schema of the array-element leg: the `(id int, arr
array<variant>, ts long)`
+ * table schema with the ELEMENT of `arr` replaced by a variant whose
shredded object declares
+ * `k string`. No DDL and no forced-schema conf can express that shape, so
it is built through the
+ * HoodieSchema API and handed to the write as `hoodie.write.schema`.
+ */
+ private def arrayElementWriteSchema(): String = {
+ // fromDDL parses `variant` on Spark 4.x only, which is all this suite
runs on.
+ val base = HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(
+ StructType.fromDDL("id int, arr array<variant>, ts long"),
"hoodie.test.record")
+ val shreddedFields = new java.util.LinkedHashMap[String, HoodieSchema]()
+ shreddedFields.put("k", HoodieSchema.create(HoodieSchemaType.STRING))
+ val shreddedElement = HoodieSchema.createVariantShreddedObject(null, null,
null, shreddedFields)
+ val fields = base.getFields.asScala.map { field =>
+ if (field.name() == "arr") {
+ HoodieSchemaField.of("arr",
+
HoodieSchema.createNullable(HoodieSchema.createArray(HoodieSchema.createNullable(shreddedElement))),
+ null, HoodieSchema.NULL_VALUE)
+ } else {
+ HoodieSchemaField.of(field.name(), field.schema(),
field.doc().orElse(null),
+ field.defaultVal().orElse(null))
+ }
+ }.asJava
+ HoodieSchema.createRecord(base.getName, "hoodie.test", null,
fields).toString
+ }
+
//
-----------------------------------------------------------------------------------------------
// A. Mixed records inside one file
//
-----------------------------------------------------------------------------------------------
@@ -818,6 +871,481 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
}
}
+ 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 nested-only
+ // legacy leg (the leg with a top-level variant beside it is re-routed
to the file-group
+ // reader instead).
+
+ // 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) {
+ withNestedOnlyVariantTable(s"nested
pushVariantIntoScan=$pushIntoScan", "cow",
+ props = Seq(NEW_FILE_GROUP_PER_COMMIT), recordTypes =
Seq(HoodieRecordType.SPARK)) {
+ (tableName, tablePath, leg) =>
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName ${nestedRowsSql(0, 5)}")
+ }
+ val shreddedInstant = latestCompletedInstant(tablePath)
+ withWriteLayout(Unshredded) {
+ spark.sql(s"insert into $tableName ${nestedRowsSql(5, 10)}")
+ }
+ val unshreddedInstant = latestCompletedInstant(tablePath)
+ // One file per commit (small.file.limit = 0) and one layout each,
so $.k is typed in the
+ // first file and residual in the second: both slots answer the
queries below.
+ assertNestedBaseLayout(tablePath, shreddedInstant, shredded = true,
leg)
+ assertNestedBaseLayout(tablePath, unshreddedInstant, shredded =
false, leg)
+
+ checkAnswer(s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName order by id")(
+ (0 until 10).map(id => Seq(id, s"x$id")): _*)
+ // A filter served out of the residual file, and one out of the
typed file.
+ checkAnswer(s"select id from $tableName where variant_get(s.inner,
'$$.k', 'string') = 'x7'")(
+ Seq(7))
+ checkAnswer(s"select id from $tableName where variant_get(s.inner,
'$$.k', 'string') = 'x2'")(
+ Seq(2))
+ checkAnswer(s"select count(*) from $tableName " +
+ s"where try_variant_get(s.inner, '$$.missing', 'string') is
null")(Seq(10))
+ checkAnswer(s"select count(*) from $tableName where s.inner is
null")(Seq(0))
+ checkAnswer(s"select id, cast(s.inner as string) from $tableName
order by id")(
+ (0 until 10).map(id => Seq(id, s"""{"k":"x$id"}""")): _*)
+ // Both arms expect the very same rows, so this is what tells them
apart: whether the
+ // rule actually rewrote s.inner into a projection struct inside the
scan.
+ val pushed = pushIntoScan.toBoolean
+ val verdict = if (pushed) "should have" else "must not have"
+ assert(variantProjectionPushedIntoScan(
+ s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName") == pushed,
+ s"[$leg] PushVariantIntoScan $verdict rewritten s.inner into a
projection struct")
+ }
+ }
+ }
+
+ // MOR: a nested-shredded base under an unshredded log update, so the
merged read has to serve
+ // ids 0-2 from the base file and ids 3-4 from the log. Before #19775 the
pushIntoScan=true arms
+ // here crashed the JVM (SIGSEGV/SIGBUS inside an unsafe copy stub, or a
java.lang.InternalError
+ // about a fault in a recent unsafe memory access): the internal reader
applied the
+ // PushVariantIntoScan projection to TOP-LEVEL fields only, so the merged
row still held a raw
+ // VariantVal at s.inner while the plan read that memory as the projected
struct s.inner.0.
+ def seedNestedMor(tableName: String, tablePath: String, leg: String): Unit
= {
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName ${nestedRowsSql(0, 5)}")
+ }
+ assertNestedBaseLayout(tablePath, latestCompletedInstant(tablePath),
shredded = true, leg)
+ // id 4's variant is nulled out on the way through the log: before
#19775 the avro-path
+ // projector emitted a non-null struct of nulls for it rather than a
NULL struct, which is
+ // what the `s.inner is null` assertion below catches.
+ withWriteLayout(Unshredded) {
+ spark.sql(s"update $tableName set " +
+ """s = named_struct('inner', parse_json(case when id = 4 then
cast(null as string) """ +
+ """else concat('{"k":"y', id, '"}') end)), ts = 1001 """ +
+ "where id >= 3")
+ }
+ }
+ def assertNestedMorReads(tableName: String, leg: String, pushed: Boolean):
Unit = {
+ // ids 0-2 come off the base file, id 3 off the log, and id 4's variant
is the log's NULL.
+ def merged(id: Int): String = if (id < 3) s"x$id" else if (id == 3) "y3"
else null
+ def mergedJson(id: Int): String = Option(merged(id)).map(k =>
s"""{"k":"$k"}""").orNull
+ checkAnswer(s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName order by id")(
+ (0 until 5).map(id => Seq(id, merged(id))): _*)
+ // A filter served out of the log row, then one out of the base row.
+ checkAnswer(s"select id from $tableName where variant_get(s.inner,
'$$.k', 'string') = 'y3'")(
+ Seq(3))
+ checkAnswer(s"select id from $tableName where variant_get(s.inner,
'$$.k', 'string') = 'x1'")(
+ Seq(1))
+ checkAnswer(s"select id, cast(s.inner as string) from $tableName order
by id")(
+ (0 until 5).map(id => Seq(id, mergedJson(id))): _*)
+ // PushVariantIntoScan rewrites `s.inner is null` onto the projection
struct itself, so a
+ // null variant projected as a struct OF nulls rather than a NULL struct
made this row
+ // disappear: the avro-block leg with the rule on is the one that failed
before the fix.
+ checkAnswer(s"select id from $tableName where s.inner is null")(Seq(4))
+ checkAnswer(s"select count(*) from $tableName where s.inner is not
null")(Seq(4))
+ // The whole struct: no extraction, so nothing is rewritten even with
pushVariantIntoScan on,
+ // and the merge is read through a plain native VariantType at depth.
`s` itself is never
+ // null - only its `inner` member is, and only for id 4 - and the
payload has to be the
+ // merged one, not a stale or nulled-out variant (VariantVal.toString is
its JSON).
+ val wholeStruct = spark.sql(s"select id, s from $tableName order by
id").collect()
+ assert(wholeStruct.length == 5, s"[$leg] whole-struct read should return
5 rows")
+ wholeStruct.foreach { row =>
+ val id = row.getInt(0)
+ assert(!row.isNullAt(1), s"[$leg] whole-struct read nulled out s for
id $id")
+ val inner = row.getStruct(1).getAs[Any]("inner")
+ assert(Option(inner).map(_.toString).orNull == mergedJson(id),
+ s"[$leg] whole-struct read of s.inner for id $id should be
${mergedJson(id)}, got $inner")
+ }
+ // Both arms expect the very same rows; only the plan tells them apart.
+ val verdict = if (pushed) "should have" else "must not have"
+ assert(variantProjectionPushedIntoScan(
+ s"select id, variant_get(s.inner, '$$.k', 'string') from $tableName")
== pushed,
+ s"[$leg] PushVariantIntoScan $verdict rewritten s.inner into a
projection struct")
+ }
+
+ Seq("true", "false").foreach { pushIntoScan =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ // Parquet log files: the log block is read by
HoodieSparkParquetReader.getUnsafeRowIterator
+ // with the projected struct threaded into the requested schema.
+ withNestedOnlyVariantTable(s"nested mor
pushVariantIntoScan=$pushIntoScan", "mor",
+ props = Seq("hoodie.compact.inline = 'false'"), recordTypes =
Seq(HoodieRecordType.SPARK)) {
+ (tableName, tablePath, leg) =>
+ seedNestedMor(tableName, tablePath, leg)
+
assert(listDataParquetFiles(tablePath).exists(_.endsWith(".log.parquet")),
+ s"[$leg] the update should have written a native parquet log file")
+ assertNestedMorReads(tableName, leg, pushIntoScan.toBoolean)
+ }
+
+ // Avro data blocks: the other projection site,
HoodieReaderContext.projectLogBlockRecords.
+ // On the current table version the append handle writes native
parquet log files whatever
+ // the block format says, so this leg only exists on a pre-native
table version.
+ withNestedOnlyVariantTable(s"nested mor avro blocks
pushVariantIntoScan=$pushIntoScan", "mor",
+ props = Seq("hoodie.compact.inline = 'false'",
+ "hoodie.write.table.version = '9'",
+ "hoodie.logfile.data.block.format = 'avro'"),
+ recordTypes = Seq(HoodieRecordType.AVRO)) { (tableName, tablePath,
leg) =>
+ seedNestedMor(tableName, tablePath, leg)
+ val blockTypes = listLogBlockTypes(tablePath)
+ assert(blockTypes.contains(HoodieLogBlockType.AVRO_DATA_BLOCK),
+ s"[$leg] expected an avro data block in the log files, found:
$blockTypes")
+ assert(!blockTypes.contains(HoodieLogBlockType.PARQUET_DATA_BLOCK),
+ s"[$leg] this leg must not write parquet data blocks, found:
$blockTypes")
+ assertNestedMorReads(tableName, leg, pushIntoScan.toBoolean)
+ }
+ }
+ }
+ }
+
+ test("Implicit widening of a sibling keeps the nested variant projection") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // A file written with s.n as int under a table that has since widened s.n
to long makes `s`
+ // an implicit type change on that file, so every member of `s` is
reconciled by
+ // SparkSchemaTransformUtils.addMissingFields. Before #19775 that walk
handed the reader the
+ // file's VariantType at s.inner instead of the PushVariantIntoScan
projection struct, and
+ // the widening Cast then cast the variant value to the projected struct:
nulls.
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "true") {
+ withNestedOnlyVariantTable("implicit widening beside a nested variant",
"cow",
+ props = Seq(NEW_FILE_GROUP_PER_COMMIT), recordTypes =
Seq(HoodieRecordType.SPARK),
+ structMembers = "inner: variant, n: int") { (tableName, tablePath,
leg) =>
+ def rowsSql(lo: Int, hi: Int, nType: String): String =
+ s"""select cast(id as int) as id,
+ | named_struct('inner', parse_json(concat('{"k":"x', id, '"}')),
+ | 'n', cast(id as $nType)) as s,
+ | 1000L as ts from range($lo, $hi, 1, 1)""".stripMargin
+ withWriteLayout(Forced("k string")) {
+ spark.sql(s"insert into $tableName ${rowsSql(0, 3, "int")}")
+ }
+ val intInstant = latestCompletedInstant(tablePath)
+ assertNestedBaseLayout(tablePath, intInstant, shredded = true, leg)
+
+ // The widening goes through the DataFrame API: a SQL insert coerces
to the table schema,
+ // while a DataFrame whose s.n is bigint evolves the table schema in
place. Every knob the
+ // write needs is an explicit option here - the write path collects
spark.hoodie.* only and
+ // drops bare hoodie.* session confs, so neither withWriteLayout nor
the small-file
+ // tblproperty of NEW_FILE_GROUP_PER_COMMIT reaches a df.write (see
layoutConfs).
+ spark.sql(rowsSql(3, 6, "bigint")).write.format("hudi")
+ .options(layoutConfs(Forced("k string")).toMap)
+ .option("hoodie.table.name", tableName)
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "ts")
+ .option("hoodie.datasource.write.operation", "insert")
+ .option("hoodie.parquet.small.file.limit", "0")
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ // The widening lands in the commit schema (a path-based relation
reads s.n back as bigint)
+ // while the SQL catalog's view of the table keeps reporting it as
int, so the reads go
+ // through a path-based view: that is the reader that sees the widened
table schema over
+ // the int file, which is what makes `s` an implicit type change there.
+ val widenedView = s"${tableName}_widened"
+
spark.read.format("hudi").load(tablePath).createOrReplaceTempView(widenedView)
+ val widenedN =
spark.table(widenedView).schema("s").dataType.asInstanceOf[StructType]("n").dataType
+ assert(widenedN == LongType,
+ s"[$leg] the DataFrame write should have widened s.n to bigint, got
$widenedN")
+
+ checkAnswer(s"select id, variant_get(s.inner, '$$.k', 'string'), s.n
from $widenedView order by id")(
+ (0 until 6).map(id => Seq(id, s"x$id", id.toLong)): _*)
+ // One filter served out of the int file and one out of the bigint
file.
+ checkAnswer(s"select id from $widenedView where variant_get(s.inner,
'$$.k', 'string') = 'x1'")(Seq(1))
+ checkAnswer(s"select id from $widenedView where variant_get(s.inner,
'$$.k', 'string') = 'x4'")(Seq(4))
+ assert(variantProjectionPushedIntoScan(
+ s"select id, variant_get(s.inner, '$$.k', 'string'), s.n from
$widenedView"),
+ s"[$leg] the projection must still be pushed into the scan")
+ }
+ }
+ }
+
+ test("An array element shredded through a declared write schema reads
natively and survives compaction") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // No DDL forces a variant that is directly an ARRAY ELEMENT - the forced
hook reaches record
+ // members only, pinned by the record-writer test above - and Spark's
PushVariantIntoScan
+ // rewrites struct paths only, so array<variant> is the shape every reader
has to handle
+ // NATIVELY, shredded or not. The only write that produces a shredded
element is one whose write
+ // schema already declares typed_value there, which the row write support
honours through
+ // hoodie.write.schema (HoodieWriteConfig.WRITE_SCHEMA_OVERRIDE). SPARK
record type only: the
+ // AVRO insert cannot take a write-schema override of a different shape
(its records are
+ // serialized under hoodie.avro.schema and deserialized under the writer
schema, which throws
+ // ArrayIndexOutOfBounds inside SerializableIndexedRecord).
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"array element shredding, $tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | arr array<variant>,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = 'ts',
+ | type = 'mor',
+ | hoodie.compact.inline = 'false'
+ | )
+ """.stripMargin)
+
+ val declaredLayout = Seq(
+ "hoodie.write.schema" -> arrayElementWriteSchema(),
+ WRITE_SHREDDING_KEY -> "true",
+ FORCE_SCHEMA_KEY -> "",
+ INFERENCE_KEY -> "false")
+ // get(arr, 1) rather than arr[1]: under ANSI mode the subscript on the
one-element array of
+ // id 2 throws instead of returning null.
+ val elementsQuery = s"select id, cast(arr[0] as string), cast(get(arr,
1) as string), size(arr) " +
+ s"from $tableName order by id"
+ // pushVariantIntoScan is swept alongside the vectorized reader although
an array element is
+ // never rewritten: both arms must read the element the same way.
+ def withReadSweep(f: => Unit): Unit = {
+ Seq("true", "false").foreach { pushIntoScan =>
+ Seq("true", "false").foreach { vectorizedReader =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" ->
pushIntoScan,
+ "spark.sql.parquet.enableVectorizedReader" ->
vectorizedReader)(f)
+ }
+ }
+ }
+
+ withSQLConf(declaredLayout: _*) {
+ spark.sql(s"""insert into $tableName values """ +
+ """(1, array(parse_json('{"k":"a1"}'), parse_json('{"k":"a2"}')),
1000), """ +
+ """(2, array(parse_json('{"k":"b1"}')), 1000)""")
+ }
+ assertVariantLayout(tablePath, shredded = true, leg, column = "arr")
+ listDataParquetFiles(tablePath).foreach { file =>
+ val elementGroup = variantGroupOf(file, "arr")
+ assert(getFieldAsGroup(elementGroup, "typed_value").containsField("k"),
+ s"[$leg] typed_value of the array element should carry
k:\n$elementGroup")
+ }
+
+ withReadSweep {
+ checkAnswer(elementsQuery)(
+ Seq(1, """{"k":"a1"}""", """{"k":"a2"}""", 2),
+ Seq(2, """{"k":"b1"}""", null, 1)
+ )
+ checkAnswer(s"select id, variant_get(arr[0], '$$.k', 'string') from
$tableName order by id")(
+ Seq(1, "a1"),
+ Seq(2, "b1")
+ )
+ checkAnswer(s"select id from $tableName where variant_get(arr[0],
'$$.k', 'string') = 'b1'")(
+ Seq(2))
+ }
+
+ // A shredded native log over the shredded base: the merged read serves
id 2 out of the log.
+ withSQLConf(declaredLayout: _*) {
+ spark.sql(s"""update $tableName set arr =
array(parse_json('{"k":"b1x"}')), ts = 1001 where id = 2""")
+ }
+
assert(listDataParquetFiles(tablePath).exists(_.endsWith(".log.parquet")),
+ s"[$leg] the update should have written a native parquet log file")
+ val updatedRows = Seq(
+ Seq(1, """{"k":"a1"}""", """{"k":"a2"}""", 2),
+ Seq(2, """{"k":"b1x"}""", null, 1))
+ withReadSweep {
+ checkAnswer(elementsQuery)(updatedRows: _*)
+ }
+ checkAnswer(s"select id, cast(arr[0] as string) from " +
+ s"hudi_query('$tableName', 'read_optimized') order by id")(
+ Seq(1, """{"k":"a1"}"""),
+ Seq(2, """{"k":"b1"}""")
+ )
+
+ // Compaction under the same declared schema: the row write support
honours it at depth again.
+ withSQLConf(declaredLayout: _*) {
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 1, leg)
+ assertNestedBaseLayout(tablePath, latestCompletedInstant(tablePath),
shredded = true, leg,
+ column = "arr")
+ checkAnswer(elementsQuery)(updatedRows: _*)
+
+ // And with no declared schema the element goes back to unshredded,
values intact.
+ withWriteLayout(Unshredded) {
+ spark.sql(s"""update $tableName set """ +
+ """arr = array(parse_json('{"k":"a1y"}'),
parse_json('{"k":"a2y"}')), ts = 1002 where id = 1""")
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 2, leg)
+ assertNestedBaseLayout(tablePath, latestCompletedInstant(tablePath),
shredded = false, leg,
+ column = "arr")
+ checkAnswer(elementsQuery)(
+ Seq(1, """{"k":"a1y"}""", """{"k":"a2y"}""", 2),
+ Seq(2, """{"k":"b1x"}""", null, 1)
+ )
+ })
+ }
+
//
-----------------------------------------------------------------------------------------------
// F2. Vectorized read decision
//
-----------------------------------------------------------------------------------------------
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
index 64d9b49fa003..3ed37a776080 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
@@ -19,10 +19,15 @@
package org.apache.spark.sql.hudi.dml.schema
-import org.apache.hudi.DataSourceReadOptions
+import org.apache.hudi.{DataSourceReadOptions, SparkAdapterSupport}
import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieLogFile
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.common.table.TableSchemaResolver
+import org.apache.hudi.common.table.log.HoodieLogFormat
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType
+import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.storage.StoragePath
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
@@ -35,9 +40,10 @@ import org.apache.parquet.hadoop.example.GroupReadSupport
import org.apache.parquet.hadoop.util.HadoopInputFile
import org.apache.parquet.schema.{GroupType, LogicalTypeAnnotation,
MessageType, Type}
import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.FileSourceScanExec
import
org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{listElement,
mapValue}
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
-import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
+import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.{getLastCommitMetadata,
getMetaClientAndFileSystemView}
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -108,29 +114,45 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
}
/**
- * Creates `(id int, s struct<inner: variant>, ts long)` and bulk-inserts
row 1 through the row
- * writer under a forced `k string` nested schema. Both forced hooks recurse
into record members
- * (the row writer's always did, the Avro write support's since the #19689
fix), so a nested
- * variant shreds on either write path; inference alone stays top-level, and
a variant that is
- * directly an array element or a map value shreds only where the write
schema already declares
- * it. This row-writer seed is what the nested-shredding legs open on.
+ * The `(id int, s struct<inner: variant>, ts long)` table of the nested
legs: the variant lives
+ * one struct member down and the table has NO top-level variant column.
Same tblproperties
+ * rendering as [[createVariantTable]]. `structMembers` is the member list
of `s`, so a leg that
+ * needs a SIBLING beside the variant (an implicit type change on one, say)
declares it here
+ * rather than building its own DDL.
*/
- private def createNestedVariantRowWriterTable(tableName: String, tablePath:
String): Unit = {
+ protected def createNestedVariantTable(tableName: String,
+ tablePath: String,
+ tableType: String,
+ props: Seq[String] = Seq.empty,
+ structMembers: String = "inner:
variant"): Unit = {
+ val extraProps = if (props.isEmpty) "" else props.mkString(",\n ", ",\n
", "")
spark.sql(
s"""
|create table $tableName (
| id int,
- | s struct<inner: variant>,
+ | s struct<$structMembers>,
| ts long
|) using hudi
| location '$tablePath'
| tblproperties (
| primaryKey = 'id',
| preCombineField = 'ts',
- | type = 'cow',
- | hoodie.datasource.write.row.writer.enable = 'true'
+ | type = '$tableType'$extraProps
| )
""".stripMargin)
+ }
+
+ /**
+ * Creates the nested table as a row-writer COW one and bulk-inserts row 1
through the row writer
+ * under a forced `k string` nested schema. Both forced hooks recurse into
record members (the row
+ * writer's always did, the Avro write support's since the #19689 fix), so a
nested variant shreds
+ * on either write path; inference alone stays top-level, and a variant that
is directly an array
+ * element or a map value shreds only where the write schema already
declares it. This row-writer
+ * seed is what the nested-shredding legs open on.
+ */
+ private def createNestedVariantRowWriterTable(tableName: String, tablePath:
String): Unit = {
+ createNestedVariantTable(tableName, tablePath, "cow",
+ props = Seq("hoodie.datasource.write.row.writer.enable = 'true'"))
withSQLConf("hoodie.spark.sql.insert.into.operation" -> "bulk_insert") {
withWriteLayout(Forced("k string")) {
spark.sql(s"""insert into $tableName values (1, named_struct('inner',
parse_json('{"k":"x1"}')), 1000)""")
@@ -149,6 +171,33 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
withTableScaffold(label, recordTypes)(createNestedVariantRowWriterTable)(f)
}
+ /**
+ * The nested-only twin of [[withVariantTable]]:
[[createNestedVariantTable]] with the table type
+ * and tblproperties of the leg, and NO seed row, so the body owns every
commit.
+ *
+ * The absence of a top-level variant is the point. Hudi's Spark-native
internal read paths only
+ * rewrite TOP-LEVEL variant columns into the full-variant projection shape
+ * (SparkAdapter.buildFullVariantReadSchema, used by
SparkFileFormatInternalRowReaderContext;
+ * Spark's own PushVariantIntoScan and the sparkRequiredSchema overlay that
follows it do go on
+ * down struct paths, #19775), so on a table that also carried a `v` column
those top-level
+ * switches would be on for the whole read and the nested path would never
be pinned on its own.
+ * With no top-level variant the reader takes the plain native path - on the
MOR RDD,
+ * HoodieMergeOnReadRDDV2.shouldRerouteVariantSplit is false as well - and
these legs really do
+ * exercise the nested read.
+ */
+ protected def withNestedOnlyVariantTable[T](label: String,
+ tableType: String,
+ props: Seq[String] = Seq.empty,
+ recordTypes:
Seq[HoodieRecordType] =
+ Seq(HoodieRecordType.AVRO,
HoodieRecordType.SPARK),
+ structMembers: String = "inner:
variant")
+ (f: (String, String, String) =>
T): Unit = {
+ withTableScaffold(label, recordTypes) { (tableName, tablePath) =>
+ createNestedVariantTable(tableName, tablePath, tableType, props = props,
+ structMembers = structMembers)
+ }(f)
+ }
+
//
---------------------------------------------------------------------------------------------
// Parquet footer helpers
//
---------------------------------------------------------------------------------------------
@@ -589,12 +638,31 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
.collect()
}
+ /**
+ * Whether the physical plan of `sql` reads a variant through a
PushVariantIntoScan projection
+ * struct: the file scan's required schema carries the projected struct at
some struct path.
+ * Pins that the rule actually fired for a true arm, so it cannot silently
become a copy of the
+ * arm with the rule off. `sparkPlan` rather than `executedPlan`: under AQE
the latter is a
+ * placeholder whose children only exist once the query runs.
+ */
+ protected def variantProjectionPushedIntoScan(sql: String): Boolean = {
+ val scans = spark.sql(sql).queryExecution.sparkPlan.collect { case scan:
FileSourceScanExec => scan }
+ assert(scans.nonEmpty, s"expected a file scan in the plan of: $sql")
+ scans.exists(_.requiredSchema.fields.exists(f =>
+ SparkAdapterSupport.sparkAdapter.containsVariantProjection(f.dataType)))
+ }
+
//
---------------------------------------------------------------------------------------------
// Write-layout toggle
//
---------------------------------------------------------------------------------------------
- /** The three write-side variant layout configs for a [[WriteLayout]]. */
- private def layoutConfs(layout: WriteLayout): Seq[(String, String)] = layout
match {
+ /**
+ * The three write-side variant layout configs for a [[WriteLayout]].
Protected because a
+ * DataFrame write needs them as explicit `.option`s:
DefaultSource.createRelation for writes
+ * collects `spark.hoodie.*` only and deliberately drops bare `hoodie.*`
session confs, so
+ * [[withWriteLayout]] around a `df.write` would silently write the default
layout.
+ */
+ protected def layoutConfs(layout: WriteLayout): Seq[(String, String)] =
layout match {
case Unshredded => Seq(
WRITE_SHREDDING_KEY -> "false",
FORCE_SCHEMA_KEY -> "",
@@ -670,6 +738,31 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
s"[$leg] expected $expected completed compaction commits, got $commits")
}
+ /**
+ * Block types of every log block in the table, read from the log files
themselves. Tests that pin
+ * a log format assert on this rather than on file names: native logs carry
a .log.parquet suffix,
+ * but an inline log file is named the same whether its data blocks are avro
or parquet.
+ */
+ protected def listLogBlockTypes(tablePath: String): Seq[HoodieLogBlockType]
= {
+ val (metaClient, fsView) = getMetaClientAndFileSystemView(tablePath)
+ val schema = new TableSchemaResolver(metaClient).getTableSchema
+ val logFiles = fsView.getAllFileSlices("").iterator().asScala
+ .flatMap(slice =>
HoodieTestUtils.getLogFileListFromFileSlice(slice).asScala).toSeq
+ assert(logFiles.nonEmpty, "expected at least one log file")
+ logFiles.flatMap { path =>
+ val reader = HoodieLogFormat.newReader(metaClient, new
HoodieLogFile(path), schema)
+ try {
+ val types = mutable.ArrayBuffer[HoodieLogBlockType]()
+ while (reader.hasNext) {
+ types += reader.next().getBlockType
+ }
+ types.toSeq
+ } finally {
+ reader.close()
+ }
+ }
+ }
+
/** The requested time of the latest completed commit-like instant, for time
travel. */
protected def latestCompletedInstant(tablePath: String): String = {
val metaClient = createMetaClient(spark, tablePath)
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
index fab2f0480dc7..75dc1e3b8e9f 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
@@ -36,10 +36,12 @@ import
org.apache.spark.sql.FileFormatUtilsForFileGroupReader.applyFiltersToPlan
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, InterpretedPredicate, Predicate, SpecializedGetters}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
BoundReference, CreateNamedStruct, Expression, GetStructField, If,
InterpretedPredicate, IsNull, Literal, Predicate, SpecializedGetters,
UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.DateFormatter
import org.apache.spark.sql.classic.ColumnConversions
import org.apache.spark.sql.execution.{PartitionedFileUtil, QueryExecution,
SQLExecution}
@@ -49,7 +51,7 @@ import
org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, Pa
import org.apache.spark.sql.hudi.SparkAdapter
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.{BaseRelation, Filter}
-import org.apache.spark.sql.types.{BinaryType, DataType, StructField,
StructType, VariantType}
+import org.apache.spark.sql.types.{BinaryType, DataType, StringType,
StructField, StructType, VariantType}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.types.variant.Variant
@@ -300,9 +302,11 @@ abstract class BaseSpark4Adapter extends SparkAdapter with
Logging {
/**
* Shared implementation behind [[SparkAdapter#buildFullVariantReadSchema]]
for the 4.x
* adapters whose parquet reader can reconstruct shredded variants (4.1+,
SPARK-54410).
- * Spark 4.0 keeps the default None: the write-side methods above have no
version gate, so
- * it does write shredded files, but its reader cannot rebuild them and the
projection
- * shape would not help.
+ * Top-level fields only, matching the whole-variant request
PushVariantIntoScan makes for a
+ * root attribute; a variant below the top level is left native, and that
reader rebuilds it
+ * at any depth from a native VariantType request (#19775). Spark 4.0 keeps
the default None:
+ * the write-side methods above have no version gate, so it does write
shredded files, but
+ * its reader cannot rebuild them and the projection shape would not help.
*/
protected final def rewriteTopLevelVariantsForFullRead(schema: StructType):
Option[StructType] = {
var rewritten = false
@@ -320,6 +324,86 @@ 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)
+ }
+ val projected = CreateNamedStruct(childExprs)
+ // A null variant has to come out as a NULL struct, not a struct of
nulls: CreateNamedStruct
+ // is never null, the parquet paths leave the field null, and
PushVariantIntoScan rewrites
+ // IsNull(v) / IsNotNull(v) onto this struct directly.
+ If(IsNull(ref), Literal(null, projected.dataType), projected)
+ 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))
+ }
+ }
+
override def createShreddedVariantWriter(
shreddedStructType: StructType,
writeStruct: Consumer[InternalRow]
diff --git
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
index c445d89f1e69..ddaf0f581476 100644
---
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
@@ -32,13 +32,11 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases,
ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{BoundReference,
CreateNamedStruct, Expression, Literal, UnsafeProjection}
-import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
+import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.trees.Origin
-import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.{METADATA_COL_ATTR_KEY,
RebaseDateTime}
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
@@ -52,7 +50,7 @@ import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
import org.apache.spark.sql.hudi.blob.{BatchedBlobReaderStrategy,
ScalarFunctions}
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.parser.{HoodieExtendedParserInterface,
HoodieSpark4_1ExtendedSqlParser}
-import org.apache.spark.sql.types.{DataType, DataTypes, Metadata,
MetadataBuilder, StructField, StructType}
+import org.apache.spark.sql.types.{DataType, DataTypes, Metadata,
MetadataBuilder, StructType}
import org.apache.spark.sql.vectorized.ColumnarBatchRow
import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel._
@@ -242,46 +240,8 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
rewriteTopLevelVariantsForFullRead(schema)
override def buildVariantProjector(sparkDataSchema: StructType,
- sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] = {
- // Quick check: any required field a variant projection struct?
- if (!sparkRequiredSchema.fields.exists(f =>
VariantMetadata.isVariantStruct(f.dataType))) {
- None
- } else {
- // Surface mismatched schemas with both field lists rather than Spark's
bare
- // IllegalArgumentException from fieldIndex.
- def lookupDataField(name: String): (Int, StructField) = {
- val idx = sparkDataSchema.getFieldIndex(name).getOrElse(
- throw new IllegalStateException(
- s"Required field '$name' is absent from sparkDataSchema; " +
- s"required=${sparkRequiredSchema.fieldNames.mkString("[", ",",
"]")}, " +
- s"data=${sparkDataSchema.fieldNames.mkString("[", ",", "]")}"))
- (idx, sparkDataSchema.fields(idx))
- }
- val exprs: Array[Expression] = sparkRequiredSchema.fields.map { rf =>
- rf.dataType match {
- case projectedStruct: StructType if
VariantMetadata.isVariantStruct(projectedStruct) =>
- val (dataIdx, dataField) = lookupDataField(rf.name)
- require(isVariantType(dataField.dataType),
- s"Expected VariantType for field '${rf.name}' in data schema,
got ${dataField.dataType}")
- val variantRef: Expression = BoundReference(dataIdx,
dataField.dataType, dataField.nullable)
- val childExprs: Seq[Expression] =
projectedStruct.fields.toSeq.flatMap { child =>
- val vm = VariantMetadata.fromMetadata(child.metadata)
- val pathLit = Literal(UTF8String.fromString(vm.path),
DataTypes.StringType)
- val tz: Option[String] = Option(vm.timeZoneId)
- val variantGet: Expression = VariantGet(variantRef, pathLit,
child.dataType, vm.failOnError, tz)
- Seq(Literal(UTF8String.fromString(child.name),
DataTypes.StringType), variantGet)
- }
- CreateNamedStruct(childExprs)
- case _ =>
- val (dataIdx, dataField) = lookupDataField(rf.name)
- BoundReference(dataIdx, dataField.dataType, dataField.nullable)
- }
- }
-
- val projection = UnsafeProjection.create(exprs.toIndexedSeq,
DataTypeUtils.toAttributes(sparkDataSchema))
- Some(row => projection(row))
- }
- }
+ sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] =
+ buildVariantProjectorForStructPaths(sparkDataSchema, sparkRequiredSchema)
// Apply LogicalTypeAnnotation.variantType((byte) 1) to the variant group,
matching parquet 1.16+'s
// SparkToParquetSchemaConverter convention.
diff --git
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
index 388582c9e963..b6425b03e5f6 100644
---
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
@@ -32,13 +32,11 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases,
ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{BoundReference,
CreateNamedStruct, Expression, Literal, UnsafeProjection}
-import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
+import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.trees.Origin
-import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.{METADATA_COL_ATTR_KEY,
RebaseDateTime}
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
@@ -52,7 +50,7 @@ import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
import org.apache.spark.sql.hudi.blob.{BatchedBlobReaderStrategy,
ScalarFunctions}
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.parser.{HoodieExtendedParserInterface,
HoodieSpark4_2ExtendedSqlParser}
-import org.apache.spark.sql.types.{DataType, DataTypes, Metadata,
MetadataBuilder, StructField, StructType}
+import org.apache.spark.sql.types.{DataType, DataTypes, Metadata,
MetadataBuilder, StructType}
import org.apache.spark.sql.vectorized.ColumnarBatchRow
import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel._
@@ -242,46 +240,8 @@ class Spark4_2Adapter extends BaseSpark4Adapter {
rewriteTopLevelVariantsForFullRead(schema)
override def buildVariantProjector(sparkDataSchema: StructType,
- sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] = {
- // Quick check: any required field a variant projection struct?
- if (!sparkRequiredSchema.fields.exists(f =>
VariantMetadata.isVariantStruct(f.dataType))) {
- None
- } else {
- // Surface mismatched schemas with both field lists rather than Spark's
bare
- // IllegalArgumentException from fieldIndex.
- def lookupDataField(name: String): (Int, StructField) = {
- val idx = sparkDataSchema.getFieldIndex(name).getOrElse(
- throw new IllegalStateException(
- s"Required field '$name' is absent from sparkDataSchema; " +
- s"required=${sparkRequiredSchema.fieldNames.mkString("[", ",",
"]")}, " +
- s"data=${sparkDataSchema.fieldNames.mkString("[", ",", "]")}"))
- (idx, sparkDataSchema.fields(idx))
- }
- val exprs: Array[Expression] = sparkRequiredSchema.fields.map { rf =>
- rf.dataType match {
- case projectedStruct: StructType if
VariantMetadata.isVariantStruct(projectedStruct) =>
- val (dataIdx, dataField) = lookupDataField(rf.name)
- require(isVariantType(dataField.dataType),
- s"Expected VariantType for field '${rf.name}' in data schema,
got ${dataField.dataType}")
- val variantRef: Expression = BoundReference(dataIdx,
dataField.dataType, dataField.nullable)
- val childExprs: Seq[Expression] =
projectedStruct.fields.toSeq.flatMap { child =>
- val vm = VariantMetadata.fromMetadata(child.metadata)
- val pathLit = Literal(UTF8String.fromString(vm.path),
DataTypes.StringType)
- val tz: Option[String] = Option(vm.timeZoneId)
- val variantGet: Expression = VariantGet(variantRef, pathLit,
child.dataType, vm.failOnError, tz)
- Seq(Literal(UTF8String.fromString(child.name),
DataTypes.StringType), variantGet)
- }
- CreateNamedStruct(childExprs)
- case _ =>
- val (dataIdx, dataField) = lookupDataField(rf.name)
- BoundReference(dataIdx, dataField.dataType, dataField.nullable)
- }
- }
-
- val projection = UnsafeProjection.create(exprs.toIndexedSeq,
DataTypeUtils.toAttributes(sparkDataSchema))
- Some(row => projection(row))
- }
- }
+ sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] =
+ buildVariantProjectorForStructPaths(sparkDataSchema, sparkRequiredSchema)
// Apply LogicalTypeAnnotation.variantType((byte) 1) to the variant group,
matching parquet 1.16+'s
// SparkToParquetSchemaConverter convention.
