[GOBBLIN-1827] Add check that if nested field is optional and has a non-null default…

gobblin-utility/src/main/java/org/apache/gobblin/util/AvroFlattener.java

@@ -23,13 +23,13 @@
 
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
-import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 
 /***
  * This class provides methods to flatten an Avro Schema to make it more optimal for ORC
@@ -402,6 +402,10 @@ private List<Schema.Field> flattenField(Schema.Field f, ImmutableList<String> pa
           }
           // Wrap the Union, since parent Union is an option
           else {
+            // If the field within the parent Union has a non-null default value, then null should not be the first member
+            if (f.hasDefaultValue() && f.defaultVal() != null) {
+              isNullFirstMember = false;
+            }
             if (isNullFirstMember) {
               flattenedFieldSchema =
                   Schema.createUnion(Arrays.asList(Schema.create(Schema.Type.NULL), flattenedFieldSchema));

gobblin-utility/src/test/java/org/apache/gobblin/util/AvroFlattenerTest.java

@@ -22,6 +22,8 @@
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
+
 
 public class AvroFlattenerTest {
 
@@ -188,4 +190,30 @@ public void testRecordWithinMapWithinMap () throws IOException {
   }
 
 
+  /**
+   * Test flattening for non-null default within an Option within another Record
+   * Record R1 {
+   *  fields: {
+   *    Union [ null,
+   *            Record 2 {
+   *              field: type
+   *              default: type
+   *            }
+   *          ]
+   *    }
+   * }
+   */
+  @Test
+  public void testNonNullDefaultWithinOptionWithinRecord () throws IOException {
+
+    Schema originalSchema = readSchemaFromJsonFile("nonNullDefaultWithinOptionWithinRecord_original.json");
+    Schema expectedSchema = readSchemaFromJsonFile("nonNullDefaultWithinOptionWithinRecord_flattened.json");
+    Schema flattenedSchema = new AvroFlattener().flatten(originalSchema, false);
+    Assert.assertEquals(AvroCompatibilityHelper.getSpecificDefaultValue(
+        flattenedSchema.getField("parentFieldUnion__unionRecordMemberFieldUnion__superNestedFieldString1")).toString(),
+        "defaultString1");
+    Assert.assertEquals(flattenedSchema.toString(), expectedSchema.toString());
+  }
+
+
 }

gobblin-utility/src/test/resources/flattenAvro/nonNullDefaultWithinOptionWithinRecord_flattened.json

@@ -0,0 +1,37 @@
+{
+  "type":"record",
+  "name":"parentRecordName",
+  "fields":[
+    {
+      "name":"parentFieldUnion__unionRecordMemberFieldUnion__superNestedFieldString1",
+      "type":[
+        "string",
+        "null"
+      ],
+      "default":"defaultString1",
+      "flatten_source":"parentFieldUnion.unionRecordMemberFieldUnion.superNestedFieldString1"
+    },
+    {
+      "name":"parentFieldUnion__unionRecordMemberFieldUnion__superNestedFieldString2",
+      "type":[
+        "string",
+        "null"
+      ],
+      "default":"defaultString2",
+      "flatten_source":"parentFieldUnion.unionRecordMemberFieldUnion.superNestedFieldString2"
+    },
+    {
+      "name":"parentFieldUnion__unionRecordMemberFieldString",
+      "type":[
+        "null",
+        "string"
+      ],
+      "default":null,
+      "flatten_source":"parentFieldUnion.unionRecordMemberFieldString"
+    },
+    {
+      "name":"parentFieldInt",
+      "type":"int"
+    }
+  ]
+}

gobblin-utility/src/test/resources/flattenAvro/nonNullDefaultWithinOptionWithinRecord_original.json

@@ -0,0 +1,33 @@
+{
+  "type" : "record",
+  "name" : "parentRecordName",
+  "fields" : [ {
+    "name" : "parentFieldUnion",
+    "type" : [ "null", {
+      "type" : "record",
+      "name" : "unionRecordMember",
+      "fields" : [ {
+        "name" : "unionRecordMemberFieldUnion",
+        "type" : [ "null", {
+          "type" : "record",
+          "name" : "superNestedRecord",
+          "fields" : [ {
+            "name" : "superNestedFieldString1",
+            "type" : "string",
+            "default": "defaultString1"
+          }, {
+            "name" : "superNestedFieldString2",
+            "type" : "string",
+            "default": "defaultString2"
+          } ]
+        } ]
+      }, {
+        "name" : "unionRecordMemberFieldString",
+        "type" : "string"
+      } ]
+    } ]
+  }, {
+    "name" : "parentFieldInt",
+    "type" : "int"
+  } ]
+}