Skip to content

Fix resolve error caused by avro record collision #161

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jogrogan merged 1 commit into from
Aug 20, 2025
Merged

Fix resolve error caused by avro record collision #161

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,14 @@ private AvroConverter() {
}

public static Schema avro(String namespace, String name, RelDataType dataType) {
// TODO: Schema generation is overly verbose today and does not support reuse, hence why we always defined a new namespace for records.
// Ideally information should be extracted from the RelDataType to allow collapsing of records into one definition.
String newNamespace = namespace + "." + name;
if (dataType.isStruct()) {
List<Schema.Field> fields = dataType.getFieldList().stream()
.map(x -> new Schema.Field(sanitize(x.getName()), avro(namespace, x.getName(), x.getType()), describe(x), null))
.map(x -> new Schema.Field(sanitize(x.getName()), avro(newNamespace, x.getName(), x.getType()), describe(x), null))
.collect(Collectors.toList());
return createAvroSchemaWithNullability(Schema.createRecord(sanitize(name), dataType.toString(), namespace, false, fields),
return createAvroSchemaWithNullability(Schema.createRecord(sanitize(name), dataType.toString(), newNamespace, false, fields),
dataType.isNullable());
} else {
switch (dataType.getSqlTypeName()) {
Expand All @@ -52,7 +55,7 @@ public static Schema avro(String namespace, String name, RelDataType dataType) {
case BINARY:
case VARBINARY:
if (dataType.getPrecision() != -1) {
return createAvroSchemaWithNullability(Schema.createFixed(sanitize(name), dataType.toString(), namespace,
return createAvroSchemaWithNullability(Schema.createFixed(sanitize(name), dataType.toString(), newNamespace,
dataType.getPrecision()), dataType.isNullable());
} else {
return createAvroTypeWithNullability(Schema.Type.BYTES, dataType.isNullable());
Expand All @@ -63,12 +66,12 @@ public static Schema avro(String namespace, String name, RelDataType dataType) {
return createAvroTypeWithNullability(Schema.Type.BOOLEAN, dataType.isNullable());
case ARRAY:
return createAvroSchemaWithNullability(
Schema.createArray(avro(null, sanitize(name) + "ArrayElement",
Schema.createArray(avro(newNamespace, sanitize(name) + "ArrayElement",
Objects.requireNonNull(dataType.getComponentType()))),
dataType.isNullable());
case MAP:
return createAvroSchemaWithNullability(
Schema.createMap(avro(null, sanitize(name) + "MapElement",
Schema.createMap(avro(newNamespace, sanitize(name) + "MapElement",
Objects.requireNonNull(dataType.getValueType()))),
dataType.isNullable());
case UNKNOWN:
Expand Down
60 changes: 55 additions & 5 deletions hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroConverterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;

import org.apache.avro.Schema;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.type.RelDataType;
Expand Down Expand Up @@ -103,7 +102,7 @@ public void testAvroKeyPayloadSchemaNoKeyOptions() {
assertNull(result.getKey()); // Key schema should be null
assertNotNull(result.getValue()); // Payload schema should not be null
assertEquals("payloadSchema", result.getValue().getName());
assertEquals("namespace", result.getValue().getNamespace());
assertEquals("namespace.payloadSchema", result.getValue().getNamespace());
assertEquals("record", result.getValue().getType().getName());
assertEquals(1, result.getValue().getFields().size());
assertEquals("field1", result.getValue().getFields().get(0).name());
Expand Down Expand Up @@ -138,14 +137,14 @@ public void testAvroKeyPayloadSchemaValidKeyOptions() {

assertNotNull(result.getKey()); // Key schema should not be null
assertEquals("keySchema", result.getKey().getName());
assertEquals("namespace", result.getKey().getNamespace());
assertEquals("namespace.keySchema", result.getKey().getNamespace());
assertEquals("record", result.getKey().getType().getName());
assertEquals(1, result.getKey().getFields().size());
assertEquals("field1", result.getKey().getFields().get(0).name()); // prefix should be stripped
assertEquals("string", result.getKey().getFields().get(0).schema().getType().getName());
assertNotNull(result.getValue()); // Payload schema should not be null
assertEquals("payloadSchema", result.getValue().getName());
assertEquals("namespace", result.getValue().getNamespace());
assertEquals("namespace.payloadSchema", result.getValue().getNamespace());
assertEquals("record", result.getValue().getType().getName());
assertEquals(1, result.getValue().getFields().size());
assertEquals("field2", result.getValue().getFields().get(0).name());
Expand All @@ -168,7 +167,7 @@ public void testAvroKeyPayloadSchemaPrimitiveKey() {
assertEquals("int", result.getKey().getType().getName());
assertNotNull(result.getValue()); // Payload schema should not be null
assertEquals("payloadSchema", result.getValue().getName());
assertEquals("namespace", result.getValue().getNamespace());
assertEquals("namespace.payloadSchema", result.getValue().getNamespace());
assertEquals("record", result.getValue().getType().getName());
assertEquals(1, result.getValue().getFields().size());
assertEquals("field1", result.getValue().getFields().get(0).name());
Expand Down Expand Up @@ -208,4 +207,55 @@ public void convertsNestedArray() {
assertEquals("field1", structElementSchema.getFields().get(0).name());
assertEquals("field2", structElementSchema.getFields().get(1).name());
}

@Test
public void handlesNamespaceInNestedArrayAndMapElements() {
RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);

// Create a "location" record type that will be reused - this mimics the real scenario
RelDataType locationType1 = typeFactory.createStructType(
List.of(typeFactory.createSqlType(SqlTypeName.VARCHAR), typeFactory.createSqlType(SqlTypeName.VARCHAR)),
List.of("countryCode", "postalCode"));

// Create another "location" record type with slightly different structure
RelDataType locationType2 = typeFactory.createStructType(
List.of(typeFactory.createSqlType(SqlTypeName.VARCHAR), typeFactory.createSqlType(SqlTypeName.INTEGER)),
List.of("countryCode", "regionCode"));

// Create structures that use these location types in different contexts
// This simulates the real scenario where multiple fields have the same name but different contexts
RelDataType profileStruct = typeFactory.createStructType(
List.of(locationType1),
List.of("location"));

RelDataType positionStruct = typeFactory.createStructType(
List.of(locationType2),
List.of("location"));

// Put both in a map structure - this creates the collision scenario
// Both will try to generate records named "location" with the same namespace
RelDataType positionsMap = typeFactory.createMapType(
typeFactory.createSqlType(SqlTypeName.VARCHAR),
positionStruct);

// Create the main record that contains both location types
RelDataType mainRecord = typeFactory.createStructType(
List.of(profileStruct, positionsMap),
List.of("profile", "positions"));

// Schema creation should succeed
Schema schema = AvroConverter.avro("com.linkedin", "MemberProfile", mainRecord);
assertNotNull(schema);

// Without the namespace-appending behavior in AvroConverter, this would fail with error "Can't redefine: com.linkedin.location"
// The issue occurs because multiple records named "location" are created with the same namespace,
// causing a collision when schema.toString(true) tries to serialize them
String schemaJson = schema.toString(true);
assertNotNull("Schema toString(true) should succeed without 'Can't redefine' errors", schemaJson);

// Verify the schema can be parsed back
Schema.Parser parser = new Schema.Parser();
Schema reparsedSchema = parser.parse(schemaJson);
assertNotNull("Generated schema must be parseable", reparsedSchema);
}
}