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 616f9ae86765 fix(common): dedupe repeated Avro fixed names (#20324)
616f9ae86765 is described below
commit 616f9ae867654d70b58674600887e4ff1205ccf2
Author: voonhous <[email protected]>
AuthorDate: Thu Oct 15 16:04:59 2026 +0800
fix(common): dedupe repeated Avro fixed names (#20324)
* fix(common): dedupe repeated Avro fixed names
AvroSchemaConverterWithTimestampNTZ is Hudi's copy of parquet-java
1.13.1's AvroSchemaConverter. In the Parquet-to-Avro direction it
names every fixed (including DECIMAL on fixed_len_byte_array) after
its field with no namespace, while repeated record names already get
a dedup namespace. Shredded variants name every leaf typed_value, so a
DECIMAL leaf collides with its enclosing typed_value record and the
read schema no longer survives the toString/parse round trip in
HoodieAvroParquetReader, failing COW upserts with
"Undefined schema: k102.typed_value".
Port PARQUET-2456 (apache/parquet-java#1304, released in 1.14.0):
records and fixed types share one name counter, and repeated names get
a <name><count> namespace. Hudi builds DECIMAL fixed types in its
logical-type branch rather than in convertFIXED_LEN_BYTE_ARRAY, so
that branch gets the same treatment.
Also restore the INT96 doc string, which the HoodieSchema migration
passed in the namespace slot (HoodieSchema.createFixed takes the
namespace second, Avro's Schema.createFixed takes the doc second).
Fixes #20307
* refactor(common): rename dedup namespace helper
Rename namespace() to nextDedupNamespace(): it bumps the shared name
counter as a side effect, which the old name did not say. Addresses
review feedback on #20324.
---
.../avro/AvroSchemaConverterWithTimestampNTZ.java | 25 +++++---
.../parquet/avro/TestAvroSchemaConverter.java | 67 +++++++++++++++++++++-
2 files changed, 84 insertions(+), 8 deletions(-)
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/AvroSchemaConverterWithTimestampNTZ.java
b/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/AvroSchemaConverterWithTimestampNTZ.java
index af7238b17de9..66c1c361fd5b 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/AvroSchemaConverterWithTimestampNTZ.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/parquet/avro/AvroSchemaConverterWithTimestampNTZ.java
@@ -354,8 +354,8 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
}
private HoodieSchema convertFields(String name, List<Type> parquetFields,
Map<String, Integer> names) {
+ String ns = nextDedupNamespace(name, names);
List<HoodieSchemaField> fields = new ArrayList<>(parquetFields.size());
- Integer nameCount = names.merge(name, 1, (oldValue, value) -> oldValue +
1);
for (Type parquetType : parquetFields) {
HoodieSchema fieldSchema = convertField(parquetType, names);
if (parquetType.isRepetition(REPEATED)) {
@@ -368,10 +368,16 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
parquetType.getName(), fieldSchema, null, (Object) null));
}
}
- HoodieSchema schema = HoodieSchema.createRecord(name, null, nameCount > 1
? name + nameCount : null, false, fields);
+ HoodieSchema schema = HoodieSchema.createRecord(name, null, ns, false,
fields);
return schema;
}
+ // PARQUET-2456: repeated record and fixed names get a dedup namespace so
the schema survives toString and parse
+ private static String nextDedupNamespace(String name, Map<String, Integer>
names) {
+ Integer nameCount = names.merge(name, 1, (oldValue, value) -> oldValue +
1);
+ return nameCount > 1 ? name + nameCount : null;
+ }
+
private HoodieSchema convertField(final Type parquetType, Map<String,
Integer> names) {
if (parquetType.isPrimitive()) {
final PrimitiveType asPrimitive = parquetType.asPrimitiveType();
@@ -381,7 +387,7 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
// Handle logical type annotations directly with HoodieSchema creation
methods
if (annotation != null) {
- HoodieSchema logicalSchema =
convertLogicalTypeAnnotationToHoodieSchema(annotation, parquetType);
+ HoodieSchema logicalSchema =
convertLogicalTypeAnnotationToHoodieSchema(annotation, parquetType, names);
if (logicalSchema != null) {
return logicalSchema;
}
@@ -405,7 +411,7 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
@Override
public HoodieSchema convertINT96(PrimitiveTypeName
primitiveTypeName) {
if (readInt96AsFixed) {
- return HoodieSchema.createFixed("INT96", "INT96 represented as
byte[12]", null, 12);
+ return HoodieSchema.createFixed("INT96", null, "INT96
represented as byte[12]", 12);
}
throw new IllegalArgumentException(
"INT96 is deprecated. As interim enable READ_INT96_AS_FIXED
flag to read as byte array.");
@@ -424,7 +430,9 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
return HoodieSchema.createUUID();
} else {
int size = parquetType.asPrimitiveType().getTypeLength();
- return HoodieSchema.createFixed(parquetType.getName(), null,
null, size);
+ String name = parquetType.getName();
+ String ns = nextDedupNamespace(name, names);
+ return HoodieSchema.createFixed(name, ns, null, size);
}
}
@Override
@@ -528,10 +536,12 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
*
* @param annotation Parquet logical type annotation
* @param parquetType The parquet type containing the annotation
+ * @param names counts of record and fixed names seen so far, used to
namespace repeated names
* @return HoodieSchema with the logical type applied, or null if no logical
type conversion is needed
* @throws IllegalArgumentException if the logical type is not compatible
with the primitive type
*/
- private HoodieSchema
convertLogicalTypeAnnotationToHoodieSchema(LogicalTypeAnnotation annotation,
Type parquetType) {
+ private HoodieSchema
convertLogicalTypeAnnotationToHoodieSchema(LogicalTypeAnnotation annotation,
Type parquetType,
+ Map<String,
Integer> names) {
if (annotation == null) {
return null;
}
@@ -545,7 +555,8 @@ public class AvroSchemaConverterWithTimestampNTZ extends
HoodieAvroParquetSchema
if (primitiveType == FIXED_LEN_BYTE_ARRAY) {
int fixedSize = parquetType.asPrimitiveType().getTypeLength();
String name = parquetType.getName();
- return java.util.Optional.of(HoodieSchema.createDecimal(name, null,
null,
+ String ns = nextDedupNamespace(name, names);
+ return java.util.Optional.of(HoodieSchema.createDecimal(name, ns,
null,
decimalLogicalType.getPrecision(),
decimalLogicalType.getScale(), fixedSize));
} else if (primitiveType == BINARY) {
return
java.util.Optional.of(HoodieSchema.createDecimal(decimalLogicalType.getPrecision(),
decimalLogicalType.getScale()));
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/parquet/avro/TestAvroSchemaConverter.java
b/hudi-hadoop-common/src/test/java/org/apache/parquet/avro/TestAvroSchemaConverter.java
index 1c468f5320c7..5167923ecb83 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/parquet/avro/TestAvroSchemaConverter.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/parquet/avro/TestAvroSchemaConverter.java
@@ -24,6 +24,7 @@ import
org.apache.hudi.common.schema.HoodieSchemaCompatibility;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
@@ -549,7 +550,7 @@ public class TestAvroSchemaConverter {
Configuration enableInt96ReadingConfig = new Configuration();
enableInt96ReadingConfig.setBoolean(AvroReadSupport.READ_INT96_AS_FIXED,
true);
- HoodieSchema int96schema = HoodieSchema.createFixed("INT96", "INT96
represented as byte[12]", null, 12);
+ HoodieSchema int96schema = HoodieSchema.createFixed("INT96", null, "INT96
represented as byte[12]", 12);
HoodieSchema schema = HoodieSchema.createRecord("myrecord", null, null,
false,
Collections.singletonList(HoodieSchemaField.of("int96_field",
int96schema, null, null)));
@@ -846,6 +847,70 @@ public class TestAvroSchemaConverter {
testParquetToAvroConversion(NEW_BEHAVIOR, schema, parquetSchema);
}
+ @Test
+ public void testReuseNamesArrays() throws Exception {
+ HoodieSchema a1 = HoodieSchema.createRecord("array", null, null, false,
+ Collections.singletonList(HoodieSchemaField.of("a4",
HoodieSchema.create(HoodieSchemaType.FLOAT))));
+ HoodieSchema a2 = HoodieSchema.createFixed("array", "array2", null, 1);
+ HoodieSchema a3 = HoodieSchema.createFixed("array", "array3", null, 1);
+ HoodieSchema schema = HoodieSchema.createRecord("Message", null, null,
false,
+ Arrays.asList(
+ HoodieSchemaField.of("a1", HoodieSchema.createArray(a1)),
+ HoodieSchemaField.of("a2", HoodieSchema.createArray(a2)),
+ HoodieSchemaField.of("a3", HoodieSchema.createArray(a3))));
+
+ String parquetSchema = "message Message {\n"
+ + " required group a1 (LIST) {\n"
+ + " repeated group array {\n"
+ + " required float a4;\n"
+ + " }\n"
+ + " }\n"
+ + " required group a2 (LIST) {\n"
+ + " repeated fixed_len_byte_array(1) array;\n"
+ + " }\n"
+ + " required group a3 (LIST) {\n"
+ + " repeated fixed_len_byte_array(1) array;\n"
+ + " }\n"
+ + "}\n";
+
+ testParquetToAvroConversion(schema, parquetSchema);
+ testParquetToAvroConversion(NEW_BEHAVIOR, schema, parquetSchema);
+ }
+
+ @Test
+ public void testReuseNamesOnFixedSurvivesStringRoundTrip() {
+ // Shredded-variant shape (GH-20307): a DECIMAL fixed typed_value nested
under a typed_value record,
+ // plus two same-named DECIMAL fixed leaves of different sizes.
+ HoodieSchema schema = getAvroSchemaConverter(new
Configuration()).convert(MessageTypeParser.parseMessageType(
+ "message m {\n"
+ + " optional group payload {\n"
+ + " required binary metadata;\n"
+ + " optional binary value;\n"
+ + " optional group typed_value {\n"
+ + " required group k10 {\n"
+ + " optional binary value;\n"
+ + " optional fixed_len_byte_array(8) typed_value
(DECIMAL(18,12));\n"
+ + " }\n"
+ + " }\n"
+ + " }\n"
+ + " optional group price { optional fixed_len_byte_array(5)
amount (DECIMAL(10,2)); }\n"
+ + " optional group tax { optional fixed_len_byte_array(8) amount
(DECIMAL(18,4)); }\n"
+ + "}\n"));
+
+ Schema avroSchema = schema.toAvroSchema();
+ assertEquals(avroSchema, new Schema.Parser().parse(avroSchema.toString()));
+
+ HoodieSchema decimalLeaf = nonNullField(nonNullField(nonNullField(schema,
"payload"), "typed_value"), "k10")
+ .getField("typed_value").get().schema().getNonNullType();
+ assertEquals("typed_value2.typed_value", decimalLeaf.getFullName());
+ assertEquals("amount", nonNullField(nonNullField(schema, "price"),
"amount").getFullName());
+ assertEquals("amount2.amount", nonNullField(nonNullField(schema, "tax"),
"amount").getFullName());
+ }
+
+ private static HoodieSchema nonNullField(HoodieSchema record, String
fieldName) {
+ return record.getField(fieldName).get().schema().getNonNullType();
+ }
+
@Test
public void testUUIDType() throws Exception {
HoodieSchema fromSchema = HoodieSchema.createRecord(
