Compatible Record In Union Lookup Should Use Unqualified Name (#574)

* Compatible record in union should use unqualified name

* Fix logic based on avro version

* add codegen

* Fix some additional union lookup logic which we missed

* Fix test

* add codegen files

---------

Co-authored-by: Amit Patil <[email protected]>

Author: pamitm

fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/FastGenericDeserializerGeneratorTest.java

@@ -33,16 +33,13 @@
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.BinaryDecoder;
-import org.apache.avro.io.BinaryEncoder;
 import org.apache.avro.io.Decoder;
-import org.apache.avro.specific.SpecificDatumWriter;
 import org.apache.avro.util.Utf8;
 import org.testng.Assert;
 import org.testng.SkipException;
 import org.testng.annotations.BeforeTest;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
-import org.testng.collections.Lists;
 import org.testng.internal.collections.Pair;
 
 import static com.linkedin.avro.fastserde.FastSerdeTestsSupport.*;
@@ -256,6 +253,154 @@ public void shouldReadFixed(Implementation implementation) {
         ((List<GenericData.Fixed>) record.get("testFixedUnionArray")).get(0).bytes());
   }
 
+  @Test(groups = {"deserializationTest"}, dataProvider = "Implementation")
+  public void shouldHandleDiffNamepspaceInRecords(Implementation implementation) {
+    //record with two fields, first optional string second a record with namespace "a.b.c" with name "innerRecordName". Inner record has one int field
+    Schema writerSchema  = Schema.parse("{\n" +
+            "  \"type\": \"record\",\n" +
+            "  \"name\": \"OuterRecord\",\n" +
+            "  \"fields\": [\n" +
+            "    {\n" +
+            "      \"name\": \"optionalString\",\n" +
+            "      \"type\": [\"null\", \"string\"],\n" +
+            "      \"default\": null\n" +
+            "    },\n" +
+            "    {\n" +
+            "      \"name\": \"innerRecord\",\n" +
+            "      \"type\": {\n" +
+            "        \"type\": \"record\",\n" +
+            "        \"name\": \"innerRecordName\",\n" +
+            "        \"namespace\": \"a.b.c\",\n" +
+            "        \"fields\": [\n" +
+            "          {\n" +
+            "            \"name\": \"intField\",\n" +
+            "            \"type\": \"int\"\n" +
+            "          }\n" +
+            "        ]\n" +
+            "      }\n" +
+            "    }\n" +
+            "  ]\n" +
+            "}");
+
+    //change namespace on inner record to "d.e.f"
+    Schema readerSchema = Schema.parse("{\n" +
+            "  \"type\": \"record\",\n" +
+            "  \"name\": \"OuterRecord\",\n" +
+            "  \"fields\": [\n" +
+            "    {\n" +
+            "      \"name\": \"optionalString\",\n" +
+            "      \"type\": [\"null\", \"string\"],\n" +
+            "      \"default\": null\n" +
+            "    },\n" +
+            "    {\n" +
+            "      \"name\": \"innerRecord\",\n" +
+            "      \"type\": [\"null\", {\n" +
+            "        \"type\": \"record\",\n" +
+            "        \"name\": \"innerRecordName\",\n" +
+            "        \"namespace\": \"d.e.f\",\n" +
+            "        \"fields\": [\n" +
+            "          {\n" +
+            "            \"name\": \"intField\",\n" +
+            "            \"type\": \"int\"\n" +
+            "          }\n" +
+            "        ]\n" +
+            "      }],\n" +
+            "      \"default\": null\n" +
+            "    }\n" +
+            "  ]\n" +
+            "}");
+
+    GenericRecord originalRecord = new GenericData.Record(writerSchema);
+    originalRecord.put("optionalString", "abc");
+    GenericRecord innerRecord = new GenericData.Record(writerSchema.getField("innerRecord").schema());
+    innerRecord.put("intField", 1);
+    originalRecord.put("innerRecord", innerRecord);
+
+    // when
+    try{
+      GenericRecord record = implementation.decode(writerSchema, readerSchema, genericDataAsDecoder(originalRecord));
+      // then
+      if(Utils.usesQualifiedNameForNamedTypedMatching()){
+        Assert.fail("1.5-1.7 don't support unqualified name for named type matching so we should have failed");
+      }
+      Assert.assertEquals(new Utf8("abc"), record.get("optionalString"));
+      GenericRecord innerRecordDecoded = (GenericRecord) record.get("innerRecord");
+      Assert.assertEquals(1, innerRecordDecoded.get("intField"));
+    } catch (Exception e){
+      if(!Utils.usesQualifiedNameForNamedTypedMatching()) {
+        Assert.fail("1.4, and 1.8+ support unqualified name for named type matching");
+      }
+    }
+  }
+
+
+  @Test(groups = {"deserializationTest"}, dataProvider = "Implementation")
+  public void shouldNotFailOnNamespaceMismatch(Implementation implementation) throws IOException {
+    // writer-side schema: "metadata" has NO namespace
+    String writerSchemaStr = "{\n" +
+            "  \"type\": \"record\",\n" +
+            "  \"name\": \"wrapper\",\n" +
+            "  \"fields\": [\n" +
+            "    {\n" +
+            "      \"name\": \"metadata\",\n" +
+            "      \"type\": [\"null\", {\n" +
+            "        \"type\": \"record\",\n" +
+            "        \"name\": \"metadata\",\n" +
+            "        \"fields\": [\n" +
+            "          {\"name\": \"fieldName\", \"type\": [\"null\", \"string\"], \"default\": null}\n" +
+            "        ]\n" +
+            "      }],\n" +
+            "      \"default\": null\n" +
+            "    }\n" +
+            "  ]\n" +
+            "}";
+
+    // reader-side schema: same nested record but WITH namespace "rtapi.surge"
+    String readerSchemaStr = "{\n" +
+            "  \"type\": \"record\",\n" +
+            "  \"name\": \"Wrapper\",\n" +
+            "  \"fields\": [\n" +
+            "    {\n" +
+            "      \"name\": \"metadata\",\n" +
+            "      \"type\": [\"null\", {\n" +
+            "        \"type\": \"record\",\n" +
+            "        \"name\": \"metadata\",\n" +
+            "        \"namespace\": \"some.other.namespace\",\n" +
+            "        \"fields\": [\n" +
+            "          {\"name\": \"fieldName\", \"type\": [\"null\", \"string\"], \"default\": null}\n" +
+            "        ]\n" +
+            "      }],\n" +
+            "      \"default\": null\n" +
+            "    }\n" +
+            "  ]\n" +
+            "}";
+
+    Schema writerSchema = AvroCompatibilityHelper.parse(writerSchemaStr);
+    Schema readerSchema = AvroCompatibilityHelper.parse(readerSchemaStr);
+
+    // Build a writer-side record instance
+    GenericRecord wrapper = new GenericData.Record(writerSchema);
+    Schema metadataSchema = writerSchema.getField("metadata").schema().getTypes().get(1);
+    GenericRecord metadataRecord = new GenericData.Record(metadataSchema);
+    metadataRecord.put("fieldName", "abc-123");
+    wrapper.put("metadata", metadataRecord);
+
+    // Attempt to deserialize – should throw AvroTypeException because of namespace mismatch
+    try{
+      GenericRecord record = implementation.decode(writerSchema, readerSchema, genericDataAsDecoder(wrapper));
+      if(Utils.usesQualifiedNameForNamedTypedMatching()){
+        Assert.fail("1.5-1.7 don't support unqualified name for named type matching so we should have failed");
+      }
+      Assert.assertEquals(((GenericRecord)record.get("metadata")).get("fieldName").toString(), "abc-123");
+    } catch (AvroTypeException e){
+        if(!Utils.usesQualifiedNameForNamedTypedMatching()) {
+            Assert.fail("1.4, and 1.8+ support unqualified name for named type matching");
+        }
+        // expected exception
+    }
+
+  }
+
   @Test(groups = {"deserializationTest"}, dataProvider = "Implementation")
   public void shouldReadEnum(Implementation implementation) {
     // given

fastserde/avro-fastserde-tests111/build/codegen/java/com/linkedin/avro/fastserde/generated/deserialization/AVRO_1_11/OuterRecord_GenericDeserializer_998347834_1261326440.java

@@ -0,0 +1,84 @@
+
+package com.linkedin.avro.fastserde.generated.deserialization.AVRO_1_11;
+
+import java.io.IOException;
+import com.linkedin.avro.fastserde.FastDeserializer;
+import com.linkedin.avro.fastserde.customized.DatumReaderCustomization;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.util.Utf8;
+
+public class OuterRecord_GenericDeserializer_998347834_1261326440
+    implements FastDeserializer<IndexedRecord>
+{
+
+    private final Schema readerSchema;
+    private final Schema optionalString0;
+    private final Schema innerRecord0;
+    private final Schema innerRecordNameRecordSchema0;
+
+    public OuterRecord_GenericDeserializer_998347834_1261326440(Schema readerSchema) {
+        this.readerSchema = readerSchema;
+        this.optionalString0 = readerSchema.getField("optionalString").schema();
+        this.innerRecord0 = readerSchema.getField("innerRecord").schema();
+        this.innerRecordNameRecordSchema0 = innerRecord0 .getTypes().get(1);
+    }
+
+    public IndexedRecord deserialize(IndexedRecord reuse, Decoder decoder, DatumReaderCustomization customization)
+        throws IOException
+    {
+        return deserializeOuterRecord0((reuse), (decoder), (customization));
+    }
+
+    public IndexedRecord deserializeOuterRecord0(Object reuse, Decoder decoder, DatumReaderCustomization customization)
+        throws IOException
+    {
+        IndexedRecord outerRecord0;
+        if ((((reuse)!= null)&&((reuse) instanceof IndexedRecord))&&(((IndexedRecord)(reuse)).getSchema() == readerSchema)) {
+            outerRecord0 = ((IndexedRecord)(reuse));
+        } else {
+            outerRecord0 = new org.apache.avro.generic.GenericData.Record(readerSchema);
+        }
+        int unionIndex0 = (decoder.readIndex());
+        if (unionIndex0 == 0) {
+            decoder.readNull();
+            outerRecord0 .put(0, null);
+        } else {
+            if (unionIndex0 == 1) {
+                Utf8 charSequence0;
+                Object oldString0 = outerRecord0 .get(0);
+                if (oldString0 instanceof Utf8) {
+                    charSequence0 = (decoder).readString(((Utf8) oldString0));
+                } else {
+                    charSequence0 = (decoder).readString(null);
+                }
+                outerRecord0 .put(0, charSequence0);
+            } else {
+                throw new RuntimeException(("Illegal union index for 'optionalString': "+ unionIndex0));
+            }
+        }
+        populate_OuterRecord0((outerRecord0), (customization), (decoder));
+        return outerRecord0;
+    }
+
+    private void populate_OuterRecord0(IndexedRecord outerRecord0, DatumReaderCustomization customization, Decoder decoder)
+        throws IOException
+    {
+        outerRecord0 .put(1, deserializeinnerRecordName0(outerRecord0 .get(1), (decoder), (customization)));
+    }
+
+    public IndexedRecord deserializeinnerRecordName0(Object reuse, Decoder decoder, DatumReaderCustomization customization)
+        throws IOException
+    {
+        IndexedRecord innerRecordName0;
+        if ((((reuse)!= null)&&((reuse) instanceof IndexedRecord))&&(((IndexedRecord)(reuse)).getSchema() == innerRecordNameRecordSchema0)) {
+            innerRecordName0 = ((IndexedRecord)(reuse));
+        } else {
+            innerRecordName0 = new org.apache.avro.generic.GenericData.Record(innerRecordNameRecordSchema0);
+        }
+        innerRecordName0 .put(0, (decoder.readInt()));
+        return innerRecordName0;
+    }
+
+}

fastserde/avro-fastserde-tests111/build/codegen/java/com/linkedin/avro/fastserde/generated/deserialization/AVRO_1_11/Wrapper_GenericDeserializer_1221519218_203960580.java

@@ -0,0 +1,87 @@
+
+package com.linkedin.avro.fastserde.generated.deserialization.AVRO_1_11;
+
+import java.io.IOException;
+import com.linkedin.avro.fastserde.FastDeserializer;
+import com.linkedin.avro.fastserde.customized.DatumReaderCustomization;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.util.Utf8;
+
+public class Wrapper_GenericDeserializer_1221519218_203960580
+    implements FastDeserializer<IndexedRecord>
+{
+
+    private final Schema readerSchema;
+    private final Schema metadata0;
+    private final Schema metadataOptionSchema0;
+    private final Schema productTypeUUID0;
+
+    public Wrapper_GenericDeserializer_1221519218_203960580(Schema readerSchema) {
+        this.readerSchema = readerSchema;
+        this.metadata0 = readerSchema.getField("metadata").schema();
+        this.metadataOptionSchema0 = metadata0 .getTypes().get(1);
+        this.productTypeUUID0 = metadataOptionSchema0 .getField("productTypeUUID").schema();
+    }
+
+    public IndexedRecord deserialize(IndexedRecord reuse, Decoder decoder, DatumReaderCustomization customization)
+        throws IOException
+    {
+        return deserializeWrapper0((reuse), (decoder), (customization));
+    }
+
+    public IndexedRecord deserializeWrapper0(Object reuse, Decoder decoder, DatumReaderCustomization customization)
+        throws IOException
+    {
+        IndexedRecord wrapper0;
+        if ((((reuse)!= null)&&((reuse) instanceof IndexedRecord))&&(((IndexedRecord)(reuse)).getSchema() == readerSchema)) {
+            wrapper0 = ((IndexedRecord)(reuse));
+        } else {
+            wrapper0 = new org.apache.avro.generic.GenericData.Record(readerSchema);
+        }
+        int unionIndex0 = (decoder.readIndex());
+        if (unionIndex0 == 0) {
+            decoder.readNull();
+            wrapper0 .put(0, null);
+        } else {
+            if (unionIndex0 == 1) {
+                wrapper0 .put(0, deserializemetadata0(wrapper0 .get(0), (decoder), (customization)));
+            } else {
+                throw new RuntimeException(("Illegal union index for 'metadata': "+ unionIndex0));
+            }
+        }
+        return wrapper0;
+    }
+
+    public IndexedRecord deserializemetadata0(Object reuse, Decoder decoder, DatumReaderCustomization customization)
+        throws IOException
+    {
+        IndexedRecord metadata1;
+        if ((((reuse)!= null)&&((reuse) instanceof IndexedRecord))&&(((IndexedRecord)(reuse)).getSchema() == metadataOptionSchema0)) {
+            metadata1 = ((IndexedRecord)(reuse));
+        } else {
+            metadata1 = new org.apache.avro.generic.GenericData.Record(metadataOptionSchema0);
+        }
+        int unionIndex1 = (decoder.readIndex());
+        if (unionIndex1 == 0) {
+            decoder.readNull();
+            metadata1 .put(0, null);
+        } else {
+            if (unionIndex1 == 1) {
+                Utf8 charSequence0;
+                Object oldString0 = metadata1 .get(0);
+                if (oldString0 instanceof Utf8) {
+                    charSequence0 = (decoder).readString(((Utf8) oldString0));
+                } else {
+                    charSequence0 = (decoder).readString(null);
+                }
+                metadata1 .put(0, charSequence0);
+            } else {
+                throw new RuntimeException(("Illegal union index for 'productTypeUUID': "+ unionIndex1));
+            }
+        }
+        return metadata1;
+    }
+
+}