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 e1dd86e968eb fix(avro): detect shredded variant base files by shape so
reconstruct… (#19582)
e1dd86e968eb is described below
commit e1dd86e968ebff25c43ad5d9774c3cf19fc6bb43
Author: voonhous <[email protected]>
AuthorDate: Thu Aug 13 18:09:03 2026 +0800
fix(avro): detect shredded variant base files by shape so reconstruct…
(#19582)
* fix(avro): detect shredded variant base files by shape so reconstruction
engages
HoodieVariantReconstruction never engaged on real files: the reader's file
schema comes from converting the parquet footer MessageType, which loses the
variant logical type, so a shredded variant group arrives as a plain
{metadata, value, typed_value} record. Detection required a
HoodieSchema.Variant with isShredded(), found none, and the reader proceeded
at the unshredded schema, silently clipping typed_value away. Any avro-path
rewrite of a shredded base file persisted the nulls; the easiest trigger is
the COW small-file merge, where a plain second INSERT bin-packs into an
existing file group (#19567). This is the second leg of #19556; the Spark
reader-context leg was #19558.
Detect the on-disk side by SHAPE, anchored by the requested side: the
requested column (table schema, logical type intact) must be a variant for
the shape match to count, so plain user structs of the same shape are
unaffected. A file schema that kept its logical type still short-circuits on
isShredded. With detection engaging, the existing fail-fast branches
(shredded reading disabled, no provider available) now fire where they
previously never triggered, turning silent corruption into loud errors.
Extracted from the shredding-inference branch (c2ec87d977d5), leaving the
inference-specific parts behind.
- Unit tests: footer-style plain-record schema engages and reconstructs end
to end; the same shape without a variant requested column stays
disengaged.
- Functional test: the issue's repro, a COW small-file merge round trip with
the shredded layout and the same-file-group bin-pack pinned.
* review(19582): make the small-file merge hand the reader a
variant-anchored requested schema
Round-1 review traced the repro path further than the fix went: the COW
small-file bin-pack runs HoodieConcatHandle -> HoodieMergeHelper, where
isStrictProjectionOf(readerSchema, writerSchema) fails on RECORD vs VARIANT
because the footer-derived reader schema lost the variant logical type. The
merge then read the base file AT the footer schema, so the requested side
had
no variant column for the shape detection to anchor on, reconstruction
stayed
disengaged, and the subsequent rewriteRecordWithNewSchema silently dropped
typed_value. Detection alone left the E2E scenario broken.
- Share the anchored predicate as VariantSchemaUtils.isShreddedVariantTarget
and add alignShreddedVariants, which swaps footer-plain shredded variant
columns for their requested form; HoodieVariantReconstruction delegates.
- HoodieMergeHelper aligns shredded variant columns before the strict
projection check, so recordSchema resolves to the writer schema, the avro
reader engages reconstruction, and the lossy rewrite is skipped.
- Pin the E2E test to the AVRO record type: the record type picks the reader
the merge uses, and this PR covers the avro leg.
- Add the requested negative test: a footer-derived plain unshredded
{metadata, value} record against a variant requested column stays
disengaged; add a compatibility test for align + strict projection.
Residual: a shredded base file merged while the writer schema evolves (added
columns, or the internal-schema transformer) still takes the rewrite path
and
drops typed_value; that needs a variant-aware record rewrite and is left to
a
follow-up.
* review(19582): close the evolving-schema leg of the variant-dropping merge
Round-2 review asked whether the schema-evolution path still drops
typed_value. It did. The previous commit aligned shredded variant columns
only inside the strict-projection check, so the fix reached one branch of
recordSchema = isPureProjection ? writerSchema : readerSchema. Whenever the
writer schema is not a strict projection of the file's - an added column, or
the Advanced Schema Evolution transformer - runMerge still handed the reader
the raw footer schema, reconstruction could not anchor on a variant
requested
column, and rewriteRecordWithNewSchema copied {metadata, value} by name and
dropped typed_value. Same silent corruption as #19567, reached by an
ordinary
ALTER TABLE ADD COLUMNS with no schema-on-read config.
- Align the reader schema once, where it is derived, so both branches and
the
schema-evolution transformer reason about the same variant-bearing form.
This also drops the inline alignment from the projection check, so the
method reads as it did before.
- Test the added-column leg end to end: shredded base file, ALTER TABLE ADD
COLUMNS, then the bin-pack merge; rows from the first commit keep their
variants and pick up a null for the new column.
- Add the requested unshredded twin of the small-file merge test, split the
way the clustering pair is, plus a unit assertion that
alignShreddedVariants
leaves a footer-derived plain {metadata, value} column untouched. Both pin
that the alignment stays a no-op for ordinary variant tables now that it
runs on every runMerge.
* review(19582): order isShreddedVariantTarget file-first like its siblings
VariantSchemaUtils.isShreddedVariantTarget took (requested, file) while
alignShreddedVariants and HoodieVariantReconstruction.create both take
(file, requested). Both parameters are HoodieSchema, so a swapped call
compiles and silently inverts the detection instead of failing.
Flip the signature to (file, requested) so the class reads one way
throughout, update the two call sites, and document the two parameters.
No behaviour change.
* review(19582): reconstruct shredded variants nested below the top level
HoodieRowParquetWriteSupport.processNestedDataType shreds variants at any
depth, but the avro read path only ever looked at top-level columns: a
variant inside a struct, an array element or a map value reached
HoodieVariantReconstruction as a plain {metadata, value, typed_value}
record that nothing matched. create() saw no target, returned null, and
the column read at the unshredded schema, so typed_value was dropped -
the same silent loss as the top-level defect, and without tripping either
fail-fast branch, since nothing at the top level looked shredded.
Make the walk recursive on both halves:
- VariantSchemaUtils gains one shared recursion over records, array
elements and map values. alignShreddedVariants keeps its contract
(file schema in, requested-aligned schema out, same instance when
nothing matches) and gains a dual, toShreddedReadSchema, which is the
schema to read the file at. stripVariantShredding recurses too, so the
output schema is unshredded all the way down.
- HoodieVariantReconstruction replaces the flat isTarget[] plus parallel
sub-schema arrays with a rebuild plan built once at create(): a variant
node per target, and record/array/map nodes that descend into it. A
position with nothing shredded below it plans to null and its value is
passed through untouched, so non-variant tables walk the same path they
did before.
Tests: nested reconstruction across all three container kinds (verified
red on the assertNotNull before the change), the nested unshredded twin
that must stay disengaged, and a nested alignment case in
TestHoodieSchemaCompatibility.
---
.../table/action/commit/HoodieMergeHelper.java | 12 +-
.../hudi/common/avro/VariantSchemaUtils.java | 224 ++++++++++++++++++---
.../schema/TestHoodieSchemaCompatibility.java | 55 +++++
.../hadoop/HoodieVariantReconstruction.java | 217 ++++++++++++++------
.../hadoop/TestHoodieVariantReconstruction.java | 192 ++++++++++++++++++
.../sql/hudi/dml/schema/TestVariantDataType.scala | 204 +++++++++++++++++++
6 files changed, 812 insertions(+), 92 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
index d04cf143640c..b9e963a7b80a 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table.action.commit;
+import org.apache.hudi.common.avro.VariantSchemaUtils;
import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieRecord;
@@ -83,7 +84,16 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper {
HoodieFileReader bootstrapFileReader = null;
HoodieSchema writerSchema = mergeHandle.getWriterSchemaWithMetaFields();
- HoodieSchema readerSchema = baseFileReader.getSchema();
+ // A shredded variant column loses its logical type through the parquet
footer, so the base
+ // file's schema surfaces it as a plain {metadata, value, typed_value}
record. Align such
+ // columns to the writer's variant form once, here, because every
downstream use of the reader
+ // schema needs the aligned form: the strict-projection check below
(RECORD vs VARIANT can
+ // never pass), and - whichever branch it lands on - the schema the reader
is handed, since
+ // HoodieVariantReconstruction anchors on the requested column being a
variant. Read at the raw
+ // footer schema instead and reconstruction stays disengaged, so the
rewrite below copies
+ // {metadata, value} by name and silently drops typed_value (#19567).
Returns the file schema
+ // untouched when no column has the shredded shape, so non-variant tables
are unaffected.
+ HoodieSchema readerSchema =
VariantSchemaUtils.alignShreddedVariants(baseFileReader.getSchema(),
writerSchema);
// In case Advanced Schema Evolution is enabled we might need to rewrite
currently
// persisted records to adhere to an evolved schema
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
index 9ef2b847e359..b2155d35eb05 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
@@ -22,6 +22,7 @@ package org.apache.hudi.common.avro;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.Option;
import java.util.ArrayList;
import java.util.List;
@@ -37,8 +38,10 @@ public class VariantSchemaUtils {
}
/**
- * Strips shredding from top-level variant fields in {@code schema},
replacing each shredded
- * variant with its unshredded form (dropping {@code typed_value}).
Non-variant fields and
+ * Strips shredding from the variant fields in {@code schema}, replacing
each shredded variant
+ * with its unshredded form (dropping {@code typed_value}). Variants nested
inside records, array
+ * elements and map values are stripped too, since the row writer shreds at
any depth
+ * ({@code HoodieRowParquetWriteSupport.processNestedDataType}). Non-variant
fields and
* already-unshredded variants pass through unchanged; returns {@code
schema} as-is when nothing
* changes.
*/
@@ -46,40 +49,213 @@ public class VariantSchemaUtils {
if (schema.getType() != HoodieSchemaType.RECORD) {
return schema;
}
+ return stripRecordVariantShredding(schema);
+ }
- List<HoodieSchemaField> fields = schema.getFields();
+ private static HoodieSchema stripRecordVariantShredding(HoodieSchema record)
{
List<HoodieSchemaField> newFields = new ArrayList<>();
boolean changed = false;
-
- for (HoodieSchemaField field : fields) {
+ for (HoodieSchemaField field : record.getFields()) {
HoodieSchema fieldSchema = field.schema();
- boolean wasNullable = fieldSchema.isNullable();
- HoodieSchema unwrapped = wasNullable ? fieldSchema.getNonNullType() :
fieldSchema;
+ HoodieSchema replacement = stripVariantShreddingAt(fieldSchema);
+ if (replacement != fieldSchema) {
+ changed = true;
+ }
+ // withSchema makes a fresh Avro Field: reusing one already bound to
this record would fail
+ // Schema.setFields with "Field already used" when building the
replacement record below.
+ newFields.add(field.withSchema(replacement));
+ }
+ if (!changed) {
+ return record;
+ }
+ return HoodieSchema.createRecord(
+ record.getAvroSchema().getName(),
+ record.getAvroSchema().getNamespace(),
+ record.getAvroSchema().getDoc(),
+ newFields);
+ }
- if (unwrapped.getType() == HoodieSchemaType.VARIANT) {
- HoodieSchema.Variant variant = (HoodieSchema.Variant) unwrapped;
- if (variant.isShredded()) {
- HoodieSchema.Variant unshredded = HoodieSchema.createVariant(
- unwrapped.getAvroSchema().getName(),
- unwrapped.getAvroSchema().getNamespace(),
- unwrapped.getAvroSchema().getDoc());
- HoodieSchema replacement = wasNullable ?
HoodieSchema.createNullable(unshredded) : unshredded;
- newFields.add(field.withSchema(replacement));
- changed = true;
- continue;
+ /** Strips shredding at one schema position, returning the argument instance
when nothing changes. */
+ private static HoodieSchema stripVariantShreddingAt(HoodieSchema schema) {
+ boolean wasNullable = schema.isNullable();
+ HoodieSchema unwrapped = wasNullable ? schema.getNonNullType() : schema;
+ HoodieSchema replacement;
+ switch (unwrapped.getType()) {
+ case VARIANT:
+ if (!((HoodieSchema.Variant) unwrapped).isShredded()) {
+ return schema;
}
+ replacement = HoodieSchema.createVariant(
+ unwrapped.getAvroSchema().getName(),
+ unwrapped.getAvroSchema().getNamespace(),
+ unwrapped.getAvroSchema().getDoc());
+ break;
+ case RECORD:
+ replacement = stripRecordVariantShredding(unwrapped);
+ break;
+ case ARRAY: {
+ HoodieSchema elementType = unwrapped.getElementType();
+ HoodieSchema strippedElement = stripVariantShreddingAt(elementType);
+ replacement = strippedElement == elementType ? unwrapped :
HoodieSchema.createArray(strippedElement);
+ break;
+ }
+ case MAP: {
+ HoodieSchema valueType = unwrapped.getValueType();
+ HoodieSchema strippedValue = stripVariantShreddingAt(valueType);
+ replacement = strippedValue == valueType ? unwrapped :
HoodieSchema.createMap(strippedValue);
+ break;
}
- newFields.add(field);
+ default:
+ return schema;
}
-
- if (!changed) {
+ if (replacement == unwrapped) {
return schema;
}
+ return wasNullable ? HoodieSchema.createNullable(replacement) :
replacement;
+ }
+
+ /**
+ * Whether this column sits shredded on disk and must be read in that shape
and reconstructed to
+ * serve the requested schema. A file schema that kept its variant logical
type answers via
+ * {@link HoodieSchema.Variant#isShredded()}; but a file schema derived from
converting the
+ * parquet footer MessageType loses the logical type (variant groups come
back as plain records),
+ * so the on-disk side is detected by SHAPE, anchored by the requested side:
the requested column
+ * (from the table schema, logical type intact) must be a variant for the
shape match to count,
+ * leaving plain user structs of the same shape alone (#19567).
+ *
+ * @param fileFieldSchema the column as it sits in the file schema
+ * @param requestedFieldSchema the same column as requested, carrying the
variant logical type
+ */
+ public static boolean isShreddedVariantTarget(HoodieSchema fileFieldSchema,
HoodieSchema requestedFieldSchema) {
+ HoodieSchema file = fileFieldSchema.getNonNullType();
+ if (file.getType() == HoodieSchemaType.VARIANT && ((HoodieSchema.Variant)
file).isShredded()) {
+ return true;
+ }
+ HoodieSchema requested = requestedFieldSchema.getNonNullType();
+ return requested.getType() == HoodieSchemaType.VARIANT &&
isShreddedVariantShape(file);
+ }
+
+ /**
+ * Returns {@code fileSchema} with each shredded variant column (per
+ * {@link #isShreddedVariantTarget}) replaced by its requested counterpart,
for projection or
+ * compatibility checks against {@code requestedSchema}. A footer-derived
shredded variant column
+ * surfaces as a plain {@code {metadata, value, typed_value}} record and so
can never look like a
+ * projection source of the requested variant, even though the readers
reconstruct it (see
+ * HoodieVariantReconstruction). Returns {@code fileSchema} as-is when
nothing matches.
+ */
+ public static HoodieSchema alignShreddedVariants(HoodieSchema fileSchema,
HoodieSchema requestedSchema) {
+ if (fileSchema.getType() != HoodieSchemaType.RECORD ||
requestedSchema.getType() != HoodieSchemaType.RECORD) {
+ return fileSchema;
+ }
+ return swapShreddedVariantFields(fileSchema, requestedSchema, true);
+ }
+
+ /**
+ * The dual of {@link #alignShreddedVariants}: returns {@code
requestedSchema} with each shredded
+ * variant column swapped to its on-disk (typed_value-bearing) form taken
from {@code fileSchema}.
+ * This is the schema to read the file at, so parquet materializes {@code
typed_value} for the
+ * reader to reconstruct from. Returns {@code requestedSchema} as-is when
nothing matches, which
+ * is how callers detect that the file has no shredded variant column to
reconstruct.
+ */
+ public static HoodieSchema toShreddedReadSchema(HoodieSchema
requestedSchema, HoodieSchema fileSchema) {
+ if (fileSchema.getType() != HoodieSchemaType.RECORD ||
requestedSchema.getType() != HoodieSchemaType.RECORD) {
+ return requestedSchema;
+ }
+ return swapShreddedVariantFields(requestedSchema, fileSchema, false);
+ }
+ /**
+ * Walks {@code base} against its matching {@code other} fields by name,
replacing every shredded
+ * variant position with the other side's schema. Recurses through records,
array elements and map
+ * values, since the row writer shreds variants at any depth
+ * ({@code HoodieRowParquetWriteSupport.processNestedDataType}). {@code
baseIsFile} says which of
+ * the two is the file side, which is what {@link #isShreddedVariantTarget}
needs to anchor
+ * detection. Returns {@code base} when nothing matches.
+ */
+ private static HoodieSchema swapShreddedVariantFields(HoodieSchema base,
HoodieSchema other, boolean baseIsFile) {
+ List<HoodieSchemaField> newFields = new ArrayList<>();
+ boolean changed = false;
+ for (HoodieSchemaField baseField : base.getFields()) {
+ HoodieSchema baseFieldSchema = baseField.schema();
+ Option<HoodieSchemaField> otherField = other.getField(baseField.name());
+ HoodieSchema replacement = otherField.isPresent()
+ ? swapShreddedVariantsAt(baseFieldSchema, otherField.get().schema(),
baseIsFile)
+ : baseFieldSchema;
+ if (replacement != baseFieldSchema) {
+ changed = true;
+ }
+ // Copy untouched fields too (withSchema makes a fresh Avro Field):
reusing a field already
+ // bound to the base record would fail Schema.setFields with "Field
already used" when
+ // building the swapped record below.
+ newFields.add(baseField.withSchema(replacement));
+ }
+ if (!changed) {
+ return base;
+ }
return HoodieSchema.createRecord(
- schema.getAvroSchema().getName(),
- schema.getAvroSchema().getNamespace(),
- schema.getAvroSchema().getDoc(),
+ base.getAvroSchema().getName(),
+ base.getAvroSchema().getNamespace(),
+ base.getAvroSchema().getDoc(),
newFields);
}
+
+ /** Swaps at one schema position, returning the {@code base} instance when
nothing changes. */
+ private static HoodieSchema swapShreddedVariantsAt(HoodieSchema base,
HoodieSchema other, boolean baseIsFile) {
+ if (baseIsFile ? isShreddedVariantTarget(base, other) :
isShreddedVariantTarget(other, base)) {
+ // Take the other side's schema wholesale, nullability included.
+ return other;
+ }
+ boolean wasNullable = base.isNullable();
+ HoodieSchema baseInner = wasNullable ? base.getNonNullType() : base;
+ HoodieSchema otherInner = other.isNullable() ? other.getNonNullType() :
other;
+ if (baseInner.getType() != otherInner.getType()) {
+ return base;
+ }
+ HoodieSchema replacement;
+ switch (baseInner.getType()) {
+ // VARIANT is deliberately absent: a variant that is not a target here
is either unshredded or
+ // has no requested-side anchor, and its typed_value internals are the
provider's business.
+ case RECORD:
+ replacement = swapShreddedVariantFields(baseInner, otherInner,
baseIsFile);
+ break;
+ case ARRAY: {
+ HoodieSchema baseElement = baseInner.getElementType();
+ HoodieSchema swappedElement = swapShreddedVariantsAt(baseElement,
otherInner.getElementType(), baseIsFile);
+ replacement = swappedElement == baseElement ? baseInner :
HoodieSchema.createArray(swappedElement);
+ break;
+ }
+ case MAP: {
+ HoodieSchema baseValue = baseInner.getValueType();
+ HoodieSchema swappedValue = swapShreddedVariantsAt(baseValue,
otherInner.getValueType(), baseIsFile);
+ replacement = swappedValue == baseValue ? baseInner :
HoodieSchema.createMap(swappedValue);
+ break;
+ }
+ default:
+ return base;
+ }
+ if (replacement == baseInner) {
+ return base;
+ }
+ return wasNullable ? HoodieSchema.createNullable(replacement) :
replacement;
+ }
+
+ /** The on-disk shredded variant shape: a record of exactly {metadata:
bytes, value: [nullable] bytes, typed_value}. */
+ private static boolean isShreddedVariantShape(HoodieSchema schema) {
+ if (schema.getType() != HoodieSchemaType.RECORD ||
schema.getFields().size() != 3) {
+ return false;
+ }
+ if
(!schema.getField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD).isPresent()) {
+ return false;
+ }
+ return isBytesField(schema, HoodieSchema.Variant.VARIANT_METADATA_FIELD)
+ && isBytesField(schema, HoodieSchema.Variant.VARIANT_VALUE_FIELD);
+ }
+
+ private static boolean isBytesField(HoodieSchema schema, String fieldName) {
+ return schema.getField(fieldName)
+ .map(HoodieSchemaField::schema)
+ .map(s -> s.isNullable() ? s.getNonNullType() : s)
+ .map(s -> s.getType() == HoodieSchemaType.BYTES)
+ .orElse(false);
+ }
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
index 0d10076a5999..5be6716c2e7a 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
@@ -18,6 +18,7 @@
package org.apache.hudi.common.schema;
+import org.apache.hudi.common.avro.VariantSchemaUtils;
import org.apache.hudi.exception.SchemaBackwardsCompatibilityException;
import org.apache.hudi.exception.SchemaCompatibilityException;
@@ -41,6 +42,7 @@ import static
org.apache.hudi.common.schema.TestHoodieSchemaUtils.SIMPLE_SCHEMA;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -215,6 +217,59 @@ public class TestHoodieSchemaCompatibility {
assertTrue(HoodieSchemaCompatibility.isStrictProjectionOf(sourceSchema,
converted));
}
+ @Test
+ public void testIsStrictProjectionWithAlignedShreddedVariants() {
+ // A shredded variant base file surfaces through the parquet footer as a
plain
+ // {metadata, value, typed_value} record, so it can never look like a
strict projection
+ // source of the table's variant column. HoodieMergeHelper aligns such
columns via
+ // VariantSchemaUtils.alignShreddedVariants before the check so the merge
reads at the
+ // writer schema and variant reconstruction can engage (#19567).
+ HoodieSchema footerShredded = HoodieSchema.createRecord("v",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("value",
HoodieSchema.createNullable(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("typed_value",
HoodieSchema.createNullable(HoodieSchemaType.INT))));
+ HoodieSchema fileSchema = HoodieSchema.createRecord("rec",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("v",
HoodieSchema.createNullable(footerShredded))));
+ HoodieSchema writerSchema = HoodieSchema.createRecord("rec",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("v",
HoodieSchema.createNullable(HoodieSchema.createVariant()))));
+
+ assertFalse(HoodieSchemaCompatibility.isStrictProjectionOf(fileSchema,
writerSchema));
+ assertTrue(HoodieSchemaCompatibility.isStrictProjectionOf(
+ VariantSchemaUtils.alignShreddedVariants(fileSchema, writerSchema),
writerSchema));
+
+ // Without a variant on the requested side there is nothing to align: a
plain user struct of
+ // the same shape passes through untouched.
+ assertSame(fileSchema,
VariantSchemaUtils.alignShreddedVariants(fileSchema, fileSchema));
+
+ // The alignment now runs on every HoodieMergeHelper.runMerge, so pin the
ordinary unshredded
+ // table too: a footer-derived plain {metadata, value} column carries no
typed_value to
+ // reconstruct, and must pass through untouched even against a variant
requested column.
+ HoodieSchema footerUnshredded = HoodieSchema.createRecord("v",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("value",
HoodieSchema.createNullable(HoodieSchemaType.BYTES))));
+ HoodieSchema unshreddedFileSchema = HoodieSchema.createRecord("rec",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("v",
HoodieSchema.createNullable(footerUnshredded))));
+ assertSame(unshreddedFileSchema,
+ VariantSchemaUtils.alignShreddedVariants(unshreddedFileSchema,
writerSchema));
+
+ // The row writer shreds variants at any depth, so the alignment has to
descend as well:
+ // a variant nested in a struct hits the same lossy rewrite when it is
left unaligned.
+ HoodieSchema nestedFileSchema = HoodieSchema.createRecord("rec",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("nested", HoodieSchema.createRecord("nested_rec",
"example.schema", null,
+ Arrays.asList(HoodieSchemaField.of("v",
HoodieSchema.createNullable(footerShredded)))))));
+ HoodieSchema nestedWriterSchema = HoodieSchema.createRecord("rec",
"example.schema", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("nested", HoodieSchema.createRecord("nested_rec",
"example.schema", null,
+ Arrays.asList(HoodieSchemaField.of("v",
HoodieSchema.createNullable(HoodieSchema.createVariant())))))));
+
assertFalse(HoodieSchemaCompatibility.isStrictProjectionOf(nestedFileSchema,
nestedWriterSchema));
+ assertTrue(HoodieSchemaCompatibility.isStrictProjectionOf(
+ VariantSchemaUtils.alignShreddedVariants(nestedFileSchema,
nestedWriterSchema), nestedWriterSchema));
+ }
+
@Test
public void testIsCompatibleProjection() {
HoodieSchema sourceSchema = HoodieSchema.parse(SOURCE_SCHEMA);
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
index 25da535e0187..a3cf0e5bd237 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
@@ -35,8 +35,9 @@ import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
-import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
/**
* Reconstructs unshredded variants when reading an already-shredded base file
on the Avro
@@ -53,23 +54,14 @@ import java.util.List;
final class HoodieVariantReconstruction {
private final HoodieSchema intermediateSchema;
- private final Schema outputAvroSchema;
private final VariantShreddingProvider provider;
- // Indexed by field position in the (requested == output) record. For target
fields, the file's
- // shredded sub-schema and the unshredded target sub-schema for rebuild;
null for non-targets.
- private final boolean[] isTarget;
- private final Schema[] shreddedSubSchemas;
- private final Schema[] unshreddedSubSchemas;
-
- private HoodieVariantReconstruction(HoodieSchema intermediateSchema, Schema
outputAvroSchema,
- VariantShreddingProvider provider, boolean[]
isTarget,
- Schema[] shreddedSubSchemas, Schema[]
unshreddedSubSchemas) {
+ private final Rebuilder rootRebuilder;
+
+ private HoodieVariantReconstruction(HoodieSchema intermediateSchema,
+ VariantShreddingProvider provider,
Rebuilder rootRebuilder) {
this.intermediateSchema = intermediateSchema;
- this.outputAvroSchema = outputAvroSchema;
this.provider = provider;
- this.isTarget = isTarget;
- this.shreddedSubSchemas = shreddedSubSchemas;
- this.unshreddedSubSchemas = unshreddedSubSchemas;
+ this.rootRebuilder = rootRebuilder;
}
/**
@@ -91,26 +83,10 @@ final class HoodieVariantReconstruction {
return null;
}
- List<HoodieSchemaField> requestedFields = requestedSchema.getFields();
- List<HoodieSchemaField> intermediateFields = new ArrayList<>();
- boolean[] isTarget = new boolean[requestedFields.size()];
- boolean anyTarget = false;
- for (int i = 0; i < requestedFields.size(); i++) {
- HoodieSchemaField requestedField = requestedFields.get(i);
- Option<HoodieSchemaField> fileField =
fileSchema.getField(requestedField.name());
- if (fileField.isPresent() &&
isShreddedVariant(fileField.get().schema())) {
- isTarget[i] = true;
- anyTarget = true;
- // Read this column in its on-disk shredded shape.
-
intermediateFields.add(requestedField.withSchema(fileField.get().schema()));
- } else {
- // Copy non-target fields too (withSchema makes a fresh Avro Field):
reusing the requested
- // field's Avro Field, already bound to the requested record, would
fail Schema.setFields with
- // "Field already used" when building the intermediate record below.
-
intermediateFields.add(requestedField.withSchema(requestedField.schema()));
- }
- }
- if (!anyTarget) {
+ // Records leave this reader unshredded; output field order matches the
requested/intermediate order.
+ HoodieSchema outputSchema =
VariantSchemaUtils.stripVariantShredding(requestedSchema);
+ Rebuilder rootRebuilder = buildRebuilder(outputSchema, fileSchema);
+ if (rootRebuilder == null) {
// No shredded variant columns in the file: nothing to reconstruct,
regardless of the flag.
return null;
}
@@ -134,25 +110,8 @@ final class HoodieVariantReconstruction {
+ " or add a provider implementation (e.g. the Spark variant module)
to the classpath.");
}
- HoodieSchema intermediateSchema = HoodieSchema.createRecord(
- requestedSchema.getAvroSchema().getName(),
- requestedSchema.getAvroSchema().getNamespace(),
- requestedSchema.getAvroSchema().getDoc(),
- intermediateFields);
- // Records leave this reader unshredded; output field order matches the
requested/intermediate order.
- HoodieSchema outputSchema =
VariantSchemaUtils.stripVariantShredding(requestedSchema);
-
- Schema[] shreddedSubSchemas = new Schema[requestedFields.size()];
- Schema[] unshreddedSubSchemas = new Schema[requestedFields.size()];
- for (int i = 0; i < requestedFields.size(); i++) {
- if (isTarget[i]) {
- shreddedSubSchemas[i] =
fileSchema.getField(requestedFields.get(i).name()).get().schema().getNonNullType().getAvroSchema();
- unshreddedSubSchemas[i] =
outputSchema.getFields().get(i).schema().getNonNullType().getAvroSchema();
- }
- }
-
- return new HoodieVariantReconstruction(intermediateSchema,
outputSchema.toAvroSchema(), provider,
- isTarget, shreddedSubSchemas, unshreddedSubSchemas);
+ return new HoodieVariantReconstruction(
+ VariantSchemaUtils.toShreddedReadSchema(requestedSchema, fileSchema),
provider, rootRebuilder);
}
/**
@@ -160,23 +119,147 @@ final class HoodieVariantReconstruction {
* a record conforming to the unshredded output schema.
*/
IndexedRecord reconstruct(IndexedRecord in) {
- GenericRecord out = new GenericData.Record(outputAvroSchema);
- for (int i = 0; i < isTarget.length; i++) {
- Object value = in.get(i);
- if (isTarget[i] && value instanceof GenericRecord) {
- out.put(i, provider.rebuildVariantRecord((GenericRecord) value,
shreddedSubSchemas[i], unshreddedSubSchemas[i]));
- } else {
- // Non-variant column, or a null variant column: pass through
unchanged.
- out.put(i, value);
+ return (IndexedRecord) rootRebuilder.rebuild(in, provider);
+ }
+
+ /**
+ * Plans the rebuild for one schema position, or returns {@code null} when
nothing below it is a
+ * shredded variant and the value can be passed through untouched. The walk
is driven by the
+ * output (requested) side because that is the shape the record read from
parquet has; the file
+ * side is matched into it by field name. Detection is anchored by the
requested side because the
+ * file schema usually comes from converting the parquet footer MessageType,
which loses the
+ * variant logical type; see VariantSchemaUtils.isShreddedVariantTarget
(#19567). Records, array
+ * elements and map values are all descended into, since the row writer
shreds variants at any
+ * depth ({@code HoodieRowParquetWriteSupport.processNestedDataType}).
+ */
+ private static Rebuilder buildRebuilder(HoodieSchema outputSchema,
HoodieSchema fileSchema) {
+ if (VariantSchemaUtils.isShreddedVariantTarget(fileSchema, outputSchema)) {
+ return new VariantRebuilder(fileSchema.getNonNullType().getAvroSchema(),
+ outputSchema.getNonNullType().getAvroSchema());
+ }
+ HoodieSchema output = outputSchema.getNonNullType();
+ HoodieSchema file = fileSchema.getNonNullType();
+ if (output.getType() != file.getType()) {
+ return null;
+ }
+ switch (output.getType()) {
+ case RECORD: {
+ List<HoodieSchemaField> outputFields = output.getFields();
+ Rebuilder[] fieldRebuilders = new Rebuilder[outputFields.size()];
+ boolean anyTarget = false;
+ for (int i = 0; i < outputFields.size(); i++) {
+ HoodieSchemaField outputField = outputFields.get(i);
+ Option<HoodieSchemaField> fileField =
file.getField(outputField.name());
+ fieldRebuilders[i] = fileField.isPresent()
+ ? buildRebuilder(outputField.schema(), fileField.get().schema())
+ : null;
+ anyTarget |= fieldRebuilders[i] != null;
+ }
+ return anyTarget ? new RecordRebuilder(output.getAvroSchema(),
fieldRebuilders) : null;
+ }
+ case ARRAY: {
+ Rebuilder elementRebuilder = buildRebuilder(output.getElementType(),
file.getElementType());
+ return elementRebuilder == null ? null : new
ArrayRebuilder(output.getAvroSchema(), elementRebuilder);
+ }
+ case MAP: {
+ Rebuilder valueRebuilder = buildRebuilder(output.getValueType(),
file.getValueType());
+ return valueRebuilder == null ? null : new
MapRebuilder(valueRebuilder);
+ }
+ default:
+ return null;
+ }
+ }
+
+ /** Rebuilds one value read at the file's shape into its unshredded output
shape. */
+ private interface Rebuilder {
+ Object rebuild(Object value, VariantShreddingProvider provider);
+ }
+
+ private static final class VariantRebuilder implements Rebuilder {
+ private final Schema shreddedSchema;
+ private final Schema unshreddedSchema;
+
+ private VariantRebuilder(Schema shreddedSchema, Schema unshreddedSchema) {
+ this.shreddedSchema = shreddedSchema;
+ this.unshreddedSchema = unshreddedSchema;
+ }
+
+ @Override
+ public Object rebuild(Object value, VariantShreddingProvider provider) {
+ // A null variant passes through unchanged.
+ return value instanceof GenericRecord
+ ? provider.rebuildVariantRecord((GenericRecord) value,
shreddedSchema, unshreddedSchema)
+ : value;
+ }
+ }
+
+ private static final class RecordRebuilder implements Rebuilder {
+ private final Schema outputSchema;
+ // Indexed by field position in the (output == intermediate) record; null
for non-targets.
+ private final Rebuilder[] fieldRebuilders;
+
+ private RecordRebuilder(Schema outputSchema, Rebuilder[] fieldRebuilders) {
+ this.outputSchema = outputSchema;
+ this.fieldRebuilders = fieldRebuilders;
+ }
+
+ @Override
+ public Object rebuild(Object value, VariantShreddingProvider provider) {
+ if (!(value instanceof IndexedRecord)) {
+ return value;
}
+ IndexedRecord in = (IndexedRecord) value;
+ GenericRecord out = new GenericData.Record(outputSchema);
+ for (int i = 0; i < fieldRebuilders.length; i++) {
+ Object fieldValue = in.get(i);
+ out.put(i, fieldRebuilders[i] == null ? fieldValue :
fieldRebuilders[i].rebuild(fieldValue, provider));
+ }
+ return out;
}
- return out;
}
- private static boolean isShreddedVariant(HoodieSchema schema) {
- HoodieSchema unwrapped = schema.getNonNullType();
- return unwrapped.getType() == HoodieSchemaType.VARIANT
- && ((HoodieSchema.Variant) unwrapped).isShredded();
+ private static final class ArrayRebuilder implements Rebuilder {
+ private final Schema outputSchema;
+ private final Rebuilder elementRebuilder;
+
+ private ArrayRebuilder(Schema outputSchema, Rebuilder elementRebuilder) {
+ this.outputSchema = outputSchema;
+ this.elementRebuilder = elementRebuilder;
+ }
+
+ @Override
+ public Object rebuild(Object value, VariantShreddingProvider provider) {
+ if (!(value instanceof List)) {
+ return value;
+ }
+ List<?> in = (List<?>) value;
+ GenericData.Array<Object> out = new GenericData.Array<>(in.size(),
outputSchema);
+ for (Object element : in) {
+ out.add(elementRebuilder.rebuild(element, provider));
+ }
+ return out;
+ }
+ }
+
+ private static final class MapRebuilder implements Rebuilder {
+ private final Rebuilder valueRebuilder;
+
+ private MapRebuilder(Rebuilder valueRebuilder) {
+ this.valueRebuilder = valueRebuilder;
+ }
+
+ @Override
+ public Object rebuild(Object value, VariantShreddingProvider provider) {
+ if (!(value instanceof Map)) {
+ return value;
+ }
+ Map<?, ?> in = (Map<?, ?>) value;
+ Map<Object, Object> out = new LinkedHashMap<>(in.size());
+ for (Map.Entry<?, ?> entry : in.entrySet()) {
+ out.put(entry.getKey(), valueRebuilder.rebuild(entry.getValue(),
provider));
+ }
+ return out;
+ }
}
private static VariantShreddingProvider loadProvider(HoodieStorage storage) {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
index f522c21d1caf..99399e361fea 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
@@ -39,6 +39,9 @@ import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -139,6 +142,195 @@ class TestHoodieVariantReconstruction {
assertNull(reconstruction.reconstruct(input).get(1));
}
+ @Test
+ void engagesOnFooterDerivedPlainShreddedShape(@TempDir Path tmp) {
+ // #19567: a real file schema comes from converting the parquet footer
MessageType, which
+ // loses the variant logical type, so the shredded column arrives as a
PLAIN record of
+ // {metadata, value, typed_value}. Detection must anchor on the requested
column being a
+ // variant and match the file side by shape, then reconstruction proceeds
as usual.
+ HoodieSchema fileSchema =
recordWithIdAndVariant(footerStylePlainShreddedSchema());
+ HoodieSchema requestedSchema =
recordWithIdAndVariant(HoodieSchema.createVariant());
+ HoodieStorage storage = storageWithReadingShredded(tmp, true);
+
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
+ TestVariantShreddingProvider.class.getName());
+
+ HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ fileSchema, requestedSchema, storage);
+ assertNotNull(reconstruction, "Plain-record shredded shape with a variant
requested column must engage");
+
+ GenericRecord shredded = new GenericData.Record(
+
reconstruction.intermediateSchema().getField("v").get().schema().getNonNullType().toAvroSchema());
+ shredded.put("metadata", ByteBuffer.wrap(new byte[] {1}));
+ shredded.put("value", null);
+ shredded.put("typed_value", 42);
+ GenericRecord input = new
GenericData.Record(reconstruction.intermediateSchema().toAvroSchema());
+ input.put("id", "record-1");
+ input.put("v", shredded);
+
+ IndexedRecord output = reconstruction.reconstruct(input);
+ assertEquals("record-1", output.get(0).toString());
+ GenericRecord variant = (GenericRecord) output.get(1);
+ assertEquals(ByteBuffer.wrap(new byte[] {1}), variant.get("metadata"));
+ assertEquals(ByteBuffer.wrap(new byte[] {42}), variant.get("value"));
+ }
+
+ @Test
+ void returnsNullForFooterDerivedPlainUnshreddedShape(@TempDir Path tmp) {
+ // The ordinary unshredded layout after footer conversion: a plain
{metadata, value} record
+ // with the variant logical type lost. There is no typed_value to
reconstruct, so detection
+ // must stay disengaged (the column reads directly at the requested
schema) even with
+ // shredded reading disabled.
+ HoodieSchema fileSchema =
recordWithIdAndVariant(footerStylePlainUnshreddedSchema());
+ HoodieSchema requestedSchema =
recordWithIdAndVariant(HoodieSchema.createVariant());
+ assertNull(HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ storageWithReadingShredded(tmp, false)));
+ }
+
+ @Test
+ void ignoresShreddedShapeWhenRequestedColumnIsNotVariant(@TempDir Path tmp) {
+ // A user struct that merely has the {metadata, value, typed_value} shape
must not be
+ // treated as a shredded variant: without the requested-side variant
anchor there is
+ // nothing to reconstruct, so create() returns null even with shredded
reading disabled.
+ HoodieSchema fileSchema =
recordWithIdAndVariant(footerStylePlainShreddedSchema());
+ HoodieSchema requestedSchema =
recordWithIdAndVariant(footerStylePlainShreddedSchema());
+ assertNull(HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ storageWithReadingShredded(tmp, false)));
+ }
+
+ @Test
+ void reconstructsShreddedVariantsNestedInRecordsArraysAndMaps(@TempDir Path
tmp) {
+ // HoodieRowParquetWriteSupport.processNestedDataType shreds variants at
any depth, so a nested
+ // variant reaches this reader as a plain {metadata, value, typed_value}
record too. Detection
+ // and rebuild must descend into records, array elements and map values,
or the nested payload
+ // is dropped the way #19567 dropped the top-level one - and silently,
since nothing at the top
+ // level looks shredded and neither fail-fast branch is reached.
+ HoodieSchema fileSchema = recordWithNestedVariants(
+ footerStylePlainShreddedSchema("nested_v"),
+ footerStylePlainShreddedSchema("element_v"),
+ footerStylePlainShreddedSchema("map_v"));
+ HoodieSchema requestedSchema = recordWithNestedVariants(
+ HoodieSchema.createVariant("nested_v", "org.apache.hudi.test", null),
+ HoodieSchema.createVariant("element_v", "org.apache.hudi.test", null),
+ HoodieSchema.createVariant("map_v", "org.apache.hudi.test", null));
+ HoodieStorage storage = storageWithReadingShredded(tmp, true);
+
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
+ TestVariantShreddingProvider.class.getName());
+
+ HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ fileSchema, requestedSchema, storage);
+ assertNotNull(reconstruction, "A variant nested below the top level must
engage reconstruction");
+
+ // The file's shredded shape must reach every nested position of the read
schema, otherwise
+ // parquet never materializes typed_value there.
+ Schema intermediateAvro =
reconstruction.intermediateSchema().toAvroSchema();
+ Schema nestedAvro = intermediateAvro.getField("nested").schema();
+ Schema itemsAvro = intermediateAvro.getField("items").schema();
+ Schema tagsAvro = intermediateAvro.getField("tags").schema();
+ assertNotNull(nestedAvro.getField("v").schema().getField("typed_value"));
+ assertNotNull(itemsAvro.getElementType().getField("typed_value"));
+ assertNotNull(tagsAvro.getValueType().getField("typed_value"));
+
+ GenericRecord nested = new GenericData.Record(nestedAvro);
+ nested.put("v", shreddedVariantRecord(nestedAvro.getField("v").schema(),
7));
+ GenericData.Array<Object> items = new GenericData.Array<>(2, itemsAvro);
+ items.add(shreddedVariantRecord(itemsAvro.getElementType(), 8));
+ // A null element must survive the descent untouched rather than blow up
the rebuild.
+ items.add(null);
+ Map<String, Object> tags = new HashMap<>();
+ tags.put("a", shreddedVariantRecord(tagsAvro.getValueType(), 9));
+
+ GenericRecord input = new GenericData.Record(intermediateAvro);
+ input.put("id", "record-1");
+ input.put("nested", nested);
+ input.put("items", items);
+ input.put("tags", tags);
+
+ IndexedRecord output = reconstruction.reconstruct(input);
+ assertEquals("record-1", output.get(0).toString());
+ // The fake provider folds typed_value into value, so a rebuilt variant
carries it there.
+ assertEquals(ByteBuffer.wrap(new byte[] {7}),
+ ((GenericRecord) ((GenericRecord)
output.get(1)).get("v")).get("value"));
+ assertEquals(ByteBuffer.wrap(new byte[] {8}),
+ ((GenericRecord) ((List<?>) output.get(2)).get(0)).get("value"));
+ assertNull(((List<?>) output.get(2)).get(1));
+ assertEquals(ByteBuffer.wrap(new byte[] {9}),
+ ((GenericRecord) ((Map<?, ?>) output.get(3)).get("a")).get("value"));
+
+ // Empty containers have nothing to rebuild and must come back empty, not
null.
+ input.put("items", new GenericData.Array<>(0, itemsAvro));
+ input.put("tags", new HashMap<String, Object>());
+ IndexedRecord emptied = reconstruction.reconstruct(input);
+ assertTrue(((List<?>) emptied.get(2)).isEmpty());
+ assertTrue(((Map<?, ?>) emptied.get(3)).isEmpty());
+ }
+
+ @Test
+ void returnsNullForNestedFooterDerivedPlainUnshreddedShape(@TempDir Path
tmp) {
+ // The nested twin of returnsNullForFooterDerivedPlainUnshreddedShape:
descending into nested
+ // positions must not make an ordinary unshredded variant look like a
reconstruction target.
+ HoodieSchema fileSchema = recordWithNestedVariants(
+ footerStylePlainUnshreddedSchema("nested_v"),
+ footerStylePlainUnshreddedSchema("element_v"),
+ footerStylePlainUnshreddedSchema("map_v"));
+ HoodieSchema requestedSchema = recordWithNestedVariants(
+ HoodieSchema.createVariant("nested_v", "org.apache.hudi.test", null),
+ HoodieSchema.createVariant("element_v", "org.apache.hudi.test", null),
+ HoodieSchema.createVariant("map_v", "org.apache.hudi.test", null));
+ assertNull(HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ storageWithReadingShredded(tmp, false)));
+ }
+
+ private static GenericRecord shreddedVariantRecord(Schema shreddedSchema,
int typedValue) {
+ GenericRecord shredded = new GenericData.Record(shreddedSchema);
+ shredded.put("metadata", ByteBuffer.wrap(new byte[] {1}));
+ shredded.put("value", null);
+ shredded.put("typed_value", typedValue);
+ return shredded;
+ }
+
+ /** A record carrying a variant inside a struct, inside an array and as a
map value. */
+ private static HoodieSchema recordWithNestedVariants(HoodieSchema
nestedVariant,
+ HoodieSchema
elementVariant,
+ HoodieSchema
mapVariant) {
+ HoodieSchema nested = HoodieSchema.createRecord("nested_record",
"org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("v", nestedVariant)));
+ return HoodieSchema.createRecord("test_nested_record",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("nested", nested),
+ HoodieSchemaField.of("items",
HoodieSchema.createArray(elementVariant)),
+ HoodieSchemaField.of("tags", HoodieSchema.createMap(mapVariant))));
+ }
+
+ /**
+ * The shape a shredded variant column has after the parquet footer
MessageType is converted
+ * back to a schema: a plain record of {metadata, value, typed_value} with
no variant logical
+ * type attached.
+ */
+ private static HoodieSchema footerStylePlainShreddedSchema() {
+ return footerStylePlainShreddedSchema("v");
+ }
+
+ private static HoodieSchema footerStylePlainShreddedSchema(String name) {
+ return HoodieSchema.createRecord(name, "org.apache.hudi.test", null,
Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("value",
HoodieSchema.createNullable(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("typed_value",
HoodieSchema.createNullable(HoodieSchemaType.INT))));
+ }
+
+ /**
+ * The shape an unshredded variant column has after the parquet footer
MessageType is converted
+ * back to a schema: a plain record of {metadata, value} with no variant
logical type attached.
+ */
+ private static HoodieSchema footerStylePlainUnshreddedSchema() {
+ return footerStylePlainUnshreddedSchema("v");
+ }
+
+ private static HoodieSchema footerStylePlainUnshreddedSchema(String name) {
+ return HoodieSchema.createRecord(name, "org.apache.hudi.test", null,
Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("value",
HoodieSchema.createNullable(HoodieSchemaType.BYTES))));
+ }
+
private static HoodieSchema recordWithIdAndVariant(HoodieSchema
variantSchema) {
return HoodieSchema.createRecord("test_record", "org.apache.hudi.test",
null, Arrays.asList(
HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
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 93392648107b..b942e5a234ce 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.fs.FSUtils
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.HoodieSchemaException
@@ -550,6 +551,209 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
}
}
+ test("Test COW small-file merge preserves shredded VARIANT values") {
+ // The #19567 repro: the second insert bin-packs into the existing small
file group, and the
+ // small-file merge (HoodieConcatHandle -> HoodieMergeHelper) rewrites the
old base file
+ // through the AVRO read path (HoodieAvroParquetReader +
HoodieVariantReconstruction), not the
+ // Spark reader context the clustering tests above exercise. Before the
fix, the footer-derived
+ // reader schema lost the variant logical type, so the merge's
strict-projection check failed
+ // and degenerated the requested schema to the footer schema itself;
reconstruction had no
+ // variant column to anchor on, and the writer-schema rewrite silently
dropped typed_value,
+ // nulling rows 1-2 after the second commit. Pinned to the AVRO record
type: that is the leg
+ // this fix covers, and the record type picks the reader the merge uses.
Unlike the clustering
+ // tests, small.file.limit stays at its default on purpose: the bin-pack
is the trigger.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withRecordType(Seq(HoodieRecordType.AVRO))(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 = 'true',
+ | hoodie.parquet.variant.force.shredding.schema.for.test = 'key
string',
+ | hoodie.index.type = 'INMEMORY'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ // Pin the trigger: the first base file must be shredded, or this
degenerates into a
+ // plain unshredded merge.
+ val firstFiles = listDataParquetFiles(tablePath)
+ assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
+ firstFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"First base file should 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)")
+
+ // Pin that the second commit went through the small-file merge: both
parquet versions
+ // belong to one file group, no second group was created.
+ val fileGroupIds = listDataParquetFiles(tablePath)
+ .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
+ assert(fileGroupIds.size == 1,
+ s"Second insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
+
+ // Rows 1 and 2 survive only if the merge carried them out of the
shredded base file;
+ // nulls here mean the AVRO read path dropped typed_value (#19567).
+ 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 COW small-file merge preserves shredded VARIANT values when the
schema evolves") {
+ // The evolving-schema leg of the same merge. Adding a column makes the
writer schema stop
+ // being a strict projection of the base file's, so runMerge takes the
other branch of
+ // recordSchema = isPureProjection ? writerSchema : readerSchema. That
branch used to hand the
+ // reader the raw footer schema, where the variant column is a plain
record, so reconstruction
+ // could not anchor and rewriteRecordWithNewSchema copied {metadata,
value} by name and dropped
+ // typed_value - the same silent corruption as the un-evolved case,
reached without any
+ // schema-on-read config. Aligning the reader schema up front covers both
branches.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
+
+ withRecordType(Seq(HoodieRecordType.AVRO))(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 = 'true',
+ | hoodie.parquet.variant.force.shredding.schema.for.test = 'key
string',
+ | hoodie.index.type = 'INMEMORY'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ val firstFiles = listDataParquetFiles(tablePath)
+ assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
+ firstFiles.foreach { filePath =>
+ val parquetSchema = readParquetSchema(filePath)
+ val variantGroup = getFieldAsGroup(parquetSchema, "v")
+ assert(variantGroup.containsField("typed_value"),
+ s"First base file should carry typed_value. Schema:\n$variantGroup")
+ }
+
+ // The added column is what breaks the strict projection: the writer
schema now carries a
+ // field the base file does not have.
+ spark.sql(s"alter table $tableName add columns (note string)")
+
+ spark.sql(s"insert into $tableName values " +
+ "(3, parse_json('{\"key\":\"value3\"}'), 1000, 'n3'), " +
+ "(4, parse_json('{\"key\":\"value4\"}'), 1000, 'n4')")
+
+ val fileGroupIds = listDataParquetFiles(tablePath)
+ .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
+ assert(fileGroupIds.size == 1,
+ s"Second insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
+
+ // Rows 1-2 keep their variants and pick up a null for the new column;
rows 3-4 carry it.
+ checkAnswer(s"select id, cast(v as string), ts, note from $tableName
order by id")(
+ Seq(1, "{\"key\":\"value1\"}", 1000, null),
+ Seq(2, "{\"key\":\"value2\"}", 1000, null),
+ Seq(3, "{\"key\":\"value3\"}", 1000, "n3"),
+ Seq(4, "{\"key\":\"value4\"}", 1000, "n4")
+ )
+ })
+ }
+
+ test("Test COW small-file merge preserves unshredded VARIANT values") {
+ // Companion to the shredded small-file merge test above, with shredding
disabled, split the
+ // same way as the clustering pair. The shredded twin only proves the
merge stopped losing
+ // typed_value; this one guards the other direction, because
alignShreddedVariants now runs on
+ // every HoodieMergeHelper.runMerge. An ordinary variant table writes the
plain {metadata,
+ // value} layout, nothing should match the shredded shape, the alignment
must be a no-op, and
+ // the bin-pack must keep round-tripping exactly as it did before the fix.
If this goes red
+ // while the shredded twin stays green, the alignment is reaching columns
it should not.
+ assume(HoodieSparkUtils.gteqSpark4_1, "Variant small-file merge read-back
requires Spark 4.1 or higher")
+
+ withRecordType(Seq(HoodieRecordType.AVRO))(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'
+ | )
+ """.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 merge
goes uncovered.
+ val firstFiles = listDataParquetFiles(tablePath)
+ assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
+ firstFiles.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)")
+
+ // Same bin-pack pin as the shredded twin: one file group, so this
really went through the
+ // small-file merge and not a fresh insert.
+ val fileGroupIds = listDataParquetFiles(tablePath)
+ .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
+ assert(fileGroupIds.size == 1,
+ s"Second insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
+
+ 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")
