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 3ba31dd37fff fix(spark): read shredded variants through internal
write-side parquet reads (#19558)
3ba31dd37fff is described below
commit 3ba31dd37fffa5521c8100d5e8f4617af5596048
Author: voonhous <[email protected]>
AuthorDate: Tue Aug 11 19:05:36 2026 +0800
fix(spark): read shredded variants through internal write-side parquet
reads (#19558)
* fix(spark): read shredded variants through internal write-side parquet
reads
Clustering, compaction base-file reads, and Spark-merger upsert merges read
parquet base files through SparkFileFormatInternalRowReaderContext without a
catalyst plan, so nothing rewrote VariantType columns the way
PushVariantIntoScan does for user queries. Against a shredded file the
reader
clipped the variant group to {metadata, value} and read value=null, and the
rewrite persisted the nulls, silently losing the variant data of every
carried-over row.
- Add SparkAdapter.buildFullVariantReadSchema, rewriting top-level
VariantType
fields into the full-variant projection struct (single child "0" at path
$)
mirroring RequestedVariantField.fullVariant; implemented for Spark
4.1/4.2,
None elsewhere (no shredded read support, no shredded writes either).
- In SparkFileFormatInternalRowReaderContext, apply the rewrite to internal
parquet base-file scans and restore native VariantType by projecting
child 0
after the read; composes with the VECTOR binary post-processing and is
inert
for Lance/ORC/log files and catalyst-provided schemas.
- Add shredded + unshredded COW clustering round-trips (small.file.limit=0
keeps the CoW small-file merge, which has a separate unfixed avro-path
leg,
out of the repro) and bulk_insert row-writer round-trips for both layouts.
Part of #19556; the avro-path leg (HoodieVariantReconstruction not engaging
on real files) is tracked on the issue.
* review(19558): address round-1 comments
- Move the buildFullVariantReadSchema implementation into BaseSpark4Adapter;
Spark 4.1/4.2 keep one-line opt-in overrides since Spark 4.0 must stay on
None (its reader cannot reconstruct shredded values).
- Reword the reader-context comment: Spark 4.0 does write shredded files
(the
write-side adapter methods have no version gate); only its read side is
missing. Note that MOR incremental, streaming and CDC reads build this
context without sparkRequiredSchema and take (and need) the same rewrite,
and that the nested-variant leg is deferred until a production write path
can produce nested shredded files.
- Extend the MOR compaction test with a second round that reads the shredded
base file produced by the first compaction through the internal reader;
rows 3 and 4 exist only in that base file.
- Assert typed_value in the pre-clustering base file so the shredded
clustering test cannot silently degrade into its unshredded twin.
* review(19558): address round-2 comments
- Add a batch incremental round trip over the shredded MOR table. On current
table versions it scans through the file-group-reader file format with a
catalyst schema (the overlay leg); the no-catalyst-schema legs (streaming
source, CDC) are tracked in #19578.
- Pin the unshredded clustering twin's layout: its pre-clustering base file
must not carry typed_value, so the twin cannot silently become a copy of
the shredded test.
* review(19558): narrow the no-catalyst-schema query legs to CDC
Under default configs only CDCFileGroupIterator builds this context without
a
catalyst schema; the streaming source and MergeOnRead relations reach their
RDD paths only with hoodie.file.group.reader.enabled=false, and batch
incremental always scans through the file format. Reword the reader-context
comment and the test note accordingly (#19578 description corrected too).
---
.../SparkFileFormatInternalRowReaderContext.scala | 79 +++++-
.../org/apache/spark/sql/hudi/SparkAdapter.scala | 17 ++
.../sql/hudi/dml/schema/TestVariantDataType.scala | 272 ++++++++++++++++++++-
.../spark/sql/adapter/BaseSpark4Adapter.scala | 25 +-
.../apache/spark/sql/adapter/Spark4_1Adapter.scala | 5 +
.../apache/spark/sql/adapter/Spark4_2Adapter.scala | 5 +
6 files changed, 396 insertions(+), 7 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 7671c64bc219..5752bcfea918 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
@@ -37,7 +37,7 @@ import org.apache.hudi.util.CloseableInternalRowIterator
import
org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter
import org.apache.spark.sql.HoodieInternalRowUtils
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{JoinedRow, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{BoundReference, Expression,
GetStructField, JoinedRow, UnsafeProjection}
import org.apache.spark.sql.execution.datasources.{PartitionedFile,
SparkColumnarFileReader}
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.internal.SQLConf
@@ -159,7 +159,40 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
structType
}
- val (readSchema, readFilters) =
getSchemaAndFiltersForRead(parquetReadStructType, hasRowIndexField)
+ // 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.
+ val isParquetBaseFile = !isInlineLog && !FSUtils.isLogFile(filePath) &&
+ HoodieFileFormat.fromFileExtension(filePath.getFileExtension) ==
HoodieFileFormat.PARQUET
+ val (readStructTypeForScan, variantOrdinals) =
+ if (sparkRequiredSchema.isEmpty && isParquetBaseFile) {
+ sparkAdapter.buildFullVariantReadSchema(parquetReadStructType) match {
+ case Some(rewritten) =>
+ val ordinals = rewritten.fields.indices
+ .filter(i => rewritten.fields(i).dataType !=
parquetReadStructType.fields(i).dataType)
+ .toSet
+ (rewritten, ordinals)
+ case None => (parquetReadStructType, Set.empty[Int])
+ }
+ } else {
+ (parquetReadStructType, Set.empty[Int])
+ }
+
+ val (readSchema, readFilters) =
getSchemaAndFiltersForRead(readStructTypeForScan, hasRowIndexField)
if (FSUtils.isLogFile(filePath)) {
// NOTE: now only primary key based filtering is supported for log files
// Position-based merging pairs log records with the RECORD_POSITIONS
bitmap by index (see
@@ -205,11 +238,21 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
readSchema, StructType(Seq.empty),
getSchemaHandler.getInternalSchemaOpt,
readFilters,
storage.getConf.asInstanceOf[StorageConfiguration[Configuration]],
tableSchemaOpt))
- // Post-process: convert binary VECTOR columns back to typed arrays
+ // Post-process: restore native VariantType from the full-variant
projection shape
+ // (child 0 of each rewritten field), then convert binary VECTOR columns
back to arrays.
+ val (variantRestoredIterator, variantRestoredSchema) = if
(variantOrdinals.nonEmpty) {
+ val restoredSchema = StructType(readSchema.fields.zipWithIndex.map {
case (f, i) =>
+ if (variantOrdinals.contains(i)) f.copy(dataType =
structType.fields(i).dataType) else f
+ })
+
(SparkFileFormatInternalRowReaderContext.wrapWithVariantRestore(rawIterator,
readSchema, variantOrdinals),
+ restoredSchema)
+ } else {
+ (rawIterator, readSchema)
+ }
if (vectorColumnInfo.nonEmpty) {
-
SparkFileFormatInternalRowReaderContext.wrapWithVectorConversion(rawIterator,
vectorColumnInfo, readSchema)
+
SparkFileFormatInternalRowReaderContext.wrapWithVectorConversion(variantRestoredIterator,
vectorColumnInfo, variantRestoredSchema)
} else {
- rawIterator
+ variantRestoredIterator
}
}
}
@@ -435,6 +478,32 @@ object SparkFileFormatInternalRowReaderContext {
VectorConversionUtils.replaceVectorColumnsWithBinary(structType, javaMap)
}
+ /**
+ * Restores native VariantType columns from the full-variant projection
shape requested for
+ * internal reads of parquet base files (see
SparkAdapter.buildFullVariantReadSchema): each
+ * rewritten column is a struct with a single child "0" holding the
reconstructed variant,
+ * so restoring is a projection of that child.
+ */
+ private[hudi] def wrapWithVariantRestore(
+ iterator: ClosableIterator[InternalRow],
+ readSchema: StructType,
+ variantOrdinals: Set[Int]): ClosableIterator[InternalRow] = {
+ val exprs: Seq[Expression] = readSchema.fields.zipWithIndex.map { case
(field, i) =>
+ val ref = BoundReference(i, field.dataType, field.nullable)
+ if (variantOrdinals.contains(i)) {
+ GetStructField(ref, 0, Some("0"))
+ } else {
+ ref: Expression
+ }
+ }.toSeq
+ val projection = UnsafeProjection.create(exprs)
+ new ClosableIterator[InternalRow] {
+ override def hasNext: Boolean = iterator.hasNext
+ override def next(): InternalRow = projection(iterator.next())
+ override def close(): Unit = iterator.close()
+ }
+ }
+
/**
* Wraps an iterator to convert binary VECTOR columns back to typed arrays.
* Unpacks bytes from FIXED_LEN_BYTE_ARRAY into GenericArrayData using the
canonical vector byte order.
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 7cc32490fbfb..f0b6292af455 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
@@ -518,6 +518,23 @@ trait SparkAdapter extends Serializable {
def buildVariantProjector(sparkDataSchema: StructType,
sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] = None
+ /**
+ * 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.
+ *
+ * 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).
+ */
+ def buildFullVariantReadSchema(schema: StructType): Option[StructType] = None
+
/**
* Generates a shredded Variant schema and marks it with write shredding
metadata.
*
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 a99393269cd1..8b270a097a3e 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
@@ -19,7 +19,7 @@
package org.apache.spark.sql.hudi.dml.schema
-import org.apache.hudi.HoodieSparkUtils
+import org.apache.hudi.{DataSourceReadOptions, HoodieSparkUtils}
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.HoodieSchemaException
import org.apache.hudi.common.testutils.HoodieTestUtils
@@ -250,9 +250,279 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
metaClient.reloadActiveTimeline()
assert(metaClient.getActiveTimeline.getCleanerTimeline.countInstants() >
0,
"Expected at least one .clean instant on the timeline after
compaction")
+
+ // Round 2: the compaction above compacted a log-only slice, so it never
read a
+ // parquet base file. Pin that the compacted base file is shredded, then
drive a
+ // second compaction that reads it through the internal reader
+ // (SparkFileFormatInternalRowReaderContext) and must carry rows 3 and
4, which
+ // exist only in that base file, forward (#19556).
+ val compactedFiles = listDataParquetFiles(tablePath)
+ assert(compactedFiles.nonEmpty, "Should have a compacted base parquet
file")
+ compactedFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"Compacted base file should carry typed_value.
Schema:\n$variantGroup")
+ }
+
+ // The v2-merged deltacommit above was the first after the compaction;
four more
+ // reach max.delta.commits = 5 and trip the second compaction inline.
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v1-r2"}'),
ts = 1003 where id = 1""")
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v2-r2"}'),
ts = 1004 where id = 2""")
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v1-r3"}'),
ts = 1005 where id = 1""")
+ spark.sql(s"""update $tableName set v = parse_json('{"key":"v2-r3"}'),
ts = 1006 where id = 2""")
+
+ metaClient.reloadActiveTimeline()
+
assertResult(2)(metaClient.getActiveTimeline.getCommitTimeline.filterCompletedInstants.countInstants)
+
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
+ Seq(1, "{\"key\":\"v1-r3\"}", 1005),
+ Seq(2, "{\"key\":\"v2-r3\"}", 1006),
+ Seq(3, "{\"key\":\"value3\"}", 1000),
+ Seq(4, "{\"key\":\"value4\"}", 1000)
+ )
+
+ // Incremental round trip over the shredded table: batch incremental
reads the
+ // shredded base file through the file-group-reader file format with a
catalyst
+ // schema, a stack nothing else in this suite pins for variant columns.
The
+ // no-catalyst-schema legs (CDC, and streaming with
+ // hoodie.file.group.reader.enabled=false) are tracked in #19578.
+ val incRows = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ .load(tablePath)
+ .selectExpr("id", "cast(v as string) as v", "ts")
+ .orderBy("id")
+ .collect()
+ assertResult(4)(incRows.length)
+ assertResult("{\"key\":\"v1-r3\"}")(incRows(0).getString(1))
+ assertResult("{\"key\":\"v2-r3\"}")(incRows(1).getString(1))
+ assertResult("{\"key\":\"value3\"}")(incRows(2).getString(1))
+ assertResult("{\"key\":\"value4\"}")(incRows(3).getString(1))
})
}
+ test("Test COW clustering preserves VARIANT values") {
+ // Same Spark 4.1 gate as the compaction test above: clustering reads the
shredded
+ // base files back through the native reader, which rejects the 3-field
shredded
+ // layout before SPARK-54410 (Spark 4.1+).
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ // Clustering rewrites ALL rows of the clustered file groups through the
internal
+ // write-side reader context (SparkReaderContextFactory ->
+ // SparkFileFormatInternalRowReaderContext), the stack whose blob
handling silently
+ // lost bytes in #19232. Nothing pinned its VARIANT behavior: this is
the first
+ // clustering coverage for the type. Shredding is forced so the rewrite
reads and
+ // rewrites the shredded layout, the default in production.
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v variant,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.parquet.variant.write.shredding.enabled = 'true',
+ | hoodie.parquet.variant.force.shredding.schema.for.test = 'key
string',
+ | hoodie.index.type = 'INMEMORY',
+ | hoodie.parquet.small.file.limit = '0',
+ | hoodie.clustering.inline = 'true',
+ | hoodie.clustering.inline.max.commits = '2'
+ | )
+ """.stripMargin)
+
+ // small.file.limit = 0 keeps the second insert in its own file group.
Otherwise the
+ // second commit bin-packs into the first file group and rewrites it
through the CoW
+ // small-file MERGE (a different internal read stack), conflating that
path's variant
+ // handling with the clustering rewrite this test isolates.
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ // The pre-clustering base file must actually carry the shredded layout;
without this
+ // check the test silently degrades into the unshredded twin below if
the forced
+ // shredding schema ever stops taking effect.
+ val preClusteringFiles = listDataParquetFiles(tablePath)
+ assert(preClusteringFiles.nonEmpty, "Should have at least one data
parquet file before clustering")
+ preClusteringFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"Pre-clustering base file should carry typed_value.
Schema:\n$variantGroup")
+ }
+
+ // Second commit trips inline clustering (max.commits = 2), which
rewrites the rows
+ // of the first commit too.
+ spark.sql(s"insert into $tableName values " +
+ "(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
+ "(4, parse_json('{\"key\":\"value4\"}'), 1000)")
+
+ // getLastClusteringInstant filters by action only, so a
REQUESTED/INFLIGHT instant
+ // satisfies isPresent; isCompleted confirms the rewrite finished.
+ val metaClient = createMetaClient(spark, tablePath)
+ val lastClustering =
metaClient.getActiveTimeline.getLastClusteringInstant
+ assert(lastClustering.isPresent && lastClustering.get.isCompleted,
+ "A COMPLETED clustering (replacecommit) instant must exist after
inline clustering; " +
+ "without a completed rewrite the round-trip below proves nothing")
+
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
+ Seq(1, "{\"key\":\"value1\"}", 1000),
+ Seq(2, "{\"key\":\"value2\"}", 1000),
+ Seq(3, "{\"key\":\"value3\"}", 1000),
+ Seq(4, "{\"key\":\"value4\"}", 1000)
+ )
+
+ // VARIANT must still surface as the native type after the clustering
rewrite.
+ val variantField = spark.table(tableName).schema.find(_.name == "v").get
+ assertResult("variant")(variantField.dataType.typeName)
+ })
+ }
+
+ test("Test COW clustering preserves unshredded VARIANT values") {
+ // Companion to the shredded clustering test above, with shredding
disabled. If this
+ // passes while the shredded one fails, the loss is specific to reading
the shredded
+ // layout inside the clustering rewrite, not to variant clustering in
general.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Variant clustering read-back
requires Spark 4.1 or higher")
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v variant,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.parquet.variant.write.shredding.enabled = 'false',
+ | hoodie.index.type = 'INMEMORY',
+ | hoodie.parquet.small.file.limit = '0',
+ | hoodie.clustering.inline = 'true',
+ | hoodie.clustering.inline.max.commits = '2'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ // Pin the layout: these files must NOT carry typed_value, or this twin
silently
+ // becomes a copy of the shredded test above and the unshredded leg of
the rewrite
+ // goes uncovered.
+ val preClusteringFiles = listDataParquetFiles(tablePath)
+ assert(preClusteringFiles.nonEmpty, "Should have at least one data
parquet file before clustering")
+ preClusteringFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(!variantGroup.containsField("typed_value"),
+ s"Unshredded base file must not carry typed_value.
Schema:\n$variantGroup")
+ }
+
+ spark.sql(s"insert into $tableName values " +
+ "(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
+ "(4, parse_json('{\"key\":\"value4\"}'), 1000)")
+
+ val metaClient = createMetaClient(spark, tablePath)
+ val lastClustering =
metaClient.getActiveTimeline.getLastClusteringInstant
+ assert(lastClustering.isPresent && lastClustering.get.isCompleted,
+ "A COMPLETED clustering (replacecommit) instant must exist after
inline clustering")
+
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
+ Seq(1, "{\"key\":\"value1\"}", 1000),
+ Seq(2, "{\"key\":\"value2\"}", 1000),
+ Seq(3, "{\"key\":\"value3\"}", 1000),
+ Seq(4, "{\"key\":\"value4\"}", 1000)
+ )
+ })
+ }
+
+ test("Test bulk_insert row-writer round-trips VARIANT") {
+ assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or
higher")
+
+ // bulk_insert takes the row-writer path (HoodieRowParquetWriteSupport),
which has its
+ // own variant writers (unshredded and shredded); no end-to-end test
covered it for
+ // either layout, only writer-level units
(TestHoodieRowParquetWriteSupportVariant).
+
+ // Unshredded bulk_insert: full round trip on any Spark 4.x.
+ withTempDir { tmp =>
+ val df = spark.sql(
+ """
+ |SELECT 1L AS id, parse_json('{"key":"value1"}') AS v, 1000L AS ts
+ |UNION ALL
+ |SELECT 2L AS id, parse_json('{"key":"value2"}') AS v, 1000L AS ts
+ |""".stripMargin)
+ df.write.format("hudi")
+ .option("hoodie.table.name", "variant_bulk_insert_unshredded")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "ts")
+ .option("hoodie.datasource.write.operation", "bulk_insert")
+ .option("hoodie.datasource.write.row.writer.enable", "true")
+ .option("hoodie.parquet.variant.write.shredding.enabled", "false")
+ .mode(SaveMode.Overwrite)
+ .save(tmp.getCanonicalPath)
+
+ val readDf = spark.read.format("hudi").load(tmp.getCanonicalPath)
+ assert(readDf.schema("v").dataType.typeName == "variant",
+ s"v should round-trip as native VariantType, got
${readDf.schema("v").dataType}")
+ val rows = readDf.selectExpr("id", "cast(v as string) as
v").orderBy("id").collect()
+ assert(rows.length == 2)
+ assert(rows(0).getString(1) == "{\"key\":\"value1\"}")
+ assert(rows(1).getString(1) == "{\"key\":\"value2\"}")
+ }
+
+ // Shredded bulk_insert exercises the row-writer shredding path end to
end; reading
+ // the shredded file back needs Spark 4.1+ (SPARK-54410).
+ if (HoodieSparkUtils.gteqSpark4_1) {
+ withTempDir { tmp =>
+ val df = spark.sql(
+ """
+ |SELECT 1L AS id, parse_json('{"key":"value1"}') AS v, 1000L AS ts
+ |UNION ALL
+ |SELECT 2L AS id, parse_json('{"key":"value2"}') AS v, 1000L AS ts
+ |""".stripMargin)
+ df.write.format("hudi")
+ .option("hoodie.table.name", "variant_bulk_insert_shredded")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "ts")
+ .option("hoodie.datasource.write.operation", "bulk_insert")
+ .option("hoodie.datasource.write.row.writer.enable", "true")
+ .option("hoodie.parquet.variant.write.shredding.enabled", "true")
+ .option("hoodie.parquet.variant.force.shredding.schema.for.test",
"key string")
+ .mode(SaveMode.Overwrite)
+ .save(tmp.getCanonicalPath)
+
+ // The written parquet must actually carry the shredded layout;
otherwise the
+ // read-back below silently validates the unshredded path a second
time.
+ val parquetFiles = listDataParquetFiles(tmp.getCanonicalPath)
+ assert(parquetFiles.nonEmpty, "Should have at least one data parquet
file")
+ parquetFiles.foreach { filePath =>
+ val schema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(schema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"bulk_insert with shredding forced should write typed_value.
Schema:\n$variantGroup")
+ }
+
+ val readDf = spark.read.format("hudi").load(tmp.getCanonicalPath)
+ val rows = readDf.selectExpr("id", "cast(v as string) as
v").orderBy("id").collect()
+ assert(rows.length == 2)
+ assert(rows(0).getString(1) == "{\"key\":\"value1\"}")
+ assert(rows(1).getString(1) == "{\"key\":\"value2\"}")
+ }
+ }
+ }
+
test("Test toHiveCompatibleSchema converts VariantType to physical struct") {
assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or
higher")
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 cf6f6981cf6c..1be6bf440a74 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
@@ -48,7 +48,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, StructType,
VariantType}
+import org.apache.spark.sql.types.{BinaryType, DataType, StructField,
StructType, VariantType}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.types.variant.Variant
@@ -296,6 +296,29 @@ abstract class BaseSpark4Adapter extends SparkAdapter with
Logging {
SparkShreddingUtils.variantShreddingSchema(dataType, isTopLevel,
isObjectField))
}
+ /**
+ * 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.
+ */
+ protected final def rewriteTopLevelVariantsForFullRead(schema: StructType):
Option[StructType] = {
+ var rewritten = false
+ val fields = schema.fields.map { f =>
+ if (isVariantType(f.dataType)) {
+ rewritten = true
+ // Mirrors RequestedVariantField.fullVariant in PushVariantIntoScan:
whole-variant
+ // access is a single child "0" at path "$" with failOnError and UTC.
+ f.copy(dataType = StructType(Array(StructField("0", VariantType,
+ metadata = VariantMetadata("$", failOnError = true, timeZoneId =
"UTC").toMetadata))))
+ } else {
+ f
+ }
+ }
+ if (rewritten) Some(StructType(fields)) else None
+ }
+
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 363d0b64446a..e16def5b9373 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
@@ -236,6 +236,11 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
VariantMetadata.isVariantStruct(structType)
}
+ // Spark 4.1 reconstructs shredded variants on read (SPARK-54410), so opt in
to the
+ // shared rewrite; Spark 4.0 stays on the default None.
+ override def buildFullVariantReadSchema(schema: StructType):
Option[StructType] =
+ rewriteTopLevelVariantsForFullRead(schema)
+
override def buildVariantProjector(sparkDataSchema: StructType,
sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] = {
// Quick check: any required field a variant projection struct?
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 b2790091b5fd..e971c4d6fc1d 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
@@ -236,6 +236,11 @@ class Spark4_2Adapter extends BaseSpark4Adapter {
VariantMetadata.isVariantStruct(structType)
}
+ // Spark 4.2 reconstructs shredded variants on read (SPARK-54410), so opt in
to the
+ // shared rewrite; Spark 4.0 stays on the default None.
+ override def buildFullVariantReadSchema(schema: StructType):
Option[StructType] =
+ rewriteTopLevelVariantsForFullRead(schema)
+
override def buildVariantProjector(sparkDataSchema: StructType,
sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] = {
// Quick check: any required field a variant projection struct?
