voonhous commented on code in PR #19777:
URL: https://github.com/apache/hudi/pull/19777#discussion_r3891487669
##########
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:
Out of scope for the DDL, but the footer-time crash goes. The read side has
the same `default` arm (`HoodieVariantReconstruction.buildRebuilder`), so
recursing here and in `buildShredder` alone would write shredded groups under
unions that the Avro reader never reconstructs; moving all three arms (plus the
shredded-file detection and the strip walks) is its own change, and the row
writer this PR mirrors never sees a union.
What changed: `applyForcedShredding` now walks the spliced schema for two
definitions of one record full name and throws `HoodieSchemaException` naming
the type and the union cause, instead of Avro's `Can't redefine` at footer
stamping on 1.11 (or a silently mis-serialized footer on 1.12). The three arms'
comments point at each other.
`testApplyForcedShreddingKeepsMultiBranchUnionsOutOfReach` pins both legs: a
type only under a union passes through untouched; the same type also at a
forced position is rejected.
--
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]
