diff --git a/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java b/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
index ead8f65..e6d3863 100644
--- a/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
+++ b/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
@@ -43,12 +43,12 @@
 import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
 import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
 import org.apache.hadoop.hive.ql.exec.vector.expressions.StringExpr;
-import org.apache.hadoop.hive.ql.util.TimestampUtils;
 import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
 import org.apache.orc.OrcProto;
 import org.apache.orc.TypeDescription;
 import org.apache.orc.TypeDescription.Category;
 import org.apache.orc.impl.reader.StripePlanner;
+import org.apache.orc.impl.reader.tree.TypeReader;
 import org.threeten.extra.chrono.HybridChronology;
 
 /**
@@ -61,9 +61,9 @@
    */
   public static class ConvertTreeReader extends TreeReader {
 
-    TreeReader fromReader;
+    TypeReader fromReader;
 
-    ConvertTreeReader(int columnId, TreeReader fromReader) throws IOException {
+    ConvertTreeReader(int columnId, TypeReader fromReader) throws IOException {
       super(columnId, null);
       this.fromReader = fromReader;
     }
@@ -249,13 +249,13 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       // Pass-thru.
       fromReader.checkEncoding(encoding);
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       // Pass-thru.
       fromReader.startStripe(planner);
     }
@@ -273,7 +273,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       // Pass-thru.
       fromReader.skipRows(items);
     }
@@ -367,7 +367,7 @@
     }
   }
 
