wombatu-kun commented on code in PR #19777:
URL: https://github.com/apache/hudi/pull/19777#discussion_r3888213495
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java:
##########
@@ -976,11 +976,26 @@ public static HoodieSchema.Vector createVector(String
name, int dimension, Vecto
* |-- 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.
Review Comment:
The 1-arg createVariantShreddedObject overload still mints the variant
record, its typed_value record and every field struct with a null namespace, so
a caller can drop into exactly the collision that deleting the 1-arg
createShreddedFieldStruct was meant to prevent. Should it go the same way, or
take a namespace?
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java:
##########
@@ -144,22 +171,144 @@ private static HoodieSchema
stripVariantShreddingAt(HoodieSchema schema) {
}
/**
- * 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}
+ */
+ public static HoodieSchema applyForcedShredding(HoodieSchema schema,
Map<String, HoodieSchema> typedValueFields) {
+ if (schema.getType() != HoodieSchemaType.RECORD || typedValueFields ==
null || typedValueFields.isEmpty()) {
+ return schema;
+ }
+ return applyForcedShreddingToRecord(schema, typedValueFields);
+ }
+
+ 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; a future change to either has to
move both.
+ */
+ 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:
Review Comment:
A record type reached both directly and through a multi-branch union is
rebuilt at one position and passed through at the other, leaving two non-equal
definitions of one Avro full name that Schema.toString rejects when the footer
schema is stamped. Recurse into union branches here and in buildShredder, or is
that shape out of scope for the test-only forced DDL?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
