Fix resolve error caused by avro record collision (#161)

Author: jogrogan

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroConverter.java

@@ -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()) {
@@ -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());
@@ -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:

hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroConverterTest.java

@@ -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;
@@ -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());
@@ -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());
@@ -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());
@@ -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);
+  }
 }