From 5249a342f52b7b4d0f0611077cb48d53a9a00da4 Mon Sep 17 00:00:00 2001
From: Uttam Kumar <ukumar@linkedin.com>
Date: Tue, 13 Feb 2024 11:37:31 -0800
Subject: [PATCH] Added handling of union type in AvscParser to handles cases
 like Map of union (#547)

---
 .../avroutil1/parser/avsc/AvscParser.java     | 10 ++++++++
 .../avroutil1/parser/avsc/AvscParserTest.java | 15 ++++++++++++
 .../schemas/RecordWithMapOfUnion.avsc         | 24 +++++++++++++++++++
 3 files changed, 49 insertions(+)
 create mode 100644 parser/src/test/resources/schemas/RecordWithMapOfUnion.avsc

diff --git a/parser/src/main/java/com/linkedin/avroutil1/parser/avsc/AvscParser.java b/parser/src/main/java/com/linkedin/avroutil1/parser/avsc/AvscParser.java
index 0ee32cd43..d2dfed0f0 100644
--- a/parser/src/main/java/com/linkedin/avroutil1/parser/avsc/AvscParser.java
+++ b/parser/src/main/java/com/linkedin/avroutil1/parser/avsc/AvscParser.java
@@ -876,6 +876,16 @@ mapSchema, locationOf(context.getUri(), literalNode), map
                 return new LiteralOrIssue(new AvroRecordLiteral(
                     recordSchema, locationOf(context.getUri(), literalNode), map
                 ));
+            case UNION:
+                AvroUnionSchema unionSchema = (AvroUnionSchema) schema;
+                LiteralOrIssue branchValueOrIssue =
+                    parseLiteral(literalNode, unionSchema.getTypes().get(0).getSchema(), fieldName, context);
+                if (branchValueOrIssue.getIssue() == null) {
+                    return branchValueOrIssue;
+                }
+                return new LiteralOrIssue(
+                    AvscIssues.badFieldDefaultValue(locationOf(context.getUri(), literalNode), literalNode.toString(),
+                        avroType, fieldName));
             default:
                 throw new UnsupportedOperationException("dont know how to parse a " + avroType + " at " + literalNode.getStartLocation()
                         + " out of a " + literalNode.getValueType() + " (" + literalNode + ")");
diff --git a/parser/src/test/java/com/linkedin/avroutil1/parser/avsc/AvscParserTest.java b/parser/src/test/java/com/linkedin/avroutil1/parser/avsc/AvscParserTest.java
index 0d2eae3d1..1d3bd4334 100644
--- a/parser/src/test/java/com/linkedin/avroutil1/parser/avsc/AvscParserTest.java
+++ b/parser/src/test/java/com/linkedin/avroutil1/parser/avsc/AvscParserTest.java
@@ -704,6 +704,21 @@ public void testRecursiveReferenceUnionWithDefault() throws Exception {
         Assert.assertEquals(recordParseResult.getIssues().size(), 0);
     }
 
+    @Test
+    public void testMapOfUnionDefault() throws Exception {
+        String recordAvsc = TestUtil.load("schemas/RecordWithMapOfUnion.avsc");
+        AvscParser avscParser = new AvscParser();
+
+        AvscParseResult recordParseResult = avscParser.parse(recordAvsc);
+        AvroRecordSchema topLevelSchema = (AvroRecordSchema) recordParseResult.getTopLevelSchema();
+
+        // Tests we don't have unparsed default values in for of AvscUnparsedLiteral.
+        new AvscSchemaWriter().writeSingle(topLevelSchema);
+
+        Assert.assertFalse(hasUnparsedDefault(topLevelSchema));
+        Assert.assertEquals(recordParseResult.getIssues().size(), 0);
+    }
+
     /**
      * Recursively goes through all fields and their schemas to check if any field has an unparsed default value
      * @param schema
diff --git a/parser/src/test/resources/schemas/RecordWithMapOfUnion.avsc b/parser/src/test/resources/schemas/RecordWithMapOfUnion.avsc
new file mode 100644
index 000000000..8d10a7352
--- /dev/null
+++ b/parser/src/test/resources/schemas/RecordWithMapOfUnion.avsc
@@ -0,0 +1,24 @@
+{
+  "type": "record",
+  "name": "RecordWithMapOfUnion",
+  "fields": [
+    {
+      "name": "myField2",
+      "type": {
+        "type": "map",
+        "values": [
+          "string",
+          {
+            "type": "array",
+            "items": "string"
+          },
+          {
+            "type": "map",
+            "values": "string"
+          }
+        ]
+      },
+      "default": {"key1": "value1"}
+    }
+  ]
+}
\ No newline at end of file