DRILL-7741: Columns are missing when using convert_from function

closes #2081

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java

@@ -332,6 +332,16 @@ protected IterOutcome handleNullInput() {
     return IterOutcome.OK_NEW_SCHEMA;
   }
 
+  @Override
+  protected IterOutcome getFinalOutcome(boolean hasMoreRecordInBoundary) {
+    // In a case of complex writers vectors are added at runtime, so the schema
+    // may change (e.g. when a batch contains new column(s) not present in previous batches)
+    if (complexWriters != null) {
+      return IterOutcome.OK_NEW_SCHEMA;
+    }
+    return super.getFinalOutcome(hasMoreRecordInBoundary);
+  }
+
   private void setupNewSchema(RecordBatch incomingBatch, int configuredBatchSize) {
     memoryManager = new ProjectMemoryManager(configuredBatchSize);
     memoryManager.init(incomingBatch, ProjectRecordBatch.this);

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

@@ -782,4 +782,26 @@ public void testUntypedPathWithUnion() throws Exception {
       resetSessionOption(ExecConstants.ENABLE_UNION_TYPE_KEY);
     }
   }
+
+  @Test
+  public void testConvertFromJson() throws Exception {
+    String fileName = "table.tsv";
+    try (BufferedWriter writer = new BufferedWriter(new FileWriter(new File(dirTestWatcher.getRootDir(), fileName)))) {
+      for (int i = 0; i < JSONRecordReader.DEFAULT_ROWS_PER_BATCH; i++) {
+        writer.write("{\"id\":\"1\"}\n");
+      }
+      writer.write("{\"id\":\"2\",\"v\":[\"abc\"]}");
+    }
+
+    String sql = "SELECT t.m.id AS id, t.m.v[0] v FROM \n" +
+        "(SELECT convert_from(columns[0], 'json') AS m FROM dfs.`%s`) t\n" +
+        "where t.m.id='2'";
+
+    testBuilder()
+        .sqlQuery(sql, fileName)
+        .unOrdered()
+        .baselineColumns("id", "v")
+        .baselineValues("2", "abc")
+        .go();
+  }
 }