-  private static TreeReader createFromInteger(int columnId,
+  private static TypeReader createFromInteger(int columnId,
                                               TypeDescription fileType,
                                               Context context) throws IOException {
     switch (fileType.getCategory()) {
@@ -1744,7 +1744,7 @@
     }
   }
 
-  private static TreeReader createAnyIntegerConvertTreeReader(int columnId,
+  private static TypeReader createAnyIntegerConvertTreeReader(int columnId,
                                                               TypeDescription fileType,
                                                               TypeDescription readerType,
                                                               Context context) throws IOException {
@@ -1798,7 +1798,7 @@
     }
   }
 
-  private static TreeReader createDoubleConvertTreeReader(int columnId,
+  private static TypeReader createDoubleConvertTreeReader(int columnId,
                                                           TypeDescription fileType,
                                                           TypeDescription readerType,
                                                           Context context) throws IOException {
@@ -1845,7 +1845,7 @@
     }
   }
 
-  private static TreeReader createDecimalConvertTreeReader(int columnId,
+  private static TypeReader createDecimalConvertTreeReader(int columnId,
                                                            TypeDescription fileType,
                                                            TypeDescription readerType,
                                                            Context context) throws IOException {
@@ -1891,7 +1891,7 @@
     }
   }
 
-  private static TreeReader createStringConvertTreeReader(int columnId,
+  private static TypeReader createStringConvertTreeReader(int columnId,
                                                           TypeDescription fileType,
                                                           TypeDescription readerType,
                                                           Context context) throws IOException {
@@ -1946,7 +1946,7 @@
     }
   }
 
-  private static TreeReader createTimestampConvertTreeReader(int columnId,
+  private static TypeReader createTimestampConvertTreeReader(int columnId,
                                                              TypeDescription fileType,
                                                              TypeDescription readerType,
                                                              Context context) throws IOException {
@@ -1995,7 +1995,7 @@
     }
   }
 
-  private static TreeReader createDateConvertTreeReader(int columnId,
+  private static TypeReader createDateConvertTreeReader(int columnId,
                                                         TypeDescription readerType,
                                                         Context context) throws IOException {
 
@@ -2036,7 +2036,7 @@
     }
   }
 
-  private static TreeReader createBinaryConvertTreeReader(int columnId,
+  private static TypeReader createBinaryConvertTreeReader(int columnId,
                                                           TypeDescription readerType,
                                                           Context context) throws IOException {
 
@@ -2199,7 +2199,7 @@
    * @return
    * @throws IOException
    */
-  public static TreeReader createConvertTreeReader(TypeDescription readerType,
+  public static TypeReader createConvertTreeReader(TypeDescription readerType,
                                                    Context context) throws IOException {
     final SchemaEvolution evolution = context.getSchemaEvolution();
 
diff --git a/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java b/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
index f5a65ba..947889e 100644
--- a/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
+++ b/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
@@ -48,6 +48,7 @@
 import org.apache.orc.TypeDescription;
 import org.apache.orc.impl.reader.ReaderEncryption;
 import org.apache.orc.impl.reader.StripePlanner;
+import org.apache.orc.impl.reader.tree.BatchReader;
 import org.apache.orc.util.BloomFilter;
 import org.apache.orc.util.BloomFilterIO;
 import org.slf4j.Logger;
@@ -78,7 +79,7 @@
   private int currentStripe = -1;
   private long rowBaseInStripe = 0;
   private long rowCountInStripe = 0;
-  private final TreeReaderFactory.TreeReader reader;
+  private final BatchReader reader;
   private final OrcIndex indexes;
   private final SargApplier sargApp;
   // an array about which row groups aren't skipped
@@ -227,8 +228,7 @@
           .setProlepticGregorian(fileReader.writerUsedProlepticGregorian(),
               fileReader.options.getConvertToProlepticGregorian())
           .setEncryption(encryption);
-    reader = TreeReaderFactory.createTreeReader(evolution.getReaderSchema(),
-        readerContext);
+    reader = TreeReaderFactory.createRootReader(evolution.getReaderSchema(), readerContext);
 
     int columns = evolution.getFileSchema().getMaximumId() + 1;
     indexes = new OrcIndex(new OrcProto.RowIndex[columns],
@@ -1109,7 +1109,7 @@
    * @throws IOException
    */
   private boolean advanceToNextRow(
-      TreeReaderFactory.TreeReader reader, long nextRow, boolean canAdvanceStripe)
+    BatchReader reader, long nextRow, boolean canAdvanceStripe)
       throws IOException {
     long nextRowInStripe = nextRow - rowBaseInStripe;
     // check for row skipping
@@ -1166,8 +1166,6 @@
       rowInStripe += batchSize;
       reader.setVectorColumnCount(batch.getDataColumnCount());
       reader.nextBatch(batch, batchSize);
-      batch.selectedInUse = false;
-      batch.size = batchSize;
       advanceToNextRow(reader, rowInStripe + rowBaseInStripe, true);
       return batch.size  != 0;
     } catch (IOException e) {
@@ -1261,7 +1259,7 @@
     }
   }
 
-  private void seekToRowEntry(TreeReaderFactory.TreeReader reader, int rowEntry)
+  private void seekToRowEntry(BatchReader reader, int rowEntry)
       throws IOException {
     OrcProto.RowIndex[] rowIndices = indexes.getRowGroupIndex();
     PositionProvider[] index = new PositionProvider[rowIndices.length];
diff --git a/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java b/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java
index cac4067..d17de68 100644
--- a/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java
+++ b/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java
@@ -48,6 +48,10 @@
 import org.apache.orc.OrcProto;
 import org.apache.orc.impl.reader.ReaderEncryption;
 import org.apache.orc.impl.reader.StripePlanner;
+import org.apache.orc.impl.reader.tree.BatchReader;
+import org.apache.orc.impl.reader.tree.PrimitiveBatchReader;
+import org.apache.orc.impl.reader.tree.StructBatchReader;
+import org.apache.orc.impl.reader.tree.TypeReader;
 import org.apache.orc.impl.writer.TimestampTreeWriter;
 
 /**
@@ -160,10 +164,9 @@
     }
   }
 
-  public abstract static class TreeReader {
+  public abstract static class TreeReader implements TypeReader {
     protected final int columnId;
     protected BitFieldReader present = null;
-    protected int vectorColumnCount;
     protected final Context context;
 
     static final long[] powerOfTenTable = {
@@ -200,14 +203,9 @@
       } else {
         present = new BitFieldReader(in);
       }
-      vectorColumnCount = -1;
     }
 
-    void setVectorColumnCount(int vectorColumnCount) {
-      this.vectorColumnCount = vectorColumnCount;
-    }
-
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
             columnId);
@@ -230,7 +228,7 @@
       }
     }
 
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       checkEncoding(planner.getEncoding(columnId));
       InStream in = planner.getStream(new StreamName(columnId,
           OrcProto.Stream.Kind.PRESENT));
@@ -271,21 +269,6 @@
       }
     }
 
-    abstract void skipRows(long rows) throws IOException;
-
-    /**
-     * Called at the top level to read into the given batch.
-     * @param batch the batch to read into
-     * @param batchSize the number of rows to read
-     * @throws IOException
-     */
-    public void nextBatch(VectorizedRowBatch batch,
-                          int batchSize) throws IOException {
-      batch.cols[0].reset();
-      batch.cols[0].ensureSize(batchSize, false);
-      nextVector(batch.cols[0], null, batchSize);
-    }
-
     /**
      * Populates the isNull vector array in the previousVector object based on
      * the present stream values. This function is called from all the child
@@ -334,6 +317,7 @@
       return present;
     }
 
+    @Override
     public int getColumnId() {
       return columnId;
     }
@@ -351,7 +335,7 @@
     }
 
     @Override
-    void skipRows(long rows) {
+    public void skipRows(long rows) {
       // PASS
     }
 
@@ -388,7 +372,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       reader = new BitFieldReader(planner.getStream(new StreamName(columnId,
           OrcProto.Stream.Kind.DATA)));
@@ -406,7 +390,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
 
@@ -437,7 +421,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       reader = new RunLengthByteReader(planner.getStream(new StreamName(columnId,
           OrcProto.Stream.Kind.DATA)));
@@ -468,7 +452,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
   }
@@ -491,7 +475,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -500,7 +484,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -533,7 +517,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
   }
@@ -556,7 +540,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -565,7 +549,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -598,7 +582,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
   }
@@ -622,7 +606,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -631,7 +615,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -664,7 +648,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
   }
@@ -684,7 +668,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -754,7 +738,7 @@
     }
 
     @Override
-    protected void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       for (int i = 0; i < items; ++i) {
         utils.readFloat(stream);
@@ -777,7 +761,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name =
           new StreamName(columnId,
@@ -847,7 +831,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       long len = items * 8;
       while (len > 0) {
@@ -877,7 +861,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -886,7 +870,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -921,7 +905,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       long lengthToSkip = 0;
       for (int i = 0; i < items; ++i) {
@@ -991,7 +975,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -1000,7 +984,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       OrcProto.ColumnEncoding.Kind kind = planner.getEncoding(columnId).getKind();
       data = createIntegerReader(kind,
@@ -1103,7 +1087,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       data.skip(items);
       nanos.skip(items);
@@ -1134,7 +1118,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -1143,7 +1127,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -1187,7 +1171,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
   }
@@ -1228,7 +1212,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -1237,7 +1221,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       valueStream = planner.getStream(new StreamName(columnId,
           OrcProto.Stream.Kind.DATA));
@@ -1339,7 +1323,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       HiveDecimalWritable scratchDecWritable = new HiveDecimalWritable();
       for (int i = 0; i < items; i++) {
@@ -1378,7 +1362,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
             columnId);
@@ -1386,7 +1370,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       InStream stream = planner.getStream(new StreamName(columnId,
           OrcProto.Stream.Kind.DATA));
@@ -1442,7 +1426,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       valueReader.skip(items);
     }
@@ -1454,7 +1438,7 @@
    * dictionary encoding was used.
    */
   public static class StringTreeReader extends TreeReader {
-    protected TreeReader reader;
+    protected TypeReader reader;
 
     StringTreeReader(int columnId, Context context) throws IOException {
       super(columnId, context);
@@ -1483,12 +1467,12 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       reader.checkEncoding(encoding);
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       // For each stripe, checks the encoding and initializes the appropriate
       // reader
       switch (planner.getEncoding(columnId).getKind()) {
@@ -1525,7 +1509,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skipRows(items);
     }
   }
@@ -1633,7 +1617,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT &&
           encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -1642,7 +1626,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       StreamName name = new StreamName(columnId,
           OrcProto.Stream.Kind.DATA);
@@ -1680,7 +1664,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       long lengthToSkip = 0;
       for (int i = 0; i < items; ++i) {
@@ -1737,7 +1721,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DICTIONARY &&
           encoding.getKind() != OrcProto.ColumnEncoding.Kind.DICTIONARY_V2) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -1746,7 +1730,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
 
       // read the dictionary blob
@@ -1890,7 +1874,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       reader.skip(countNonNulls(items));
     }
 
@@ -2011,7 +1995,7 @@
   }
 
   public static class StructTreeReader extends TreeReader {
-    protected final TreeReader[] fields;
+    public final TypeReader[] fields;
 
     protected StructTreeReader(int columnId,
                                TypeDescription readerSchema,
@@ -2026,14 +2010,14 @@
       }
     }
 
-    public TreeReader[] getChildReaders() {
+    public TypeReader[] getChildReaders() {
       return fields;
     }
 
     protected StructTreeReader(int columnId, InStream present,
                                Context context,
                                OrcProto.ColumnEncoding encoding,
-                               TreeReader[] childReaders) throws IOException {
+                               TypeReader[] childReaders) throws IOException {
       super(columnId, present, context);
       if (encoding != null) {
         checkEncoding(encoding);
@@ -2044,7 +2028,7 @@
     @Override
     public void seek(PositionProvider[] index) throws IOException {
       super.seek(index);
-      for (TreeReader kid : fields) {
+      for (TypeReader kid : fields) {
         if (kid != null) {
           kid.seek(index);
         }
@@ -2052,20 +2036,6 @@
     }
 
     @Override
-    public void nextBatch(VectorizedRowBatch batch,
-                          int batchSize) throws IOException {
-      for(int i=0; i < fields.length &&
-          (vectorColumnCount == -1 || i < vectorColumnCount); ++i) {
-        ColumnVector colVector = batch.cols[i];
-        if (colVector != null) {
-          colVector.reset();
-          colVector.ensureSize((int) batchSize, false);
-          fields[i].nextVector(colVector, null, batchSize);
-        }
-      }
-    }
-
-    @Override
     public void nextVector(ColumnVector previousVector,
                            boolean[] isNull,
                            final int batchSize) throws IOException {
@@ -2085,9 +2055,9 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
-      for (TreeReader field : fields) {
+      for (TypeReader field : fields) {
         if (field != null) {
           field.startStripe(planner);
         }
@@ -2095,9 +2065,9 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
-      for (TreeReader field : fields) {
+      for (TypeReader field : fields) {
         if (field != null) {
           field.skipRows(items);
         }
@@ -2106,7 +2076,7 @@
   }
 
   public static class UnionTreeReader extends TreeReader {
-    protected final TreeReader[] fields;
+    protected final TypeReader[] fields;
     protected RunLengthByteReader tags;
 
     protected UnionTreeReader(int fileColumn,
@@ -2125,7 +2095,7 @@
     protected UnionTreeReader(int columnId, InStream present,
                               Context context,
                               OrcProto.ColumnEncoding encoding,
-                              TreeReader[] childReaders) throws IOException {
+                              TypeReader[] childReaders) throws IOException {
       super(columnId, present, context);
       if (encoding != null) {
         checkEncoding(encoding);
@@ -2137,7 +2107,7 @@
     public void seek(PositionProvider[] index) throws IOException {
       super.seek(index);
       tags.seek(index[columnId]);
-      for (TreeReader kid : fields) {
+      for (TypeReader kid : fields) {
         kid.seek(index);
       }
     }
@@ -2165,11 +2135,11 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       tags = new RunLengthByteReader(planner.getStream(new StreamName(columnId,
           OrcProto.Stream.Kind.DATA)));
-      for (TreeReader field : fields) {
+      for (TypeReader field : fields) {
         if (field != null) {
           field.startStripe(planner);
         }
@@ -2177,7 +2147,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       long[] counts = new long[fields.length];
       for (int i = 0; i < items; ++i) {
@@ -2190,7 +2160,7 @@
   }
 
   public static class ListTreeReader extends TreeReader {
-    protected final TreeReader elementReader;
+    protected final TypeReader elementReader;
     protected IntegerReader lengths = null;
 
     protected ListTreeReader(int fileColumn,
@@ -2206,7 +2176,7 @@
                              Context context,
                              InStream data,
                              OrcProto.ColumnEncoding encoding,
-                             TreeReader elementReader) throws IOException {
+                             TypeReader elementReader) throws IOException {
       super(columnId, present, context);
       if (data != null && encoding != null) {
         checkEncoding(encoding);
@@ -2247,7 +2217,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -2256,7 +2226,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       lengths = createIntegerReader(planner.getEncoding(columnId).getKind(),
           planner.getStream(new StreamName(columnId,
@@ -2267,7 +2237,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       long childSkip = 0;
       for (long i = 0; i < items; ++i) {
@@ -2278,8 +2248,8 @@
   }
 
   public static class MapTreeReader extends TreeReader {
-    protected final TreeReader keyReader;
-    protected final TreeReader valueReader;
+    protected final TypeReader keyReader;
+    protected final TypeReader valueReader;
     protected IntegerReader lengths = null;
 
     protected MapTreeReader(int fileColumn,
@@ -2297,8 +2267,8 @@
                             Context context,
                             InStream data,
                             OrcProto.ColumnEncoding encoding,
-                            TreeReader keyReader,
-                            TreeReader valueReader) throws IOException {
+                            TypeReader keyReader,
+                            TypeReader valueReader) throws IOException {
       super(columnId, present, context);
       if (data != null && encoding != null) {
         checkEncoding(encoding);
@@ -2342,7 +2312,7 @@
     }
 
     @Override
-    void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
+    public void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException {
       if ((encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT) &&
           (encoding.getKind() != OrcProto.ColumnEncoding.Kind.DIRECT_V2)) {
         throw new IOException("Unknown encoding " + encoding + " in column " +
@@ -2351,7 +2321,7 @@
     }
 
     @Override
-    void startStripe(StripePlanner planner) throws IOException {
+    public void startStripe(StripePlanner planner) throws IOException {
       super.startStripe(planner);
       lengths = createIntegerReader(planner.getEncoding(columnId).getKind(),
           planner.getStream(new StreamName(columnId,
@@ -2365,7 +2335,7 @@
     }
 
     @Override
-    void skipRows(long items) throws IOException {
+    public void skipRows(long items) throws IOException {
       items = countNonNulls(items);
       long childSkip = 0;
       for (long i = 0; i < items; ++i) {
@@ -2376,7 +2346,7 @@
     }
   }
 
-  public static TreeReader createTreeReader(TypeDescription readerType,
+  public static TypeReader createTreeReader(TypeDescription readerType,
                                             Context context
                                             ) throws IOException {
     OrcFile.Version version = context.getFileFormat();
@@ -2446,4 +2416,14 @@
             readerTypeCategory);
     }
   }
+
+  public static BatchReader createRootReader(TypeDescription readerType, Context context)
+          throws IOException {
+    TypeReader reader = createTreeReader(readerType, context);
+    if (reader instanceof StructTreeReader) {
+      return new StructBatchReader((StructTreeReader) reader);
+    } else {
+      return new PrimitiveBatchReader(reader);
+    }
+  }
 }
diff --git a/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java b/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java
index 463a70c..d8d08fb 100644
--- a/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java
+++ b/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java
@@ -494,15 +494,14 @@
     return result;
   }
 
-  private static class StreamInformation {
-    final OrcProto.Stream.Kind kind;
-    final int column;
-    final long offset;
-    final long length;
-    BufferChunk firstChunk;
+  public static class StreamInformation {
+    public final OrcProto.Stream.Kind kind;
+    public final int column;
+    public final long offset;
+    public final long length;
+    public BufferChunk firstChunk;
 
-    StreamInformation(OrcProto.Stream.Kind kind,
-                      int column, long offset, long length) {
+    public StreamInformation(OrcProto.Stream.Kind kind, int column, long offset, long length) {
       this.kind = kind;
       this.column = column;
       this.offset = offset;
diff --git a/java/core/src/java/org/apache/orc/impl/reader/tree/BatchReader.java b/java/core/src/java/org/apache/orc/impl/reader/tree/BatchReader.java
new file mode 100644
index 0000000..710d2cd
--- /dev/null
+++ b/java/core/src/java/org/apache/orc/impl/reader/tree/BatchReader.java
@@ -0,0 +1,70 @@
+/*
+ * 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
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.orc.impl.reader.tree;
+
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.impl.PositionProvider;
+import org.apache.orc.impl.reader.StripePlanner;
+
+import java.io.IOException;
+
+
+/**
+ * The top level interface that the reader uses to read the columns from the
+ * ORC file.
+ */
+public abstract class BatchReader {
+  // The row type reader
+  public final TypeReader rootType;
+
+  protected int vectorColumnCount = -1;
+
+  public BatchReader(TypeReader rootType) {
+    this.rootType = rootType;
+  }
+
+  public void startStripe(StripePlanner planner) throws IOException {
+    rootType.startStripe(planner);
+  }
+
+  public void setVectorColumnCount(int vectorColumnCount) {
+    this.vectorColumnCount = vectorColumnCount;
+  }
+
+  /**
+   * Read the next batch of data from the file.
+   * @param batch     the batch to read into
+   * @param batchSize the number of rows to read
+   * @throws IOException errors reading the file
+   */
+  public abstract void nextBatch(VectorizedRowBatch batch,
+                                 int batchSize) throws IOException;
+
+  protected void resetBatch(VectorizedRowBatch batch, int batchSize) {
+    batch.selectedInUse = false;
+    batch.size = batchSize;
+  }
+
+  public void skipRows(long rows) throws IOException {
+    rootType.skipRows(rows);
+  }
+
+  public void seek(PositionProvider[] index) throws IOException {
+    rootType.seek(index);
+  }
+}
diff --git a/java/core/src/java/org/apache/orc/impl/reader/tree/PrimitiveBatchReader.java b/java/core/src/java/org/apache/orc/impl/reader/tree/PrimitiveBatchReader.java
new file mode 100644
index 0000000..669b6a5
--- /dev/null
+++ b/java/core/src/java/org/apache/orc/impl/reader/tree/PrimitiveBatchReader.java
@@ -0,0 +1,37 @@
+/*
+ * 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
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.orc.impl.reader.tree;
+
+import java.io.IOException;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+public class PrimitiveBatchReader extends BatchReader {
+
+  public PrimitiveBatchReader(TypeReader rowReader) {
+    super(rowReader);
+  }
+
+  @Override
+  public void nextBatch(VectorizedRowBatch batch,
+                        int batchSize) throws IOException {
+  batch.cols[0].reset();
+  batch.cols[0].ensureSize(batchSize, false);
+  rootType.nextVector(batch.cols[0], null, batchSize);
+  resetBatch(batch, batchSize);
+  }
+}
diff --git a/java/core/src/java/org/apache/orc/impl/reader/tree/StructBatchReader.java b/java/core/src/java/org/apache/orc/impl/reader/tree/StructBatchReader.java
new file mode 100644
index 0000000..9503436
--- /dev/null
+++ b/java/core/src/java/org/apache/orc/impl/reader/tree/StructBatchReader.java
@@ -0,0 +1,46 @@
+/*
+ * 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
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.orc.impl.reader.tree;
+
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.impl.TreeReaderFactory;
+
+import java.io.IOException;
+
+public class StructBatchReader extends BatchReader {
+
+  public StructBatchReader(TreeReaderFactory.StructTreeReader rowReader) {
+    super(rowReader);
+  }
+
+  @Override
+  public void nextBatch(VectorizedRowBatch batch, int batchSize) throws IOException {
+    TypeReader[] children = ((TreeReaderFactory.StructTreeReader) rootType).fields;
+    for (int i = 0; i < children.length &&
+                    (vectorColumnCount == -1 || i < vectorColumnCount); ++i) {
+      ColumnVector colVector = batch.cols[i];
+      if (colVector != null) {
+        colVector.reset();
+        colVector.ensureSize(batchSize, false);
+        children[i].nextVector(colVector, null, batchSize);
+      }
+    }
+    resetBatch(batch, batchSize);
+  }
+}
diff --git a/java/core/src/java/org/apache/orc/impl/reader/tree/TypeReader.java b/java/core/src/java/org/apache/orc/impl/reader/tree/TypeReader.java
new file mode 100644
index 0000000..5932bee
--- /dev/null
+++ b/java/core/src/java/org/apache/orc/impl/reader/tree/TypeReader.java
@@ -0,0 +1,43 @@
+/*
+ * 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
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.orc.impl.reader.tree;
+
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.orc.OrcProto;
+import org.apache.orc.impl.PositionProvider;
+import org.apache.orc.impl.reader.StripePlanner;
+
+import java.io.IOException;
+
+public interface TypeReader {
+  void checkEncoding(OrcProto.ColumnEncoding encoding) throws IOException;
+
+  void startStripe(StripePlanner planner) throws IOException;
+
+  void seek(PositionProvider[] index) throws IOException;
+
+  void seek(PositionProvider index) throws IOException;
+
+  void skipRows(long rows) throws IOException;
+
+  void nextVector(ColumnVector previous,
+                  boolean[] isNull,
+                  int batchSize) throws IOException;
+
+  int getColumnId();
+}
diff --git a/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java b/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java
index a0981a6..be7b616 100644
--- a/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java
+++ b/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java
@@ -53,6 +53,9 @@
 import org.apache.orc.Writer;
 import org.apache.orc.impl.reader.ReaderEncryption;
 import org.apache.orc.impl.reader.StripePlanner;
+import org.apache.orc.impl.reader.tree.BatchReader;
+import org.apache.orc.impl.reader.tree.StructBatchReader;
+import org.apache.orc.impl.reader.tree.TypeReader;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -1641,13 +1644,14 @@
 
     TreeReaderFactory.Context treeContext =
         new TreeReaderFactory.ReaderContext().setSchemaEvolution(evo);
-    TreeReaderFactory.TreeReader reader =
-        TreeReaderFactory.createTreeReader(readType, treeContext);
+    BatchReader reader =
+        TreeReaderFactory.createRootReader(readType, treeContext);
 
     // check to make sure the tree reader is built right
-    assertEquals(TreeReaderFactory.StructTreeReader.class, reader.getClass());
-    TreeReaderFactory.TreeReader[] children =
-        ((TreeReaderFactory.StructTreeReader) reader).getChildReaders();
+    assertEquals(StructBatchReader.class, reader.getClass());
+    assertEquals(TreeReaderFactory.StructTreeReader.class, reader.rootType.getClass());
+    TypeReader[] children =
+        ((TreeReaderFactory.StructTreeReader) reader.rootType).getChildReaders();
     assertEquals(3, children.length);
     assertEquals(TreeReaderFactory.NullTreeReader.class, children[0].getClass());
     assertEquals(TreeReaderFactory.StringTreeReader.class, children[1].getClass());
@@ -1694,12 +1698,12 @@
 
     TreeReaderFactory.Context treeContext =
         new TreeReaderFactory.ReaderContext().setSchemaEvolution(evo);
-    TreeReaderFactory.TreeReader reader =
+    TypeReader reader =
         TreeReaderFactory.createTreeReader(readType, treeContext);
 
     // check to make sure the tree reader is built right
     assertEquals(TreeReaderFactory.StructTreeReader.class, reader.getClass());
-    TreeReaderFactory.TreeReader[] children =
+    TypeReader[] children =
         ((TreeReaderFactory.StructTreeReader) reader).getChildReaders();
     assertEquals(2, children.length);
     assertEquals(TreeReaderFactory.IntTreeReader.class, children[0].getClass());