You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/31 08:27:29 UTC
(hudi) branch master updated: fix(variant): shred nested variants on the Avro write path for parity… (#19777)
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 6103a52e5c8b fix(variant): shred nested variants on the Avro write path for parity… (#19777)
6103a52e5c8b is described below
commit 6103a52e5c8b272095e3f8eae241fb1fda33bf17
Author: voonhous <vo...@gmail.com>
AuthorDate: Mon Aug 31 16:27:22 2026 +0800
fix(variant): shred nested variants on the Avro write path for parity… (#19777)
* fix(variant): shred nested variants on the Avro write path for parity with the row writer
HoodieRowParquetWriteSupport shreds a variant at any depth, while
HoodieAvroWriteSupport only touched top-level fields, so a nested
variant silently declined to the unshredded layout on the AVRO record
type. Same table, same config, two on-disk layouts.
- VariantSchemaUtils.applyForcedShredding reaches variant record
members at any depth, mirroring the row writer; a bare array element
or map value stays unforced on both paths. Generated records get a
Hudi-owned namespace, so a reused record type or a DDL field named
like a user type no longer trips Avro's "Can't redefine". A record
type with a variant member reached both at a forced position and
inside a multi-branch union, which the DDL does not reach on either
the write or the read side, is rejected at splice time with a clear
error instead of failing at footer stamping.
- HoodieAvroWriteSupport shreds values wherever the effective schema
declares typed_value, through a Shredder tree with by-name field
matching. A declared nested typed_value used to crash at the first
write.
- stripVariantShreddingByShape (the footer fallback) recurses, so a
nested typed_value cannot leak into the table schema and Hive sync.
- supportBatch applies the top-level variant policy at any depth, so
nested-variant tables read row-based on Spark 4.1+ like top-level
ones (#18605).
- HoodieSchema.createShreddedFieldStruct and
createVariantShreddedObject take the namespace explicitly; the
overloads that minted records with a null namespace are gone.
Out of scope: inference at depth; arrays keep their per-writer list
layout; the parquet-avro < 1.14 single-field-element read limitation
is #19782. The Spark-native internal read path is #19775.
Tests extend TestHoodieAvroWriteSupportShredding (with a value-level
parquet round trip), TestVariantSchemaUtils, TestHoodieSchema and
TestVariantShreddingMixedLayouts.
Closes #19689. Part of #18937.
---
.../hudi/common/avro/VariantSchemaUtils.java | 369 +++++++++++---
.../hudi/common/config/HoodieStorageConfig.java | 6 +-
.../apache/hudi/common/schema/HoodieSchema.java | 59 ++-
.../hudi/common/table/TableSchemaResolver.java | 5 +-
.../hudi/common/avro/TestVariantSchemaUtils.java | 223 +++++++-
.../hudi/common/schema/TestHoodieSchema.java | 33 +-
.../apache/hudi/avro/HoodieAvroWriteSupport.java | 350 +++++++------
.../hadoop/HoodieVariantReconstruction.java | 7 +-
.../avro/TestHoodieAvroWriteSupportShredding.java | 565 ++++++++++++++++++++-
.../hadoop/TestHoodieVariantReconstruction.java | 14 +-
.../hudi/hadoop/HiveHoodieReaderContext.java | 2 +-
.../hudi/hadoop/TestHoodieParquetInputFormat.java | 10 +-
.../hudi/hadoop/testutils/InputFormatTestUtil.java | 19 +-
.../HoodieFileGroupReaderBasedFileFormat.scala | 38 +-
.../parquet/VariantParquetTestFixtures.scala | 28 +-
.../schema/TestVariantShreddingMixedLayouts.scala | 157 +++++-
.../dml/schema/VariantShreddingTestSupport.scala | 37 +-
.../TestHoodieVariantReconstructionRoundTrip.java | 6 +-
.../TestSpark4VariantShreddingProvider.java | 6 +-
.../TestHoodieRowParquetWriteSupportVariant.java | 14 +-
.../TestSpark40HoodieParquetReadSupport.scala | 25 +-
21 files changed, 1604 insertions(+), 369 deletions(-)
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 8d6e40e4cf90..eef5d07401a7 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
@@ -26,11 +26,15 @@ 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.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.schema.internal.HoodieSchemaException;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.avro.Schema;
+
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -50,10 +54,37 @@ public class VariantSchemaUtils {
/**
* Namespace of the shredded record types {@link #applyInferredShredding} generates. Hudi-owned
* so the generated names ({@code <column>_variant}) cannot collide with a user-declared record
- * type of the same simple name in the table schema.
+ * type of the same simple name in the table schema. Deliberately distinct from
+ * {@link #FORCED_VARIANT_NAMESPACE}: the two splices generate the same simple name for the same
+ * column, so a shared namespace would let an inferred record and a forced one collide on one
+ * full name while carrying different typed_value schemas.
*/
private static final String INFERRED_VARIANT_NAMESPACE = "hoodie.variant.inferred";
+ /**
+ * Namespace prefix of the shredded record types {@link #applyForcedShredding} generates; the
+ * enclosing record's full name is appended, so a forced variant is named
+ * {@code hoodie.variant.forced.<enclosing record full name>.<field>_variant}.
+ *
+ * <p>The name has to be a function of the (enclosing record type, field name) pair and of
+ * nothing else. {@code Schema.toString()}, which is what gets stamped into the parquet footer as
+ * {@code parquet.avro.schema}, throws "Can't redefine" when two NON-equal record definitions
+ * share a full name, and emits the second of two EQUAL ones as a bare name reference. A record
+ * type reused under two fields is rebuilt once per position by this splice, so the two rebuilds
+ * must come out EQUAL; keying the generated name on the enclosing record TYPE does that, whereas
+ * a dotted-path name ({@code a.v} against {@code b.v}) would make them differ and fail at file
+ * open. Two distinct record types with a same-named variant member still land in distinct
+ * namespaces, which is exactly Avro's own notion of identity.
+ *
+ * <p>The records generated under the variant carry this namespace too, so a DDL field name is
+ * free to match a user-declared record type in the table schema: the {@code typed_value} record
+ * sits directly in it, and the {@code {value, typed_value}} struct per DDL field one level below
+ * that, under the {@code typed_value} record itself (see
+ * {@link HoodieSchema#createShreddedFieldStruct(String, String, HoodieSchema)} for why a DDL
+ * field spelled {@code typed_value} or {@code <column>_variant} needs the extra level).
+ */
+ private static final String FORCED_VARIANT_NAMESPACE = "hoodie.variant.forced";
+
private VariantSchemaUtils() {
}
@@ -144,22 +175,205 @@ public class VariantSchemaUtils {
}
/**
- * Strips {@code typed_value} from top-level fields that have the variant SHAPE but lost the
- * variant logical type, i.e. plain records of {@code {metadata: bytes, value: [nullable]
- * bytes, typed_value}} (see {@link #isShreddedVariantShape}). Parquet-footer-derived schemas
- * come back this way (the converter does not attach the variant logical type), so
+ * Splices the forced test-DDL shredding schema into every variant that is a RECORD MEMBER of
+ * {@code schema}, at any depth: top-level fields, members of nested records, and members of
+ * records reached through array elements and map values. The inverse direction of
+ * {@link #stripVariantShredding}. Shredded and unshredded variants alike are replaced, because
+ * the DDL overrides whatever the write schema declared. Returns {@code schema} as-is when it is
+ * not a record, when {@code typedValueFields} is empty, or when nothing below it is a variant.
+ *
+ * <p>A variant that is DIRECTLY an array element or a map value is deliberately NOT forced,
+ * mirroring {@code HoodieRowParquetWriteSupport}: its forced-DDL arm sits in
+ * {@code generateShreddedSchema}, which walks record fields, while the array and map arms of
+ * {@code processNestedDataType} shred an element or value only when the write schema itself
+ * declares a {@code typed_value} there. Both write supports keeping the same reach is what makes
+ * a table's on-disk layout independent of the record type it was written with.
+ *
+ * <p>Value-level shredding is schema-driven at any position ({@code HoodieAvroWriteSupport}
+ * walks the effective schema this method returns), so a hand-authored write schema that declares
+ * {@code typed_value} on a bare element or value still shreds; only this DDL hook stops at
+ * record members.
+ *
+ * @param schema the writer schema
+ * @param typedValueFields the DDL fields (name to type) every forced variant is shredded on, as
+ * parsed from {@code hoodie.parquet.variant.force.shredding.schema.for.test}
+ * @throws HoodieSchemaException when the splice would leave two definitions of one record full
+ * name, see {@code checkOneDefinitionPerRecordName}
+ */
+ public static HoodieSchema applyForcedShredding(HoodieSchema schema, Map<String, HoodieSchema> typedValueFields) {
+ if (schema.getType() != HoodieSchemaType.RECORD || typedValueFields == null || typedValueFields.isEmpty()) {
+ return schema;
+ }
+ HoodieSchema forced = applyForcedShreddingToRecord(schema, typedValueFields);
+ if (forced != schema) {
+ checkOneDefinitionPerRecordName(forced.getAvroSchema(), new HashMap<>());
+ }
+ return forced;
+ }
+
+ private static HoodieSchema applyForcedShreddingToRecord(HoodieSchema record, Map<String, HoodieSchema> typedValueFields) {
+ List<HoodieSchemaField> fields = record.getFields();
+ // Built lazily, as in stripRecordVariantShredding: every record without a variant member walks
+ // this method and must not pay for a field copy it will throw away.
+ List<HoodieSchemaField> newFields = null;
+ for (int i = 0; i < fields.size(); i++) {
+ HoodieSchemaField field = fields.get(i);
+ HoodieSchema fieldSchema = field.schema();
+ // A record member is the one position the DDL is allowed to shred; see applyForcedShredding.
+ HoodieSchema replacement = applyForcedShreddingAt(fieldSchema, typedValueFields, record, field.name(), true);
+ if (replacement != fieldSchema && newFields == null) {
+ newFields = copyFieldsBefore(fields, i);
+ }
+ if (newFields != null) {
+ // 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 (newFields == null) {
+ return record;
+ }
+ return HoodieSchema.createRecord(
+ record.getAvroSchema().getName(),
+ record.getAvroSchema().getNamespace(),
+ record.getAvroSchema().getDoc(),
+ newFields);
+ }
+
+ /**
+ * Applies the forced DDL at one schema position, returning the argument instance when nothing
+ * changes. {@code enclosingRecord} and {@code fieldName} are the record member this position was
+ * reached from and name the generated record (see {@link #FORCED_VARIANT_NAMESPACE}).
+ * {@code variantAllowed} is false once the walk has stepped through an array element or a map
+ * value, where the DDL does not reach; it goes back to true for the members of any record found
+ * there, so {@code array<struct<v variant>>} shreds while {@code array<variant>} does not.
+ *
+ * <p>A genuine multi-branch UNION is out of scope. {@code getNonNullType} only unwraps the
+ * nullable two-branch form, so anything else comes back a UNION, hits {@code default} and passes
+ * through untouched -- a variant inside it is never forced, at any depth below it.
+ * {@code HoodieAvroWriteSupport.buildShredder} has the same arm, so the schema splice and the
+ * value walk agree on what a union does, and so does the read side
+ * ({@code HoodieVariantReconstruction.buildRebuilder}), which is why recursing here alone is not
+ * an option: it would write shredded groups under unions that the Avro reader never
+ * reconstructs. A future change to any of the three arms has to move all of them. The one shape
+ * the pass-through cannot represent -- a record type with a variant member reached both here and
+ * at a forced position, which would leave two definitions of one name -- is rejected up front by
+ * {@link #applyForcedShredding}.
+ */
+ private static HoodieSchema applyForcedShreddingAt(HoodieSchema schema,
+ Map<String, HoodieSchema> typedValueFields,
+ HoodieSchema enclosingRecord,
+ String fieldName,
+ boolean variantAllowed) {
+ boolean wasNullable = schema.isNullable();
+ HoodieSchema unwrapped = wasNullable ? schema.getNonNullType() : schema;
+ HoodieSchema replacement;
+ switch (unwrapped.getType()) {
+ case VARIANT:
+ if (!variantAllowed) {
+ return schema;
+ }
+ replacement = HoodieSchema.createVariantShreddedObject(
+ fieldName + "_variant",
+ FORCED_VARIANT_NAMESPACE + "." + enclosingRecord.getFullName(),
+ unwrapped.getAvroSchema().getDoc(),
+ typedValueFields);
+ break;
+ case RECORD:
+ replacement = applyForcedShreddingToRecord(unwrapped, typedValueFields);
+ break;
+ case ARRAY: {
+ HoodieSchema elementType = unwrapped.getElementType();
+ HoodieSchema shreddedElement = applyForcedShreddingAt(elementType, typedValueFields, enclosingRecord, fieldName, false);
+ replacement = shreddedElement == elementType ? unwrapped : HoodieSchema.createArray(shreddedElement);
+ break;
+ }
+ case MAP: {
+ HoodieSchema valueType = unwrapped.getValueType();
+ HoodieSchema shreddedValue = applyForcedShreddingAt(valueType, typedValueFields, enclosingRecord, fieldName, false);
+ replacement = shreddedValue == valueType ? unwrapped : HoodieSchema.createMap(shreddedValue);
+ break;
+ }
+ default:
+ return schema;
+ }
+ if (replacement == unwrapped) {
+ return schema;
+ }
+ return wasNullable ? HoodieSchema.createNullable(replacement) : replacement;
+ }
+
+ /**
+ * Rejects a spliced schema that defines one record full name twice with different fields. That
+ * is what Avro's {@code Schema.toString()} refuses ("Can't redefine") once the file schema is
+ * stamped into the parquet footer on avro 1.11, and what avro 1.12 turns into a footer that
+ * parses back into another schema (the second definition is written as a bare name reference),
+ * so the check cannot be left to serialization. The one way the splice produces such a pair is a
+ * record type with a variant member reached both at a position the DDL forces and inside a
+ * multi-branch union, which {@link #applyForcedShreddingAt} passes through untouched; the error
+ * says so. A record already seen is not descended into again, so a type reused at many
+ * positions costs one equality check per reuse, and a recursive type terminates.
+ */
+ private static void checkOneDefinitionPerRecordName(Schema schema, Map<String, Schema> definitions) {
+ switch (schema.getType()) {
+ case RECORD: {
+ Schema previous = definitions.putIfAbsent(schema.getFullName(), schema);
+ if (previous != null) {
+ if (!previous.equals(schema)) {
+ throw new HoodieSchemaException(String.format(
+ "Forced variant shredding (%s) produced two different definitions of record type '%s'. "
+ + "A record type with a variant member reached both at a position the DDL forces and inside "
+ + "a multi-branch union does this: the DDL does not reach into unions (nor does the Avro read "
+ + "path), so the type is rebuilt at the first position and kept as-is under the union, and Avro "
+ + "allows one definition per name in a file schema. Give the two positions distinct record "
+ + "types or take the variant out of the union.",
+ HoodieStorageConfig.PARQUET_VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key(), schema.getFullName()));
+ }
+ return;
+ }
+ for (Schema.Field field : schema.getFields()) {
+ checkOneDefinitionPerRecordName(field.schema(), definitions);
+ }
+ return;
+ }
+ case ARRAY:
+ checkOneDefinitionPerRecordName(schema.getElementType(), definitions);
+ return;
+ case MAP:
+ checkOneDefinitionPerRecordName(schema.getValueType(), definitions);
+ return;
+ case UNION:
+ for (Schema branch : schema.getTypes()) {
+ checkOneDefinitionPerRecordName(branch, definitions);
+ }
+ return;
+ default:
+ return;
+ }
+ }
+
+ /**
+ * Strips {@code typed_value} from fields that have the variant SHAPE but lost the variant
+ * logical type, i.e. plain records of {@code {metadata: bytes, value: [nullable] bytes,
+ * typed_value}} (see {@link #isShreddedVariantShape}). Parquet-footer-derived schemas come back
+ * this way (the converter does not attach the variant logical type), so
* {@link #stripVariantShredding} alone cannot see them. Used by the table-schema footer
* fallback only; returns {@code schema} as-is when nothing matches.
*
+ * <p>The walk recurses through records, array elements and map values because both write
+ * supports shred below the top level: the forced-shredding DDL reaches every variant that is a
+ * record member at any depth on the ROW path
+ * ({@code HoodieRowParquetWriteSupport.processNestedDataType}) and on the AVRO path
+ * ({@link #applyForcedShredding}), and a write schema that declares {@code typed_value} on a
+ * bare array element or map value shreds it on either path.
+ *
* <p>Unlike {@link #isShreddedVariantTarget}, the match here has NO requested-side anchor:
* the footer fallback runs precisely when no table schema is available to anchor on, so a
* plain user struct that happens to have exactly this shape is stripped too (a documented,
* accepted false positive: {@code metadata} plus {@code typed_value} is the variant spec's
- * vocabulary). Top-level fields only, matching the scope of
- * {@link #getInferableVariantColumns}: inference never shreds a nested variant, and
- * {@code HoodieAvroWriteSupport.applyForcedShreddingSchema} walks top-level fields only. The row
- * writer shreds at any depth its write schema asks it to, forced DDL or not (see
- * {@link #swapShreddedVariantFields}); those nested layouts are not covered by this fallback.
+ * vocabulary). Recursing extends that false positive from top-level fields to every depth. A
+ * match is terminal: the members of a variant group are the spec's own fields, not user columns
+ * that could hold a nested variant of their own.
*
* <p>The shape check also admits the spec's two-field {@code {metadata, typed_value}} form (a
* writer may omit {@code value} when every row is typed). Stripping that would leave a
@@ -170,53 +384,93 @@ public class VariantSchemaUtils {
if (schema.getType() != HoodieSchemaType.RECORD) {
return schema;
}
+ return stripRecordVariantShreddingByShape(schema);
+ }
- List<HoodieSchemaField> newFields = new ArrayList<>();
- boolean changed = false;
-
- for (HoodieSchemaField field : schema.getFields()) {
+ private static HoodieSchema stripRecordVariantShreddingByShape(HoodieSchema record) {
+ List<HoodieSchemaField> fields = record.getFields();
+ // Built lazily, as in stripRecordVariantShredding: this fallback runs on every table-schema
+ // resolution from a data file, and the overwhelmingly common schema matches nothing.
+ List<HoodieSchemaField> newFields = null;
+ for (int i = 0; i < fields.size(); i++) {
+ HoodieSchemaField field = fields.get(i);
HoodieSchema fieldSchema = field.schema();
- boolean wasNullable = fieldSchema.isNullable();
- HoodieSchema unwrapped = wasNullable ? fieldSchema.getNonNullType() : fieldSchema;
-
- if (isShreddedVariantShape(unwrapped)) {
- List<HoodieSchemaField> strippedFields = new ArrayList<>();
- for (HoodieSchemaField member : unwrapped.getFields()) {
- if (!HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD.equals(member.name())) {
- strippedFields.add(HoodieSchemaUtils.createNewSchemaField(member));
- }
- }
- if (!unwrapped.getField(HoodieSchema.Variant.VARIANT_VALUE_FIELD).isPresent()) {
- // Null default, like every other optional field on this path (the footer converter
- // attaches one, and kept members carry theirs over): HoodieSchemaField#equals compares
- // defaults, so omitting it would make the stripped schema differ from the one a
- // three-field file yields for the same column.
- strippedFields.add(HoodieSchemaField.of(
- HoodieSchema.Variant.VARIANT_VALUE_FIELD, HoodieSchema.createNullable(HoodieSchemaType.BYTES),
- null, HoodieSchema.NULL_VALUE));
- }
- HoodieSchema stripped = HoodieSchema.createRecord(
- unwrapped.getAvroSchema().getName(),
- unwrapped.getAvroSchema().getNamespace(),
- unwrapped.getAvroSchema().getDoc(),
- strippedFields);
- HoodieSchema replacement = wasNullable ? HoodieSchema.createNullable(stripped) : stripped;
- newFields.add(HoodieSchemaUtils.createNewSchemaField(field.withSchema(replacement)));
- changed = true;
- } else {
- newFields.add(HoodieSchemaUtils.createNewSchemaField(field));
+ HoodieSchema replacement = stripVariantShreddingByShapeAt(fieldSchema);
+ if (replacement != fieldSchema && newFields == null) {
+ newFields = copyFieldsBefore(fields, i);
+ }
+ if (newFields != null) {
+ // 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 (newFields == null) {
+ return record;
+ }
+ return HoodieSchema.createRecord(
+ record.getAvroSchema().getName(),
+ record.getAvroSchema().getNamespace(),
+ record.getAvroSchema().getDoc(),
+ newFields);
+ }
- if (!changed) {
+ /** Strips by shape at one schema position, returning the argument instance when nothing changes. */
+ private static HoodieSchema stripVariantShreddingByShapeAt(HoodieSchema schema) {
+ boolean wasNullable = schema.isNullable();
+ HoodieSchema unwrapped = wasNullable ? schema.getNonNullType() : schema;
+ HoodieSchema replacement;
+ switch (unwrapped.getType()) {
+ case RECORD:
+ // A match is terminal: whatever sits under a variant group belongs to the shredding spec,
+ // not to the user, so the walk never descends into one.
+ replacement = isShreddedVariantShape(unwrapped)
+ ? stripShreddedVariantShape(unwrapped)
+ : stripRecordVariantShreddingByShape(unwrapped);
+ break;
+ case ARRAY: {
+ HoodieSchema elementType = unwrapped.getElementType();
+ HoodieSchema strippedElement = stripVariantShreddingByShapeAt(elementType);
+ replacement = strippedElement == elementType ? unwrapped : HoodieSchema.createArray(strippedElement);
+ break;
+ }
+ case MAP: {
+ HoodieSchema valueType = unwrapped.getValueType();
+ HoodieSchema strippedValue = stripVariantShreddingByShapeAt(valueType);
+ replacement = strippedValue == valueType ? unwrapped : HoodieSchema.createMap(strippedValue);
+ break;
+ }
+ default:
+ return schema;
+ }
+ if (replacement == unwrapped) {
return schema;
}
+ return wasNullable ? HoodieSchema.createNullable(replacement) : replacement;
+ }
+ /** Rebuilds one shape-matched variant group in the unshredded {@code {metadata, value}} form. */
+ private static HoodieSchema stripShreddedVariantShape(HoodieSchema variantShaped) {
+ List<HoodieSchemaField> strippedFields = new ArrayList<>();
+ for (HoodieSchemaField member : variantShaped.getFields()) {
+ if (!HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD.equals(member.name())) {
+ strippedFields.add(HoodieSchemaUtils.createNewSchemaField(member));
+ }
+ }
+ if (!variantShaped.getField(HoodieSchema.Variant.VARIANT_VALUE_FIELD).isPresent()) {
+ // Null default, like every other optional field on this path (the footer converter
+ // attaches one, and kept members carry theirs over): HoodieSchemaField#equals compares
+ // defaults, so omitting it would make the stripped schema differ from the one a
+ // three-field file yields for the same column.
+ strippedFields.add(HoodieSchemaField.of(
+ HoodieSchema.Variant.VARIANT_VALUE_FIELD, HoodieSchema.createNullable(HoodieSchemaType.BYTES),
+ null, HoodieSchema.NULL_VALUE));
+ }
return HoodieSchema.createRecord(
- schema.getAvroSchema().getName(),
- schema.getAvroSchema().getNamespace(),
- schema.getAvroSchema().getDoc(),
- newFields);
+ variantShaped.getAvroSchema().getName(),
+ variantShaped.getAvroSchema().getNamespace(),
+ variantShaped.getAvroSchema().getDoc(),
+ strippedFields);
}
/**
@@ -428,15 +682,16 @@ public class VariantSchemaUtils {
* file side, which is what {@link #isShreddedVariantTarget} needs to anchor detection. Returns
* {@code base} when nothing matches.
*
- * <p>The walk recurses through records, array elements and map values because the row writer
- * shreds at any depth its write schema asks it to
- * ({@code HoodieRowParquetWriteSupport.processNestedDataType} recurses into structs, array
- * elements and map values, and {@code generateShreddedSchema} re-reads the forced-shredding DDL
- * on every entry, so {@code struct<v variant>} plus
- * {@code hoodie.parquet.variant.force.shredding.schema.for.test} does shred at depth on the ROW
- * path). The AVRO path is the narrower one: {@code HoodieAvroWriteSupport.applyForcedShreddingSchema}
- * walks top-level fields only, so on that path a nested shredded column needs a hand-authored
- * write schema that declares {@code typed_value} below the top level.
+ * <p>The walk recurses through records, array elements and map values because both write
+ * supports shred at any depth their write schema asks them to. The forced-shredding DDL
+ * ({@code hoodie.parquet.variant.force.shredding.schema.for.test}) force-shreds every variant
+ * that is a record member at any depth on the ROW path
+ * ({@code HoodieRowParquetWriteSupport.generateShreddedSchema} re-reads the DDL on every entry
+ * and {@code processNestedDataType} recurses into structs, array elements and map values) and,
+ * since the #19689 fix, on the AVRO path too ({@link #applyForcedShredding}), so
+ * {@code struct<v variant>} comes out shredded whichever record type wrote it. A variant that is
+ * DIRECTLY an array element or a map value is force-shredded by neither; it shreds on either
+ * path only when the write schema itself declares {@code typed_value} there.
*/
private static HoodieSchema swapShreddedVariantFields(HoodieSchema base, HoodieSchema other, boolean baseIsFile) {
List<HoodieSchemaField> baseFields = base.getFields();
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
index 6f2fc87c6073..391ea62a571e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
@@ -281,7 +281,11 @@ public class HoodieStorageConfig extends HoodieConfig {
+ "Spark's internal spark.sql.variant.forceShreddingSchemaForTest. "
+ "The value should be a DDL-format schema string (e.g., 'a int, b string, c decimal(15, 1)'). "
+ "When set and write shredding is enabled, this schema overrides the schema-driven shredding "
- + "configuration for all variant columns.");
+ + "configuration for all variant columns. Applies to variant columns at any depth on both the "
+ + "Spark row write path and the Avro record write path: top-level columns and struct members, "
+ + "including members of structs nested under arrays and maps. A variant that is directly an "
+ + "array element or a map value is not forced; it shreds only when the write schema itself "
+ + "declares a typed_value there.");
public static final ConfigProperty<Boolean> PARQUET_VARIANT_ALLOW_READING_SHREDDED = ConfigProperty
.key("hoodie.parquet.variant.allow.reading.shredded")
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
index 3ec7025bb6a1..5f6cec1bc483 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
@@ -976,11 +976,26 @@ public class HoodieSchema implements Serializable {
* |-- typed_value: <fieldType> (nullable)
* </pre></p>
*
+ * <p>The record is named after the shredded field, so a caller that generates it from a user
+ * schema has to put it in a namespace it owns, and one strictly below the variant's own. With a
+ * null namespace the struct's full name is the bare field name, which collides with a
+ * user-declared record type of that name; in the variant's own namespace a field spelled
+ * {@code typed_value} or {@code <column>_variant} collides with the generated records that sit
+ * there. Either collision breaks {@code Schema.toString()} -- what gets stamped into the parquet
+ * footer: on avro 1.11 it throws "Can't redefine" at file open, and on avro 1.12 it emits the
+ * second definition as a bare reference to the first, i.e. writes a footer schema that parses
+ * back into something else. {@link #createVariantShreddedObject} therefore passes the full name
+ * of the enclosing {@code typed_value} record. A null namespace is fine for a caller with
+ * nothing to give -- a struct built standalone, as tests do, has no surrounding schema for the
+ * bare name to collide with; anything generating structs into a user schema owes them one.
+ *
* @param fieldName the name for the record (used as the Avro record name)
+ * @param namespace the namespace of the generated record (null only for a struct built
+ * standalone, with no surrounding schema, per above)
* @param fieldType the schema for the typed_value within this field
* @return a new HoodieSchema representing the shredded field struct
*/
- public static HoodieSchema createShreddedFieldStruct(String fieldName, HoodieSchema fieldType) {
+ public static HoodieSchema createShreddedFieldStruct(String fieldName, String namespace, HoodieSchema fieldType) {
ValidationUtils.checkArgument(fieldName != null && !fieldName.isEmpty(), "Field name cannot be null or empty");
ValidationUtils.checkArgument(fieldType != null, "Field type cannot be null");
List<HoodieSchemaField> fields = Arrays.asList(
@@ -997,7 +1012,7 @@ public class HoodieSchema implements Serializable {
NULL_VALUE
)
);
- return HoodieSchema.createRecord(fieldName, null, null, fields);
+ return HoodieSchema.createRecord(fieldName, namespace, null, fields);
}
/**
@@ -1010,7 +1025,7 @@ public class HoodieSchema implements Serializable {
* fields.put("a", HoodieSchema.create(HoodieSchemaType.INT));
* fields.put("b", HoodieSchema.create(HoodieSchemaType.STRING));
* fields.put("c", HoodieSchema.createDecimal(15, 1));
- * HoodieSchema.Variant variant = HoodieSchema.createVariantShreddedObject(fields);
+ * HoodieSchema.Variant variant = HoodieSchema.createVariantShreddedObject("v_variant", "com.example.rec", null, fields);
* }</pre></p>
*
* <p>Produces the following structure:
@@ -1030,18 +1045,19 @@ public class HoodieSchema implements Serializable {
* | | |-- typed_value: decimal(15,1) (nullable)
* </pre></p>
*
- * @param shreddedFields Map of field names to their typed value schemas. Use LinkedHashMap for ordered fields.
- * @return a new HoodieSchema.Variant with properly nested typed_value
- */
- public static HoodieSchema.Variant createVariantShreddedObject(Map<String, HoodieSchema> shreddedFields) {
- return createVariantShreddedObject(null, null, null, shreddedFields);
- }
-
- /**
- * Creates a shredded Variant schema for an object type with custom name, namespace, and documentation.
+ * <p>There is deliberately no overload without a namespace: the variant record, its
+ * {@code typed_value} record and every per-field struct are named records, and a null namespace
+ * puts them all at the bare names {@code variant}, {@code typed_value} and the field names, where
+ * two differently shredded variants in one schema, or a user record type spelled like a DDL
+ * field, collide; see {@link #createShreddedFieldStruct(String, String, HoodieSchema)} for what
+ * a collision costs. A null namespace is fine only for a variant built standalone, with no
+ * surrounding schema, as tests do; a caller generating into a user schema owes it one, and the
+ * one production caller ({@code VariantSchemaUtils.applyForcedShredding}) does.
*
* @param name the variant record name (can be null, defaults to "variant")
- * @param namespace the namespace (can be null)
+ * @param namespace the namespace of the variant record and its {@code typed_value} record; the per-field
+ * structs go one level below, under {@code <namespace>.typed_value} (null only for a
+ * standalone variant, per above)
* @param doc the documentation (can be null)
* @param shreddedFields Map of field names to their typed value schemas. Use LinkedHashMap for ordered fields.
* @return a new HoodieSchema.Variant with properly nested typed_value
@@ -1054,7 +1070,22 @@ public class HoodieSchema implements Serializable {
// Build typed_value fields, each wrapped in the spec-compliant {value, typed_value} struct
List<HoodieSchemaField> typedValueFields = new ArrayList<>();
for (Map.Entry<String, HoodieSchema> entry : shreddedFields.entrySet()) {
- HoodieSchema fieldStruct = createShreddedFieldStruct(entry.getKey(), entry.getValue());
+ // The field structs go one level below the variant, under the typed_value record that holds
+ // them, so their full name is <namespace>.typed_value.<field>. Their names come from the
+ // DDL/inferred field names, which are free to match a user-declared record type in the table
+ // schema -- and equally free to be spelled "typed_value" or "<column>_variant", the names of
+ // the two records generated in the variant's own namespace. The extra level clears the field
+ // structs of all three; see createShreddedFieldStruct for what a collision costs. It does
+ // not cover the two generated records themselves, which own <namespace>.typed_value and
+ // <namespace>.<column>_variant outright: a user record type declared with either full name
+ // still collides. Accepted -- the only caller generating from a user schema
+ // (VariantSchemaUtils.applyForcedShredding, behind a test-only config) puts them under
+ // hoodie.variant.forced.<enclosing record>, so reaching that residual means declaring a
+ // record type inside a Hudi-owned namespace.
+ HoodieSchema fieldStruct = createShreddedFieldStruct(
+ entry.getKey(),
+ namespace == null ? null : namespace + "." + Variant.VARIANT_TYPED_VALUE_FIELD,
+ entry.getValue());
typedValueFields.add(HoodieSchemaField.of(
entry.getKey(),
HoodieSchema.createNullable(fieldStruct),
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java b/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
index eff6f6a7130e..5ff906cd8b5e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
@@ -118,8 +118,9 @@ public class TableSchemaResolver {
private Option<HoodieSchema> getTableSchemaFromDataFileInternal() {
// typed_value is a per-file physical layout (possibly inferred per file) and is stripped
// from the resolved table schema. Footer-derived schemas lose the variant logical type, so
- // variant columns are also stripped by shape; that fallback walks top-level fields only, so
- // a nested variant the row writer shredded at depth can still surface here.
+ // variant columns are also stripped by shape. Both strips recurse through records, array
+ // elements and map values, so typed_value never surfaces here no matter which writer shredded
+ // the column or at what depth.
return getTableParquetSchemaFromDataFile()
.map(VariantSchemaUtils::stripVariantShredding)
.map(VariantSchemaUtils::stripVariantShreddingByShape);
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
index 089b01f5515d..eb3d00cceadb 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
@@ -25,12 +25,15 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaTestUtils;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.schema.internal.HoodieSchemaException;
+import org.apache.avro.Schema;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -39,6 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestVariantSchemaUtils {
@@ -52,6 +56,24 @@ public class TestVariantSchemaUtils {
"pre_shredded", null, null, HoodieSchema.create(HoodieSchemaType.LONG)))));
}
+ /** The parsed form of a {@code hoodie.parquet.variant.force.shredding.schema.for.test} DDL. */
+ private static Map<String, HoodieSchema> forcedDdl() {
+ Map<String, HoodieSchema> ddl = new LinkedHashMap<>();
+ ddl.put("k", HoodieSchema.create(HoodieSchemaType.STRING));
+ return ddl;
+ }
+
+ /** Asserts that {@code schema} is a variant shredded on exactly the {@link #forcedDdl} fields. */
+ private static void assertShreddedOnDdl(HoodieSchema schema) {
+ HoodieSchema.Variant variant = (HoodieSchema.Variant) schema;
+ assertTrue(variant.isShredded());
+ assertEquals(Collections.singletonList("k"), fieldNames(variant.getTypedValueField().get().getNonNullType()));
+ }
+
+ private static List<String> fieldNames(HoodieSchema record) {
+ return record.getFields().stream().map(HoodieSchemaField::name).collect(Collectors.toList());
+ }
+
private static HoodieConfig inferenceEnabledConfig() {
HoodieConfig config = new HoodieConfig();
config.setValue(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED, "true");
@@ -96,9 +118,10 @@ public class TestVariantSchemaUtils {
@Test
public void testGetInferableVariantColumnsIsTopLevelOnly() {
- // Inference never shreds a nested variant (struct member, array element, map value): the
- // footer-fallback strip and the write supports' hooks share that top-level scope, so a
- // recursive walk here would shred what nothing else accounts for.
+ // Inference is top-level by design and is now the only hook that is: the footer-fallback
+ // strip and both write supports' forced-DDL hooks recurse into record members at any depth
+ // (VariantSchemaUtils.applyForcedShredding), so a nested variant here is a scope decision
+ // about per-file inference, not a consistency gap with the rest of the shredding code.
HoodieSchema schema = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
HoodieSchemaField.of("top", HoodieSchema.createVariant()),
HoodieSchemaField.of("s", HoodieSchema.createRecord("s_rec", "ns", null,
@@ -147,7 +170,7 @@ public class TestVariantSchemaUtils {
// Spec-form object typed_value for v1; v2 intentionally absent (declined).
Map<String, HoodieSchema> typedFields = new LinkedHashMap<>();
typedFields.put("a", HoodieSchema.create(HoodieSchemaType.INT));
- inferred.put("v1", HoodieSchema.createVariantShreddedObject(typedFields).getTypedValueField().get());
+ inferred.put("v1", HoodieSchema.createVariantShreddedObject(null, null, null, typedFields).getTypedValueField().get());
HoodieSchema spliced = VariantSchemaUtils.applyInferredShredding(schema, inferred);
assertNotEquals(schema, spliced);
@@ -194,23 +217,179 @@ public class TestVariantSchemaUtils {
Collections.singletonMap("id", HoodieSchema.create(HoodieSchemaType.LONG))));
}
+ @Test
+ public void testApplyForcedShreddingReachesRecordMembersAtAnyDepth() {
+ HoodieSchema schema = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("pre", HoodieSchema.createVariantShredded(
+ "pre_shredded", null, null, HoodieSchema.create(HoodieSchemaType.LONG))),
+ // s and items are nullable so the record and array arms have to rewrap their replacement.
+ HoodieSchemaField.of("s", HoodieSchemaTestUtils.createNullableRecord("s_rec",
+ HoodieSchemaField.of("inner", HoodieSchema.createNullable(HoodieSchema.createVariant()),
+ null, HoodieSchema.NULL_VALUE)), null, HoodieSchema.NULL_VALUE),
+ HoodieSchemaTestUtils.createNullableArrayField("items", HoodieSchemaTestUtils.createRecord("item_rec",
+ HoodieSchemaField.of("v", HoodieSchema.createVariant("item_variant", null, null)))),
+ HoodieSchemaTestUtils.createMapField("m", HoodieSchemaTestUtils.createRecord("map_rec",
+ HoodieSchemaField.of("v", HoodieSchema.createVariant("map_variant", null, null)))),
+ HoodieSchemaTestUtils.createArrayField("arr", HoodieSchema.createVariant("arr_variant", null, null)),
+ HoodieSchemaTestUtils.createMapField("mv", HoodieSchema.createVariant("mv_variant", null, null))));
+
+ HoodieSchema forced = VariantSchemaUtils.applyForcedShredding(schema, forcedDdl());
+
+ // Every variant that is a record member is forced, at any depth and through collections, and
+ // the DDL overrides a variant that was already shredded on something else.
+ assertShreddedOnDdl(forced.getField("pre").get().schema());
+ HoodieSchema s = forced.getField("s").get().schema();
+ assertTrue(s.isNullable(), "a rebuilt record stays nullable");
+ HoodieSchema inner = s.getNonNullType().getField("inner").get().schema();
+ assertTrue(inner.isNullable(), "nullability is preserved around the spliced variant");
+ assertShreddedOnDdl(inner.getNonNullType());
+ HoodieSchema items = forced.getField("items").get().schema();
+ assertTrue(items.isNullable(), "a rebuilt array stays nullable");
+ assertShreddedOnDdl(items.getNonNullType().getElementType().getField("v").get().schema());
+ assertShreddedOnDdl(forced.getField("m").get().schema().getValueType().getField("v").get().schema());
+
+ // A variant that is DIRECTLY an array element or a map value is not forced, matching
+ // HoodieRowParquetWriteSupport.processNestedDataType: its collection arms shred only a
+ // typed_value the write schema itself declares.
+ assertFalse(((HoodieSchema.Variant) forced.getField("arr").get().schema().getElementType()).isShredded());
+ assertFalse(((HoodieSchema.Variant) forced.getField("mv").get().schema().getValueType()).isShredded());
+ assertEquals(HoodieSchemaType.STRING, forced.getField("id").get().schema().getType());
+
+ // The strip is the inverse at every depth.
+ HoodieSchema stripped = VariantSchemaUtils.stripVariantShredding(forced);
+ assertFalse(((HoodieSchema.Variant) stripped.getField("s").get().schema().getNonNullType()
+ .getField("inner").get().schema().getNonNullType()).isShredded());
+ assertFalse(((HoodieSchema.Variant) stripped.getField("items").get().schema().getNonNullType()
+ .getElementType().getField("v").get().schema()).isShredded());
+ }
+
+ @Test
+ public void testApplyForcedShreddingIdentityWhenNothingMatches() {
+ HoodieSchema noVariants = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaTestUtils.createArrayField("a", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("s", HoodieSchemaTestUtils.createRecord("s_rec",
+ HoodieSchemaField.of("n", HoodieSchema.create(HoodieSchemaType.LONG))))));
+ assertSame(noVariants, VariantSchemaUtils.applyForcedShredding(noVariants, forcedDdl()));
+ // An empty DDL is a no-op (no forced config set), and a non-record schema has no member to
+ // splice into.
+ HoodieSchema withVariants = schemaWithVariants();
+ assertSame(withVariants, VariantSchemaUtils.applyForcedShredding(withVariants, Collections.emptyMap()));
+ HoodieSchema bareVariant = HoodieSchema.createVariant();
+ assertSame(bareVariant, VariantSchemaUtils.applyForcedShredding(bareVariant, forcedDdl()));
+ }
+
+ @Test
+ public void testApplyForcedShreddingNamesVariantsPerEnclosingRecordType() {
+ // One record type used under two fields is rebuilt once per position. Avro's Schema.toString,
+ // which is what lands in the parquet footer as parquet.avro.schema, throws "Can't redefine"
+ // for two non-equal definitions sharing a full name and emits equal ones as a name reference,
+ // so the two rebuilds have to come out EQUAL. Naming the generated record after the enclosing
+ // record TYPE does that; a dotted path (a.v against b.v) would not.
+ HoodieSchema reused = HoodieSchemaTestUtils.createRecord("reused",
+ HoodieSchemaField.of("v", HoodieSchema.createVariant()));
+ HoodieSchema schema = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("a", reused),
+ HoodieSchemaField.of("b", reused)));
+
+ HoodieSchema forced = VariantSchemaUtils.applyForcedShredding(schema, forcedDdl());
+ assertEquals(forced.getField("a").get().schema(), forced.getField("b").get().schema());
+ // Serializing is the actual file-open failure mode the naming rule exists for: two non-equal
+ // rebuilds of one record type would throw "Can't redefine" here, not at splice time.
+ assertEquals(forced.toAvroSchema(), new Schema.Parser().parse(forced.toString()));
+
+ // Two DIFFERENT record types with a same-named variant member stay in distinct namespaces.
+ HoodieSchema twoTypes = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("x", HoodieSchemaTestUtils.createRecord("x_rec",
+ HoodieSchemaField.of("v", HoodieSchema.createVariant()))),
+ HoodieSchemaField.of("y", HoodieSchemaTestUtils.createRecord("y_rec",
+ HoodieSchemaField.of("v", HoodieSchema.createVariant())))));
+ HoodieSchema forcedTwo = VariantSchemaUtils.applyForcedShredding(twoTypes, forcedDdl());
+ assertNotEquals(
+ forcedTwo.getField("x").get().schema().getField("v").get().schema().getFullName(),
+ forcedTwo.getField("y").get().schema().getField("v").get().schema().getFullName());
+
+ // The records generated UNDER the variant are named after the DDL fields, so they have to
+ // carry the Hudi-owned namespace as well: a user record type named like a DDL field ("k"
+ // here) otherwise shares the bare name with the generated {value, typed_value} struct and
+ // serializing throws "Can't redefine: k". The namespace has to be one level BELOW the variant
+ // too, since a DDL field is equally free to be spelled "typed_value" or "<column>_variant" --
+ // the names of the typed_value record and of the variant record generated for column "v".
+ Map<String, HoodieSchema> clashingDdl = new LinkedHashMap<>();
+ clashingDdl.put("k", HoodieSchema.create(HoodieSchemaType.STRING));
+ clashingDdl.put("typed_value", HoodieSchema.create(HoodieSchemaType.STRING));
+ clashingDdl.put("v_variant", HoodieSchema.create(HoodieSchemaType.STRING));
+ HoodieSchema ddlNameClash = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("k", HoodieSchemaTestUtils.createRecord("k",
+ HoodieSchemaField.of("n", HoodieSchema.create(HoodieSchemaType.LONG)))),
+ HoodieSchemaField.of("v", HoodieSchema.createVariant()),
+ // A second variant column in the same record: both columns generate a typed_value record
+ // under the one namespace, so that name is defined twice and has to serialize as a
+ // reference, which only holds while the two definitions stay equal.
+ HoodieSchemaField.of("v2", HoodieSchema.createVariant())));
+ HoodieSchema forcedClash = VariantSchemaUtils.applyForcedShredding(ddlNameClash, clashingDdl);
+ // Equality, not just a successful parse: avro 1.11 throws "Can't redefine" on the clash while
+ // avro 1.12 emits the second definition as a bare reference to the first, which parses fine
+ // and yields a different schema.
+ assertEquals(forcedClash.toAvroSchema(), new Schema.Parser().parse(forcedClash.toString()));
+ }
+
+ @Test
+ public void testApplyForcedShreddingKeepsMultiBranchUnionsOutOfReach() {
+ // The DDL does not reach into a multi-branch union, on the schema side, in the value walk and
+ // on the Avro read path alike (HoodieVariantReconstruction.buildRebuilder has the same default
+ // arm): a record type with a variant member that sits only under one is passed through untouched.
+ HoodieSchema reused = HoodieSchemaTestUtils.createRecord("reused",
+ HoodieSchemaField.of("v", HoodieSchema.createVariant()));
+ HoodieSchema unionWithReused = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL), reused, HoodieSchema.create(HoodieSchemaType.STRING));
+ HoodieSchema underUnionOnly = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("u", unionWithReused, null, HoodieSchema.NULL_VALUE)));
+ assertSame(underUnionOnly, VariantSchemaUtils.applyForcedShredding(underUnionOnly, forcedDdl()));
+
+ // The same type reached at a forced position too would be rebuilt there and kept as-is under
+ // the union: two definitions of one full name, which Avro refuses when the footer schema is
+ // stamped ("Can't redefine" on 1.11; 1.12 writes a name reference that parses back into
+ // another schema). The splice rejects it up front and names the type and the cause.
+ HoodieSchema both = HoodieSchema.createRecord("rec", "ns", null, Arrays.asList(
+ HoodieSchemaField.of("a", reused),
+ HoodieSchemaField.of("u", unionWithReused, null, HoodieSchema.NULL_VALUE)));
+ HoodieSchemaException failure = assertThrows(HoodieSchemaException.class,
+ () -> VariantSchemaUtils.applyForcedShredding(both, forcedDdl()));
+ assertTrue(failure.getMessage().contains("'" + reused.getFullName() + "'")
+ && failure.getMessage().contains("multi-branch union"), failure.getMessage());
+ }
+
@Test
public void testStripVariantShreddingByShape() {
// Footer-derived schemas lose the variant logical type: a shredded variant comes back as a
- // plain record {metadata: bytes, value: nullable bytes, typed_value}.
- HoodieSchema footerVariant = HoodieSchemaTestUtils.createPlainShreddedVariantRecord("v",
- HoodieSchema.createRecord("tv", null, null,
- Collections.singletonList(HoodieSchemaField.of("a", HoodieSchema.create(HoodieSchemaType.LONG)))));
+ // plain record {metadata: bytes, value: nullable bytes, typed_value}. Both write supports
+ // shred below the top level (record members at any depth, and a declared write schema can
+ // shred a bare array element or map value), so the fallback walks the whole tree: the fields
+ // below cover a shredded shape as a top-level column, a struct member, an array element and a
+ // map value.
+ HoodieSchema typedValue = HoodieSchema.createRecord("tv", null, null,
+ Collections.singletonList(HoodieSchemaField.of("a", HoodieSchema.create(HoodieSchemaType.LONG))));
HoodieSchema schema = HoodieSchema.createRecord("rec", null, null, Arrays.asList(
HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.STRING)),
- HoodieSchemaField.of("v", HoodieSchema.createNullable(footerVariant))));
+ HoodieSchemaField.of("v", HoodieSchema.createNullable(
+ HoodieSchemaTestUtils.createPlainShreddedVariantRecord("v", typedValue))),
+ HoodieSchemaField.of("s", HoodieSchemaTestUtils.createRecord("s_rec",
+ HoodieSchemaField.of("inner", HoodieSchemaTestUtils.createPlainShreddedVariantRecord("inner_v", typedValue)))),
+ HoodieSchemaTestUtils.createArrayField("items",
+ HoodieSchemaTestUtils.createPlainShreddedVariantRecord("item_v", typedValue)),
+ HoodieSchemaTestUtils.createMapField("m",
+ HoodieSchemaTestUtils.createPlainShreddedVariantRecord("map_v", typedValue))));
HoodieSchema stripped = VariantSchemaUtils.stripVariantShreddingByShape(schema);
- HoodieSchema v = stripped.getField("v").get().schema().getNonNullType();
- assertEquals(2, v.getFields().size());
- assertFalse(v.getField("typed_value").isPresent());
- assertTrue(v.getField("metadata").isPresent());
- assertTrue(v.getField("value").isPresent());
+ assertEquals(Arrays.asList("metadata", "value"),
+ fieldNames(stripped.getField("v").get().schema().getNonNullType()));
+ assertEquals(Arrays.asList("metadata", "value"),
+ fieldNames(stripped.getField("s").get().schema().getField("inner").get().schema()));
+ assertEquals(Arrays.asList("metadata", "value"), fieldNames(stripped.getField("items").get().schema().getElementType()));
+ assertEquals(Arrays.asList("metadata", "value"), fieldNames(stripped.getField("m").get().schema().getValueType()));
// Non-variant-shaped records pass through; identity when nothing matches.
assertEquals(HoodieSchemaType.STRING, stripped.getField("id").get().schema().getType());
assertSame(stripped, VariantSchemaUtils.stripVariantShreddingByShape(stripped));
@@ -218,14 +397,19 @@ public class TestVariantSchemaUtils {
// Documented limitation: the footer fallback has no table schema to anchor on, so unlike the
// read path's isShreddedVariantTarget / alignShreddedVariants (which keep a same-shaped user
// struct intact, see TestHoodieSchemaCompatibility) this strip cannot tell a shredded variant
- // from a plain user struct of the same shape and strips both. Pinned so that adding an anchor
- // later is a deliberate change.
+ // from a plain user struct of the same shape and strips both, at every depth it walks. Pinned
+ // so that adding an anchor later is a deliberate change.
HoodieSchema userStruct = HoodieSchemaTestUtils.createPlainShreddedVariantRecord("user_struct",
HoodieSchema.create(HoodieSchemaType.LONG));
HoodieSchema userStructStripped = VariantSchemaUtils.stripVariantShreddingByShape(HoodieSchema.createRecord(
"rec", null, null, Collections.singletonList(HoodieSchemaField.of("s", userStruct)))).getField("s").get().schema();
- assertEquals(Arrays.asList("metadata", "value"),
- userStructStripped.getFields().stream().map(HoodieSchemaField::name).collect(Collectors.toList()));
+ assertEquals(Arrays.asList("metadata", "value"), fieldNames(userStructStripped));
+ // Same false positive one level down, since the walk recurses.
+ HoodieSchema nestedUserStructStripped = VariantSchemaUtils.stripVariantShreddingByShape(HoodieSchema.createRecord(
+ "rec", null, null, Collections.singletonList(HoodieSchemaField.of("outer",
+ HoodieSchemaTestUtils.createRecord("outer_rec", HoodieSchemaField.of("s", userStruct))))))
+ .getField("outer").get().schema().getField("s").get().schema();
+ assertEquals(Arrays.asList("metadata", "value"), fieldNames(nestedUserStructStripped));
}
@Test
@@ -239,8 +423,7 @@ public class TestVariantSchemaUtils {
Collections.singletonList(HoodieSchemaField.of("v", twoFieldVariant)));
HoodieSchema v = VariantSchemaUtils.stripVariantShreddingByShape(schema).getField("v").get().schema();
- assertEquals(Arrays.asList("metadata", "value"),
- v.getFields().stream().map(HoodieSchemaField::name).collect(Collectors.toList()));
+ assertEquals(Arrays.asList("metadata", "value"), fieldNames(v));
assertTrue(v.getField("value").get().schema().isNullable());
assertEquals(HoodieSchemaType.BYTES, v.getField("value").get().schema().getNonNullType().getType());
// The synthesized field must carry a null default like every other footer-derived optional field.
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
index efbd0720c345..fded6184026a 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
@@ -2896,7 +2896,7 @@ public class TestHoodieSchema {
@Test
public void testCreateShreddedFieldStruct() {
- HoodieSchema fieldStruct = HoodieSchema.createShreddedFieldStruct("age", HoodieSchema.create(HoodieSchemaType.INT));
+ HoodieSchema fieldStruct = HoodieSchema.createShreddedFieldStruct("age", null, HoodieSchema.create(HoodieSchemaType.INT));
assertNotNull(fieldStruct);
assertEquals(HoodieSchemaType.RECORD, fieldStruct.getType());
@@ -2919,7 +2919,7 @@ public class TestHoodieSchema {
@Test
public void testCreateShreddedFieldStructWithDecimal() {
HoodieSchema decimalSchema = HoodieSchema.createDecimal(15, 1);
- HoodieSchema fieldStruct = HoodieSchema.createShreddedFieldStruct("price", decimalSchema);
+ HoodieSchema fieldStruct = HoodieSchema.createShreddedFieldStruct("price", null, decimalSchema);
assertNotNull(fieldStruct);
List<HoodieSchemaField> fields = fieldStruct.getFields();
@@ -2939,7 +2939,7 @@ public class TestHoodieSchema {
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("c", HoodieSchema.createDecimal(15, 1));
- HoodieSchema.Variant variant = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variant = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
assertNotNull(variant);
assertInstanceOf(HoodieSchema.Variant.class, variant);
@@ -3006,6 +3006,11 @@ public class TestHoodieSchema {
assertEquals("org.apache.hudi", variant.getAvroSchema().getNamespace());
assertTrue(variant.isShredded());
assertTrue(TestHoodieSchema.isVariantSchema(variant.getAvroSchema()));
+ // The struct generated per shredded field sits one level below the variant's namespace, under
+ // the typed_value record that holds it, so a field name cannot collide with the two records
+ // generated in that namespace; see createShreddedFieldStruct.
+ assertEquals("org.apache.hudi.typed_value.age",
+ variant.getTypedValueField().get().getField("age").get().schema().getNonNullType().getFullName());
}
@Test
@@ -3014,7 +3019,7 @@ public class TestHoodieSchema {
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.INT));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.STRING));
- HoodieSchema.Variant original = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant original = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
String jsonSchema = original.toString();
// Parse back from JSON
@@ -3049,7 +3054,7 @@ public class TestHoodieSchema {
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("c", HoodieSchema.createDecimal(15, 1));
- HoodieSchema.Variant variant = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variant = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
// getPlainTypedValueSchema should unwrap the nested {value, typed_value} structs
Option<HoodieSchema> plainOpt = variant.getPlainTypedValueSchema();
@@ -3100,10 +3105,10 @@ public class TestHoodieSchema {
// Both record levels are named "typed_value", as the schema converters produce them.
HoodieSchema innerObject = HoodieSchema.createRecord("typed_value", "inner.ns", null,
Collections.singletonList(HoodieSchemaField.of("b",
- HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("b_wrapper", HoodieSchema.create(HoodieSchemaType.LONG))))));
+ HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("b_wrapper", null, HoodieSchema.create(HoodieSchemaType.LONG))))));
HoodieSchema topTypedValue = HoodieSchema.createRecord("typed_value", "outer.ns", null,
Collections.singletonList(HoodieSchemaField.of("a",
- HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("a_wrapper", innerObject)))));
+ HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("a_wrapper", null, innerObject)))));
// Nullable typed_value, as produced by the inferred-shredding splice.
HoodieSchema.Variant variant = HoodieSchema.createVariantShredded(HoodieSchema.createNullable(topTypedValue));
@@ -3135,16 +3140,16 @@ public class TestHoodieSchema {
// gave both leaves "typed_value_x_y_z_plain").
HoodieSchema leafObject = HoodieSchema.createRecord("typed_value", "leaf.ns", null,
Collections.singletonList(HoodieSchemaField.of("c",
- HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("c_wrapper", HoodieSchema.create(HoodieSchemaType.LONG))))));
+ HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("c_wrapper", null, HoodieSchema.create(HoodieSchemaType.LONG))))));
HoodieSchema underXy = HoodieSchema.createRecord("typed_value", "a.ns", null,
Collections.singletonList(HoodieSchemaField.of("z",
- HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("z_wrapper", leafObject)))));
+ HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("z_wrapper", null, leafObject)))));
HoodieSchema underX = HoodieSchema.createRecord("typed_value", "b.ns", null,
Collections.singletonList(HoodieSchemaField.of("y_z",
- HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("y_z_wrapper", leafObject)))));
+ HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("y_z_wrapper", null, leafObject)))));
HoodieSchema topTypedValue = HoodieSchema.createRecord("typed_value", "outer.ns", null, Arrays.asList(
- HoodieSchemaField.of("x_y", HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("x_y_wrapper", underXy))),
- HoodieSchemaField.of("x", HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("x_wrapper", underX)))));
+ HoodieSchemaField.of("x_y", HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("x_y_wrapper", null, underXy))),
+ HoodieSchemaField.of("x", HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("x_wrapper", null, underX)))));
HoodieSchema plain = HoodieSchema.createVariantShredded(topTypedValue).getPlainTypedValueSchema().get();
HoodieSchema zLeaf = nestedRecord(plain, "x_y", "z");
@@ -3172,7 +3177,7 @@ public class TestHoodieSchema {
Collections.singletonList(
HoodieSchemaField.of("value", HoodieSchema.createNullable(HoodieSchemaType.BYTES))));
HoodieSchema topTypedValue = HoodieSchema.createRecord("typed_value", null, null, Arrays.asList(
- HoodieSchemaField.of("a", HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("a_wrapper", HoodieSchema.create(HoodieSchemaType.INT)))),
+ HoodieSchemaField.of("a", HoodieSchema.createNullable(HoodieSchema.createShreddedFieldStruct("a_wrapper", null, HoodieSchema.create(HoodieSchemaType.INT)))),
HoodieSchemaField.of("u", HoodieSchema.createNullable(valueOnlyWrapper))));
HoodieSchema.Variant variant = HoodieSchema.createVariantShredded(topTypedValue);
@@ -3190,7 +3195,7 @@ public class TestHoodieSchema {
public void testGetPlainTypedValueSchemaArrayTypedValue() {
// Spec form for arrays: typed_value = array<wrapper{value, typed_value: string}>
HoodieSchema arrayTypedValue = HoodieSchema.createArray(
- HoodieSchema.createShreddedFieldStruct("element_wrapper", HoodieSchema.create(HoodieSchemaType.STRING)));
+ HoodieSchema.createShreddedFieldStruct("element_wrapper", null, HoodieSchema.create(HoodieSchemaType.STRING)));
HoodieSchema.Variant variant = HoodieSchema.createVariantShredded(arrayTypedValue);
HoodieSchema plain = variant.getPlainTypedValueSchema().get();
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/avro/HoodieAvroWriteSupport.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/avro/HoodieAvroWriteSupport.java
index e95386f94cb6..e8cc1d941bfd 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/avro/HoodieAvroWriteSupport.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/avro/HoodieAvroWriteSupport.java
@@ -29,7 +29,6 @@ import org.apache.hudi.common.config.HoodieStorageConfig;
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.schema.HoodieSchemaUtils;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
@@ -44,7 +43,7 @@ import org.apache.parquet.avro.AvroWriteSupport;
import org.apache.parquet.hadoop.api.WriteSupport;
import org.apache.parquet.schema.MessageType;
-import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -63,7 +62,9 @@ import static org.apache.hudi.common.config.HoodieStorageConfig.PARQUET_VARIANT_
*
* <p>When variant columns are configured for shredding (via {@link HoodieSchema.Variant#isShredded()}),
* this class transforms variant records at write time to populate {@code typed_value} columns
- * by parsing variant binary data using a {@link VariantShreddingProvider} loaded via reflection.</p>
+ * by parsing variant binary data using a {@link VariantShreddingProvider} loaded via reflection.
+ * A variant is transformed wherever the effective schema declares {@code typed_value} - a record
+ * field at any depth, an array element or a map value - mirroring the row write path.</p>
*/
public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
@@ -72,20 +73,10 @@ public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
protected final Properties properties;
/**
- * Whether variant write shredding is enabled via config.
+ * Plans the value-level transform for the whole record, or is null when the effective schema
+ * declares no shredded variant anywhere and records can be written untouched.
*/
- private final boolean variantWriteShreddingEnabled;
-
- /**
- * The effective (possibly shredded) Avro schema used for writing.
- */
- private final Schema effectiveAvroSchema;
-
- /**
- * Variant fields that need shredding, keyed by their index in the effective schema.
- * Empty if no shredding is needed.
- */
- private final Map<Integer, ShreddedVariantField> shreddedVariantFields;
+ private final Shredder rootShredder;
/**
* Provider for variant shredding (loaded via reflection). Null if no shredding is needed.
@@ -107,45 +98,17 @@ public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
footerMetadata.put(HoodieSchema.VECTOR_COLUMNS_METADATA_KEY, vectorMeta);
}
- this.effectiveAvroSchema = effectiveSchema.toAvroSchema();
- this.variantWriteShreddingEnabled = Boolean.parseBoolean(
+ boolean shreddingEnabled = Boolean.parseBoolean(
properties.getProperty(PARQUET_VARIANT_WRITE_SHREDDING_ENABLED.key(),
String.valueOf(PARQUET_VARIANT_WRITE_SHREDDING_ENABLED.defaultValue())));
- // When shredding is enabled, collect the variant fields that need shredding.
- Map<Integer, ShreddedVariantField> shreddedFields = new LinkedHashMap<>();
-
- if (effectiveSchema.getType() == HoodieSchemaType.RECORD) {
- List<HoodieSchemaField> fields = effectiveSchema.getFields();
- for (int i = 0; i < fields.size(); i++) {
- HoodieSchemaField field = fields.get(i);
- HoodieSchema fieldSchema = field.schema();
- // Unwrap nullable union to get the underlying type
- if (fieldSchema.isNullable()) {
- fieldSchema = fieldSchema.getNonNullType();
- }
- if (fieldSchema.getType() != HoodieSchemaType.VARIANT) {
- continue;
- }
- if (variantWriteShreddingEnabled) {
- HoodieSchema.Variant variant = (HoodieSchema.Variant) fieldSchema;
- if (variant.isShredded() && variant.getTypedValueField().isPresent()) {
- // Get the Avro sub-schema for this variant field from the effective schema
- Schema fieldAvroSchema = effectiveAvroSchema.getFields().get(i).schema();
- // Unwrap nullable union
- if (fieldAvroSchema.getType() == Schema.Type.UNION) {
- fieldAvroSchema = getNonNullFromUnion(fieldAvroSchema);
- }
- shreddedFields.put(i, new ShreddedVariantField(fieldAvroSchema, variant));
- }
- }
- }
- }
-
- this.shreddedVariantFields = shreddedFields;
+ // When shredding is enabled, plan the transform for every position the effective schema
+ // declares shredded; null means there is none and records pass through untouched.
+ this.rootShredder = shreddingEnabled ? buildShredder(effectiveSchema) : null;
- // Load shredding provider via reflection if needed
- if (!shreddedVariantFields.isEmpty()) {
+ // Load shredding provider via reflection if needed. A schema that is shredded only below the
+ // top level needs it just as much, so this is keyed off the whole tree, not the root fields.
+ if (rootShredder != null) {
String providerClass = properties.getProperty(PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key());
if (providerClass == null || providerClass.isEmpty()) {
throw new HoodieException("Variant write shredding is enabled and the write schema requires shredding "
@@ -167,9 +130,12 @@ public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
* unused typed_value columns.</p>
*
* <p>When shredding is enabled and a forced shredding schema is configured via
- * {@link HoodieStorageConfig#PARQUET_VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST},
- * all variant fields are replaced with shredded variants using the forced schema.
- * This handles the case where the input schema is unshredded but shredding is desired.</p>
+ * {@link HoodieStorageConfig#PARQUET_VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST}, every variant
+ * that is a record member - at the top level or at any depth, including records nested under
+ * arrays and maps - is replaced with a shredded variant using the forced schema. This handles
+ * the case where the input schema is unshredded but shredding is desired. A variant that is
+ * directly an array element or a map value is left alone, mirroring the row write path; such a
+ * position only shreds when the write schema itself declares {@code typed_value} there.</p>
*
* <p>When shredding is enabled without a forced schema, the schema is returned as-is
* (already-shredded variants stay shredded, unshredded variants stay unshredded).</p>
@@ -212,55 +178,182 @@ public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
@SuppressWarnings("unchecked")
@Override
public void write(T record) {
- if (!shreddedVariantFields.isEmpty() && shreddingProvider != null) {
- super.write((T) shredRecord((IndexedRecord) record));
- } else {
- super.write(record);
- }
+ super.write(rootShredder == null ? record : (T) rootShredder.shred(record, shreddingProvider));
}
/**
- * Builds a shredded copy of {@code inputRecord}: variant fields configured for shredding are
- * transformed via the {@link VariantShreddingProvider} to populate {@code typed_value}; all other
- * fields are copied across as-is.
+ * Plans the shredding transform for one position of the effective schema, or returns {@code null}
+ * when nothing below it is a shredded variant and the value can be copied by reference. The
+ * mirror image of {@code HoodieVariantReconstruction.buildRebuilder} on the read side.
+ *
+ * <p>Walking values (and not just splicing the schema) is mandatory: {@link AvroWriteSupport}
+ * serializes through {@link ConvertingGenericData}, which inherits the positional
+ * {@code GenericData.getField}, so a {@code typed_value} spliced into a nested variant while the
+ * record below stays the untransformed {@code {metadata, value}} would have parquet read index 2
+ * off a 2-field record and fail the write.</p>
+ *
+ * <p>A genuine multi-branch union is out of scope, deliberately: {@code getNonNullType()} only
+ * unwraps the nullable two-branch form, so anything else comes back a UNION and lands in
+ * {@code default}, leaving any variant inside it unshredded. The schema side agrees -
+ * {@code VariantSchemaUtils.applyForcedShreddingAt} has the same {@code default} arm, and so does
+ * the read side in {@code HoodieVariantReconstruction.buildRebuilder} - so the three cannot drift
+ * apart on their own; changing any one has to be a deliberate change to all of them. The one
+ * shape the schema-side pass-through cannot represent (a record type with a variant member both
+ * under a union and at a forced position) is rejected at splice time, before this walk runs.</p>
*/
- private GenericRecord shredRecord(IndexedRecord inputRecord) {
- GenericRecord shreddedRecord = new GenericData.Record(effectiveAvroSchema);
-
- // Copy all fields, transforming variant fields that need shredding
- List<Schema.Field> effectiveFields = effectiveAvroSchema.getFields();
- Schema inputSchema = inputRecord.getSchema();
-
- for (int i = 0; i < effectiveFields.size(); i++) {
- Schema.Field effectiveField = effectiveFields.get(i);
- String fieldName = effectiveField.name();
- Schema.Field inputField = inputSchema.getField(fieldName);
- if (inputField == null) {
- continue;
+ private static Shredder buildShredder(HoodieSchema effective) {
+ HoodieSchema unwrapped = effective.getNonNullType();
+ switch (unwrapped.getType()) {
+ case VARIANT: {
+ HoodieSchema.Variant variant = (HoodieSchema.Variant) unwrapped;
+ return variant.isShredded() && variant.getTypedValueField().isPresent()
+ ? new VariantShredder(unwrapped.getAvroSchema(), variant) : null;
+ }
+ case RECORD: {
+ List<HoodieSchemaField> fields = unwrapped.getFields();
+ String[] fieldNames = new String[fields.size()];
+ Shredder[] fieldShredders = new Shredder[fields.size()];
+ boolean anyShredded = false;
+ for (int i = 0; i < fields.size(); i++) {
+ fieldNames[i] = fields.get(i).name();
+ fieldShredders[i] = buildShredder(fields.get(i).schema());
+ anyShredded |= fieldShredders[i] != null;
+ }
+ return anyShredded
+ ? new RecordShredder(unwrapped.getAvroSchema(), fieldNames, fieldShredders) : null;
}
+ case ARRAY: {
+ Shredder elementShredder = buildShredder(unwrapped.getElementType());
+ return elementShredder == null
+ ? null : new ArrayShredder(unwrapped.getAvroSchema(), elementShredder);
+ }
+ case MAP: {
+ Shredder valueShredder = buildShredder(unwrapped.getValueType());
+ return valueShredder == null ? null : new MapShredder(valueShredder);
+ }
+ default:
+ return null;
+ }
+ }
+
+ /** Transforms one value into the shape the effective schema declares for its position. */
+ private interface Shredder {
+ Object shred(Object value, VariantShreddingProvider provider);
+ }
+
+ private static final class VariantShredder implements Shredder {
+ private final Schema shreddedAvroSchema;
+ private final HoodieSchema.Variant variant;
+
+ private VariantShredder(Schema shreddedAvroSchema, HoodieSchema.Variant variant) {
+ this.shreddedAvroSchema = shreddedAvroSchema;
+ this.variant = variant;
+ }
- ShreddedVariantField shreddedField = shreddedVariantFields.get(i);
- if (shreddedField != null) {
- // This is a shredded variant field - transform it
- Object fieldValue = inputRecord.get(inputField.pos());
- if (fieldValue instanceof GenericRecord) {
- GenericRecord variantRecord = (GenericRecord) fieldValue;
- GenericRecord shreddedVariant = shreddingProvider.shredVariantRecord(
- variantRecord,
- shreddedField.avroSchema,
- shreddedField.hoodieSchema);
- shreddedRecord.put(i, shreddedVariant);
- } else {
- // Null or unexpected type - copy as-is
- shreddedRecord.put(i, fieldValue);
+ @Override
+ public Object shred(Object value, VariantShreddingProvider provider) {
+ // A null variant, or an unexpected type, passes through unchanged.
+ return value instanceof GenericRecord
+ ? provider.shredVariantRecord((GenericRecord) value, shreddedAvroSchema, variant)
+ : value;
+ }
+ }
+
+ private static final class RecordShredder implements Shredder {
+ private final Schema effectiveSchema;
+ private final String[] fieldNames;
+ // Indexed by field position in the effective record; null for fields copied by reference.
+ private final Shredder[] fieldShredders;
+ // Input records are not rewritten before they reach the writer (HoodieAvroIndexedRecord passes
+ // them through) and nested records carry their own schema, so fields are matched by NAME.
+ // One writer sees one input schema instance per level, so a single-entry cache resolves the
+ // name lookups once; the write support is single-threaded.
+ private Schema cachedInputSchema;
+ private int[] cachedInputPositions;
+
+ private RecordShredder(Schema effectiveSchema, String[] fieldNames, Shredder[] fieldShredders) {
+ this.effectiveSchema = effectiveSchema;
+ this.fieldNames = fieldNames;
+ this.fieldShredders = fieldShredders;
+ }
+
+ @Override
+ public Object shred(Object value, VariantShreddingProvider provider) {
+ if (!(value instanceof IndexedRecord)) {
+ return value;
+ }
+ IndexedRecord in = (IndexedRecord) value;
+ int[] inputPositions = inputPositionsFor(in.getSchema());
+ GenericRecord out = new GenericData.Record(effectiveSchema);
+ for (int i = 0; i < inputPositions.length; i++) {
+ if (inputPositions[i] < 0) {
+ // The input does not carry this field; leave it null rather than guessing a position.
+ continue;
}
- } else {
- // Non-variant field - copy as-is
- shreddedRecord.put(i, inputRecord.get(inputField.pos()));
+ Object fieldValue = in.get(inputPositions[i]);
+ out.put(i, fieldShredders[i] == null
+ ? fieldValue : fieldShredders[i].shred(fieldValue, provider));
}
+ return out;
}
- return shreddedRecord;
+ private int[] inputPositionsFor(Schema inputSchema) {
+ if (inputSchema == cachedInputSchema) {
+ return cachedInputPositions;
+ }
+ int[] inputPositions = new int[fieldNames.length];
+ for (int i = 0; i < fieldNames.length; i++) {
+ Schema.Field inputField = inputSchema.getField(fieldNames[i]);
+ inputPositions[i] = inputField == null ? -1 : inputField.pos();
+ }
+ cachedInputSchema = inputSchema;
+ cachedInputPositions = inputPositions;
+ return inputPositions;
+ }
+ }
+
+ private static final class ArrayShredder implements Shredder {
+ private final Schema effectiveSchema;
+ private final Shredder elementShredder;
+
+ private ArrayShredder(Schema effectiveSchema, Shredder elementShredder) {
+ this.effectiveSchema = effectiveSchema;
+ this.elementShredder = elementShredder;
+ }
+
+ @Override
+ public Object shred(Object value, VariantShreddingProvider provider) {
+ if (!(value instanceof Collection)) {
+ return value;
+ }
+ Collection<?> in = (Collection<?>) value;
+ GenericData.Array<Object> out = new GenericData.Array<>(in.size(), effectiveSchema);
+ for (Object element : in) {
+ out.add(elementShredder.shred(element, provider));
+ }
+ return out;
+ }
+ }
+
+ private static final class MapShredder implements Shredder {
+ private final Shredder valueShredder;
+
+ private MapShredder(Shredder valueShredder) {
+ this.valueShredder = valueShredder;
+ }
+
+ @Override
+ public Object shred(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(), valueShredder.shred(entry.getValue(), provider));
+ }
+ return out;
+ }
}
@Override
@@ -283,69 +376,16 @@ public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
footerMetadata.put(key, value);
}
- /**
- * Bundles the Avro sub-schema and {@link HoodieSchema.Variant} for a shredded variant field,
- * keyed by effective-schema field index in {@link #shreddedVariantFields}.
- */
- private static final class ShreddedVariantField {
- private final Schema avroSchema;
- private final HoodieSchema.Variant hoodieSchema;
-
- ShreddedVariantField(Schema avroSchema, HoodieSchema.Variant hoodieSchema) {
- this.avroSchema = avroSchema;
- this.hoodieSchema = hoodieSchema;
- }
- }
-
private static final Pattern DECIMAL_PATTERN = Pattern.compile(
"decimal\\s*\\(\\s*(\\d+)\\s*,\\s*(\\d+)\\s*\\)");
/**
- * Applies a forced shredding schema to all variant fields in the given schema.
- * The forced schema DDL (e.g., {@code "a int, b string"}) defines the typed_value
+ * Applies a forced shredding schema to the variant record members of the given schema, at any
+ * depth. The forced schema DDL (e.g., {@code "a int, b string"}) defines the typed_value
* fields that will be added to each variant column.
*/
private static HoodieSchema applyForcedShreddingSchema(HoodieSchema schema, String ddl) {
- if (schema.getType() != HoodieSchemaType.RECORD) {
- return schema;
- }
-
- Map<String, HoodieSchema> shreddedFields = parseShreddingDDL(ddl);
-
- List<HoodieSchemaField> fields = schema.getFields();
- List<HoodieSchemaField> newFields = new ArrayList<>();
- boolean changed = false;
-
- for (HoodieSchemaField field : fields) {
- HoodieSchema fieldSchema = field.schema();
- boolean wasNullable = fieldSchema.isNullable();
- HoodieSchema unwrapped = wasNullable ? fieldSchema.getNonNullType() : fieldSchema;
-
- if (unwrapped.getType() == HoodieSchemaType.VARIANT) {
- HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(
- unwrapped.getAvroSchema().getName(),
- unwrapped.getAvroSchema().getNamespace(),
- unwrapped.getAvroSchema().getDoc(),
- shreddedFields);
- HoodieSchema replacement = wasNullable
- ? HoodieSchema.createNullable(shreddedVariant) : shreddedVariant;
- // replacement already mirrors the field's original nullability, so withSchema alone suffices.
- newFields.add(HoodieSchemaUtils.createNewSchemaField(field.withSchema(replacement)));
- changed = true;
- } else {
- newFields.add(HoodieSchemaUtils.createNewSchemaField(field));
- }
- }
-
- if (!changed) {
- return schema;
- }
-
- return HoodieSchema.createRecord(
- schema.getAvroSchema().getName(),
- schema.getAvroSchema().getNamespace(),
- schema.getAvroSchema().getDoc(),
- newFields);
+ return VariantSchemaUtils.applyForcedShredding(schema, parseShreddingDDL(ddl));
}
/**
@@ -399,18 +439,6 @@ public class HoodieAvroWriteSupport<T> extends AvroWriteSupport<T> {
}
}
- /**
- * Extracts the non-null type from a union schema.
- */
- private static Schema getNonNullFromUnion(Schema unionSchema) {
- for (Schema type : unionSchema.getTypes()) {
- if (type.getType() != Schema.Type.NULL) {
- return type;
- }
- }
- throw new IllegalArgumentException("Union schema does not contain a non-null type: " + unionSchema);
- }
-
private static class HoodieBloomFilterAvroWriteSupport extends HoodieBloomFilterWriteSupport<String> {
public HoodieBloomFilterAvroWriteSupport(BloomFilter bloomFilter) {
super(bloomFilter);
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 09fa1648580c..75b64a8afa56 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
@@ -130,10 +130,11 @@ final class HoodieVariantReconstruction {
* 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, matching what the row writer can emit; see
+ * elements and map values are all descended into, matching what the writers can emit; see
* {@code VariantSchemaUtils.swapShreddedVariantFields} for what actually produces a nested
- * shredded file today (the row path shreds at depth off the forced-shredding property; the AVRO
- * path needs a hand-authored write schema).
+ * shredded file today (both the row and the AVRO write path shred variant record members at any
+ * depth off the forced-shredding property; a variant that is directly an array element or a map
+ * value shreds on neither path unless the write schema itself declares typed_value there).
*/
private static Rebuilder buildRebuilder(HoodieSchema outputSchema, HoodieSchema fileSchema) {
if (VariantSchemaUtils.isShreddedVariantTarget(fileSchema, outputSchema)) {
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
index 653e9cbf9a7b..e4ae84422fbc 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
@@ -19,33 +19,75 @@
package org.apache.hudi.avro;
+import org.apache.hudi.common.avro.VariantShreddingProvider;
+import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
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.table.HoodieTableConfig;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.core.io.storage.HoodieAvroFileWriter;
+import org.apache.hudi.core.io.storage.HoodieFileWriter;
+import org.apache.hudi.core.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.avro.AvroParquetReader;
import org.apache.parquet.avro.AvroSchemaConverterWithTimestampNTZ;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.avro.AvroSchemaUtils.getNonNullTypeFromUnion;
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.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
class TestHoodieAvroWriteSupportShredding {
+ /** The variant value bytes {@link StubShreddingProvider} shreds; anything else stays a residual. */
+ private static final byte[] TYPED_MARKER = "typed".getBytes(StandardCharsets.UTF_8);
+ /** The variant value bytes {@link StubShreddingProvider} declines outright, returning null. */
+ private static final byte[] DECLINE_MARKER = "decline".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] RESIDUAL_VALUE = "residual".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] VARIANT_METADATA = new byte[] {1, 0, 0};
+ private static final String SHREDDED_LEAF = "shredded";
+
+ @TempDir
+ java.nio.file.Path tmpDir;
+
/**
* The forced-shredding DDL must tolerate the commas inside parameterized types such as
* {@code decimal(15, 1)}, which is the documented example on
@@ -57,19 +99,245 @@ class TestHoodieAvroWriteSupportShredding {
HoodieSchema effective = HoodieAvroWriteSupport.generateEffectiveSchema(
singleVariantRecord(), forcedShreddingProps("a int, b string, c decimal(15, 1)"));
- HoodieSchema variantField = effective.getFields().get(0).schema();
- HoodieSchema variant = variantField.isNullable() ? variantField.getNonNullType() : variantField;
- HoodieSchema typedValueField = variant.getFields().stream()
- .filter(f -> HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD.equals(f.name()))
- .map(HoodieSchemaField::schema)
- .findFirst()
- .orElseThrow(() -> new AssertionError("shredded variant is missing typed_value: " + variant));
- HoodieSchema typedValue = typedValueField.isNullable() ? typedValueField.getNonNullType() : typedValueField;
+ assertShredded(effective.getFields().get(0).schema(), "forced top-level variant",
+ Arrays.asList("a", "b", "c"));
+ }
- List<String> shreddedFieldNames = typedValue.getFields().stream()
- .map(HoodieSchemaField::name)
- .collect(Collectors.toList());
- assertEquals(Arrays.asList("a", "b", "c"), shreddedFieldNames);
+ /**
+ * The forced-shredding DDL has to reach every variant that is a record member, at any depth and
+ * including records carried by an array or a map, so that the AVRO write path lays out the same
+ * file the row writer does for the same table and config. A variant that is directly an array
+ * element or a map value is left unshredded on purpose on both paths: the DDL says what
+ * typed_value holds, not where, and forcing a collection leaf is out of scope.
+ */
+ @Test
+ void forcedShreddingReachesVariantRecordMembersAtEveryDepth() {
+ HoodieSchema effective = HoodieAvroWriteSupport.generateEffectiveSchema(
+ nestedVariantRecord(), forcedShreddingProps("a int, b string"));
+
+ // Which of the six positions the DDL reaches is the position matrix in
+ // TestVariantSchemaUtils#testApplyForcedShreddingReachesRecordMembersAtAnyDepth, and restating it
+ // here would only pin the same splice twice. What this test owns is the leg past it: the DDL
+ // string reaches the splice through generateEffectiveSchema, and the spliced typed_value
+ // survives the list and map wrappers the generic parquet-avro converter generates.
+ MessageType parquet = new AvroSchemaConverterWithTimestampNTZ().convert(effective);
+ assertTrue(parquet.getType("s").asGroupType().getType("inner").asGroupType().containsField("typed_value"),
+ "expected a shredded group at s.inner: " + parquet);
+ assertTrue(listElementOf(parquet.getType("items").asGroupType()).getType("v").asGroupType()
+ .containsField("typed_value"), "expected a shredded group at the items element: " + parquet);
+ assertTrue(mapValueOf(parquet.getType("m").asGroupType()).getType("v").asGroupType()
+ .containsField("typed_value"), "expected a shredded group at the m value: " + parquet);
+ assertFalse(listElementOf(parquet.getType("arr").asGroupType()).containsField("typed_value"),
+ "a bare array element must stay unshredded: " + parquet);
+ assertFalse(mapValueOf(parquet.getType("mv").asGroupType()).containsField("typed_value"),
+ "a bare map value must stay unshredded: " + parquet);
+ }
+
+ /**
+ * The value-level half of the same contract. Splicing typed_value into the schema only pays off if
+ * the writer also rewrites the records it is handed, at every position the effective schema
+ * declares shredded: {@link org.apache.parquet.avro.AvroWriteSupport} reads fields positionally, so
+ * a variant left in its {@code {metadata, value}} shape under a schema that declares typed_value
+ * fails the write outright. This drives a real parquet write with a stub provider - no Spark on
+ * this module's classpath - and reads the file back raw so typed_value is visible exactly as it
+ * was stored.
+ *
+ * <p>The positions come from the schema, which the test above pins; what the rows add are the
+ * cases the value walk itself can get wrong: the residual fallback, a provider that declines a
+ * variant and hands back null, a null nullable struct, a null array column and a null map column,
+ * a null map value, a null variant carried by an array element, a nested record whose fields
+ * arrive in a different order, and one that does not carry a field at all.</p>
+ */
+ @Test
+ void writesNestedVariantsShreddedAtEveryPositionTheEffectiveSchemaDeclares() throws Exception {
+ HoodieSchema table = roundTripRecord();
+ Properties props = forcedShreddingProps("k string");
+ // The write support loads the provider reflectively and this module ships none, so point it at
+ // the stub below instead of the Spark implementation.
+ props.setProperty(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
+ StubShreddingProvider.class.getName());
+
+ // The writer factory converts its own generateEffectiveSchema result into the file's MessageType
+ // while the write support constructor recomputes the effective schema for the records it builds.
+ // A cheap guard that the two calls agree, down to the generated record names - which the Avro
+ // schema string carries and a MessageType would drop.
+ assertEquals(
+ HoodieAvroWriteSupport.generateEffectiveSchema(table, props).toAvroSchema().toString(),
+ HoodieAvroWriteSupport.generateEffectiveSchema(table, props).toAvroSchema().toString(),
+ "generateEffectiveSchema must be deterministic for the same schema and properties");
+
+ Schema tableAvro = table.toAvroSchema();
+ Schema structAvro = getNonNullTypeFromUnion(tableAvro.getField("s").schema());
+ Schema itemAvro = getNonNullTypeFromUnion(tableAvro.getField("items").schema()).getElementType();
+ Schema mapValueAvro = getNonNullTypeFromUnion(
+ getNonNullTypeFromUnion(tableAvro.getField("m").schema()).getValueType());
+ Schema variantAvro = tableAvro.getField("v").schema();
+ // Input records reach the writer untransformed and carry their own schema, so the same field
+ // names in another order - and a record missing one entirely - have to resolve by name.
+ Schema reorderedAvro = HoodieSchema.createRecord("s_reordered", "org.apache.hudi.test", null,
+ Arrays.asList(
+ HoodieSchemaField.of("n", HoodieSchema.createNullable(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("inner", HoodieSchema.createVariant()))).toAvroSchema();
+ Schema withoutNAvro = HoodieSchema.createRecord("s_without_n", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("inner", HoodieSchema.createVariant()))).toAvroSchema();
+
+ List<GenericRecord> rows = Arrays.asList(
+ // Every declared position shredded, plus the bare array element that is not declared.
+ recordOf(tableAvro,
+ "id", 0,
+ "s", recordOf(structAvro, "inner", variant(variantAvro, TYPED_MARKER), "n", 7),
+ "items", Arrays.asList(
+ recordOf(itemAvro, "v", variant(variantAvro, TYPED_MARKER), "label", "first"),
+ recordOf(itemAvro, "v", variant(variantAvro, TYPED_MARKER), "label", "second")),
+ "m", Collections.singletonMap("a", recordOf(mapValueAvro, "v", variant(variantAvro, TYPED_MARKER))),
+ "arr", Collections.singletonList(variant(variantAvro, TYPED_MARKER)),
+ "v", variant(variantAvro, TYPED_MARKER)),
+ // The residual fallback at depth, a null variant under an array element, and a variant the
+ // provider declines at a nullable position.
+ recordOf(tableAvro,
+ "id", 1,
+ "s", recordOf(structAvro, "inner", variant(variantAvro, RESIDUAL_VALUE), "n", 1),
+ "items", Arrays.asList(
+ recordOf(itemAvro, "v", null, "label", "no variant"),
+ recordOf(itemAvro, "v", variant(variantAvro, DECLINE_MARKER), "label", "declined")),
+ "m", Collections.emptyMap(),
+ "arr", Collections.emptyList(),
+ "v", variant(variantAvro, TYPED_MARKER)),
+ // A null nullable struct, and a null array and a null map beside it: there is no record,
+ // element or map value below any of them to walk, so each has to pass through untouched.
+ recordOf(tableAvro,
+ "id", 2,
+ "s", null,
+ "items", null,
+ "m", null,
+ "arr", Collections.emptyList(),
+ "v", variant(variantAvro, TYPED_MARKER)),
+ // The same struct with its fields declared in the other order.
+ recordOf(tableAvro,
+ "id", 3,
+ "s", recordOf(reorderedAvro, "n", 3, "inner", variant(variantAvro, TYPED_MARKER)),
+ "items", Collections.emptyList(),
+ "m", Collections.emptyMap(),
+ "arr", Collections.emptyList(),
+ "v", variant(variantAvro, TYPED_MARKER)),
+ // An input struct that does not carry n at all.
+ recordOf(tableAvro,
+ "id", 4,
+ "s", recordOf(withoutNAvro, "inner", variant(variantAvro, TYPED_MARKER)),
+ "items", Collections.emptyList(),
+ "m", Collections.emptyMap(),
+ "arr", Collections.emptyList(),
+ "v", variant(variantAvro, TYPED_MARKER)),
+ // A null map value: like the null struct, there is no record below it to walk.
+ recordOf(tableAvro,
+ "id", 5,
+ "s", null,
+ "items", Collections.emptyList(),
+ "m", Collections.singletonMap("a", null),
+ "arr", Collections.emptyList(),
+ "v", variant(variantAvro, TYPED_MARKER)));
+
+ List<GenericRecord> readBack = writeAndReadBack(table, props, "shredded.parquet", rows);
+ assertEquals(rows.size(), readBack.size(), "every row must round trip");
+
+ GenericRecord shredded = readBack.get(0);
+ GenericRecord struct = (GenericRecord) shredded.get("s");
+ assertTypedValue((GenericRecord) struct.get("inner"), "s.inner");
+ assertEquals(Integer.valueOf(7), struct.get("n"), "a non-variant sibling must survive the rewrite");
+ List<?> items = (List<?>) shredded.get("items");
+ assertTypedValue((GenericRecord) ((GenericRecord) items.get(0)).get("v"), "items[0].v");
+ assertTypedValue((GenericRecord) ((GenericRecord) items.get(1)).get("v"), "items[1].v");
+ assertEquals("second", String.valueOf(((GenericRecord) items.get(1)).get("label")),
+ "a non-variant sibling inside an array element must survive the rewrite");
+ assertTypedValue((GenericRecord) ((GenericRecord) mapValue((Map<?, ?>) shredded.get("m"), "a")).get("v"),
+ "m[a].v");
+ assertTypedValue((GenericRecord) shredded.get("v"), "top-level v");
+ // The bare array element is the one position the DDL leaves alone: the file declares no
+ // typed_value column there at all, and the value has to arrive untouched.
+ GenericRecord bareElement = (GenericRecord) ((List<?>) shredded.get("arr")).get(0);
+ assertNull(bareElement.getSchema().getField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD),
+ "a bare array element must stay unshredded on disk: " + bareElement.getSchema());
+ assertEquals(ByteBuffer.wrap(TYPED_MARKER), bareElement.get("value"), "bare array element value");
+
+ GenericRecord residual = readBack.get(1);
+ GenericRecord residualInner = (GenericRecord) ((GenericRecord) residual.get("s")).get("inner");
+ assertNull(residualInner.get("typed_value"), "a value the provider does not match must stay a residual");
+ assertEquals(ByteBuffer.wrap(RESIDUAL_VALUE), residualInner.get("value"), "residual value at depth");
+ List<?> residualItems = (List<?>) residual.get("items");
+ assertNull(((GenericRecord) residualItems.get(0)).get("v"),
+ "a null variant under an array element must pass through as null");
+ assertNull(((GenericRecord) residualItems.get(1)).get("v"),
+ "a variant the provider declines must land as null at a nullable position");
+
+ GenericRecord nullCollections = readBack.get(2);
+ assertNull(nullCollections.get("s"), "a null nullable struct must pass through as null");
+ assertNull(nullCollections.get("items"), "a null array column must pass through as null");
+ assertNull(nullCollections.get("m"), "a null map column must pass through as null");
+
+ GenericRecord reordered = (GenericRecord) readBack.get(3).get("s");
+ assertTypedValue((GenericRecord) reordered.get("inner"), "s.inner from a reordered input record");
+ assertEquals(Integer.valueOf(3), reordered.get("n"), "input fields must be matched by name, not position");
+
+ GenericRecord withoutN = (GenericRecord) readBack.get(4).get("s");
+ assertTypedValue((GenericRecord) withoutN.get("inner"), "s.inner from an input record without n");
+ assertNull(withoutN.get("n"), "a field the input does not carry must be left null");
+
+ assertNull(mapValue((Map<?, ?>) readBack.get(5).get("m"), "a"),
+ "a null map value must pass through as null");
+
+ // The same declined variant at a NON-nullable position is not a null the file can hold:
+ // rt_map_value_record.v is a required group, so parquet-avro fails the write outright rather
+ // than storing anything. Its own file, so the rows above stay unaffected.
+ GenericRecord declinedAtRequiredPosition = recordOf(tableAvro,
+ "id", 6,
+ "s", null,
+ "items", Collections.emptyList(),
+ "m", Collections.singletonMap("a",
+ recordOf(mapValueAvro, "v", variant(variantAvro, DECLINE_MARKER))),
+ "arr", Collections.emptyList(),
+ "v", variant(variantAvro, TYPED_MARKER));
+ Exception failure = assertThrows(Exception.class,
+ () -> writeAndReadBack(table, props, "declined.parquet",
+ Collections.singletonList(declinedAtRequiredPosition)),
+ "a declined variant at a required position must fail the write, not write a null");
+ assertTrue(messageChain(failure).contains("Null-value for required field"),
+ "expected parquet-avro's required-field failure, got: " + messageChain(failure));
+ }
+
+ /**
+ * The provider gate is keyed off the whole schema tree, not the root fields. A table whose only
+ * shredded variant sits below the top level needs a provider just as much as one shredded at the
+ * root, and having none configured has to fail at writer construction rather than NPE on the
+ * first record. Master only threw when a ROOT field was shredded, so a nested-only schema is
+ * exactly the failure mode this change introduces.
+ */
+ @Test
+ void nestedOnlyShreddedSchemaWithoutAProviderFailsAtConstruction() {
+ // Hand-built rather than nestedVariantRecord() plus a forced DDL, which would shred the same
+ // positions: a schema that already carries typed_value with no DDL to apply is the only input
+ // that reaches generateEffectiveSchema's "enabled, no forced DDL, take the schema as-is" arm.
+ Map<String, HoodieSchema> ddlFields = new LinkedHashMap<>();
+ ddlFields.put("k", HoodieSchema.create(HoodieSchemaType.STRING));
+ HoodieSchema table = HoodieSchema.createRecord(
+ "nested_only_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("s", HoodieSchema.createRecord(
+ "nested_only_s_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("inner",
+ HoodieSchema.createVariantShreddedObject(
+ "inner_variant", "org.apache.hudi.test", null, ddlFields)))))));
+
+ // Shredding on, no forced DDL (the schema is already shredded) and no provider named: this
+ // module ships none, and the write support does not auto-detect - only the factory does.
+ Properties props = new Properties();
+ props.setProperty(HoodieStorageConfig.PARQUET_VARIANT_WRITE_SHREDDING_ENABLED.key(), "true");
+ MessageType messageType = new AvroSchemaConverterWithTimestampNTZ()
+ .convert(HoodieAvroWriteSupport.generateEffectiveSchema(table, props));
+
+ HoodieException failure = assertThrows(HoodieException.class,
+ () -> new HoodieAvroWriteSupport(messageType, table, Option.empty(), props),
+ "a variant shredded only below the top level still needs a shredding provider");
+ assertTrue(failure.getMessage().contains("no VariantShreddingProvider"),
+ "expected the missing-provider message, got: " + failure.getMessage());
}
/**
@@ -78,12 +346,15 @@ class TestHoodieAvroWriteSupportShredding {
* without tripping Avro's "Field already used". Rebuilding a record while reusing a
* {@code Schema.Field} still bound to the source record throws, so any table with a shredded
* variant AND at least one other column failed here. #18938 fixed exactly that defect in the
- * sibling HoodieVariantReconstruction and left this twin behind. Nested variants must be
- * stripped too, but this AVRO path's own hook is not what puts one there -
- * {@code applyForcedShreddingSchema} walks top-level fields only. The ROW write path shreds at
- * any depth its forced-shredding DDL asks (HoodieRowParquetWriteSupport.processNestedDataType),
- * so a clustering/compaction schema read back from such a file can carry typed_value below the
- * top level, which is why the nested leg is pinned here at unit level.
+ * sibling HoodieVariantReconstruction and left this twin behind. The strip is the exact inverse
+ * of the forced splice and has to reach as far: both writers shred variant record members at any
+ * depth off the forced-shredding DDL, and a hand-authored write schema can put typed_value on a
+ * bare array element or map value too, which no DDL forces. A clustering/compaction schema read
+ * back from such a file carries typed_value below the top level, so every arm is pinned here.
+ * No arm here is evidence for this change. The top-level and nested-record arms are master's test
+ * verbatim, and master already recursed into records; the array-element and map-value arms pin
+ * pre-existing behaviour of {@code VariantSchemaUtils#stripVariantShreddingAt} and pass on master
+ * too. The whole test is regression cover for the strip direction while the forced splice grows.
*/
@Test
void disablingShreddingStripsTypedValueAtEveryDepth() {
@@ -95,7 +366,11 @@ class TestHoodieAvroWriteSupportShredding {
HoodieSchemaField.of("nested", HoodieSchema.createRecord(
"nested_record", "org.apache.hudi.test", null,
Collections.singletonList(HoodieSchemaField.of("nv", HoodieSchema.createVariantShredded(
- "nv", "org.apache.hudi.test", null, HoodieSchema.create(HoodieSchemaType.INT))))))));
+ "nv", "org.apache.hudi.test", null, HoodieSchema.create(HoodieSchemaType.INT)))))),
+ HoodieSchemaField.of("items", HoodieSchema.createArray(HoodieSchema.createVariantShredded(
+ "iv", "org.apache.hudi.test", null, HoodieSchema.create(HoodieSchemaType.INT)))),
+ HoodieSchemaField.of("tags", HoodieSchema.createMap(HoodieSchema.createVariantShredded(
+ "tv", "org.apache.hudi.test", null, HoodieSchema.create(HoodieSchemaType.INT))))));
Properties props = new Properties();
props.setProperty(HoodieStorageConfig.PARQUET_VARIANT_WRITE_SHREDDING_ENABLED.key(), "false");
@@ -107,6 +382,8 @@ class TestHoodieAvroWriteSupportShredding {
assertEquals("id", effective.getFields().get(0).name(), "non-variant fields must survive the rebuild");
assertUnshredded(effective.getField("v").get().schema(), "top-level variant");
assertUnshredded(effective.getField("nested").get().schema().getField("nv").get().schema(), "nested variant");
+ assertUnshredded(effective.getField("items").get().schema().getElementType(), "array element variant");
+ assertUnshredded(effective.getField("tags").get().schema().getValueType(), "map value variant");
}
private static void assertUnshredded(HoodieSchema fieldSchema, String label) {
@@ -115,6 +392,38 @@ class TestHoodieAvroWriteSupportShredding {
assertFalse(((HoodieSchema.Variant) unwrapped).isShredded(), label + " should no longer be shredded");
}
+ private static void assertShredded(HoodieSchema fieldSchema, String label,
+ List<String> expectedTypedValueFields) {
+ HoodieSchema unwrapped = fieldSchema.isNullable() ? fieldSchema.getNonNullType() : fieldSchema;
+ assertInstanceOf(HoodieSchema.Variant.class, unwrapped, label + " should be a variant");
+ HoodieSchema typedValueField = unwrapped.getFields().stream()
+ .filter(f -> HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD.equals(f.name()))
+ .map(HoodieSchemaField::schema)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError(label + " is missing typed_value: " + unwrapped));
+ HoodieSchema typedValue = typedValueField.isNullable() ? typedValueField.getNonNullType() : typedValueField;
+
+ List<String> shreddedFieldNames = typedValue.getFields().stream()
+ .map(HoodieSchemaField::name)
+ .collect(Collectors.toList());
+ assertEquals(expectedTypedValueFields, shreddedFieldNames, label + " typed_value members");
+ }
+
+ /**
+ * The element type of a parquet LIST group. One converter, one layout: every MessageType here
+ * comes from the no-arg {@link AvroSchemaConverterWithTimestampNTZ}, which leaves
+ * {@code parquet.avro.write-old-list-structure} at its default and so always emits the 2-level
+ * form ({@code repeated group array}); a 3-level {@code list/element} arm would be dead code.
+ */
+ private static GroupType listElementOf(GroupType listGroup) {
+ return listGroup.getType(0).asGroupType();
+ }
+
+ /** The value type of a parquet MAP group ({@code key_value/value}). */
+ private static GroupType mapValueOf(GroupType mapGroup) {
+ return mapGroup.getType(0).asGroupType().getType("value").asGroupType();
+ }
+
/**
* A shredded variant written via the Avro path must carry the parquet VARIANT logical type on the
* variant group so external readers recognize it (mirroring the row-writer path and native Spark).
@@ -149,6 +458,157 @@ class TestHoodieAvroWriteSupportShredding {
Collections.singletonList(HoodieSchemaField.of("v", HoodieSchema.createVariant())));
}
+ /**
+ * A record carrying a variant at every position the write path can meet one: a struct member,
+ * a struct member reached through an array and through a map, and the two positions the forced
+ * DDL must leave alone - a bare array element and a bare map value.
+ */
+ private static HoodieSchema nestedVariantRecord() {
+ HoodieSchema struct = HoodieSchema.createRecord("s_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("inner", HoodieSchema.createVariant())));
+ HoodieSchema itemStruct = HoodieSchema.createRecord("item_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("v", HoodieSchema.createVariant())));
+ HoodieSchema mapValueStruct = HoodieSchema.createRecord("map_value_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("v", HoodieSchema.createVariant())));
+ return HoodieSchema.createRecord("test_nested_record", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("s", struct),
+ HoodieSchemaField.of("items", HoodieSchema.createArray(itemStruct)),
+ HoodieSchemaField.of("m", HoodieSchema.createMap(mapValueStruct)),
+ HoodieSchemaField.of("arr", HoodieSchema.createArray(HoodieSchema.createVariant())),
+ HoodieSchemaField.of("mv", HoodieSchema.createMap(HoodieSchema.createVariant()))));
+ }
+
+ /**
+ * The round-trip table schema: a variant under a nullable struct (next to a plain sibling), under
+ * a struct in an array and under a nullable struct in a map, one directly under an array - the
+ * position the DDL leaves unshredded - and one at the top level as the control.
+ */
+ private static HoodieSchema roundTripRecord() {
+ HoodieSchema struct = HoodieSchema.createRecord("rt_s_record", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("inner", HoodieSchema.createVariant()),
+ // Nullable so that an input record without it is legal rather than a broken write.
+ HoodieSchemaField.of("n", HoodieSchema.createNullable(HoodieSchemaType.INT))));
+ // The sibling is not decoration, and not a shredding concern. The Avro write path emits 2-level
+ // parquet lists (parquet-avro's default parquet.avro.write-old-list-structure=true; the row
+ // writer emits 3-level), and parquet-avro before 1.14 decides whether a single-field repeated
+ // group is the element or a synthetic wrapper with a check that is sensitive to the Avro record
+ // NAME (AvroRecordConverter#isElementType), which never matches the name Hudi gives the element
+ // record. So on the parquet-avro this module builds against (1.13.1) Hudi's own
+ // HoodieAvroParquetReader cannot read array<struct<single field>> at all, shredded or not: a
+ // known reader limitation on parquet-avro < 1.14, closed as won't-fix in apache/hudi#19782 with
+ // the workarounds recorded there (write 3-level lists via
+ // parquet.avro.write-old-list-structure=false, or give the element a second field).
+ // parquet-avro 1.14+ recognizes the group by its "array" name instead (PARQUET-2450). Hence the
+ // sibling here.
+ HoodieSchema itemStruct = HoodieSchema.createRecord("rt_item_record", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("v", HoodieSchema.createNullable(HoodieSchema.createVariant())),
+ HoodieSchemaField.of("label", HoodieSchema.createNullable(HoodieSchemaType.STRING))));
+ // The struct is nullable below so a null map value is a legal row; the variant inside it is not,
+ // which makes it the required position a declined variant has to fail the write at.
+ HoodieSchema mapValueStruct = HoodieSchema.createRecord("rt_map_value_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("v", HoodieSchema.createVariant())));
+ // items and m are nullable columns, which is the ordinary shape a Spark table gives a collection
+ // and the only input that reaches the array and map shredders' pass-through arms: a null there
+ // is neither a Collection nor a Map, so it has to be handed on rather than walked.
+ return HoodieSchema.createRecord("test_round_trip_record", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("s", HoodieSchema.createNullable(struct)),
+ HoodieSchemaField.of("items", HoodieSchema.createNullable(HoodieSchema.createArray(itemStruct))),
+ HoodieSchemaField.of("m", HoodieSchema.createNullable(
+ HoodieSchema.createMap(HoodieSchema.createNullable(mapValueStruct)))),
+ HoodieSchemaField.of("arr", HoodieSchema.createArray(HoodieSchema.createVariant())),
+ HoodieSchemaField.of("v", HoodieSchema.createVariant())));
+ }
+
+ /**
+ * Writes the rows through the public writer factory and reads the file back raw. Going through
+ * {@link HoodieFileWriterFactory} rather than a hand-built write support is the point: it is the
+ * production route, and it owns both halves the round trip depends on - it carries the properties
+ * into {@link HoodieAvroWriteSupport#generateEffectiveSchema} and converts that same effective
+ * schema into the file's MessageType. {@link MetaFieldsMode#NONE} keeps the writer from adding the
+ * Hudi meta columns, so the file holds only the table's own.
+ *
+ * <p>The read is deliberately not HoodieAvroParquetReader: that one rebuilds the variants through
+ * the provider, which would hide the typed_value columns under test. parquet-avro stores the write
+ * schema in the footer, so the records come back at the effective (shredded) schema.</p>
+ */
+ private List<GenericRecord> writeAndReadBack(HoodieSchema table, Properties props, String fileName,
+ List<GenericRecord> rows) throws Exception {
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmpDir.toString());
+ StoragePath path = new StoragePath(tmpDir.resolve(fileName).toAbsolutePath().toString());
+ HoodieConfig config = new HoodieConfig(TypedProperties.copy(props));
+ config.setValue(HoodieTableConfig.META_FIELDS_MODE, MetaFieldsMode.NONE.name());
+ // Bypassing HoodieWriteConfig means no engine sets the codec, and the factory no longer
+ // defaults it (#19685), so this low-level path has to.
+ config.setValue(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME, "zstd");
+ try (HoodieFileWriter writer = HoodieFileWriterFactory.getFileWriter(
+ "000", path, storage, config, table, new LocalTaskContextSupplier(),
+ HoodieRecord.HoodieRecordType.AVRO)) {
+ for (GenericRecord row : rows) {
+ ((HoodieAvroFileWriter) writer).writeAvro(String.valueOf(row.get("id")), row);
+ }
+ }
+
+ Configuration conf = new Configuration();
+ List<GenericRecord> readBack = new ArrayList<>();
+ try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(
+ HadoopInputFile.fromPath(new Path(path.toUri()), conf)).withConf(conf).build()) {
+ for (GenericRecord next = reader.read(); next != null; next = reader.read()) {
+ readBack.add(next);
+ }
+ }
+ return readBack;
+ }
+
+ /** A record of {@code schema} built from {@code name, value} pairs. */
+ private static GenericRecord recordOf(Schema schema, Object... nameValuePairs) {
+ GenericRecord record = new GenericData.Record(schema);
+ for (int i = 0; i < nameValuePairs.length; i += 2) {
+ record.put((String) nameValuePairs[i], nameValuePairs[i + 1]);
+ }
+ return record;
+ }
+
+ /** An unshredded variant record, the shape every input record carries. */
+ private static GenericRecord variant(Schema variantSchema, byte[] valueBytes) {
+ return recordOf(variantSchema,
+ HoodieSchema.Variant.VARIANT_METADATA_FIELD, ByteBuffer.wrap(VARIANT_METADATA),
+ HoodieSchema.Variant.VARIANT_VALUE_FIELD, ByteBuffer.wrap(valueBytes));
+ }
+
+ private static void assertTypedValue(GenericRecord variantRecord, String label) {
+ GenericRecord typedValue = (GenericRecord) variantRecord.get("typed_value");
+ assertNotNull(typedValue, label + " should have been shredded into typed_value");
+ assertNull(variantRecord.get("value"), label + " residual must be null once typed_value holds the value");
+ GenericRecord ddlField = (GenericRecord) typedValue.get("k");
+ assertNotNull(ddlField, label + " is missing the DDL field k: " + typedValue);
+ assertEquals(SHREDDED_LEAF, String.valueOf(ddlField.get("typed_value")), label + " typed_value leaf");
+ }
+
+ /**
+ * The value under {@code key}, which may be null - the entry is looked up first so that a present
+ * key with a null value is distinguishable from a missing key. parquet-avro hands map keys back as
+ * Utf8, which never equals a String key.
+ */
+ private static Object mapValue(Map<?, ?> map, String key) {
+ return map.entrySet().stream()
+ .filter(entry -> key.equals(String.valueOf(entry.getKey())))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("no entry " + key + " in " + map))
+ .getValue();
+ }
+
+ /** The whole cause chain rendered, so an assertion does not depend on which layer wraps a failure. */
+ private static String messageChain(Throwable throwable) {
+ StringBuilder chain = new StringBuilder();
+ for (Throwable current = throwable; current != null && current != current.getCause();
+ current = current.getCause()) {
+ chain.append(current).append('\n');
+ }
+ return chain.toString();
+ }
+
/** Write-support properties that shred every variant column with the given forced test DDL. */
private static Properties forcedShreddingProps(String forcedShreddingDdl) {
Properties props = new Properties();
@@ -166,4 +626,71 @@ class TestHoodieAvroWriteSupportShredding {
return false;
}
}
+
+ /**
+ * Stands in for the engine provider the write support loads reflectively, so the value walk can be
+ * driven without Spark on the classpath. The rule is deterministic and exercises all three
+ * outcomes a real provider has: a variant whose value bytes are {@link #TYPED_MARKER} shreds -
+ * every DDL field struct gets its typed_value leaf and the top-level residual goes null;
+ * {@link #DECLINE_MARKER} returns null, which is what {@code Spark4VariantShreddingProvider} does
+ * for a variant whose value or metadata is null, and the caller writes that null straight into the
+ * record; anything else keeps its value in the residual with typed_value left null. Must be public
+ * with a no-arg constructor for {@code ReflectionUtils.loadClass}.
+ */
+ public static class StubShreddingProvider implements VariantShreddingProvider {
+
+ @Override
+ public GenericRecord shredVariantRecord(GenericRecord unshreddedVariant, Schema shreddedSchema,
+ HoodieSchema.Variant variantSchema) {
+ ByteBuffer value = (ByteBuffer) unshreddedVariant.get(HoodieSchema.Variant.VARIANT_VALUE_FIELD);
+ if (isMarker(value, DECLINE_MARKER)) {
+ // The real provider declines the same way (null value or metadata), and VariantShredder
+ // passes the null straight into the record it is building.
+ return null;
+ }
+ GenericRecord shredded = new GenericData.Record(shreddedSchema);
+ shredded.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
+ unshreddedVariant.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD));
+ if (!isMarker(value, TYPED_MARKER)) {
+ // The residual fallback: typed_value stays null and the binary is carried as-is.
+ shredded.put(HoodieSchema.Variant.VARIANT_VALUE_FIELD, value);
+ return shredded;
+ }
+ Schema typedValueSchema = getNonNullTypeFromUnion(shreddedSchema.getField(
+ HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD).schema());
+ GenericRecord typedValue = new GenericData.Record(typedValueSchema);
+ for (Schema.Field field : typedValueSchema.getFields()) {
+ // Each DDL field is a {value, typed_value} struct; only the typed leaf is populated.
+ Schema fieldStruct = getNonNullTypeFromUnion(field.schema());
+ // Every DDL this test forces is `k string`, so the leaf is always a string.
+ Schema leafSchema = getNonNullTypeFromUnion(
+ fieldStruct.getField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD).schema());
+ if (leafSchema.getType() != Schema.Type.STRING) {
+ throw new UnsupportedOperationException("no stub value for " + leafSchema);
+ }
+ GenericRecord leaf = new GenericData.Record(fieldStruct);
+ leaf.put(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, SHREDDED_LEAF);
+ typedValue.put(field.name(), leaf);
+ }
+ shredded.put(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, typedValue);
+ return shredded;
+ }
+
+ @Override
+ public GenericRecord rebuildVariantRecord(GenericRecord shreddedVariant, Schema shreddedSchema,
+ Schema unshreddedSchema) {
+ throw new UnsupportedOperationException(
+ "the round trip reads raw parquet; the read half is TestHoodieVariantReconstruction's");
+ }
+
+ private static boolean isMarker(ByteBuffer value, byte[] marker) {
+ if (value == null) {
+ return false;
+ }
+ ByteBuffer copy = value.duplicate();
+ byte[] bytes = new byte[copy.remaining()];
+ copy.get(bytes);
+ return Arrays.equals(bytes, marker);
+ }
+ }
}
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 571379740f4f..99d1cf45f2fa 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
@@ -261,11 +261,12 @@ class TestHoodieVariantReconstruction {
@Test
void reconstructsShreddedVariantsNestedInRecordsArraysAndMaps(@TempDir Path tmp) {
- // The row writer shreds at any depth its write schema asks it to
- // (HoodieRowParquetWriteSupport.processNestedDataType), so a nested variant can reach this
- // reader as a plain {metadata, value, typed_value} record too. On the AVRO write path such a
- // file needs a hand-authored write schema - HoodieAvroWriteSupport.applyForcedShreddingSchema
- // walks top-level fields only - which is why this is a unit test and not an end-to-end one.
+ // Both write paths shred at any depth their write schema asks them to
+ // (HoodieRowParquetWriteSupport.processNestedDataType, VariantSchemaUtils.applyForcedShredding),
+ // so a nested variant can reach this reader as a plain {metadata, value, typed_value} record too.
+ // The bare array element / map value below is the shape neither forced-shredding hook produces -
+ // only a hand-authored write schema does - which is why this is a unit test and not an
+ // end-to-end one.
// 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.
@@ -406,7 +407,8 @@ class TestHoodieVariantReconstruction {
@Override
public GenericRecord shredVariantRecord(
GenericRecord unshreddedVariant, Schema shreddedSchema, HoodieSchema.Variant variantSchema) {
- throw new UnsupportedOperationException("Not used by the reconstruction test");
+ throw new UnsupportedOperationException("the reconstruction test only rebuilds; the shred-side "
+ + "stub is TestHoodieAvroWriteSupportShredding.StubShreddingProvider");
}
@Override
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
index 441753a626cb..0b0ae00117b0 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
@@ -164,7 +164,7 @@ public class HiveHoodieReaderContext extends HoodieReaderContext<ArrayWritable>
// payload in typed_value, which the projection drops). Detection is shape-based on the
// footer schema and anchored on the requested column being a variant, so plain user structs
// of the same shape are left alone. toShreddedReadSchema recurses through structs, array
- // elements and map values, matching the row writer, which shreds nested variants too.
+ // elements and map values, matching both write paths, which shred nested variants too.
// The flagged columns are split by Hive's read column names on the outer conf, since the
// per-file copy below gets requiredSchema's names from setSchemas and cannot tell the two
// apart: a column Hive selected fails as Hive-visible nulls; a column only requiredSchema
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java
index 07ee28dd29ac..08e98b959d4b 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java
@@ -810,7 +810,7 @@ public class TestHoodieParquetInputFormat {
@Test
public void testLegacyReaderGuardHonoursNestedPruningAndArrayWrappers() throws Exception {
- // A shredded variant below the top level (the row writer shreds at depth) fails too, naming
+ // A shredded variant below the top level (both write paths shred at depth) fails too, naming
// the struct column that holds it. The Hive type carries a sibling `other` the file does not
// hold - columns.types comes from the table - so the nested-pruning legs below have a path to
// project that never reaches the shredded group.
@@ -866,10 +866,10 @@ public class TestHoodieParquetInputFormat {
assertTrue(arrayPrunedFailure.getMessage().contains("'a'"),
"The error must name the column holding the array of shredded variants, got: " + arrayPrunedFailure.getMessage());
- // The 3-level list/element layout, which is what the row writer - the only production writer
- // that shreds inside a collection - emits. Here the repeated group IS a collection level, so
- // the walk has to see through it as well and still land the shredded group at `a`. The layout
- // is pinned on the footer, or the leg could silently degrade back to the 2-level one above.
+ // The 3-level list/element layout, which is what the row writer emits while the Avro write path
+ // emits the 2-level form above. Here the repeated group IS a collection level, so the walk has
+ // to see through it as well and still land the shredded group at `a`. The layout is pinned on
+ // the footer, or the leg could silently degrade back to the 2-level one.
StoragePath threeLevelPath =
InputFormatTestUtil.writeArrayShreddedVariantParquetFile(basePath, "array_shredded_3level.parquet", false, true);
GroupType repeated = fileSchemaOf(threeLevelPath).getType("a").asGroupType().getType(0).asGroupType();
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java
index 0ea8ed524761..fb40535a6275 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java
@@ -553,7 +553,7 @@ public class InputFormatTestUtil {
/** A shredded variant schema whose typed_value carries one string field {@code key}. */
static HoodieSchema.Variant shreddedVariantSchema() {
- return HoodieSchema.createVariantShreddedObject(
+ return HoodieSchema.createVariantShreddedObject(null, null, null,
Collections.singletonMap("key", HoodieSchema.create(HoodieSchemaType.STRING)));
}
@@ -601,7 +601,8 @@ public class InputFormatTestUtil {
/**
* Writes a one-row parquet file with an {@code id} column and a struct column {@code s} whose
- * {@code inner} member is a shredded variant: the shape the row writer produces at depth.
+ * {@code inner} member is a shredded variant: the shape both write paths produce at depth, since
+ * the forced DDL reaches a variant record member at any depth on the row and the Avro path alike.
*/
public static StoragePath writeNestedShreddedVariantParquetFile(java.nio.file.Path dir, String fileName) throws IOException {
return writeNestedShreddedVariantParquetFile(dir, fileName, false);
@@ -646,7 +647,8 @@ public class InputFormatTestUtil {
/**
* Writes a one-row parquet file with an {@code id} column and an {@code a} column holding an
- * array of shredded variants: the shape the row writer produces inside a collection.
+ * array of shredded variants: the shape either write path produces from a write schema that
+ * declares it, a bare array element being the one position the forced DDL leaves alone on both.
*
* <p>parquet-avro 1.13 writes the 2-level list layout by default
* ({@code parquet.avro.write-old-list-structure} is true unless set), so the element group is
@@ -671,8 +673,8 @@ public class InputFormatTestUtil {
/**
* As {@link #writeArrayShreddedVariantParquetFile(java.nio.file.Path, String, boolean)}, with
* {@code threeLevelList} switching the layout from parquet-avro's default 2-level {@code array}
- * to the 3-level {@code list}/{@code element} one, which is what the Spark row writer - the only
- * production writer that shreds inside a collection - emits.
+ * to the 3-level {@code list}/{@code element} one, which is what the Spark row writer emits; the
+ * Avro write path emits the 2-level form, so both layouts reach the guard.
*/
public static StoragePath writeArrayShreddedVariantParquetFile(java.nio.file.Path dir, String fileName,
boolean wrapElementInStruct,
@@ -750,9 +752,10 @@ public class InputFormatTestUtil {
/**
* Writes a one-row parquet file with an {@code id} column and an {@code m} column holding a
- * {@code map<string, variant>} whose value is shredded: the shape the row writer produces inside
- * a map. A map's entry level, key and value are levels Hive's dotted paths never name, so the
- * shredded group has to land at the column's own path {@code m}.
+ * {@code map<string, variant>} whose value is shredded: the shape either write path produces from
+ * a write schema that declares it, a bare map value being the other position the forced DDL
+ * leaves alone on both. A map's entry level, key and value are levels Hive's dotted paths never
+ * name, so the shredded group has to land at the column's own path {@code m}.
*/
public static StoragePath writeMapShreddedVariantParquetFile(java.nio.file.Path dir, String fileName) throws IOException {
HoodieSchema.Variant shreddedVariant = shreddedVariantSchema();
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index 1ebfdf379a82..aba5c31bc1a4 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -57,7 +57,7 @@ import org.apache.spark.sql.execution.vectorized.{OffHeapColumnVector, OnHeapCol
import org.apache.spark.sql.hudi.MultipleColumnarFileFormatReader
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnarBatchUtils}
import org.apache.spark.util.SerializableConfiguration
@@ -135,6 +135,25 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
}
}
+ /**
+ * True when `dataType` itself, or anything it nests, satisfies `predicate`. A struct that
+ * matches is reported without descending into it, so a variant projection struct is answered
+ * as one type rather than as its pushed-down extraction fields.
+ */
+ private def containsType(dataType: DataType, predicate: DataType => Boolean): Boolean = {
+ predicate(dataType) || (dataType match {
+ case s: StructType => s.fields.exists(f => containsType(f.dataType, predicate))
+ case a: ArrayType => containsType(a.elementType, predicate)
+ case m: MapType => containsType(m.valueType, predicate)
+ case _ => false
+ })
+ }
+
+ private def isVariantProjection(dataType: DataType): Boolean = dataType match {
+ case s: StructType => sparkAdapter.isVariantProjectionStruct(s)
+ case _ => false
+ }
+
/**
* Checks if the file format supports vectorized reading, please refer to SPARK-40918.
*
@@ -153,8 +172,7 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
supportVectorizedRead = false
supportReturningBatch = false
false
- } else if (schema.fields.exists(f => f.dataType.isInstanceOf[StructType]
- && sparkAdapter.isVariantProjectionStruct(f.dataType.asInstanceOf[StructType]))) {
+ } else if (schema.fields.exists(f => containsType(f.dataType, isVariantProjection))) {
// Spark 4.1's PushVariantIntoScan rewrites a variant column to a struct of pushed-down
// extractions. The Spark vectorized parquet reader treats this as a nested type change
// (data column is VariantType, required is a struct) and refuses to read in vectorized
@@ -162,9 +180,21 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
supportVectorizedRead = false
supportReturningBatch = false
false
- } else if (HoodieSparkUtils.gteqSpark4_1 && schema.fields.exists(f => sparkAdapter.isVariantType(f.dataType))) {
+ } else if (HoodieSparkUtils.gteqSpark4_1
+ && schema.fields.exists(f => containsType(f.dataType, sparkAdapter.isVariantType))) {
// #18605: Spark 4.1's vectorized variant read produces UnsafeRow encodings that SIGBUS
// during RangePartitioner sampling. Force row-based reads. Spark 4.0 unaffected.
+ //
+ // Both checks above walk the schema instead of scanning top-level fields only. Spark's
+ // ParquetUtils.isBatchReadSupported treats VariantType as an atomic type and, once
+ // spark.sql.parquet.enableNestedColumnVectorizedReader is on - on by default since Spark
+ // 3.4.0 (added in 3.3.0, off) - nested columns as batch-readable, so at stock settings a
+ // variant reached the vectorized reader whenever it sat inside a struct/array/map even
+ // though a top-level one did not. The SIGBUS is in the UnsafeRow encoding of the vectorized
+ // variant vectors that RangePartitioner samples, and it samples WHOLE rows, so a variant
+ // carried inside a struct is exposed exactly the same way under any range-partitioned query.
+ // The cost is that nested-variant tables lose vectorization on 4.1+ by default, exactly as
+ // top-level ones do.
supportVectorizedRead = false
supportReturningBatch = false
false
diff --git a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantParquetTestFixtures.scala b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantParquetTestFixtures.scala
index 7af09eb76e64..1ea7bcff2b4c 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantParquetTestFixtures.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VariantParquetTestFixtures.scala
@@ -19,14 +19,15 @@
package org.apache.spark.sql.execution.datasources.parquet
-import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
+import org.apache.parquet.schema.{GroupType, LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
/**
- * The parquet layouts the shredded-variant guards are tested against, shared by the suites that
- * cover them: [[TestParquetSchemaEvolutionUtils]] here and TestSpark40HoodieParquetReadSupport in
- * hudi-spark4.0.x (which pulls this module's test-jar). One definition per shape keeps the two
- * guards pinned against the same files.
+ * The parquet layouts the shredded-variant guards are tested against, plus the LIST/MAP walking
+ * rules production resolves them by. The layouts are shared by [[TestParquetSchemaEvolutionUtils]]
+ * here and TestSpark40HoodieParquetReadSupport in hudi-spark4.0.x; the walking rules are used by
+ * VariantShreddingTestSupport in hudi-spark. Both of those out-of-module suites pull this module's
+ * test-jar. One definition per shape keeps the guards pinned against the same files.
*
* Only parquet types are named here, so the object stays usable from every Spark-version module.
*/
@@ -74,4 +75,21 @@ object VariantParquetTestFixtures {
.addField(value)
.named("key_value"))
.named(name)
+
+ /**
+ * The element of a parquet LIST group, resolved by the same function the read-side guards use,
+ * so a test walking a file cannot drift from the rule production applies: both the 3-level
+ * layout the Spark writer emits (group -> repeated "list" -> element) and the 2-level one
+ * parquet-avro emits (the repeated group, named "array" or "<field>_tuple", IS the element) are
+ * covered there. A shape the rule does not recognize is a broken fixture, so it fails here
+ * rather than returning the None production uses to stop a walk without failing a read.
+ */
+ def listElement(list: GroupType): Type =
+ ParquetSchemaEvolutionUtils.parquetListElement(list).getOrElse(
+ throw new IllegalArgumentException(s"not a parquet LIST layout the read guards resolve:\n$list"))
+
+ /** The value of a parquet MAP group, per ParquetSchemaEvolutionUtils.parquetMapValue. */
+ def mapValue(map: GroupType): Type =
+ ParquetSchemaEvolutionUtils.parquetMapValue(map).getOrElse(
+ throw new IllegalArgumentException(s"not a parquet MAP layout the read guards resolve:\n$map"))
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
index a0c0c07c6f2a..4535a04e6261 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
@@ -19,14 +19,17 @@
package org.apache.spark.sql.hudi.dml.schema
-import org.apache.hudi.HoodieSparkUtils
+import org.apache.hudi.{HoodieSchemaConversionUtils, HoodieSparkUtils, HoodieTableSchema, SparkAdapterSupport}
+import org.apache.hudi.common.model.HoodieFileFormat
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter
import org.apache.hudi.testutils.DataSourceTestUtils
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
/**
* Mixed-layout variant shredding matrix: files with DIFFERENT typed_value layouts in one table,
@@ -655,9 +658,9 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
// DDL request that struct and fail in Spark before any Hudi hook.
}
- // The guard recurses: a NESTED shredded variant (struct<inner: variant>, written by the
- // bulk-insert row writer, the one production writer that shreds below the top level) fails
- // fast too, instead of slipping past a top-level-only walk.
+ // The guard recurses: a NESTED shredded variant (struct<inner: variant>, seeded here by the
+ // bulk-insert row writer, though both write paths shred below the top level) fails fast too,
+ // instead of slipping past a top-level-only walk.
withNestedVariantTable("nested schema-on-read", recordTypes = Seq(HoodieRecordType.SPARK)) {
(tableName, tablePath, leg) =>
assertVariantLayout(tablePath, shredded = true, s"$leg setup", column = "s.inner")
@@ -740,9 +743,9 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
test("Row-writer forced shredding of a nested variant reads back and merges") {
assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
- // The bulk-insert row writer is the only production writer that shreds a NESTED variant
- // (the forced hook of the AVRO write support is top-level only, and #18961's inference
- // never shreds nested columns).
+ // Both forced hooks shred a NESTED variant - the row writer's and, since the #19689 fix, the
+ // AVRO write support's - so this leg is about the row-writer seed and what the merge below
+ // does to it; #18961's inference never shreds nested columns either way.
withNestedVariantTable("nested forced shredding") { (tableName, tablePath, leg) =>
val files = listDataParquetFiles(tablePath)
assert(files.size == 1, s"[$leg] expected one base file, got $files")
@@ -751,21 +754,12 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
assert(getFieldAsGroup(innerGroup, "typed_value").containsField("k"),
s"[$leg] nested typed_value should carry k:\n$innerGroup")
- // #18605 history: the batch-disabling guards in HoodieFileGroupReaderBasedFileFormat are
- // top-level-only, so a NESTED variant still reaches the vectorized reader. The session conf
- // does pick the reader - HoodieFileGroupReaderBasedFileFormat.supportBatch reads
- // sparkSession.sessionState.conf and ParquetUtils.isBatchReadSupportedForSchema gates on
- // spark.sql.parquet.enableVectorizedReader, and only afterwards does
- // buildReaderWithPartitionValues write that decision back into the conf - so sweep it and
- // pin both readers. Every nested-variant read bug so far (HUDI-7190, HUDI-8803, #18605) is
- // vectorized-only, which leaves the row-based leg as the control.
- Seq("true", "false").foreach { vectorizedReader =>
- withSQLConf("spark.sql.parquet.enableVectorizedReader" -> vectorizedReader) {
- checkAnswer(s"select id, cast(s.inner as string) from $tableName")(
- Seq(1, """{"k":"x1"}""")
- )
- }
- }
+ // One read, whatever spark.sql.parquet.enableVectorizedReader says: supportBatch vetoes
+ // batch reads for a variant at any depth before that conf is consulted, and section F2 pins
+ // that decision directly.
+ checkAnswer(s"select id, cast(s.inner as string) from $tableName")(
+ Seq(1, """{"k":"x1"}""")
+ )
// A plain insert bin-packs into the same file group: the small-file merge must read the
// nested-shredded base back (nested reconstruction on the AVRO record type leg).
@@ -773,6 +767,13 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
spark.sql(s"""insert into $tableName values (2, named_struct('inner', parse_json('{"k":"x2"}')), 1000)""")
}
assertSingleFileGroup(tablePath, leg)
+ // The merge rewrote the whole file group under the INCOMING layout, so the layout below is
+ // the record writer's, not the row writer's seed. On the AVRO leg that write goes through
+ // HoodieAvroWriteSupport, whose forced hook now reaches the nested member; before the #19689
+ // fix it silently rewrote s.inner unshredded and nothing here noticed. It pins the
+ // small-file MERGE handle's write, while the record-writer test below pins a fresh INSERT
+ // handle, so the two are not interchangeable.
+ assertVariantLayout(tablePath, shredded = true, leg, column = "s.inner")
checkAnswer(s"select id, cast(s.inner as string) from $tableName order by id")(
Seq(1, """{"k":"x1"}"""),
Seq(2, """{"k":"x2"}""")
@@ -780,6 +781,110 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
}
}
+ test("Both record types force-shred a nested variant through the record writer") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // A plain insert, so the write goes through the record writers rather than the row writer: the
+ // AVRO leg is HoodieAvroWriteSupport end to end, the SPARK leg HoodieRowParquetWriteSupport,
+ // and the #19689 fix is what made their forced hooks agree. The DDL reaches a variant that is
+ // a record MEMBER at any depth - s.inner, and the inner of the struct element of items - but
+ // never one that is directly a collection element, which is why arr stays unshredded on both
+ // paths.
+ withVariantTable("record-writer nested forced shredding", "cow",
+ extraCols = "s struct<inner: variant>, items array<struct<inner: variant>>, arr array<variant>") {
+ (tableName, tablePath, leg) =>
+ withWriteLayout(Forced("k string")) {
+ spark.sql(
+ s"""insert into $tableName values (1, parse_json('{"k":"top"}'),
+ | named_struct('inner', parse_json('{"k":"nested"}')),
+ | array(named_struct('inner', parse_json('{"k":"element"}'))),
+ | array(parse_json('{"k":"bare"}')), 1000)""".stripMargin)
+ }
+
+ val files = listDataParquetFiles(tablePath)
+ assert(files.size == 1, s"[$leg] expected one base file, got $files")
+ Seq("s.inner", "items.inner").foreach { column =>
+ assertVariantLayout(tablePath, shredded = true, leg, column = column)
+ val group = variantGroupOf(files.head, column)
+ assert(getFieldAsGroup(group, "typed_value").containsField("k"),
+ s"[$leg] typed_value of $column should carry k:\n$group")
+ }
+ assertVariantLayout(tablePath, shredded = false, leg, column = "arr")
+
+ checkAnswer(s"select id, cast(v as string), cast(s.inner as string), " +
+ s"cast(items[0].inner as string), cast(arr[0] as string) from $tableName")(
+ Seq(1, """{"k":"top"}""", """{"k":"nested"}""", """{"k":"element"}""", """{"k":"bare"}""")
+ )
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // F2. Vectorized read decision
+ // -----------------------------------------------------------------------------------------------
+
+ test("supportBatch turns off vectorization for a variant at any depth") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // The read legs above take one read and let supportBatch decide, whatever
+ // spark.sql.parquet.enableVectorizedReader says; this pins the decision itself, which is what
+ // vetoes their batch reads in the first place. Spark's ParquetUtils.isBatchReadSupported
+ // calls VariantType atomic and, once spark.sql.parquet.enableNestedColumnVectorizedReader is
+ // on, nested columns batch-readable, so a variant one struct, array or map deep used to reach
+ // the vectorized reader even though a top-level one did not - while #18605's SIGBUS is in the
+ // UnsafeRow encoding of the variant vectors RangePartitioner samples, and it samples WHOLE
+ // rows, which a nesting level does nothing to hide. That setting has been on by default since
+ // Spark 3.4.0 (added in 3.3.0, off), so the hole was reachable at stock settings and not only
+ // once a user opted in.
+ // The withSQLConf below turns it and the vectorized reader itself on as an explicit pin rather
+ // than as an opt-in - the shared test session does not promise Spark's defaults - so the
+ // variant-free twins are vectorized and only the guard can say no.
+ //
+ // Nothing below opens a table: supportBatch decides off the schema it is handed, so the
+ // constructor arguments only have to be well-formed.
+ val unusedSchema = StructType.fromDDL("id int, ts long")
+ val format = new HoodieFileGroupReaderBasedFileFormat(
+ "unused", HoodieTableSchema(unusedSchema,
+ HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(unusedSchema, "unused", "hoodie.test")),
+ "unused", "000", Seq.empty, isMOR = false, isBootstrap = false, isIncremental = false,
+ "", shouldUseRecordPosition = false, Seq.empty, isMultipleBaseFileFormatsEnabled = false,
+ HoodieFileFormat.PARQUET)
+
+ withSQLConf(
+ "spark.sql.parquet.enableVectorizedReader" -> "true",
+ "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true") {
+ // Each shape is asserted against its own variant-free twin, so a schema that lost vectorization
+ // for some other reason cannot pass as the variant guard doing its job.
+ def shapes(elementType: String): Seq[String] = Seq(
+ s"id int, s struct<inner: $elementType>",
+ s"id int, arr array<$elementType>",
+ s"id int, m map<string, $elementType>")
+
+ shapes("string").zip(shapes("variant")).foreach { case (control, nested) =>
+ assert(format.supportBatch(spark, StructType.fromDDL(control)),
+ s"the variant-free $control must stay vectorized")
+ assert(!format.supportBatch(spark, StructType.fromDDL(nested)),
+ s"$nested must fall back to row-based reads")
+ }
+
+ // And the same for the other arm, PushVariantIntoScan's projection struct. A field whose own
+ // type is that struct is the control the top-level-only check already matched; the recursion
+ // is what the leg below adds, where the struct sits one member deeper.
+ val projection = SparkAdapterSupport.sparkAdapter.buildFullVariantReadSchema(StructType.fromDDL("v variant"))
+ assert(projection.isDefined, "Spark 4.1+ must rewrite a top-level variant for a full read")
+ val projectionStruct = projection.get.fields.head.dataType
+ val topLevelProjection = StructType(Array(
+ StructField("id", IntegerType),
+ StructField("s", projectionStruct)))
+ assert(!format.supportBatch(spark, topLevelProjection),
+ "a top-level variant projection struct must fall back to row-based reads")
+ val nestedProjection = StructType(Array(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Array(StructField("inner", projectionStruct))))))
+ assert(!format.supportBatch(spark, nestedProjection),
+ "a nested variant projection struct must fall back to row-based reads")
+ }
+ }
+
// -----------------------------------------------------------------------------------------------
// G. Schema evolution
// -----------------------------------------------------------------------------------------------
@@ -1056,7 +1161,9 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
// Inference is top-level only (VariantSchemaUtils.getInferableVariantColumns), so the nested
// variant keeps the plain {metadata, value} shape in the very file where the top-level one
- // shredded. Only the forced hook shreds below the top level, and only through the row writer.
+ // shredded. Only the forced hooks shred below the top level - both of them, into record
+ // members at any depth; a variant that is directly a collection element needs a write schema
+ // that already declares it.
val nestedSourceSql = s"select cast(id as int) as id, parse_json(${ObjA.jsonExpr}) as v, " +
s"named_struct('inner', parse_json(${ObjA.jsonExpr})) as s, 1000L as ts from range(0, 4, 1, 1)"
@@ -1076,8 +1183,8 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
assertNestedUnshredded(tableName, tablePath, leg)
}
- // The row writer is the one production writer that CAN shred a nested variant; under
- // inference it must not, because inference never asks it to.
+ // The row writer shreds a nested variant when the forced DDL asks it to; under inference it
+ // must not, because inference never asks.
withVariantTable("nested inference row writer", "cow", props = Seq(NEW_FILE_GROUP_PER_COMMIT),
extraCols = "s struct<inner: variant>", recordTypes = Seq(HoodieRecordType.SPARK)) {
(tableName, tablePath, leg) =>
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
index 1221b7cd5ea2..64d9b49fa003 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
@@ -33,8 +33,9 @@ import org.apache.parquet.hadoop.{ParquetFileReader, ParquetReader}
import org.apache.parquet.hadoop.api.ReadSupport
import org.apache.parquet.hadoop.example.GroupReadSupport
import org.apache.parquet.hadoop.util.HadoopInputFile
-import org.apache.parquet.schema.{GroupType, MessageType, Type}
+import org.apache.parquet.schema.{GroupType, LogicalTypeAnnotation, MessageType, Type}
import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{listElement, mapValue}
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
@@ -108,9 +109,11 @@ trait VariantShreddingTestSupport { self: HoodieSparkSqlTestBase =>
/**
* Creates `(id int, s struct<inner: variant>, ts long)` and bulk-inserts row 1 through the row
- * writer under a forced `k string` nested schema: the bulk-insert row writer is the only
- * production writer that shreds a variant BELOW the top level, so every nested-shredding leg
- * needs exactly this table and this seed row.
+ * writer under a forced `k string` nested schema. Both forced hooks recurse into record members
+ * (the row writer's always did, the Avro write support's since the #19689 fix), so a nested
+ * variant shreds on either write path; inference alone stays top-level, and a variant that is
+ * directly an array element or a map value shreds only where the write schema already declares
+ * it. This row-writer seed is what the nested-shredding legs open on.
*/
private def createNestedVariantRowWriterTable(tableName: String, tablePath: String): Unit = {
spark.sql(
@@ -189,11 +192,31 @@ trait VariantShreddingTestSupport { self: HoodieSparkSqlTestBase =>
/**
* The variant group of `column` inside one parquet file. `column` may be a dotted path
- * ("s.inner"), walked one struct level per segment, so a nested variant is inspected the same
- * way as a top-level one.
+ * ("s.inner"), walked one member per segment, so a nested variant is inspected the same way as
+ * a top-level one. A segment that lands on a LIST or MAP group is stepped through its wrapper
+ * to what it holds, so "items.inner" reaches the `inner` member of the element struct of an
+ * `array<struct<inner: variant>>` and "arr" reaches the element of an `array<variant>`.
*/
protected def variantGroupOf(filePath: String, column: String): GroupType =
- column.split('.').foldLeft(readParquetSchema(filePath): GroupType)(getFieldAsGroup)
+ column.split('.').foldLeft(readParquetSchema(filePath): GroupType) {
+ (group, segment) => collectionElementOf(getFieldAsGroup(group, segment))
+ }
+
+ /**
+ * Steps a LIST or MAP group down to the group it holds - a list element or a map value - and
+ * returns anything else, a variant group included, unchanged. The two write paths disagree on
+ * the list layout (the Spark writer emits the 3-level `list`/`element` shape, parquet-avro the
+ * 2-level one whose repeated group, named "array" or "<field>_tuple", IS the element), so the
+ * step delegates to the read-side guards' own rule rather than restating it.
+ */
+ private def collectionElementOf(group: GroupType): GroupType =
+ group.getLogicalTypeAnnotation match {
+ case _: LogicalTypeAnnotation.ListLogicalTypeAnnotation =>
+ collectionElementOf(listElement(group).asGroupType())
+ case _: LogicalTypeAnnotation.MapLogicalTypeAnnotation =>
+ collectionElementOf(mapValue(group).asGroupType())
+ case _ => group
+ }
/**
* Pins the on-disk layout of `column` (a name or a dotted path) across every data parquet file.
diff --git a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
index 7e79e80721c9..870c8501ace0 100644
--- a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
+++ b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
@@ -76,7 +76,7 @@ class TestHoodieVariantReconstructionRoundTrip {
Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
- HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
HoodieSchema.Variant unshreddedVariant = HoodieSchema.createVariant();
HoodieSchema fileSchema = HoodieSchema.createRecord("r", "org.apache.hudi.test", null, Arrays.asList(
@@ -126,7 +126,7 @@ class TestHoodieVariantReconstructionRoundTrip {
Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
- HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
HoodieSchema.Variant unshreddedVariant = HoodieSchema.createVariant();
// Shred with the real provider first, then drop the (null) top-level residual, which yields exactly
@@ -221,7 +221,7 @@ class TestHoodieVariantReconstructionRoundTrip {
Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
- HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
HoodieSchema.Variant unshreddedVariant = HoodieSchema.createVariant();
// Data fields are nullable with null defaults, as Hudi table schemas declare them; the
// skeleton read requests the full schema and fills the data columns with the defaults.
diff --git a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
index 8abb9bf992fe..edbf9f270420 100644
--- a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
+++ b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
@@ -283,7 +283,7 @@ class TestSpark4VariantShreddingProvider {
Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
- assertRoundTrips("{\"a\":\"x\",\"b\":5}", HoodieSchema.createVariantShreddedObject(shreddedFields));
+ assertRoundTrips("{\"a\":\"x\",\"b\":5}", HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields));
}
@Test
@@ -293,7 +293,7 @@ class TestSpark4VariantShreddingProvider {
Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
- HoodieSchema.Variant shredded = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant shredded = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
Variant variant = VariantBuilder.parseJson("{\"a\":\"x\",\"c\":99}", false);
GenericRecord shreddedRecord = shred(variant, shredded);
@@ -317,7 +317,7 @@ class TestSpark4VariantShreddingProvider {
// the residual, its typed_value stays null, and the top-level value stays null.
Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.LONG));
- HoodieSchema.Variant shredded = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant shredded = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
Variant variant = VariantBuilder.parseJson("{\"a\":\"str\"}", false);
GenericRecord shreddedRecord = shred(variant, shredded);
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupportVariant.java b/hudi-spark-datasource/hudi-spark4.0.x/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupportVariant.java
index 98326bab2645..75e025623a8e 100644
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupportVariant.java
+++ b/hudi-spark-datasource/hudi-spark4.0.x/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupportVariant.java
@@ -295,7 +295,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
LinkedHashMap<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.INT));
shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.STRING));
- HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
// Create a nested struct containing the variant
HoodieSchema nestedStructSchema = HoodieSchema.createRecord("NestedStruct", null, null,
@@ -350,7 +350,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
// Setup: Create shredded variant
LinkedHashMap<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("x", HoodieSchema.create(HoodieSchemaType.LONG));
- HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
// Create deeply nested structure: level2 -> level1 -> variant
HoodieSchema level2Schema = HoodieSchema.createRecord("Level2", null, null,
@@ -410,7 +410,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
// Setup: Array of shredded variants
LinkedHashMap<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("item", HoodieSchema.create(HoodieSchemaType.STRING));
- HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
HoodieSchema arraySchema = HoodieSchema.createArray(variantSchema);
HoodieSchema recordSchema = HoodieSchema.createRecord("record", null, null,
@@ -451,7 +451,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
// Setup: Variant in struct, inside array, inside struct
LinkedHashMap<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("data", HoodieSchema.create(HoodieSchemaType.INT));
- HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
// Inner struct with variant
HoodieSchema innerStructSchema = HoodieSchema.createRecord("InnerStruct", null, null,
@@ -523,7 +523,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
// Setup: Map with nullable shredded variant values
LinkedHashMap<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("score", HoodieSchema.create(HoodieSchemaType.INT));
- HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
// Wrap the variant in a nullable union, as would occur in real schemas
HoodieSchema nullableVariantSchema = HoodieSchema.createNullable(variantSchema);
@@ -569,7 +569,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
// Setup: Array with nullable shredded variant elements
LinkedHashMap<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
shreddedFields.put("tag", HoodieSchema.create(HoodieSchemaType.STRING));
- HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant variantSchema = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
// Wrap the variant in a nullable union
HoodieSchema nullableVariantSchema = HoodieSchema.createNullable(variantSchema);
@@ -710,7 +710,7 @@ public class TestHoodieRowParquetWriteSupportVariant {
@Test
public void testEmptyShreddedFieldsMapThrows() {
assertThrows(IllegalArgumentException.class,
- () -> HoodieSchema.createVariantShreddedObject(Collections.emptyMap()),
+ () -> HoodieSchema.createVariantShreddedObject(null, null, null, Collections.emptyMap()),
"Empty shredded fields map should throw IllegalArgumentException");
}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
index fa89bcdfcd2f..1e61d93180e1 100644
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
+++ b/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
@@ -22,7 +22,7 @@ import org.apache.hudi.exception.HoodieException
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.parquet.schema.Types
import org.apache.spark.sql.execution.datasources.VariantMetadata
-import org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant, stringKeyMap, threeLevelList, unshreddedVariant}
+import org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant, stringKeyMap, threeLevelList, twoLevelList, unshreddedVariant}
import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType, MapType, MetadataBuilder, StringType, StructField, StructType, VariantType}
import org.junit.jupiter.api.{Assertions, Test}
@@ -128,9 +128,11 @@ class TestSpark40HoodieParquetReadSupport {
}
/**
- * A shredded variant inside an array is unreadable too. The walk resolves the parquet element
- * through the 3-level list layout and reports it as v.element; the plain-struct catalyst twin
- * over the same file is left alone.
+ * A shredded variant inside an array is unreadable too, in both list layouts a Hudi base file
+ * can carry: the 3-level list the row writer emits, reported as v.element, and the 2-level
+ * "array" list the Avro write path emits (parquet-avro's default
+ * parquet.avro.write-old-list-structure=true), where the repeated group is itself the element
+ * record. The plain-struct catalyst twin over the same file is left alone.
*/
@Test
def testRejectShreddedVariantsFailsFastOnVariantInArray(): Unit = {
@@ -146,6 +148,21 @@ class TestSpark40HoodieParquetReadSupport {
val structSchema = new StructType().add("v", ArrayType(plainStructTwin))
Spark40HoodieParquetReadSupport.rejectShreddedVariants(schema, Some(structSchema))
+
+ // The Avro writer's own layout for the array position it does shred, array<struct<variant>>:
+ // the repeated group named "array" IS the element record. A walk that only knew the 3-level
+ // layout would unwrap that single-field group, pair the variant group itself against
+ // catalyst's struct<inner: variant>, match no field name and let the file through.
+ val avroSchema = Types.buildMessage()
+ .addField(twoLevelList("v", "array", shreddedVariant("inner")))
+ .named("test")
+ val avroVariantSchema =
+ new StructType().add("v", ArrayType(new StructType().add("inner", VariantType)))
+ val avroFailure = Assertions.assertThrows(classOf[HoodieException],
+ () => Spark40HoodieParquetReadSupport.rejectShreddedVariants(avroSchema, Some(avroVariantSchema)))
+ Assertions.assertTrue(
+ avroFailure.getMessage.contains("shredded variant") && avroFailure.getMessage.contains("'v.element.inner'"),
+ s"The 2-level list error must name the shredded variant column, got: ${avroFailure.getMessage}")
}
/**