DRILL-4733: max(dir0) reading more columns than necessary

exec/java-exec/src/main/java/org/apache/drill/exec/store/ImplicitColumnExplorer.java

@@ -135,12 +135,6 @@ private void init() {
           tableColumns.add(column);
         }
       }
-
-      // We must make sure to pass a table column(not to be confused with partition column) to the underlying record
-      // reader.
-      if (tableColumns.size() == 0) {
-        tableColumns.add(AbstractRecordReader.STAR_COLUMN);
-      }
     }
   }
 

exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java

@@ -41,6 +41,7 @@
 import org.apache.drill.exec.record.CloseableRecordBatch;
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.store.AbstractRecordReader;
 import org.apache.drill.exec.store.ImplicitColumnExplorer;
 import org.apache.drill.exec.store.RecordReader;
 import org.apache.drill.exec.store.RecordWriter;
@@ -126,8 +127,12 @@ CloseableRecordBatch getReaderBatch(FragmentContext context, EasySubScan scan) t
     final ImplicitColumnExplorer columnExplorer = new ImplicitColumnExplorer(context, scan.getColumns());
 
     if (!columnExplorer.isSelectAllColumns()) {
+      // We must make sure to pass a table column (not to be confused with implicit column) to the underlying record reader.
+      List<SchemaPath> tableColumns =
+          columnExplorer.getTableColumns().size() == 0 ?
+              Lists.<SchemaPath>newArrayList(AbstractRecordReader.STAR_COLUMN) : columnExplorer.getTableColumns();
       scan = new EasySubScan(scan.getUserName(), scan.getWorkUnits(), scan.getFormatPlugin(),
-          columnExplorer.getTableColumns(), scan.getSelectionRoot());
+          tableColumns, scan.getSelectionRoot());
       scan.setOperatorId(scan.getOperatorId());
     }
 

exec/java-exec/src/test/java/org/apache/drill/exec/store/TestImplicitFileColumns.java

@@ -20,6 +20,7 @@
 import com.google.common.base.Charsets;
 import com.google.common.io.Files;
 import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.util.TestTools;
 import org.apache.drill.exec.util.JsonStringArrayList;
 import org.apache.drill.exec.util.Text;
 import org.apache.hadoop.fs.Path;
@@ -110,4 +111,20 @@ public void testImplicitColumnsForParquet() throws Exception {
         .go();
   }
 
+  @Test // DRILL-4733
+  public void testMultilevelParquetWithSchemaChange() throws Exception {
+    try {
+      test("alter session set `planner.enable_decimal_data_type` = true");
+      testBuilder()
+          .sqlQuery(String.format("select max(dir0) as max_dir from dfs_test.`%s/src/test/resources/multilevel/parquetWithSchemaChange`",
+              TestTools.getWorkingPath()))
+          .unOrdered()
+          .baselineColumns("max_dir")
+          .baselineValues("voter50")
+          .go();
+    } finally {
+      test("alter session set `planner.enable_decimal_data_type` = false");
+    }
+  }
+
 }