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 b1639e09e11b fix(spark): read shredded variants through the CDC and
legacy streaming paths, and restore partition values there (#19583)
b1639e09e11b is described below
commit b1639e09e11b71587f2f44ac4f6ccd23909505d2
Author: voonhous <[email protected]>
AuthorDate: Thu Aug 13 16:32:44 2026 +0800
fix(spark): read shredded variants through the CDC and legacy streaming
paths, and restore partition values there (#19583)
* fix(spark): read shredded variants through the CDC and legacy streaming
paths
Writing the coverage #19578 asked for surfaced three defects of the #19556
null-read family on the query paths that do not go through catalyst:
- HoodieMergeOnReadRDDV2 base-only splits bypass the internal reader context
and read through the plain skip-merging base reader, so a shredded base
file with no logs read null variants on the legacy
(hoodie.file.group.reader.enabled=false) streaming/relation path. Splits
with variant columns now take the file-group reader branch instead.
- CDCFileGroupIterator's BASE_FILE_INSERT case reads the new base file
directly with the plain table schema, so CDC after-images of insert
commits read null variants under every supplemental logging mode. The
direct read now applies the same full-variant rewrite and restore as the
reader context (helpers factored out of
SparkFileFormatInternalRowReaderContext; context behavior unchanged).
- InternalRowToJsonStringConverter had no VariantType case, so CDC images
serialized the VariantVal bean (raw bytes as base64) instead of the
variant's JSON. Variant columns now embed as real JSON nodes via
VariantVal.toString.
Tests: CDC round trip over a shredded COW table (insert + update,
OP_KEY_ONLY and DATA_BEFORE_AFTER, shredded layout pinned; single-row table
so the in-flight avro merge-path fix #19582 is not a prerequisite), and a
legacy-path MOR streaming round trip covering both the base-only and merged
split branches over a compacted shredded base file.
* review(19583): address round-1 comments
Main code:
- HoodieMergeOnReadRDDV2: gate the base-only re-route on
sparkAdapter.buildFullVariantReadSchema(...).isDefined instead of the mere
presence of a variant column. It is None below Spark 4.1, where the file
group reader reads the same nulls, so re-routing only lost the fast path.
- HoodieMergeOnReadRDDV2: keep the fast path for splits carrying partition
values parsed off the partition path. Only requiredSchemaReaderSkipMerging
appends those (drop.partition.columns, extract-from-path, bootstrap fast
read); the file group reader branch builds its PartitionedFile with
InternalRow.empty and an empty partition schema, so re-routing such a
split
would trade null variants for null partition columns. The same gap on the
merged branch predates this change and is left for a follow-up.
- InternalRowToJsonStringConverter: match the variant column on
dt.typeName rather than SparkAdapter.isVariantType. The guard runs for
every
non-string/array/map/struct field, and resolving the adapter needs a
version
module that is absent from hudi-spark-common's own test classpath: 13 of
the
14 TestInternalRowToJsonStringConverter cases errored with
ClassNotFoundException: Spark4_2Adapter. Also fall back to the raw
rendering
when readTree rejects the variant JSON (a non-finite double renders as a
bare
NaN/Infinity token), so a CDC query degrades instead of failing.
Tests:
- TestInternalRowToJsonStringConverter: cover the JSON embedding and the
malformed-JSON fallback; suite goes 14/14 (was 13 errors before the fix).
- TestStreamingSource: write through format("hudi"), not the fully qualified
name. Only Spark4DefaultSource overrides supportsDataType to accept
VariantType and it is reachable solely via the registered short name, so
the
write was dying with UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE before the read.
Add the testLegacyIncrementalStreamSource plan guard so the test cannot
pass
through a silent fallback to the file group reader. Correct the coverage
comment (the second batch is a log-only split, not a merged one) and add a
second testStream whose own checkpoint replays from INIT, which is what
actually produces a base + log slice on this path.
- TestVariantDataType: sweep the CDC round trip over shredded and unshredded
layouts rather than adding a second copy, so the rewrite branch is
covered on
both sides. Pin the test to HoodieRecordType.SPARK: a cdc table always
writes
through FileGroupReaderBasedMergeHandle, whose reader context follows the
merger's record type, and the AVRO leg's shredded base-file read is the
separate defect tracked as #19567/#19582. The previous claim that a
single-row
table removed that prerequisite was wrong: what fails is the CDC before
image
being written with value=null, which does not depend on carried-over rows.
Verified on Spark 4.1: TestVariantDataType 13/13 (2 cancelled as Spark-3
only),
TestStreamingSource 15/15. TestStreamingSource's suite-level
leaked-file-stream
abort reproduces identically on the pre-PR revision, so it is not from this
change.
* fix(spark): append partition values on the MOR RDD file-group-reader
branch
HoodieMergeOnReadRDDV2's file-group-reader branch sourced every projected
column from the data files, so it returned NULL partition columns whenever
those columns are not persisted there: drop.partition.columns, read-side
extraction from the partition path, or a bootstrap data-queries-only read.
SparkFileFormatInternalRowReaderContext builds its PartitionedFile with
InternalRow.empty and an empty partition schema, and the builder's
withPartitionPath only feeds the bootstrap merge, so nothing injected them.
Only the skip-merging fast path ever appended them, via
appendPartitionValues.
The output row shape was already correct -- TableSchemaResolver re-appends
dropped partition columns to the table schema, so the required schema keeps
them and the reader merely fills them with nulls -- which makes this a value
substitution at existing ordinals, with no column added, moved or dropped.
- Carry the parsed values on HoodieMergeOnReadFileSplit, resolved on the
driver
by the same routine the skip-merging reader relies on. A log-only slice
has no
base file to carry them, so they come off a log file's path; that uses
HoodieLogFile#getPath, since pathInfo is transient and null there.
- Splice them in with one per-split UnsafeProjection over a JoinedRow. Bound
references rather than literals: literals are inlined into generated
code, so
every distinct partition value would miss Spark's codegen cache. Mirrors
HoodieFileGroupReaderBasedFileFormat.appendPartitionAndProject.
- No-op unless it has to run: the split carries no values when the columns
are
read from the data files, the ordinal array is empty for a non-partitioned
table (including the metadata table), and unprojected partition columns
resolve to -1. The metadata-table Avro leg is untouched.
With the branch fixed, the base-only re-route added earlier in this PR no
longer has to avoid it, so its partition-value escape hatch is dropped and
splits with shredded variant columns re-route unconditionally.
Reachable today only from legacy MOR streaming
(hoodie.file.group.reader.enabled
=false); batch MOR always scans through
HoodieFileGroupReaderBasedFileFormat,
which appends partition values itself.
Tests, in TestLegacyParquetReadPath since it already constructs the legacy
relations directly:
- MOR snapshot over drop.partition.columns with a partial upsert, so one
file
group merges base plus log while the others stay base-only. Verified red
before this change (base-served p0 rows came back null) and green after.
- MOR incremental over a log-only slice. This one passes with or without the
splicing above, so it is not a repro: its rows come from log records,
which do
carry the partition column because HUDI-6926 makes a MOR upsert ignore
drop.partition.columns. It pins the log-only resolution off
HoodieLogFile#getPath and agreement with the file-group-reader oracle.
Noted
as such in the test.
* review(19583): correct the variant image fallback rationale
Round-2 review pointed out the stated trigger was wrong, and it is: Spark
guards the double and float arms of Variant.toJsonImpl with isFinite and
sends
the non-finite one through appendQuoted, so a variant renders NaN and
Infinity
QUOTED and readTree never sees a bare token. Verified against the bytecode
and
by rendering hand-built variants on Spark 4.0.2 and 4.1.1.
The fallback still earns its place, for a different reason. Jackson's
default
StreamReadConstraints cap field names at 50k chars, strings at 20M and
nesting
at 1000 levels, while a variant may hold all three well inside its own 128MB
limit, and castToVariant applies no such gate. All three were reproduced and
all three arrive as StreamConstraintsException, a JsonProcessingException,
so
the existing catch covers them.
- Reword the comment to name the real triggers.
- Record why value.toString stays outside the try: it throws
MALFORMED_VARIANT
(a SparkRuntimeException, which JsonProcessingException does not cover) on
corrupt bytes. That is a data-integrity signal an operator needs, not a
rendering quirk to swallow into a CDC image, and there would be no
rendering
left to fall back to anyway.
- Retarget the test, which asserted a rendering production cannot produce,
onto
over-deep nesting, and rename it accordingly. It expects the value as a
JSON
string, so it only passes when the fallback actually fires.
* review(19583): embed the variant image verbatim and correct two comments
Round-3 review. The write-side gap is real: Jackson enforces a nesting cap
on
output too (StreamWriteConstraints, also 1000 levels), inside convert's
writeValueAsString and therefore outside the branch's catch, so a variant
deep
enough to clear the read limit but not the write limit once the image's own
object levels are added still failed the query. Embedding the validated
rendering as a RawValue closes it: that goes out through
JsonGenerator.writeRawValue, which keeps no nesting context at all.
Measured while confirming this, on jackson 2.18.2/2.20.0/2.21.2: the window
is
one level per enclosing object for an object-rendered variant and one more
for
an array-rendered one, because jackson-core's WriterBasedJsonGenerator
validates the PARENT depth in writeStartObject(Object) and so tolerates one
extra level. A variant at the top of the image rendering as an array fails
at
exactly depth 1000, which is what the new test pins.
RawValue is byte-identical to the parsed-node path for every rendering
Variant.toJson can produce -- checked against the real renderer across
decimals, doubles, escaping, key order and all scalar shapes -- and it
drops a
parse-then-rebuild round trip. It does mean unvalidated text would be
spliced
verbatim, so the gate is tightened to match:
- FAIL_ON_TRAILING_TOKENS on the mapper, since readTree otherwise parses a
valid prefix and leaves the rest unconsumed;
- an explicit MissingNode check, since readTree answers blank input with
MissingNode rather than throwing, and RawValue would emit nothing at all.
Neither is producible by Variant.toJson, which always emits exactly one
complete value; they are defence in depth for the verbatim embed.
Also, from the same review:
- MergeOnReadSnapshotRelation: the split's partition values are non-empty on
any of shouldExtractPartitionValuesFromPartitionPath's three triggers, not
only when the columns are omitted from the data files -- the
extract-from-path
read option applies to tables that persist them too. Comment corrected.
- TestVariantDataType: the compaction test still pointed at #19578 as
tracking
the no-catalyst-schema legs, which this PR closes and covers. Repointed
at the
CDC round trip and the TestStreamingSource test.
Verified: TestInternalRowToJsonStringConverter 16/16, with the depth test
red
before the RawValue change (StreamWriteConstraints exceeded) and green
after;
TestVariantDataType 13/13, so the CDC round trip still reads its images back
through get_json_object.
---
.../SparkFileFormatInternalRowReaderContext.scala | 52 ++++++----
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 11 ++-
.../org/apache/hudi/HoodieMergeOnReadRDDV2.scala | 77 ++++++++++++++-
.../apache/hudi/MergeOnReadSnapshotRelation.scala | 15 ++-
.../org/apache/hudi/cdc/CDCFileGroupIterator.scala | 20 +++-
.../cdc/InternalRowToJsonStringConverter.scala | 45 +++++++++
.../cdc/TestInternalRowToJsonStringConverter.scala | 79 ++++++++++++++-
.../functional/TestLegacyParquetReadPath.scala | 106 ++++++++++++++++++++-
.../hudi/functional/TestStreamingSource.scala | 105 +++++++++++++++++++-
.../sql/hudi/dml/schema/TestVariantDataType.scala | 106 ++++++++++++++++++++-
10 files changed, 574 insertions(+), 42 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 5752bcfea918..acab0f7a0bbc 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
@@ -180,14 +180,8 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
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])
- }
+
SparkFileFormatInternalRowReaderContext.fullVariantReadSchemaWithOrdinals(parquetReadStructType)
+ .getOrElse((parquetReadStructType, Set.empty[Int]))
} else {
(parquetReadStructType, Set.empty[Int])
}
@@ -479,15 +473,28 @@ object SparkFileFormatInternalRowReaderContext {
}
/**
- * 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.
+ * 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).
*/
- private[hudi] def wrapWithVariantRestore(
- iterator: ClosableIterator[InternalRow],
- readSchema: StructType,
- variantOrdinals: Set[Int]): ClosableIterator[InternalRow] = {
+ private[hudi] def fullVariantReadSchemaWithOrdinals(structType: StructType):
Option[(StructType, Set[Int])] = {
+
SparkAdapterSupport.sparkAdapter.buildFullVariantReadSchema(structType).map {
rewritten =>
+ val ordinals = rewritten.fields.indices
+ .filter(i => rewritten.fields(i).dataType !=
structType.fields(i).dataType)
+ .toSet
+ (rewritten, ordinals)
+ }
+ }
+
+ /**
+ * Projection restoring native VariantType columns from the full-variant
projection shape:
+ * each rewritten column is a struct with a single child "0" holding the
reconstructed
+ * variant, so restoring is a projection of that child. NOTE:
UnsafeProjection reuses its
+ * output buffer; callers that buffer rows must copy them.
+ */
+ private[hudi] def variantRestoreProjection(readSchema: StructType,
variantOrdinals: Set[Int]): UnsafeProjection = {
val exprs: Seq[Expression] = readSchema.fields.zipWithIndex.map { case
(field, i) =>
val ref = BoundReference(i, field.dataType, field.nullable)
if (variantOrdinals.contains(i)) {
@@ -496,7 +503,18 @@ object SparkFileFormatInternalRowReaderContext {
ref: Expression
}
}.toSeq
- val projection = UnsafeProjection.create(exprs)
+ UnsafeProjection.create(exprs)
+ }
+
+ /**
+ * Restores native VariantType columns from the full-variant projection
shape requested for
+ * internal reads of parquet base files (see
SparkAdapter.buildFullVariantReadSchema).
+ */
+ private[hudi] def wrapWithVariantRestore(
+ iterator: ClosableIterator[InternalRow],
+ readSchema: StructType,
+ variantOrdinals: Set[Int]): ClosableIterator[InternalRow] = {
+ val projection = variantRestoreProjection(readSchema, variantOrdinals)
new ClosableIterator[InternalRow] {
override def hasNext: Boolean = iterator.hasNext
override def next(): InternalRow = projection(iterator.next())
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
index e1e01103bdf2..3a0a363fa2eb 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
@@ -427,11 +427,14 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
* and pass this reader on parquet file. So that, we can query the partition
columns.
*/
+ protected def getPartitionColumnsAsInternalRow(filePath: StoragePath):
InternalRow =
+ getPartitionColumnsAsInternalRowInternal(filePath, metaClient.getBasePath,
shouldExtractPartitionValuesFromPartitionPath)
+
protected def getPartitionColumnsAsInternalRow(file: StoragePathInfo):
InternalRow =
- getPartitionColumnsAsInternalRowInternal(file, metaClient.getBasePath,
shouldExtractPartitionValuesFromPartitionPath)
+ getPartitionColumnsAsInternalRow(file.getPath)
protected def getPartitionColumnValuesAsInternalRow(file: StoragePathInfo):
InternalRow =
- getPartitionColumnsAsInternalRowInternal(file,
+ getPartitionColumnsAsInternalRowInternal(file.getPath,
metaClient.getBasePath, extractPartitionValuesFromPartitionPath = true)
protected def usePartitionValueExtractorOnRead(optParams: Map[String,
String], sparkSession: SparkSession): Boolean = {
@@ -439,11 +442,11 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
DataSourceReadOptions.USE_PARTITION_VALUE_EXTRACTOR_ON_READ.defaultValue).toBoolean
}
- protected def getPartitionColumnsAsInternalRowInternal(file:
StoragePathInfo, basePath: StoragePath,
+ protected def getPartitionColumnsAsInternalRowInternal(filePath:
StoragePath, basePath: StoragePath,
extractPartitionValuesFromPartitionPath: Boolean): InternalRow = {
if (extractPartitionValuesFromPartitionPath) {
val tablePathWithoutScheme = basePath.getPathWithoutSchemeAndAuthority
- val partitionPathWithoutScheme =
file.getPath.getParent.getPathWithoutSchemeAndAuthority
+ val partitionPathWithoutScheme =
filePath.getParent.getPathWithoutSchemeAndAuthority
val relativePath =
tablePathWithoutScheme.toUri.relativize(partitionPathWithoutScheme.toUri).toString
val timeZoneId = conf.get("timeZone",
sparkSession.sessionState.conf.sessionLocalTimeZone)
val rowValues = HoodieSparkUtils.parsePartitionColumnValues(
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 3ae45815c679..267b4e4389a6 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
@@ -33,6 +33,7 @@ import org.apache.hudi.common.table.log.InstantRange.RangeType
import org.apache.hudi.common.table.read.{HoodieFileGroupReader,
HoodieRecordReader}
import org.apache.hudi.common.table.read.lsm.{HoodieLsmFileGroupReader,
LsmReaderUtils}
import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.common.util.ValidationUtils.checkState
import org.apache.hudi.common.util.collection.ClosableIterator
import
org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
import
org.apache.hudi.metadata.HoodieTableMetadata.getDataTableBasePathFromMetadataTable
@@ -47,7 +48,7 @@ import org.apache.spark.{HoodieSparkInputMetricsUtils,
Partition, SerializableWr
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.UnsafeProjection
+import org.apache.spark.sql.catalyst.expressions.{BoundReference, Expression,
JoinedRow, UnsafeProjection}
import org.apache.spark.sql.execution.datasources.{FileFormat,
SparkColumnarFileReader}
import org.apache.spark.sql.hudi.MultipleColumnarFileFormatReader
import org.apache.spark.sql.internal.SQLConf
@@ -148,12 +149,35 @@ 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.
+ private val shouldRerouteVariantSplit: Boolean =
+
sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined
+
+ // Ordinal each table partition column occupies in the required schema, or
-1 when it is not
+ // projected. Indexed by the table-config partition-field order, i.e. the
very order in which
+ // HoodieBaseRelation#getPartitionColumnsAsInternalRow emits the values
carried by a split.
+ // Resolved on the driver: only this array is shipped to the executors.
+ private val partitionColumnOrdinals: Array[Int] = {
+ val caseSensitive = sqlConf.caseSensitiveAnalysis
+ val requiredFieldNames = requiredSchema.structTypeSchema.fieldNames
+
metaClient.getTableConfig.getPartitionFields.orElse(Array.empty[String]).map {
partitionColumn =>
+ requiredFieldNames.indexWhere(fieldName =>
+ if (caseSensitive) fieldName == partitionColumn else
fieldName.equalsIgnoreCase(partitionColumn))
+ }
+ }
+
override def compute(split: Partition, context: TaskContext):
Iterator[InternalRow] = {
val partition = split.asInstanceOf[HoodieMergeOnReadPartition]
val bytesReadCallback =
HoodieSparkInputMetricsUtils.getFSBytesReadOnThreadCallback()
val iter: Iterator[InternalRow] = partition.split match {
- case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty =>
+ case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty &&
!shouldRerouteVariantSplit =>
val projectedReader =
projectReader(fileReaders.requiredSchemaReaderSkipMerging,
requiredSchema.structTypeSchema)
projectedReader(dataFileOnlySplit.dataFile.get)
@@ -220,7 +244,7 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
.withInternalSchemaOpt(HOption.ofNullable(tableSchema.internalSchema.orNull))
.build()
}
- convertCloseableIterator(fileGroupReader.getClosableIterator)
+ convertCloseableIterator(fileGroupReader.getClosableIterator,
partition.split.partitionValues)
}
}
@@ -265,16 +289,59 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
}
}
- private def convertCloseableIterator(closeableFileGroupRecordIterator:
ClosableIterator[InternalRow]): Iterator[InternalRow] = {
+ private def convertCloseableIterator(closeableFileGroupRecordIterator:
ClosableIterator[InternalRow],
+ partitionValues: InternalRow):
Iterator[InternalRow] = {
+ // NOTE: built here, i.e. once per split on the executor -- a projection
is not serializable.
+ val mapper: InternalRow => InternalRow =
partitionValueMapper(partitionValues).getOrElse(identity)
new Iterator[InternalRow] with Closeable {
override def hasNext: Boolean = closeableFileGroupRecordIterator.hasNext
- override def next(): InternalRow =
closeableFileGroupRecordIterator.next()
+ override def next(): InternalRow =
mapper(closeableFileGroupRecordIterator.next())
override def close(): Unit = closeableFileGroupRecordIterator.close()
}
}
+ /**
+ * Builds the mapper splicing a split's partition values into the rows the
file-group reader
+ * produces, or None when there is nothing to splice.
+ *
+ * The file-group reader sources every projected column from the data files,
so it hands back
+ * nulls for the partition columns whenever those are not persisted there
+ * ([["hoodie.datasource.write.drop.partition.columns"]], read-side
extraction from the partition
+ * path, or a bootstrap data-queries-only read). Only the values parsed off
the partition path
+ * carry them, which is what a split holds.
+ *
+ * Binding at existing ordinals is sufficient because the output row shape
is already correct:
+ * [[org.apache.hudi.common.table.TableSchemaResolver]] re-appends dropped
partition columns to
+ * the table schema, so the required schema keeps them and the reader merely
fills them with
+ * nulls. No column is added, moved or dropped here.
+ *
+ * @param partitionValues values parsed off the partition path, in
table-config partition-field
+ * order (matching [[partitionColumnOrdinals]]),
empty when the partition
+ * columns are read from the data files as usual
+ */
+ private def partitionValueMapper(partitionValues: InternalRow):
Option[InternalRow => InternalRow] = {
+ if (partitionValues.numFields == 0 || partitionColumnOrdinals.forall(_ <
0)) {
+ None
+ } else {
+ checkState(partitionValues.numFields == partitionColumnOrdinals.length,
+ s"Expected ${partitionColumnOrdinals.length} partition values but got
${partitionValues.numFields}")
+ val requiredFields = requiredSchema.structTypeSchema.fields
+ // NOTE: The partition values are bound as references into the right
half of a JoinedRow rather
+ // than substituted as literals: literals are inlined into the
generated code, so every
+ // distinct partition value would miss Spark's codegen cache.
+ val projectedFields: Seq[Expression] = requiredFields.zipWithIndex.map {
case (field, ordinal) =>
+ val partitionFieldIdx = partitionColumnOrdinals.indexOf(ordinal)
+ val boundOrdinal = if (partitionFieldIdx >= 0) requiredFields.length +
partitionFieldIdx else ordinal
+ BoundReference(boundOrdinal, field.dataType, field.nullable)
+ }.toSeq
+ val projection = UnsafeProjection.create(projectedFields)
+ val joinedRow = new JoinedRow()
+ Some((row: InternalRow) => projection.apply(joinedRow(row,
partitionValues)))
+ }
+ }
+
private def withInputMetrics(iter: Iterator[InternalRow],
closeableIter: Iterator[InternalRow],
context: TaskContext,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
index aa96512ca18f..f44891619274 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
@@ -37,7 +37,8 @@ import org.apache.spark.sql.types.StructType
import scala.collection.JavaConverters._
case class HoodieMergeOnReadFileSplit(dataFile: Option[PartitionedFile],
- logFiles: List[HoodieLogFile]) extends
HoodieFileSplit
+ logFiles: List[HoodieLogFile],
+ partitionValues: InternalRow =
InternalRow.empty) extends HoodieFileSplit
case class MergeOnReadSnapshotRelation(override val sqlContext: SQLContext,
override val optParams: Map[String,
String],
@@ -132,7 +133,17 @@ abstract class BaseMergeOnReadSnapshotRelation(sqlContext:
SQLContext,
getPartitionColumnsAsInternalRow(file.getPathInfo),
file.getPathInfo.getPath, 0, file.getFileSize)
}
- HoodieMergeOnReadFileSplit(partitionedBaseFile, logFiles)
+ // Non-empty exactly when a reader has to take the partition columns off
the path rather than the
+ // data files, i.e. on any of
shouldExtractPartitionValuesFromPartitionPath's triggers: dropped
+ // partition columns, the extract-from-path read option (which also
covers tables that do persist
+ // them), or a bootstrap data-queries-only read. A log-only slice still
lives in the partition
+ // directory, so a log file's path resolves them just as the base file
does.
+ // NOTE: HoodieLogFile#getPathInfo is transient and null for log files
built from a path.
+ val partitionValues =
partitionedBaseFile.map(_.partitionValues).getOrElse {
+ logFiles.headOption.map(f =>
getPartitionColumnsAsInternalRow(f.getPath)).getOrElse(InternalRow.empty)
+ }
+
+ HoodieMergeOnReadFileSplit(partitionedBaseFile, logFiles,
partitionValues)
}.toList
}
}
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 44630e8d5bbc..b845dbe38d35 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,8 +378,24 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
val pf = sparkPartitionedFileUtils.createPartitionedFile(
InternalRow.empty, absCDCPath, 0, fileStatus.getLength)
- recordIter = baseFileReader.read(pf,
originTableSchema.structTypeSchema, new StructType(),
- toJavaOption(originTableSchema.internalSchema), Seq.empty, conf,
tableSchemaOpt)
+ // 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.
+ val baseRows = SparkFileFormatInternalRowReaderContext
+
.fullVariantReadSchemaWithOrdinals(originTableSchema.structTypeSchema) match {
+ case Some((rewritten, ordinals)) =>
+ val restore =
SparkFileFormatInternalRowReaderContext.variantRestoreProjection(rewritten,
ordinals)
+ baseFileReader.read(pf, rewritten, new StructType(),
+ toJavaOption(originTableSchema.internalSchema), Seq.empty,
conf, tableSchemaOpt)
+ .map(row => restore(row).copy(): InternalRow)
+ case None =>
+ baseFileReader.read(pf, originTableSchema.structTypeSchema, new
StructType(),
+ toJavaOption(originTableSchema.internalSchema), Seq.empty,
conf, tableSchemaOpt)
+ }
+ recordIter = baseRows
.map(record => BufferedRecords.fromEngineRecord(record, schema,
readerContext.getRecordContext, orderingFieldNames, false))
case BASE_FILE_DELETE =>
assert(currentCDCFileSplit.getBeforeFileSlice.isPresent)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
index 85c69548533d..4d01831e9a8c 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
@@ -19,7 +19,9 @@
package org.apache.hudi.cdc
import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.core.JsonProcessingException
import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.databind.util.RawValue
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util.{ArrayData, MapData}
@@ -32,6 +34,9 @@ class InternalRowToJsonStringConverter(schema: StructType) {
val _mapper = new ObjectMapper
_mapper.setSerializationInclusion(Include.NON_ABSENT)
_mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ // The variant branch below embeds its input verbatim once readTree
accepts it, and readTree on
+ // its own is happy to parse a valid prefix and leave the rest unconsumed.
+ _mapper.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, true)
_mapper.registerModule(DefaultScalaModule)
_mapper
}
@@ -90,6 +95,37 @@ class InternalRowToJsonStringConverter(schema: StructType) {
structMap.toMap
case _ => value // fallback
}
+ case dt if dt.typeName ==
InternalRowToJsonStringConverter.VARIANT_TYPE_NAME =>
+ // VariantVal.toString renders the variant as JSON; embed that
rendering verbatim so the
+ // image carries the variant's structure. Falling through to the
default would serialize
+ // the VariantVal bean, i.e. its raw value/metadata bytes as base64.
+ // Matched on the type name rather than SparkAdapter.isVariantType:
this guard is
+ // evaluated for every non-string/array/map/struct field, and
resolving the adapter
+ // needs a version module that is not on hudi-spark-common's own
test classpath.
+ val variantJson = value.toString
+ try {
+ // readTree is a validation gate only, and its result is
discarded: embedding the parsed
+ // node instead would re-serialize the tree through the generator,
and Jackson's write-side
+ // nesting cap (StreamWriteConstraints, also 1000) is enforced by
writeValueAsString in
+ // convert, outside this block. A variant deep enough to clear the
read limit but not the
+ // write limit once the image's own object levels are added would
fail the query there.
+ // RawValue goes out through JsonGenerator.writeRawValue, which
keeps no nesting context.
+ val parsed = mapper.readTree(variantJson)
+ // readTree accepts blank input as a MissingNode instead of
throwing, and RawValue would
+ // then emit nothing at all, leaving a malformed image.
Trailing-token input is refused by
+ // FAIL_ON_TRAILING_TOKENS above, which readTree does not check on
its own.
+ if (parsed == null || parsed.isMissingNode) variantJson else new
RawValue(variantJson)
+ } catch {
+ // A variant can hold a field name, string or nesting depth past
Jackson's default
+ // StreamReadConstraints (50k chars, 20M chars, 1000 levels) while
staying well inside
+ // the variant size limit, and all three arrive here as
StreamConstraintsException. A
+ // CDC image is diagnostic data rather than the table's data, so
keep the rendering as
+ // a plain string instead of failing the query over it.
+ // NOTE: value.toString is deliberately outside this block. It
throws MALFORMED_VARIANT
+ // on corrupt bytes, which is a data-integrity problem an operator
has to see, not a
+ // rendering quirk to paper over -- and there would be no
rendering left to fall back to.
+ case _: JsonProcessingException => variantJson
+ }
case _ =>
// For primitive types and other unsupported types, return as is
value
@@ -97,3 +133,12 @@ class InternalRowToJsonStringConverter(schema: StructType) {
}
}
}
+
+object InternalRowToJsonStringConverter {
+
+ /**
+ * Type name of Spark's VariantType. Matched by name so this module, which
also compiles
+ * against Spark 3 where the type does not exist, needs neither the symbol
nor a SparkAdapter.
+ */
+ private val VARIANT_TYPE_NAME = "variant"
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
index 0c9bb613184f..7ea8cae4bc40 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
@@ -18,15 +18,16 @@
package org.apache.hudi.cdc
-import org.apache.hudi.HoodieTableSchema
+import org.apache.hudi.{HoodieSparkUtils, HoodieTableSchema}
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData}
-import org.apache.spark.sql.types.{ArrayType, DataTypes, MapType, Metadata,
StructField, StructType}
+import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, MapType,
Metadata, StructField, StructType}
import org.apache.spark.unsafe.types.UTF8String
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.Test
class TestInternalRowToJsonStringConverter {
@@ -167,6 +168,80 @@ class TestInternalRowToJsonStringConverter {
assertEquals("""{"id":8,"name":"empty_map_test","properties":{}}""",
converted.toString)
}
+ @Test
+ def variantColumnEmbedsItsJson(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_0, "VariantType requires Spark 4.0
or higher")
+ // VariantVal.toString renders the variant as JSON, so the image must
carry that structure
+ // rather than a bean rendering of the raw value/metadata bytes.
+ val row = InternalRow.fromSeq(Seq(1,
variantRendering("""{"key":"value1"}""")))
+ assertEquals("""{"id":1,"v":{"key":"value1"}}""",
+ new
InternalRowToJsonStringConverter(variantSchema.structTypeSchema).convert(row).toString)
+ }
+
+ @Test
+ def variantColumnFallsBackToRawRenderingWhenJacksonRejectsIt(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_0, "VariantType requires Spark 4.0
or higher")
+ // Jackson's default StreamReadConstraints cap nesting at 1000 levels,
field names at 50k chars
+ // and strings at 20M; a variant can exceed all three well inside its own
size limit, and
+ // VariantVal.toString renders it faithfully. The image keeps that
rendering rather than failing
+ // the whole CDC query. Nesting is the cheapest of the three to provoke.
+ val tooDeeplyNested = ("[" * 1001) + "1" + ("]" * 1001)
+ val row = InternalRow.fromSeq(Seq(1, variantRendering(tooDeeplyNested)))
+ assertEquals(s"""{"id":1,"v":"$tooDeeplyNested"}""",
+ new
InternalRowToJsonStringConverter(variantSchema.structTypeSchema).convert(row).toString)
+ }
+
+ @Test
+ def variantColumnSurvivesJacksonWriteSideNestingCap(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_0, "VariantType requires Spark 4.0
or higher")
+ // Jackson caps nesting on the write side too, and the image adds its own
object level on top of
+ // the variant, so a depth the read side accepts could still fail in
writeValueAsString -- outside
+ // the fallback, taking the whole CDC query with it. The rendering is
embedded verbatim rather
+ // than as a parsed node precisely so that write-side accounting never
sees it.
+ val atReadLimit = ("[" * 1000) + "1" + ("]" * 1000)
+ val row = InternalRow.fromSeq(Seq(1, variantRendering(atReadLimit)))
+ val image = new
InternalRowToJsonStringConverter(variantSchema.structTypeSchema).convert(row).toString
+ // Embedded as a real array, i.e. neither the fallback nor a write-side
failure.
+ assertEquals(s"""{"id":1,"v":$atReadLimit}""", image)
+ }
+
+ @Test
+ def variantColumnFallsBackWhenRenderingHasTrailingTokens(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_0, "VariantType requires Spark 4.0
or higher")
+ // readTree parses a valid prefix and leaves the rest, so without
FAIL_ON_TRAILING_TOKENS a
+ // verbatim embed would splice the trailing text straight into the image
and corrupt it.
+ val row = InternalRow.fromSeq(Seq(1, variantRendering("""{"a":1}
trailing""")))
+ assertEquals("""{"id":1,"v":"{\"a\":1} trailing"}""",
+ new
InternalRowToJsonStringConverter(variantSchema.structTypeSchema).convert(row).toString)
+ }
+
+ /**
+ * Stands in for a VariantVal: the converter reaches the value only through
toString, which is
+ * what renders a real variant as JSON. Constructing a genuine VariantVal
here would need
+ * Spark-4-only symbols, and this module compiles against Spark 3 as well;
the end-to-end
+ * behaviour over real variants is covered by the CDC round trip in
TestVariantDataType.
+ *
+ * Note this stands in only for renderings a real variant can produce. Spark
quotes non-finite
+ * doubles (Variant.toJsonImpl guards them with isFinite and takes the
appendQuoted arm), so no
+ * variant renders a bare NaN or Infinity token.
+ */
+ private def variantRendering(json: String): AnyRef = new AnyRef {
+ override def toString: String = json
+ }
+
+ private def variantSchema: HoodieTableSchema = {
+ val structTypeSchema = new StructType(Array[StructField](
+ StructField("id", DataTypes.IntegerType, nullable = false,
Metadata.empty),
+ StructField("v", DataType.fromDDL("variant"), nullable = true,
Metadata.empty)))
+ val avroSchemaStr: String =
+ """{"type": "record", "name": "test", "fields": [
+ |{"name": "id", "type": "int"},
+ |{"name": "v", "type": {"type": "record", "name": "variant",
"logicalType": "variant",
+ | "fields": [{"name": "metadata", "type": "bytes"}, {"name": "value",
"type": "bytes"}]}}
+ |]}""".stripMargin
+ HoodieTableSchema(structTypeSchema, HoodieSchema.parse(avroSchemaStr),
Option.empty[InternalSchema])
+ }
+
private def hoodieTableSchema: HoodieTableSchema = {
val structTypeSchema = new StructType(Array[StructField](
StructField("uuid", DataTypes.IntegerType, nullable = false,
Metadata.empty),
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLegacyParquetReadPath.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLegacyParquetReadPath.scala
index fac97a0cbecb..1cac6b2f7f25 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLegacyParquetReadPath.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLegacyParquetReadPath.scala
@@ -17,11 +17,11 @@
package org.apache.hudi.functional
-import org.apache.hudi.{BaseFileOnlyRelation, DataSourceReadOptions,
DataSourceWriteOptions, IncrementalRelationV1, IncrementalRelationV2,
ScalaAssertionSupport}
+import org.apache.hudi.{BaseFileOnlyRelation, DataSourceReadOptions,
DataSourceWriteOptions, IncrementalRelationV1, IncrementalRelationV2,
MergeOnReadIncrementalRelationV2, MergeOnReadSnapshotRelation,
ScalaAssertionSupport}
import org.apache.hudi.common.config.HoodieReaderConfig
import org.apache.hudi.common.table.HoodieTableConfig
import org.apache.hudi.common.table.log.InstantRange.RangeType
-import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
import org.apache.hudi.testutils.HoodieSparkClientTestBase
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
@@ -156,6 +156,36 @@ class TestLegacyParquetReadPath extends
HoodieSparkClientTestBase with ScalaAsse
assertEquals(expected, actual)
}
+ // Inline compaction is switched on automatically for batch MOR writes
(HoodieSparkSqlWriter), so it
+ // is pinned off here to keep the second deltacommit in a log file rather
than a rewritten base file.
+ private val morDropPartitionColumnsOpts = Map(
+ DataSourceWriteOptions.TABLE_TYPE.key ->
DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL,
+ HoodieTableConfig.DROP_PARTITION_COLUMNS.key -> "true",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false")
+
+ /**
+ * MOR table whose base files do not carry the partition column, so its
values exist only on the
+ * partition path. Deltacommit 1 inserts ids 1..30 across p0/p1/p2;
deltacommit 2 upserts a strict
+ * subset of the p0 rows, so the p0 file group gains a log file while p1/p2
stay base-only.
+ *
+ * Both "strict subset" and "single partition" are deliberate: the p1/p2
groups keep the
+ * skip-merging fast path in the same query, and the p0 rows left untouched
by deltacommit 2 are
+ * served from the base file, which is where the partition column is
missing. Note that
+ * HoodieSparkSqlWriter forces drop.partition.columns off for MOR upserts
(HUDI-6926), so the log
+ * records themselves do carry the column -- only the base-file records do
not.
+ */
+ private def writeMorDropPartitionColumnsCommits(): Unit = {
+ writeBatch(makeRows(1 to 30, ts = 1L, i => i * 10L),
+ DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
morDropPartitionColumnsOpts)
+ writeBatch(makeRows((1 to 30).filter(_ % 6 == 0), ts = 2L, i => i * 100L),
+ DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
morDropPartitionColumnsOpts)
+ }
+
+ /** Distinct `partition` values, rendering a null (the symptom under test)
as "null" so that the
+ * assertion reports it rather than failing while sorting. */
+ private def distinctPartitions(df: DataFrame): Seq[String] =
+ df.select("partition").distinct().collect().map(r =>
String.valueOf(r.get(0))).sorted.toSeq
+
private def fgReaderDf(extraOpts: Map[String, String] = Map.empty):
DataFrame =
spark.read.format("hudi")
.option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key, "true")
@@ -193,6 +223,16 @@ class TestLegacyParquetReadPath extends
HoodieSparkClientTestBase with ScalaAsse
spark.baseRelationToDataFrame(hadoopFsRelation)
}
+ /**
+ * Legacy MOR snapshot scan through [[MergeOnReadSnapshotRelation]] and
[[org.apache.hudi.HoodieMergeOnReadRDDV2]]:
+ * base-only splits take the skip-merging fast path, splits carrying log
files the file-group-reader branch.
+ */
+ private def legacyMorSnapshotDf(extraOpts: Map[String, String] = Map.empty):
DataFrame = {
+ val metaClient = createMetaClient(spark, basePath)
+ spark.baseRelationToDataFrame(
+ MergeOnReadSnapshotRelation(sqlContext, legacyReadOpts(extraOpts),
metaClient, None))
+ }
+
/**
* Whether the legacy parquet format engages its vectorized (batch) reader
for the table's
* schema. Because the schema carries a nested struct and an array, batch
support additionally
@@ -270,9 +310,7 @@ class TestLegacyParquetReadPath extends
HoodieSparkClientTestBase with ScalaAsse
assertSameRows(newReaderDf, legacyFileFormatDf(extractOpts))
assertSameRows(newReaderDf, legacyRelationDf(extractOpts))
- val partitions = legacyFileFormatDf(extractOpts)
-
.select("partition").distinct().collect().map(_.getString(0)).sorted.toSeq
- assertEquals(Seq("p0", "p1", "p2"), partitions)
+ assertEquals(Seq("p0", "p1", "p2"),
distinctPartitions(legacyFileFormatDf(extractOpts)))
}
@Test
@@ -470,4 +508,62 @@ class TestLegacyParquetReadPath extends
HoodieSparkClientTestBase with ScalaAsse
assertSameRows(newReaderDf, legacyDf)
}
}
+
+ @Test
+ def testMorSnapshotReadWithDroppedPartitionColumns(): Unit = {
+ writeMorDropPartitionColumnsCommits()
+
+ // The p0 file group carries a log file, so it is served by the
file-group-reader branch of
+ // HoodieMergeOnReadRDDV2, which sources every projected column from the
files -- and the p0 base
+ // file, holding the rows deltacommit 2 left alone, does not have the
partition column. The p1/p2
+ // groups stay base-only in the very same query, so the skip-merging fast
path -- the only one
+ // that ever appended the parsed values -- is pinned unregressed at the
same time.
+ val newReaderDf = fgReaderDf()
+ assertEquals(30, newReaderDf.count())
+ // Had the updates landed in a rewritten base file instead of a log, the
read-optimized query
+ // would already show them and no split in this table would take the
merging branch at all.
+ val readOptimizedDf = fgReaderDf(
+ Map(DataSourceReadOptions.QUERY_TYPE.key ->
DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL))
+ assertEquals(0, readOptimizedDf.filter(col("ts") === 2L).count())
+
+ val legacyDf = legacyMorSnapshotDf()
+ assertSameRows(newReaderDf, legacyDf)
+ assertEquals(Seq("p0", "p1", "p2"), distinctPartitions(legacyDf))
+
+ // The p0 rows deltacommit 2 left alone are the ones served straight from
the base file, where the
+ // partition column does not exist; they must come back as p0 and not as
null.
+ val untouchedP0Ids = (1 to 30).filter(i => i % 3 == 0 && i % 6 !=
0).map(_.toString)
+ val baseServedP0Df = legacyDf.filter(col("ts") === 1L &&
col("id").isin(untouchedP0Ids: _*))
+ assertEquals(untouchedP0Ids.size.toLong, baseServedP0Df.count())
+ assertEquals(Seq("p0"), distinctPartitions(baseServedP0Df))
+ }
+
+ @Test
+ def testMorIncrementalReadWithDroppedPartitionColumnsOnLogOnlySlice(): Unit
= {
+ writeMorDropPartitionColumnsCommits()
+
+ val metaClient = createMetaClient(spark, basePath)
+ val firstInstant =
metaClient.getCommitsTimeline.filterCompletedInstants.firstInstant.get
+
+ // Start-exclusive span covering only the update deltacommit. Its
file-system view is built from
+ // the files that deltacommit touched -- log files alone -- so the slice
carries no base file and
+ // the split has to resolve its partition values off a log file's path
instead.
+ //
+ // NOTE: this test passes with or without the splicing, so unlike the
snapshot test above it is
+ // not a repro of the null-partition bug. The rows here come from log
records, and those DO carry
+ // the partition column: HUDI-6926 makes a MOR upsert ignore
drop.partition.columns. What it does
+ // pin is the log-only resolution itself -- reading the values off
HoodieLogFile#getPath, whose
+ // pathInfo is null for these log files -- and that the result agrees with
the file-group-reader
+ // oracle.
+ val incOpts = Map(
+ DataSourceReadOptions.QUERY_TYPE.key ->
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL,
+ DataSourceReadOptions.START_COMMIT.key -> firstInstant.getCompletionTime)
+
+ val newReaderDf = fgReaderDf(incOpts)
+ val legacyDf = spark.baseRelationToDataFrame(
+ MergeOnReadIncrementalRelationV2(sqlContext, legacyReadOpts(incOpts),
metaClient, None, RangeType.OPEN_CLOSED))
+
+ assertSameRows(newReaderDf, legacyDf)
+ assertEquals(Seq("p0"), distinctPartitions(legacyDf))
+ }
}
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 ff0ce57d2b87..3c943037f2e9 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
@@ -17,9 +17,9 @@
package org.apache.hudi.functional
-import org.apache.hudi.DataSourceReadOptions
+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}
+import org.apache.hudi.DataSourceWriteOptions.{ORDERING_FIELDS,
RECORDKEY_FIELD, TABLE_TYPE}
import org.apache.hudi.common.config.HoodieReaderConfig
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.common.model.HoodieTableType.{COPY_ON_WRITE,
MERGE_ON_READ}
@@ -30,7 +30,7 @@ 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.spark.sql.{Row, SaveMode}
+import org.apache.spark.sql.{DataFrame, Row, SaveMode}
import org.apache.spark.sql.streaming.StreamTest
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
@@ -381,6 +381,105 @@ 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.
+ 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"
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(MERGE_ON_READ)
+ .setTableName(getTableName(tablePath))
+ .setRecordKeyFields("id")
+ .setOrderingFields("ts")
+
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
+
+ // 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 = {
+ // The write must use the short name: it resolves to
Spark4DefaultSource, the only
+ // provider overriding CreatableRelationProvider.supportsDataType to
accept VariantType.
+ // The fully qualified "org.apache.hudi" resolves to DefaultSource
(short name "hudi_v1"),
+ // which has no override, so DataSource.planForWriting on Spark 4.x
rejects the variant
+ // column with UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE.
+ spark.sql(valuesSql).write.format("hudi")
+ .options(commonOptions)
+ .option(TBL_NAME.key, getTableName(tablePath))
+ .option(TABLE_TYPE.key, MERGE_ON_READ.name)
+ .option("hoodie.index.type", "INMEMORY")
+ .option("hoodie.parquet.variant.write.shredding.enabled", "true")
+ .option("hoodie.parquet.variant.force.shredding.schema.for.test",
"key string")
+ .option(HoodieCompactionConfig.INLINE_COMPACT.key, compact.toString)
+ .option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key,
"2")
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ }
+
+ // 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")
+
+ // 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
+ // legacy branch is dropped and the source silently falls back to the
file group reader
+ // path, which scans a HadoopFsRelation ("FileScan" / "Scan parquet")
instead.
+ val assertLegacyRddPlan = AssertOnQuery { q =>
+ val plan = q.lastExecution.executedPlan.toString
+ assertTrue(plan.contains("Scan ExistingRDD"),
+ "expected the legacy RDD-backed incremental batch, but got plan: " +
plan)
+ assertTrue(!plan.contains("FileScan"),
+ "expected no file-group-reader HadoopFsRelation scan, but got plan:
" + plan)
+ 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)
+
+ 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.
+ AssertOnQuery { q => q.processAllAvailable(); true },
+ assertLegacyRddPlan,
+ CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1\"}", 1000L), Row(2,
"{\"key\":\"v2\"}", 1000L)),
+ lastOnly = true, isSorted = false),
+ StopStream,
+
+ // Log-only split: MergeOnReadIncrementalRelationV2 builds its file
system view out of the
+ // 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)
+ true
+ },
+ StartStream(),
+ AssertOnQuery { q => q.processAllAvailable(); true },
+ CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1-updated\"}", 1001L)),
lastOnly = true, isSorted = false)
+ )
+
+ // Merged split: a fresh testStream over a fresh streaming DataFrame
gets its own
+ // checkpoint, so it replays from the INIT offset and its first batch
spans the compaction
+ // commit and the update deltacommit together. Only such a span puts the
shredded base file
+ // and the update log file in one affected-file list, giving the base +
log slice that
+ // neither batch above produces.
+ testStream(variantStreamDf())(
+ AssertOnQuery { q => q.processAllAvailable(); true },
+ assertLegacyRddPlan,
+ CheckAnswerRows(Seq(Row(1, "{\"key\":\"v1-updated\"}", 1001L), Row(2,
"{\"key\":\"v2\"}", 1000L)),
+ lastOnly = true, isSorted = 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/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index 8b270a097a3e..93392648107b 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
@@ -20,6 +20,7 @@
package org.apache.spark.sql.hudi.dml.schema
import org.apache.hudi.{DataSourceReadOptions, HoodieSparkUtils}
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.HoodieSchemaException
import org.apache.hudi.common.testutils.HoodieTestUtils
@@ -285,8 +286,10 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
// 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.
+ // no-catalyst-schema legs are covered by the CDC round trip below and,
for
+ // streaming with hoodie.file.group.reader.enabled=false, by
+ // TestStreamingSource#"test mor stream source reads shredded variant
with legacy
+ // file group reader disabled".
val incRows = spark.read.format("hudi")
.option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
.option(DataSourceReadOptions.START_COMMIT.key, "000")
@@ -448,6 +451,105 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
+ test("Test CDC captures VARIANT values from shredded and unshredded base
files") {
+ // #19578: CDC is the only default-config query path that builds the
internal reader
+ // context without a catalyst schema, and its BASE_FILE_INSERT case
additionally reads
+ // the new base file directly, bypassing the context, so it needs its own
full-variant
+ // rewrite. That rewrite is picked from the spark adapter and the
requested schema alone,
+ // never from the file, so the unshredded leg runs through it too; both
layouts are swept
+ // below so neither side of that branch goes uncovered.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ // 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.
+ Seq(true, false).foreach { shredded =>
+ Seq("OP_KEY_ONLY", "DATA_BEFORE_AFTER").foreach { loggingMode =>
+ // SPARK is pinned instead of sweeping both record types: 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, whose shredded read is the separate defect
tracked as
+ // #19567/#19582; SPARK routes it through
SparkFileFormatInternalRowReaderContext,
+ // fixed in #19558. The CDC read path under test is independent of
that choice.
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ val leg = s"$loggingMode, shredded=$shredded"
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ // The forced shredding schema belongs to the shredded leg only; the
unshredded leg
+ // must reach the writer with shredding off and no forced schema.
+ val forceShreddingProp =
+ if (shredded)
"hoodie.parquet.variant.force.shredding.schema.for.test = 'key string'," else ""
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v variant,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | 'hoodie.table.cdc.enabled' = 'true',
+ | 'hoodie.table.cdc.supplemental.logging.mode' =
'$loggingMode',
+ | hoodie.parquet.variant.write.shredding.enabled = '$shredded',
+ | $forceShreddingProp
+ | hoodie.index.type = 'INMEMORY'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"""insert into $tableName values (1,
parse_json('{"key":"value1"}'), 1000)""")
+
+ // Pin the layout the CDC reads: without this, one leg silently
degrades into a
+ // second copy of the other and its half of the rewrite branch goes
uncovered.
+ val baseFiles = listDataParquetFiles(tablePath)
+ assert(baseFiles.nonEmpty, s"[$leg] should have a base parquet file
after the insert")
+ baseFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ if (shredded) {
+ assert(variantGroup.containsField("typed_value"),
+ s"[$leg] base file should carry typed_value.
Schema:\n$variantGroup")
+ } else {
+ assert(!variantGroup.containsField("typed_value"),
+ s"[$leg] base file must not carry typed_value.
Schema:\n$variantGroup")
+ }
+ }
+
+ spark.sql(s"""update $tableName set v =
parse_json('{"key":"value2"}'), ts = 1001 where id = 1""")
+
+ val cdc = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.INCREMENTAL_FORMAT.key,
DataSourceReadOptions.INCREMENTAL_FORMAT_CDC_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ .load(tablePath)
+
+ // Insert leg: the after-image comes from BASE_FILE_INSERT's direct
read of the
+ // base file; a null here means that read dropped the variant
payload.
+ val insertRows = cdc.where("op = 'i'")
+ .selectExpr("get_json_object(after, '$.v.key') as
after_key").collect()
+ assert(insertRows.length == 1, s"[$leg] expected exactly one insert
cdc row")
+ assert(insertRows(0).getString(0) == "value1",
+ s"[$leg] insert after-image lost the variant payload:
${insertRows(0)}")
+
+ // Update leg: images come from slice reads (OP_KEY_ONLY) or the cdc
log
+ // (DATA_BEFORE_AFTER); both must carry the variant payloads.
+ val updateRows = cdc.where("op = 'u'")
+ .selectExpr(
+ "get_json_object(before, '$.v.key') as before_key",
+ "get_json_object(after, '$.v.key') as after_key")
+ .collect()
+ assert(updateRows.length == 1, s"[$leg] expected exactly one update
cdc row")
+ assert(updateRows(0).getString(0) == "value1",
+ s"[$leg] update before-image lost the variant payload:
${updateRows(0)}")
+ assert(updateRows(0).getString(1) == "value2",
+ s"[$leg] update after-image lost the variant payload:
${updateRows(0)}")
+ })
+ }
+ }
+ }
+
test("Test bulk_insert row-writer round-trips VARIANT") {
assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or
higher")
