do not add suffixes unnecessarily

Author: ruolin59

coral-schema/src/main/java/com/linkedin/coral/schema/avro/SchemaUtilities.java

@@ -748,20 +748,23 @@ private static Schema setupNestedNamespaceForRecord(@Nonnull Schema schema, @Non
    * @return Map from record name -> ordered list of original namespaces (only for records with collisions)
    */
   private static Map<String, List<String>> detectNamespaceCollisions(@Nonnull Schema schema) {
-    Map<String, List<String>> recordNameToOriginalNamespaces = new LinkedHashMap<>();
+    Map<String, List<String>> recordKeyToOriginalNamespaces = new LinkedHashMap<>();
 
-    // Scan all fields to collect record types and their original namespaces in order
+    // Scan all fields to collect record types with their parent paths
     for (Schema.Field field : schema.getFields()) {
-      collectRecordTypes(field.schema(), recordNameToOriginalNamespaces);
+      collectRecordTypes(field.schema(), "", recordKeyToOriginalNamespaces);
     }
 
-    // Filter to only records that have collisions (multiple different original namespaces)
+    // Build collision map: record name -> list of original namespaces (only for actual collisions)
     Map<String, List<String>> collisions = new LinkedHashMap<>();
-    for (Map.Entry<String, List<String>> entry : recordNameToOriginalNamespaces.entrySet()) {
+    for (Map.Entry<String, List<String>> entry : recordKeyToOriginalNamespaces.entrySet()) {
       // Use a Set to check uniqueness while preserving order in List
       Set<String> uniqueNamespaces = new LinkedHashSet<>(entry.getValue());
       if (uniqueNamespaces.size() > 1) {
-        collisions.put(entry.getKey(), entry.getValue());
+        // Extract just the record name from the key (format: "parentPath::recordName")
+        String key = entry.getKey();
+        String recordName = key.substring(key.lastIndexOf("::") + 2);
+        collisions.put(recordName, entry.getValue());
       }
     }
 
@@ -771,32 +774,40 @@ private static Map<String, List<String>> detectNamespaceCollisions(@Nonnull Sche
   /**
    * Recursively collects all record types and their original namespaces from a schema.
    * This traverses through unions, arrays, and maps to find all nested record types.
-   * The order of collection matches the order records appear in the schema.
+   * Records are keyed by their parent path to ensure only records at the same hierarchical
+   * level are considered for collision detection.
    * 
    * @param schema The schema to scan
-   * @param recordNameToNamespaces Map to populate with record name -> ordered list of original namespaces
+   * @param parentPath The hierarchical path to this schema element (e.g., "Parent.Child")
+   * @param recordKeyToNamespaces Map to populate with (parentPath::recordName) -> ordered list of original namespaces
    */
-  private static void collectRecordTypes(@Nonnull Schema schema,
-      @Nonnull Map<String, List<String>> recordNameToNamespaces) {
+  private static void collectRecordTypes(@Nonnull Schema schema, @Nonnull String parentPath,
+      @Nonnull Map<String, List<String>> recordKeyToNamespaces) {
     switch (schema.getType()) {
       case RECORD:
         String originalNamespace = schema.getNamespace() != null ? schema.getNamespace() : "";
-        recordNameToNamespaces.computeIfAbsent(schema.getName(), k -> new ArrayList<>()).add(originalNamespace);
-        // Recursively collect records from this record's fields to detect deeply nested collisions
+        String recordName = schema.getName();
+        // Create a unique key combining parent path and record name
+        String key = parentPath + "::" + recordName;
+        recordKeyToNamespaces.computeIfAbsent(key, k -> new ArrayList<>()).add(originalNamespace);
+        
+        // Recursively collect records from this record's fields
+        // Update parent path to include this record
+        String newParentPath = parentPath.isEmpty() ? recordName : parentPath + "." + recordName;
         for (Schema.Field field : schema.getFields()) {
-          collectRecordTypes(field.schema(), recordNameToNamespaces);
+          collectRecordTypes(field.schema(), newParentPath, recordKeyToNamespaces);
         }
         break;
       case UNION:
         for (Schema type : schema.getTypes()) {
-          collectRecordTypes(type, recordNameToNamespaces);
+          collectRecordTypes(type, parentPath, recordKeyToNamespaces);
         }
         break;
       case ARRAY:
-        collectRecordTypes(schema.getElementType(), recordNameToNamespaces);
+        collectRecordTypes(schema.getElementType(), parentPath, recordKeyToNamespaces);
         break;
       case MAP:
-        collectRecordTypes(schema.getValueType(), recordNameToNamespaces);
+        collectRecordTypes(schema.getValueType(), parentPath, recordKeyToNamespaces);
         break;
       case ENUM:
       case BOOLEAN:

coral-schema/src/test/java/com/linkedin/coral/schema/avro/SchemaUtilitiesTests.java

@@ -214,11 +214,17 @@ public void testSetupNameAndNamespaceDetectsDeeplyNestedCollisions() {
     Schema collidingRecord2 = SchemaBuilder.record("CollidingRecord").namespace("com.bar.v2").fields().name("field2")
         .type().intType().noDefault().endRecord();
 
+    // Create a "Metadata" record that appears at a different hierarchical level
+    // This should NOT get a suffix since it's not colliding with anything at its level
+    Schema metadataRecord = SchemaBuilder.record("Metadata").namespace("com.original").fields().name("version")
+        .type().stringType().noDefault().endRecord();
+
     // Create an intermediate record that contains both colliding records
     // This represents the middle layer in the nesting hierarchy
     Schema intermediateRecord =
         SchemaBuilder.record("IntermediateRecord").namespace("com.intermediate").fields().name("collidingField1")
-            .type(collidingRecord1).noDefault().name("collidingField2").type(collidingRecord2).noDefault().endRecord();
+            .type(collidingRecord1).noDefault().name("collidingField2").type(collidingRecord2).noDefault()
+            .name("metadata").type(metadataRecord).noDefault().endRecord();
 
     // Create top-level parent schema that contains the intermediate record
     Schema parentSchema = SchemaBuilder.record("ParentRecord").namespace("com.parent").fields()
@@ -233,14 +239,17 @@ public void testSetupNameAndNamespaceDetectsDeeplyNestedCollisions() {
 
     Schema.Field collidingField1 = intermediateSchema.getField("collidingField1");
     Schema.Field collidingField2 = intermediateSchema.getField("collidingField2");
+    Schema.Field metadataField = intermediateSchema.getField("metadata");
 
     Schema resultColliding1 = collidingField1.schema();
     Schema resultColliding2 = collidingField2.schema();
+    Schema resultMetadata = metadataField.schema();
 
     String namespace1 = resultColliding1.getNamespace();
     String namespace2 = resultColliding2.getNamespace();
+    String metadataNamespace = resultMetadata.getNamespace();
 
-    // Both records have the same name
+    // Both colliding records have the same name
     Assert.assertEquals(resultColliding1.getName(), "CollidingRecord");
     Assert.assertEquals(resultColliding2.getName(), "CollidingRecord");
 
@@ -254,5 +263,13 @@ public void testSetupNameAndNamespaceDetectsDeeplyNestedCollisions() {
         "First colliding record namespace should have numeric suffix. Got: " + namespace1);
     Assert.assertTrue(namespace2.endsWith("-0") || namespace2.endsWith("-1"),
         "Second colliding record namespace should have numeric suffix. Got: " + namespace2);
+
+    // Verify that the non-colliding Metadata record does NOT have a numeric suffix
+    Assert.assertEquals(resultMetadata.getName(), "Metadata");
+    Assert.assertFalse(metadataNamespace.matches(".*-\\d+$"),
+        "Metadata record should NOT have numeric suffix since it's not colliding at its level. Got: "
+            + metadataNamespace);
+    Assert.assertTrue(metadataNamespace.contains("IntermediateRecord"),
+        "Metadata namespace should follow hierarchical naming. Got: " + metadataNamespace);
   }
 }