DRILL-5523: Revert if condition in UnionAllRecordBatch changed in DRI…

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java

@@ -204,7 +204,17 @@ private IterOutcome doWork() throws ClassTransformationException, IOException, S
         MajorType outputFieldType = outputFields.get(index).getType();
         MaterializedField outputField = MaterializedField.create(outputPath.getAsUnescapedPath(), outputFieldType);
 
-        if (outputFields.get(index).getPath().equals(inputPath.getAsUnescapedPath())) {
+        /*
+          todo: Fix if condition when DRILL-4824 is merged
+          If condition should be changed to:
+          `if (outputFields.get(index).getPath().equals(inputPath.getAsUnescapedPath())) {`
+          DRILL-5419 has changed condition to correct one but this caused regression (DRILL-5521).
+          Root cause is missing indication of child column in map types when it is null.
+          DRILL-4824 is re-working json reader implementation, including map types and will fix this problem.
+          Reverting condition to previous one to avoid regression till DRILL-4824 is merged.
+          Unit test - TestJsonReader.testKvgenWithUnionAll().
+         */
+        if (outputFields.get(index).getPath().equals(inputPath)) {
           ValueVector vvOut = container.addOrGet(outputField);
           TransferPair tp = vvIn.makeTransferPair(vvOut);
           transfers.add(tp);

exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestJsonReader.java

@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -721,4 +721,23 @@ public void testFlattenEmptyArrayWithUnionType() throws Exception {
       testNoResult("alter session reset `exec.enable_union_type`");
     }
   }
+
+  @Test // DRILL-5521
+  public void testKvgenWithUnionAll() throws Exception {
+    File directory = new File(BaseTestQuery.getTempDir("json/input"));
+    try {
+      directory.mkdirs();
+      String fileName = "map.json";
+      try (BufferedWriter writer = new BufferedWriter(new FileWriter(new File(directory, fileName)))) {
+        writer.write("{\"rk\": \"a\", \"m\": {\"a\":\"1\"}}");
+      }
+
+      String query = String.format("select kvgen(m) as res from (select m from dfs_test.`%s/%s` union all " +
+          "select convert_from('{\"a\" : null}' ,'json') as m from (values(1)))", directory.toPath().toString(), fileName);
+      assertEquals("Row count should match", 2, testSql(query));
+
+    } finally {
+      org.apache.commons.io.FileUtils.deleteQuietly(directory);
+    }
+  }
 }