From dbb097a7410638b8808bb4650615ca5759284a87 Mon Sep 17 00:00:00 2001
From: Jason Altekruse <altekrusejason@gmail.com>
Date: Wed, 6 Jan 2016 15:35:11 -0600
Subject: [PATCH 01/15] Parquet-400: Fixed issue reading some files from HDFS
 and S3 when using Hadoop 2.x

The problem was not handling the case where a read request returns less
than the requested number of bytes. The FSDataInputStream lacks an
API equivalent for readFully when using ByteBuffers, which used to solve
this problem when using byte arrays as the destination. This has been
fixed by including a loop to manually request the remaining bytes until
everything has been read.
---
 .../java/org/apache/parquet/hadoop/ParquetFileReader.java   | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 55ed5ee050..24e430eac2 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -725,7 +725,7 @@ protected PageHeader readPageHeader() throws IOException {
         // to allow reading older files (using dictionary) we need this.
         // usually 13 to 19 bytes are missing
         // if the last page is smaller than this, the page header itself is truncated in the buffer.
-        this.byteBuf.rewind(); // resetting the buffer to the position before we got the error
+        this.byteBuf.position(initialPos); // resetting the buffer to the position before we got the error
         LOG.info("completing the column chunk to read the page header");
         pageHeader = Util.readPageHeader(new SequenceInputStream(this, f)); // trying again from the buffer + remainder of the stream.
       }
@@ -815,7 +815,9 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
       List<Chunk> result = new ArrayList<Chunk>(chunks.size());
       f.seek(offset);
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      CompatibilityUtil.getBuf(f, chunksByteBuffer, length);
+      while (chunksByteBuffer.hasRemaining()) {
+        CompatibilityUtil.getBuf(f, chunksByteBuffer, chunksByteBuffer.remaining());
+      }
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
       int currentChunkOffset = 0;

From da5c5c91155f70487e65a790f19626b138cde2f5 Mon Sep 17 00:00:00 2001
From: Daniel Weeks <dweeks@netflix.com>
Date: Wed, 27 Jan 2016 13:16:43 -0800
Subject: [PATCH 02/15] Create blacklist for FileSystems that don't work well
 with bytebuffer impl

---
 .../parquet/hadoop/ParquetFileReader.java     | 25 ++++++++++++++++---
 1 file changed, 22 insertions(+), 3 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 24e430eac2..0c8b635ddd 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -96,6 +96,10 @@ public class ParquetFileReader implements Closeable {
 
   public static String PARQUET_READ_PARALLELISM = "parquet.metadata.read.parallelism";
 
+  //URI Schemes to blacklist for bytebuffer read.
+  public static final String PARQUET_BYTEBUFFER_BLACKLIST = "parquet.bytebuffer.fs.blacklist";
+  public static final String[] PARQUET_BYTEBUFFER_BLACKLIST_DEFAULT = {"s3", "s3n", "s3a"};
+
   private static ParquetMetadataConverter converter = new ParquetMetadataConverter();
 
   /**
@@ -472,6 +476,7 @@ static ParquetFileReader open(Configuration conf, Path file) throws IOException
   private final FileMetaData fileMetaData;
   private final String createdBy;
   private final ByteBufferAllocator allocator;
+  private final boolean disableByteBufferRead;
 
   private int currentBlock = 0;
 
@@ -506,6 +511,10 @@ public ParquetFileReader(
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(configuration, 0);
     this.allocator = new HeapByteBufferAllocator();
+
+    //Bypass ByteBuffer read path for S3 FileSystems.  See PARQUET-400.
+    List<String> fsBlackList = Arrays.asList(configuration.getStrings(PARQUET_BYTEBUFFER_BLACKLIST, PARQUET_BYTEBUFFER_BLACKLIST_DEFAULT));
+    this.disableByteBufferRead = fsBlackList.contains(filePath.toUri().getScheme());
   }
 
   public void appendTo(ParquetFileWriter writer) throws IOException {
@@ -814,10 +823,20 @@ public void addChunk(ChunkDescriptor descriptor) {
     public List<Chunk> readAll(FSDataInputStream f) throws IOException {
       List<Chunk> result = new ArrayList<Chunk>(chunks.size());
       f.seek(offset);
-      ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      while (chunksByteBuffer.hasRemaining()) {
-        CompatibilityUtil.getBuf(f, chunksByteBuffer, chunksByteBuffer.remaining());
+
+      //Allocate the bytebuffer based on whether the FS can support it.
+      ByteBuffer chunksByteBuffer;
+      if(disableByteBufferRead) {
+        byte[] chunkBytes = new byte[length];
+        f.readFully(chunkBytes);
+        chunksByteBuffer = ByteBuffer.wrap(chunkBytes);
+      } else {
+        chunksByteBuffer = allocator.allocate(length);
+        while (chunksByteBuffer.hasRemaining()) {
+          CompatibilityUtil.getBuf(f, chunksByteBuffer, chunksByteBuffer.remaining());
+        }
       }
+
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
       int currentChunkOffset = 0;

From f35c772831dd73737a69e2192c8d814ee1dbe07d Mon Sep 17 00:00:00 2001
From: Jason Altekruse <altekrusejason@gmail.com>
Date: Wed, 20 Apr 2016 16:36:38 -0700
Subject: [PATCH 03/15] remove unused parameter

---
 .../java/org/apache/parquet/hadoop/ParquetFileReader.java   | 2 +-
 .../org/apache/parquet/hadoop/util/CompatibilityUtil.java   | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 0c8b635ddd..2b0a862a06 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -833,7 +833,7 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
       } else {
         chunksByteBuffer = allocator.allocate(length);
         while (chunksByteBuffer.hasRemaining()) {
-          CompatibilityUtil.getBuf(f, chunksByteBuffer, chunksByteBuffer.remaining());
+          CompatibilityUtil.getBuf(f, chunksByteBuffer);
         }
       }
 
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index bacf222a24..f57197225c 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -76,7 +76,7 @@ private static Object invoke(Method method, String errorMsg, Object instance, Ob
     }
   }
 
-  public static int getBuf(FSDataInputStream f, ByteBuffer readBuf, int maxSize) throws IOException {
+  public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int res;
     if (useV21) {
       try {
@@ -88,7 +88,7 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf, int maxSize) t
           // be a reasonable check to make to see if the interface is
           // present but not implemented and we should be falling back
           useV21 = false;
-          return getBuf(f, readBuf, maxSize);
+          return getBuf(f, readBuf);
         } else if (e.getCause() instanceof IOException) {
           throw (IOException) e.getCause();
         } else {
@@ -105,7 +105,7 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf, int maxSize) t
         throw new ShouldNeverHappenException(e);
       }
     } else {
-      byte[] buf = new byte[maxSize];
+      byte[] buf = new byte[readBuf.remaining()];
       res = f.read(buf);
       readBuf.put(buf, 0, res);
     }

From 96406d824c52e7d49fd9cfe73ef1bef460049637 Mon Sep 17 00:00:00 2001
From: Jason Altekruse <altekrusejason@gmail.com>
Date: Thu, 21 Apr 2016 09:54:09 -0700
Subject: [PATCH 04/15] WIP addressing comments

Conflicts:
	parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
---
 .../hadoop/util/CompatibilityUtil.java        | 20 +++++++------------
 1 file changed, 7 insertions(+), 13 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index f57197225c..0b3b4de94f 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -66,16 +66,6 @@ private V21FileAPI() throws ReflectiveOperationException {
     }
   }
 
-  private static Object invoke(Method method, String errorMsg, Object instance, Object... args) {
-    try {
-      return method.invoke(instance, args);
-    } catch (IllegalAccessException e) {
-      throw new IllegalArgumentException(errorMsg, e);
-    } catch (InvocationTargetException e) {
-      throw new IllegalArgumentException(errorMsg, e);
-    }
-  }
-
   public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int res;
     if (useV21) {
@@ -105,9 +95,13 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
         throw new ShouldNeverHappenException(e);
       }
     } else {
-      byte[] buf = new byte[readBuf.remaining()];
-      res = f.read(buf);
-      readBuf.put(buf, 0, res);
+      if (readBuf.hasArray()) {
+        res = f.read(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
+      } else {
+        byte[] buf = new byte[readBuf.remaining()];
+        res = f.read(buf);
+        readBuf.put(buf, 0, res);
+      }
     }
     return res;
   }

From e800f20cc5c056df9a877ecfde96c26ce1ffebc2 Mon Sep 17 00:00:00 2001
From: Jason Altekruse <altekrusejason@gmail.com>
Date: Mon, 23 May 2016 15:31:47 -0700
Subject: [PATCH 05/15] Fix infinite loop bug caused by not updating bytebuffer
 position.

---
 .../java/org/apache/parquet/hadoop/util/CompatibilityUtil.java  | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index 0b3b4de94f..4289e1a265 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -96,7 +96,9 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
       }
     } else {
       if (readBuf.hasArray()) {
+        int initPos = readBuf.position();
         res = f.read(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
+        readBuf.position(initPos + res);
       } else {
         byte[] buf = new byte[readBuf.remaining()];
         res = f.read(buf);

From 84606590a28ce86caaf3778fd60a9439ff36d87e Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Tue, 24 May 2016 13:57:47 -0700
Subject: [PATCH 06/15] Switch to readFully in compatUtil, remove loop

---
 .../parquet/hadoop/ParquetFileReader.java      | 18 ++----------------
 .../parquet/hadoop/util/CompatibilityUtil.java | 12 ++++++++++--
 2 files changed, 12 insertions(+), 18 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index c924609cf7..2708377b19 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -490,7 +490,6 @@ public static ParquetFileReader open(Configuration conf, Path file, ParquetMetad
   private final Map<ColumnPath, ColumnDescriptor> paths = new HashMap<ColumnPath, ColumnDescriptor>();
   private final FileMetaData fileMetaData; // may be null
   private final ByteBufferAllocator allocator;
-  private final boolean disableByteBufferRead;
   private final Configuration conf;
 
   // not final. in some cases, this may be lazily loaded for backward-compat.
@@ -534,10 +533,6 @@ public ParquetFileReader(
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(configuration, 0);
     this.allocator = new HeapByteBufferAllocator();
-
-    //Bypass ByteBuffer read path for S3 FileSystems.  See PARQUET-400.
-    List<String> fsBlackList = Arrays.asList(configuration.getStrings(PARQUET_BYTEBUFFER_BLACKLIST, PARQUET_BYTEBUFFER_BLACKLIST_DEFAULT));
-    this.disableByteBufferRead = fsBlackList.contains(filePath.toUri().getScheme());
   }
 
   /**
@@ -1050,17 +1045,8 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
       f.seek(offset);
 
       //Allocate the bytebuffer based on whether the FS can support it.
-      ByteBuffer chunksByteBuffer;
-      if(disableByteBufferRead) {
-        byte[] chunkBytes = new byte[length];
-        f.readFully(chunkBytes);
-        chunksByteBuffer = ByteBuffer.wrap(chunkBytes);
-      } else {
-        chunksByteBuffer = allocator.allocate(length);
-        while (chunksByteBuffer.hasRemaining()) {
-          CompatibilityUtil.getBuf(f, chunksByteBuffer);
-        }
-      }
+      ByteBuffer chunksByteBuffer = allocator.allocate(length);
+      CompatibilityUtil.getBuf(f, chunksByteBuffer);
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index 4289e1a265..f611486a0c 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -66,6 +66,12 @@ private V21FileAPI() throws ReflectiveOperationException {
     }
   }
 
+  /**
+   * This method attempts to read into the provided readBuffer, readBuffer.remaining() bytes.
+   * If the underlying InputStream supports read directly into ByteBuffer we go ahead and invoke that.
+   * Else we fall back to directly calling readFully() on the underlying stream.
+   * @return Number of bytes read - should be readBuf.remaining()
+   */
   public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int res;
     if (useV21) {
@@ -97,11 +103,13 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
     } else {
       if (readBuf.hasArray()) {
         int initPos = readBuf.position();
-        res = f.read(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
+        res = readBuf.remaining();
+        f.readFully(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
         readBuf.position(initPos + res);
       } else {
         byte[] buf = new byte[readBuf.remaining()];
-        res = f.read(buf);
+        res = readBuf.remaining();
+        f.readFully(buf);
         readBuf.put(buf, 0, res);
       }
     }

From 41807ee8b4da989ba37b55e79a49d939e4162606 Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Tue, 24 May 2016 14:46:34 -0700
Subject: [PATCH 07/15] More logging

---
 .../java/org/apache/parquet/hadoop/ParquetFileReader.java  | 3 ++-
 .../org/apache/parquet/hadoop/util/CompatibilityUtil.java  | 7 +++++++
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 2708377b19..9e532624b3 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -1046,7 +1046,8 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
 
       //Allocate the bytebuffer based on whether the FS can support it.
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      CompatibilityUtil.getBuf(f, chunksByteBuffer);
+      int res = CompatibilityUtil.getBuf(f, chunksByteBuffer);
+      LOG.info("Requested to read: " + length + " ended up reading: " + res);
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index f611486a0c..b24841f299 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -19,6 +19,7 @@
 package org.apache.parquet.hadoop.util;
 
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.parquet.Log;
 import org.apache.parquet.ShouldNeverHappenException;
 
 import java.io.IOException;
@@ -33,6 +34,8 @@ public class CompatibilityUtil {
   private static boolean useV21;
   public static final V21FileAPI fileAPI;
 
+  private static final Log LOG = Log.getLog(CompatibilityUtil.class);
+
   private static class V21FileAPI {
     private final Method PROVIDE_BUF_READ_METHOD;
     private final Class<?> FSDataInputStreamCls;
@@ -76,6 +79,7 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
     int res;
     if (useV21) {
       try {
+        LOG.info("Trying to use v21 apis");
         res = (Integer) fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
       } catch (InvocationTargetException e) {
         if (e.getCause() instanceof UnsupportedOperationException) {
@@ -83,6 +87,7 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
           // can choose to throw UnsupportedOperationException, so this should
           // be a reasonable check to make to see if the interface is
           // present but not implemented and we should be falling back
+          LOG.info("Failed to read data using v21 APIs: ", e);
           useV21 = false;
           return getBuf(f, readBuf);
         } else if (e.getCause() instanceof IOException) {
@@ -104,11 +109,13 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
       if (readBuf.hasArray()) {
         int initPos = readBuf.position();
         res = readBuf.remaining();
+        LOG.info("Reading into readBuf's array. initPos: " + initPos + " remaining: " + readBuf.remaining());
         f.readFully(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
         readBuf.position(initPos + res);
       } else {
         byte[] buf = new byte[readBuf.remaining()];
         res = readBuf.remaining();
+        LOG.info("Reading into new array. remaining: " + readBuf.remaining());
         f.readFully(buf);
         readBuf.put(buf, 0, res);
       }

From 79e1d238ce4e7f1e8db61a7ce0de60a57e086851 Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Tue, 24 May 2016 15:19:52 -0700
Subject: [PATCH 08/15] Loop in reader

---
 .../java/org/apache/parquet/hadoop/ParquetFileReader.java   | 6 ++++--
 .../org/apache/parquet/hadoop/util/CompatibilityUtil.java   | 2 +-
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 9e532624b3..4b507d6f1d 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -1046,8 +1046,10 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
 
       //Allocate the bytebuffer based on whether the FS can support it.
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      int res = CompatibilityUtil.getBuf(f, chunksByteBuffer);
-      LOG.info("Requested to read: " + length + " ended up reading: " + res);
+      while (chunksByteBuffer.hasRemaining()) {
+        int res = CompatibilityUtil.getBuf(f, chunksByteBuffer);
+        LOG.info("Requested to read: " + length + " ended up reading: " + res);
+      }
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index b24841f299..8796df63be 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -79,7 +79,7 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
     int res;
     if (useV21) {
       try {
-        LOG.info("Trying to use v21 apis");
+        LOG.info("Trying to use v21 apis, initPos: " + readBuf.position() + " remaining: " + readBuf.remaining());
         res = (Integer) fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
       } catch (InvocationTargetException e) {
         if (e.getCause() instanceof UnsupportedOperationException) {

From 29b2452ab7cc3710e7219786737ee8d0c91fe182 Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Tue, 24 May 2016 16:55:40 -0700
Subject: [PATCH 09/15] Minor updates

---
 .../org/apache/parquet/hadoop/ParquetFileReader.java |  5 +----
 .../parquet/hadoop/util/CompatibilityUtil.java       | 12 ++++--------
 2 files changed, 5 insertions(+), 12 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 4b507d6f1d..2708377b19 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -1046,10 +1046,7 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
 
       //Allocate the bytebuffer based on whether the FS can support it.
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      while (chunksByteBuffer.hasRemaining()) {
-        int res = CompatibilityUtil.getBuf(f, chunksByteBuffer);
-        LOG.info("Requested to read: " + length + " ended up reading: " + res);
-      }
+      CompatibilityUtil.getBuf(f, chunksByteBuffer);
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index 8796df63be..14ad60d2a1 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -19,7 +19,6 @@
 package org.apache.parquet.hadoop.util;
 
 import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.parquet.Log;
 import org.apache.parquet.ShouldNeverHappenException;
 
 import java.io.IOException;
@@ -34,8 +33,6 @@ public class CompatibilityUtil {
   private static boolean useV21;
   public static final V21FileAPI fileAPI;
 
-  private static final Log LOG = Log.getLog(CompatibilityUtil.class);
-
   private static class V21FileAPI {
     private final Method PROVIDE_BUF_READ_METHOD;
     private final Class<?> FSDataInputStreamCls;
@@ -79,15 +76,16 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
     int res;
     if (useV21) {
       try {
-        LOG.info("Trying to use v21 apis, initPos: " + readBuf.position() + " remaining: " + readBuf.remaining());
-        res = (Integer) fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
+        res = readBuf.remaining();
+        while (readBuf.hasRemaining()) {
+          fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
+        }
       } catch (InvocationTargetException e) {
         if (e.getCause() instanceof UnsupportedOperationException) {
           // the FSDataInputStream docs say specifically that implementations
           // can choose to throw UnsupportedOperationException, so this should
           // be a reasonable check to make to see if the interface is
           // present but not implemented and we should be falling back
-          LOG.info("Failed to read data using v21 APIs: ", e);
           useV21 = false;
           return getBuf(f, readBuf);
         } else if (e.getCause() instanceof IOException) {
@@ -109,13 +107,11 @@ public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOExcep
       if (readBuf.hasArray()) {
         int initPos = readBuf.position();
         res = readBuf.remaining();
-        LOG.info("Reading into readBuf's array. initPos: " + initPos + " remaining: " + readBuf.remaining());
         f.readFully(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
         readBuf.position(initPos + res);
       } else {
         byte[] buf = new byte[readBuf.remaining()];
         res = readBuf.remaining();
-        LOG.info("Reading into new array. remaining: " + readBuf.remaining());
         f.readFully(buf);
         readBuf.put(buf, 0, res);
       }

From cab228f9cc2e8403c8b52f5943b44621b86d4fa7 Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Thu, 26 May 2016 09:33:41 -0700
Subject: [PATCH 10/15] Add some tests, refactor code a bit

---
 .../parquet/hadoop/ParquetFileReader.java     |  12 +-
 .../hadoop/util/CompatibilityUtil.java        | 131 +++++++++++-------
 .../hadoop/util/TestCompatibilityUtil.java    |  93 +++++++++++++
 3 files changed, 183 insertions(+), 53 deletions(-)
 create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 2708377b19..0b84684f4e 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -102,9 +102,9 @@ public class ParquetFileReader implements Closeable {
 
   public static String PARQUET_READ_PARALLELISM = "parquet.metadata.read.parallelism";
 
-  //URI Schemes to blacklist for bytebuffer read.
-  public static final String PARQUET_BYTEBUFFER_BLACKLIST = "parquet.bytebuffer.fs.blacklist";
-  public static final String[] PARQUET_BYTEBUFFER_BLACKLIST_DEFAULT = {"s3", "s3n", "s3a"};
+  // configure if we want to use Hadoop's V2 read(bytebuffer). If true, we try to read using the
+  // new Hadoop read(ByteBuffer) api. Else, we skip.
+  public static String PARQUET_HADOOP_BYTEBUFFER_READ = "parquet.read.useByteBuffer";
 
   private static ParquetMetadataConverter converter = new ParquetMetadataConverter();
 
@@ -491,6 +491,7 @@ public static ParquetFileReader open(Configuration conf, Path file, ParquetMetad
   private final FileMetaData fileMetaData; // may be null
   private final ByteBufferAllocator allocator;
   private final Configuration conf;
+  private final CompatibilityUtil compatibilityUtil;
 
   // not final. in some cases, this may be lazily loaded for backward-compat.
   private ParquetMetadata footer;
@@ -533,6 +534,7 @@ public ParquetFileReader(
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(configuration, 0);
     this.allocator = new HeapByteBufferAllocator();
+    this.compatibilityUtil = new CompatibilityUtil(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
   }
 
   /**
@@ -565,6 +567,7 @@ public ParquetFileReader(Configuration conf, Path file, MetadataFilter filter) t
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(conf, 0);
     this.allocator = new HeapByteBufferAllocator();
+    this.compatibilityUtil = new CompatibilityUtil(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
   }
 
   /**
@@ -588,6 +591,7 @@ public ParquetFileReader(Configuration conf, Path file, ParquetMetadata footer)
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(conf, 0);
     this.allocator = new HeapByteBufferAllocator();
+    this.compatibilityUtil = new CompatibilityUtil(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
   }
 
   public ParquetMetadata getFooter() {
@@ -1046,7 +1050,7 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
 
       //Allocate the bytebuffer based on whether the FS can support it.
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      CompatibilityUtil.getBuf(f, chunksByteBuffer);
+      compatibilityUtil.getBuf(f, chunksByteBuffer);
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index 14ad60d2a1..4ddd3eea2f 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -19,6 +19,7 @@
 package org.apache.parquet.hadoop.util;
 
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.parquet.Log;
 import org.apache.parquet.ShouldNeverHappenException;
 
 import java.io.IOException;
@@ -28,11 +29,6 @@
 
 public class CompatibilityUtil {
 
-  // Will be set to true if the implementation of FSDataInputSteam supports
-  // the 2.x APIs, in particular reading using a provided ByteBuffer
-  private static boolean useV21;
-  public static final V21FileAPI fileAPI;
-
   private static class V21FileAPI {
     private final Method PROVIDE_BUF_READ_METHOD;
     private final Class<?> FSDataInputStreamCls;
@@ -43,19 +39,39 @@ private V21FileAPI() throws ReflectiveOperationException {
       PROVIDE_BUF_READ_METHOD = FSDataInputStreamCls.getMethod("read", ByteBuffer.class);
     }
   }
-  
-  static {
+
+  // Will be set to true if the implementation of FSDataInputSteam supports
+  // the 2.x APIs, in particular reading using a provided ByteBuffer
+  private boolean useV2;
+  private V21FileAPI fileAPI;
+
+  private static final Log LOG = Log.getLog(CompatibilityUtil.class);
+
+  public CompatibilityUtil(boolean useV2) {
+    if ( useV2 && !isHadoop2x() ) {
+      LOG.info("Can't read Hadoop 2x classes, will be using 1x read APIs");
+      this.useV2 = false;
+    } else {
+      this.useV2 = useV2;
+    }
+
+    initializeFileAPI(this.useV2);
+  }
+
+  private boolean isHadoop2x() {
     // Test to see if a class from the Hadoop 2.x API is available
-    boolean v21 = true;
+    boolean v2 = true;
     try {
       Class.forName("org.apache.hadoop.io.compress.DirectDecompressor");
     } catch (ClassNotFoundException cnfe) {
-      v21 = false;
+      v2 = false;
     }
+    return v2;
+  }
 
-    useV21 = v21;
+  private void initializeFileAPI(boolean useV21) {
     try {
-      if (v21) {
+      if (useV21) {
         fileAPI = new V21FileAPI();
       } else {
         fileAPI = null;
@@ -72,50 +88,67 @@ private V21FileAPI() throws ReflectiveOperationException {
    * Else we fall back to directly calling readFully() on the underlying stream.
    * @return Number of bytes read - should be readBuf.remaining()
    */
-  public static int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+  public int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int res;
-    if (useV21) {
-      try {
-        res = readBuf.remaining();
-        while (readBuf.hasRemaining()) {
-          fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
-        }
-      } catch (InvocationTargetException e) {
-        if (e.getCause() instanceof UnsupportedOperationException) {
-          // the FSDataInputStream docs say specifically that implementations
-          // can choose to throw UnsupportedOperationException, so this should
-          // be a reasonable check to make to see if the interface is
-          // present but not implemented and we should be falling back
-          useV21 = false;
-          return getBuf(f, readBuf);
-        } else if (e.getCause() instanceof IOException) {
-          throw (IOException) e.getCause();
-        } else {
-          // To handle any cases where a Runtime exception occurs and provide
-          // some additional context information. A stacktrace would just give
-          // a line number, this at least tells them we were using the version
-          // of the read method designed for using a ByteBuffer.
-          throw new IOException("Error reading out of an FSDataInputStream " +
-              "using the Hadoop 2 ByteBuffer based read method.", e.getCause());
-        }
-      } catch (IllegalAccessException e) {
-        // This method is public because it is defined in an interface,
-        // there should be no problems accessing it
-        throw new ShouldNeverHappenException(e);
-      }
+    if (useV2) {
+      res = readWithByteBuffer(f, readBuf);
     } else {
       if (readBuf.hasArray()) {
-        int initPos = readBuf.position();
-        res = readBuf.remaining();
-        f.readFully(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
-        readBuf.position(initPos + res);
+        res = readWithExistingArray(f, readBuf);
       } else {
-        byte[] buf = new byte[readBuf.remaining()];
-        res = readBuf.remaining();
-        f.readFully(buf);
-        readBuf.put(buf, 0, res);
+        res = readWithNewArray(f, readBuf);
       }
     }
     return res;
   }
+
+  private int readWithByteBuffer(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int remaining = readBuf.remaining();
+    try {
+      while (readBuf.hasRemaining()) {
+        fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
+      }
+    } catch (InvocationTargetException e) {
+      if (e.getCause() instanceof UnsupportedOperationException) {
+        // the FSDataInputStream docs say specifically that implementations
+        // can choose to throw UnsupportedOperationException, so this should
+        // be a reasonable check to make to see if the interface is
+        // present but not implemented and we should be falling back
+        useV2 = false;
+        return getBuf(f, readBuf);
+      } else if (e.getCause() instanceof IOException) {
+        throw (IOException) e.getCause();
+      } else {
+        // To handle any cases where a Runtime exception occurs and provide
+        // some additional context information. A stacktrace would just give
+        // a line number, this at least tells them we were using the version
+        // of the read method designed for using a ByteBuffer.
+        throw new IOException("Error reading out of an FSDataInputStream " +
+          "using the Hadoop 2 ByteBuffer based read method.", e.getCause());
+      }
+    } catch (IllegalAccessException e) {
+      // This method is public because it is defined in an interface,
+      // there should be no problems accessing it
+      throw new ShouldNeverHappenException(e);
+    }
+
+    return remaining;
+  }
+
+  private int readWithExistingArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int initPos = readBuf.position();
+    int remaining = readBuf.remaining();
+    f.readFully(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
+    readBuf.position(initPos + remaining);
+    return remaining;
+  }
+
+  private int readWithNewArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int remaining = readBuf.remaining();
+    byte[] buf = new byte[readBuf.remaining()];
+    f.readFully(buf);
+    readBuf.put(buf, 0, remaining);
+    return remaining;
+  }
+
 }
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java
new file mode 100644
index 0000000000..436df9f369
--- /dev/null
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java
@@ -0,0 +1,93 @@
+/*
+ * 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.parquet.hadoop.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import org.junit.Test;
+
+import junit.framework.Assert;
+
+public class TestCompatibilityUtil {
+
+  private static final byte [] TEST_ARRAY = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
+
+  private static class MockInputStream extends ByteArrayInputStream
+    implements Seekable, PositionedReadable {
+    public MockInputStream(byte[] buf) {
+      super(buf);
+    }
+
+    // empty implementation for unused methods
+    public int read(long position, byte[] buffer, int offset, int length) { return -1; }
+    public void readFully(long position, byte[] buffer, int offset, int length) {}
+    public void readFully(long position, byte[] buffer) {}
+    public void seek(long position) {}
+    public long getPos() { return 0; }
+    public boolean seekToNewSource(long targetPos) { return false; }
+  }
+
+  @Test
+  public void testGetBufWithArray() throws Exception {
+    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocate(10);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    int readCount = compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    Assert.assertEquals("Mismatching no of chars read", 10, readCount);
+    Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
+  }
+
+  @Test
+  public void testGetBufWithoutArray() throws Exception {
+    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocateDirect(10);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    int readCount = compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    Assert.assertEquals("Mismatching no of chars read", 10, readCount);
+    Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
+  }
+
+  @Test
+  public void testGetBufWithSmallerBuffer() throws Exception {
+    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocate(5);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    int readCount = compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    Assert.assertEquals("Mismatching no of chars read", 5, readCount);
+    Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
+  }
+
+  @Test(expected = EOFException.class)
+  public void testGetBufWithLargerBuffer() throws Exception {
+    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocate(50);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    // this throws an exception as we are trying to read 50 chars and have only 10
+    compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+  }
+}

From 9225171ada4dcb435b0f8ee2efb5d2dde27bac8c Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Fri, 17 Jun 2016 18:56:58 -0700
Subject: [PATCH 11/15] Fix minor comments

---
 .../org/apache/parquet/hadoop/ParquetFileReader.java     | 9 ++++++---
 .../apache/parquet/hadoop/util/CompatibilityUtil.java    | 4 ++--
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 0b84684f4e..50595c2bc3 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -102,9 +102,12 @@ public class ParquetFileReader implements Closeable {
 
   public static String PARQUET_READ_PARALLELISM = "parquet.metadata.read.parallelism";
 
-  // configure if we want to use Hadoop's V2 read(bytebuffer). If true, we try to read using the
-  // new Hadoop read(ByteBuffer) api. Else, we skip.
-  public static String PARQUET_HADOOP_BYTEBUFFER_READ = "parquet.read.useByteBuffer";
+  // configure if we want to use Hadoop's V2 read(bytebuffer) API.
+  // If true, we try to read using the new Hadoop read(ByteBuffer) api. This reads data into the provided
+  // byteBuffer and allows us to potentially take advantage zero-copy read path in Hadoop.
+  // Else, we skip and either read into the byteBuffer's array (if its present) or allocate one and copy
+  // into it.
+  public static String PARQUET_HADOOP_BYTEBUFFER_READ = "parquet.read.use.byte.buffer";
 
   private static ParquetMetadataConverter converter = new ParquetMetadataConverter();
 
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index 4ddd3eea2f..95ad21c4fb 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -138,14 +138,14 @@ private int readWithByteBuffer(FSDataInputStream f, ByteBuffer readBuf) throws I
   private int readWithExistingArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int initPos = readBuf.position();
     int remaining = readBuf.remaining();
-    f.readFully(readBuf.array(), readBuf.arrayOffset(), readBuf.remaining());
+    f.readFully(readBuf.array(), readBuf.arrayOffset(), remaining);
     readBuf.position(initPos + remaining);
     return remaining;
   }
 
   private int readWithNewArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int remaining = readBuf.remaining();
-    byte[] buf = new byte[readBuf.remaining()];
+    byte[] buf = new byte[remaining];
     f.readFully(buf);
     readBuf.put(buf, 0, remaining);
     return remaining;

From d7caf1087fc9492e4d04843480846006515ebf6c Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Mon, 20 Jun 2016 13:30:09 -0700
Subject: [PATCH 12/15] Add parquet-hadoop2

---
 parquet-hadoop2/pom.xml | 62 +++++++++++++++++++++++++++++++++++++++++
 pom.xml                 |  2 ++
 2 files changed, 64 insertions(+)
 create mode 100644 parquet-hadoop2/pom.xml

diff --git a/parquet-hadoop2/pom.xml b/parquet-hadoop2/pom.xml
new file mode 100644
index 0000000000..1548286fc1
--- /dev/null
+++ b/parquet-hadoop2/pom.xml
@@ -0,0 +1,62 @@
+<!--
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.parquet</groupId>
+        <artifactId>parquet</artifactId>
+        <relativePath>../pom.xml</relativePath>
+        <version>1.8.2-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>parquet-hadoop2</artifactId>
+    <packaging>jar</packaging>
+    <name>Apache Parquet Hadoop2</name>
+    <url>https://parquet.apache.org</url>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.parquet</groupId>
+            <artifactId>parquet-column</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>${hadoop2.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-enforcer-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 512bf3735e..6e5666918e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -78,6 +78,7 @@
     <jackson.package>org.codehaus.jackson</jackson.package>
     <shade.prefix>shaded.parquet</shade.prefix>
     <hadoop.version>1.1.0</hadoop.version>
+    <hadoop2.version>2.3.0</hadoop2.version>
     <cascading.version>2.5.3</cascading.version>
     <cascading3.version>3.0.3</cascading3.version>
     <parquet.format.version>2.3.1</parquet.format.version>
@@ -117,6 +118,7 @@
     <module>parquet-hive</module>
     <module>parquet-hive-bundle</module>
     <module>parquet-tools</module>
+    <module>parquet-hadoop2</module>
   </modules>
 
   <dependencies>

From dbf9fccb86dff335577ab0ca934f46e5f8261ed5 Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Mon, 20 Jun 2016 17:34:10 -0700
Subject: [PATCH 13/15] Get rid of reflection in compatUtil's read call

---
 .../parquet/hadoop/ParquetFileReader.java     |  11 +-
 .../hadoop/util/CompatibilityReader.java      |  42 ++++++
 .../hadoop/util/CompatibilityReaderV1.java    |  58 ++++++++
 .../hadoop/util/CompatibilityUtil.java        | 125 +++---------------
 ...il.java => TestCompatibilityReaderV1.java} |  41 ++++--
 parquet-hadoop2/pom.xml                       |   5 +
 .../hadoop/util/CompatibilityReaderV2.java    |  46 +++++++
 .../util/TestCompatibilityReaderV2.java       | 107 +++++++++++++++
 8 files changed, 309 insertions(+), 126 deletions(-)
 create mode 100644 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java
 create mode 100644 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java
 rename parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/{TestCompatibilityUtil.java => TestCompatibilityReaderV1.java} (67%)
 create mode 100644 parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
 create mode 100644 parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 50595c2bc3..8e779e732b 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -62,6 +62,7 @@
 import org.apache.parquet.column.page.DictionaryPageReadStore;
 import org.apache.parquet.filter2.compat.FilterCompat;
 import org.apache.parquet.filter2.compat.RowGroupFilter;
+import org.apache.parquet.hadoop.util.CompatibilityReader;
 import org.apache.parquet.hadoop.util.CompatibilityUtil;
 
 import org.apache.parquet.Log;
@@ -494,7 +495,7 @@ public static ParquetFileReader open(Configuration conf, Path file, ParquetMetad
   private final FileMetaData fileMetaData; // may be null
   private final ByteBufferAllocator allocator;
   private final Configuration conf;
-  private final CompatibilityUtil compatibilityUtil;
+  private final CompatibilityReader compatibilityReader;
 
   // not final. in some cases, this may be lazily loaded for backward-compat.
   private ParquetMetadata footer;
@@ -537,7 +538,7 @@ public ParquetFileReader(
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(configuration, 0);
     this.allocator = new HeapByteBufferAllocator();
-    this.compatibilityUtil = new CompatibilityUtil(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
+    this.compatibilityReader = CompatibilityUtil.getHadoopReader(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
   }
 
   /**
@@ -570,7 +571,7 @@ public ParquetFileReader(Configuration conf, Path file, MetadataFilter filter) t
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(conf, 0);
     this.allocator = new HeapByteBufferAllocator();
-    this.compatibilityUtil = new CompatibilityUtil(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
+    this.compatibilityReader = CompatibilityUtil.getHadoopReader(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
   }
 
   /**
@@ -594,7 +595,7 @@ public ParquetFileReader(Configuration conf, Path file, ParquetMetadata footer)
     // the codec factory to get decompressors
     this.codecFactory = new CodecFactory(conf, 0);
     this.allocator = new HeapByteBufferAllocator();
-    this.compatibilityUtil = new CompatibilityUtil(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
+    this.compatibilityReader = CompatibilityUtil.getHadoopReader(conf.getBoolean(PARQUET_HADOOP_BYTEBUFFER_READ, true));
   }
 
   public ParquetMetadata getFooter() {
@@ -1053,7 +1054,7 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
 
       //Allocate the bytebuffer based on whether the FS can support it.
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      compatibilityUtil.getBuf(f, chunksByteBuffer);
+      compatibilityReader.readBuf(f, chunksByteBuffer);
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java
new file mode 100644
index 0000000000..8ad4ca4d48
--- /dev/null
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java
@@ -0,0 +1,42 @@
+/*
+ * 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.parquet.hadoop.util;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+
+/**
+ * Allows us to use either Hadoop V1 / V2 read APIs to read data without reflection. HadoopV2 implementation
+ * of this interface resides in a module with Hadoop2 dependencies.
+ * Note: classes that implement this interface are instantiated using reflection and must thus have a
+ * default constructor.
+ */
+public interface CompatibilityReader {
+
+  /**
+   * This method attempts to read into the provided readBuffer, readBuffer.remaining() bytes.
+   * @return Number of bytes read - should be readBuf.remaining()
+   * @throws EOFException if readBuf.remaining() is greater than the number of bytes available to
+   * read on the FSDataInputStream f.
+   */
+  int readBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException;
+}
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java
new file mode 100644
index 0000000000..8bed0c034b
--- /dev/null
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java
@@ -0,0 +1,58 @@
+/*
+ * 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.parquet.hadoop.util;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+
+/**
+ * Uses Hadoop V1's readFully(byte[], ...) APIs to read data.
+ */
+public class CompatibilityReaderV1 implements CompatibilityReader {
+
+  @Override
+  public int readBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int res;
+    if (readBuf.hasArray()) {
+      res = readWithExistingArray(f, readBuf);
+    } else {
+      res = readWithNewArray(f, readBuf);
+    }
+
+    return res;
+  }
+
+  private int readWithExistingArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int initPos = readBuf.position();
+    int remaining = readBuf.remaining();
+    f.readFully(readBuf.array(), readBuf.arrayOffset(), remaining);
+    readBuf.position(initPos + remaining);
+    return remaining;
+  }
+
+  private int readWithNewArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int remaining = readBuf.remaining();
+    byte[] buf = new byte[remaining];
+    f.readFully(buf);
+    readBuf.put(buf, 0, remaining);
+    return remaining;
+  }
+}
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
index 95ad21c4fb..330e3d4474 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityUtil.java
@@ -18,48 +18,30 @@
  */
 package org.apache.parquet.hadoop.util;
 
-import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.parquet.Log;
-import org.apache.parquet.ShouldNeverHappenException;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
 
 public class CompatibilityUtil {
 
-  private static class V21FileAPI {
-    private final Method PROVIDE_BUF_READ_METHOD;
-    private final Class<?> FSDataInputStreamCls;
-
-    private V21FileAPI() throws ReflectiveOperationException {
-      final String PACKAGE = "org.apache.hadoop";
-      FSDataInputStreamCls = Class.forName(PACKAGE + ".fs.FSDataInputStream");
-      PROVIDE_BUF_READ_METHOD = FSDataInputStreamCls.getMethod("read", ByteBuffer.class);
-    }
-  }
-
-  // Will be set to true if the implementation of FSDataInputSteam supports
-  // the 2.x APIs, in particular reading using a provided ByteBuffer
-  private boolean useV2;
-  private V21FileAPI fileAPI;
+  private static final String READER_V2_CLASS = "org.apache.parquet.hadoop.util.CompatibilityReaderV2";
 
   private static final Log LOG = Log.getLog(CompatibilityUtil.class);
 
-  public CompatibilityUtil(boolean useV2) {
-    if ( useV2 && !isHadoop2x() ) {
+  public static CompatibilityReader getHadoopReader(boolean useV2) {
+    if (!useV2) {
+      return new CompatibilityReaderV1();
+    }
+
+    if (!isHadoop2x()) {
       LOG.info("Can't read Hadoop 2x classes, will be using 1x read APIs");
-      this.useV2 = false;
-    } else {
-      this.useV2 = useV2;
+      return new CompatibilityReaderV1();
     }
 
-    initializeFileAPI(this.useV2);
+    return newV2Reader();
   }
 
-  private boolean isHadoop2x() {
-    // Test to see if a class from the Hadoop 2.x API is available
+  // Test to see if a class from the Hadoop 2.x API is available
+  // If it is, we try to instantiate the V2 CompatibilityReader.
+  private static boolean isHadoop2x() {
     boolean v2 = true;
     try {
       Class.forName("org.apache.hadoop.io.compress.DirectDecompressor");
@@ -69,86 +51,13 @@ private boolean isHadoop2x() {
     return v2;
   }
 
-  private void initializeFileAPI(boolean useV21) {
+  private static CompatibilityReader newV2Reader() {
     try {
-      if (useV21) {
-        fileAPI = new V21FileAPI();
-      } else {
-        fileAPI = null;
-      }
-
+      Class<?> reader = Class.forName(READER_V2_CLASS);
+      return (CompatibilityReader)reader.newInstance();
     } catch (ReflectiveOperationException e) {
-      throw new IllegalArgumentException("Error finding appropriate interfaces using reflection.", e);
-    }
-  }
-
-  /**
-   * This method attempts to read into the provided readBuffer, readBuffer.remaining() bytes.
-   * If the underlying InputStream supports read directly into ByteBuffer we go ahead and invoke that.
-   * Else we fall back to directly calling readFully() on the underlying stream.
-   * @return Number of bytes read - should be readBuf.remaining()
-   */
-  public int getBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
-    int res;
-    if (useV2) {
-      res = readWithByteBuffer(f, readBuf);
-    } else {
-      if (readBuf.hasArray()) {
-        res = readWithExistingArray(f, readBuf);
-      } else {
-        res = readWithNewArray(f, readBuf);
-      }
+      LOG.warn("Unable to instantiate Hadoop V2 compatibility reader class: " + READER_V2_CLASS + " , will be using 1x read APIs", e);
+      return new CompatibilityReaderV1();
     }
-    return res;
   }
-
-  private int readWithByteBuffer(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
-    int remaining = readBuf.remaining();
-    try {
-      while (readBuf.hasRemaining()) {
-        fileAPI.PROVIDE_BUF_READ_METHOD.invoke(f, readBuf);
-      }
-    } catch (InvocationTargetException e) {
-      if (e.getCause() instanceof UnsupportedOperationException) {
-        // the FSDataInputStream docs say specifically that implementations
-        // can choose to throw UnsupportedOperationException, so this should
-        // be a reasonable check to make to see if the interface is
-        // present but not implemented and we should be falling back
-        useV2 = false;
-        return getBuf(f, readBuf);
-      } else if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      } else {
-        // To handle any cases where a Runtime exception occurs and provide
-        // some additional context information. A stacktrace would just give
-        // a line number, this at least tells them we were using the version
-        // of the read method designed for using a ByteBuffer.
-        throw new IOException("Error reading out of an FSDataInputStream " +
-          "using the Hadoop 2 ByteBuffer based read method.", e.getCause());
-      }
-    } catch (IllegalAccessException e) {
-      // This method is public because it is defined in an interface,
-      // there should be no problems accessing it
-      throw new ShouldNeverHappenException(e);
-    }
-
-    return remaining;
-  }
-
-  private int readWithExistingArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
-    int initPos = readBuf.position();
-    int remaining = readBuf.remaining();
-    f.readFully(readBuf.array(), readBuf.arrayOffset(), remaining);
-    readBuf.position(initPos + remaining);
-    return remaining;
-  }
-
-  private int readWithNewArray(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
-    int remaining = readBuf.remaining();
-    byte[] buf = new byte[remaining];
-    f.readFully(buf);
-    readBuf.put(buf, 0, remaining);
-    return remaining;
-  }
-
 }
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java
similarity index 67%
rename from parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java
rename to parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java
index 436df9f369..6eeb1064f2 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityUtil.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java
@@ -29,7 +29,7 @@
 
 import junit.framework.Assert;
 
-public class TestCompatibilityUtil {
+public class TestCompatibilityReaderV1 {
 
   private static final byte [] TEST_ARRAY = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
 
@@ -48,46 +48,61 @@ public void seek(long position) {}
     public boolean seekToNewSource(long targetPos) { return false; }
   }
 
+  // confirm writer version when flag = false
   @Test
-  public void testGetBufWithArray() throws Exception {
-    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+  public void testReaderFlagOff() {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(false);
+    Assert.assertEquals("Incorrect CompatibilityReader instantiated", CompatibilityReaderV1.class, reader.getClass());
+  }
+
+  // confirm writer version when flag is true but we're on hadoop 1.x
+  @Test
+  public void testReaderFlagTrueHadoopV1() {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(true);
+    Assert.assertEquals("Incorrect CompatibilityReader instantiated", CompatibilityReaderV1.class, reader.getClass());
+  }
+
+  @Test
+  public void testReadBufWithArray() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(false);
     ByteBuffer byteBuffer = ByteBuffer.allocate(10);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 10, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
 
   @Test
-  public void testGetBufWithoutArray() throws Exception {
-    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+  public void testReadBufWithoutArray() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(false);
     ByteBuffer byteBuffer = ByteBuffer.allocateDirect(10);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 10, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
 
   @Test
-  public void testGetBufWithSmallerBuffer() throws Exception {
-    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+  public void testReadBufWithSmallerBuffer() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(false);
     ByteBuffer byteBuffer = ByteBuffer.allocate(5);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 5, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
 
   @Test(expected = EOFException.class)
-  public void testGetBufWithLargerBuffer() throws Exception {
-    CompatibilityUtil compatibilityUtil = new CompatibilityUtil(true);
+  public void testReadBufWithLargerBuffer() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(false);
     ByteBuffer byteBuffer = ByteBuffer.allocate(50);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
     // this throws an exception as we are trying to read 50 chars and have only 10
-    compatibilityUtil.getBuf(fsDataInputStream, byteBuffer);
+    reader.readBuf(fsDataInputStream, byteBuffer);
   }
+
 }
diff --git a/parquet-hadoop2/pom.xml b/parquet-hadoop2/pom.xml
index 1548286fc1..fbbede17a0 100644
--- a/parquet-hadoop2/pom.xml
+++ b/parquet-hadoop2/pom.xml
@@ -45,6 +45,11 @@
             <version>${hadoop2.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.parquet</groupId>
+            <artifactId>parquet-hadoop</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java b/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
new file mode 100644
index 0000000000..bcd28f4c50
--- /dev/null
+++ b/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.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.parquet.hadoop.util;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+
+/**
+ * Uses the Hadoop V2 read(ByteBuffer) APIs to read data.
+ */
+public class CompatibilityReaderV2 implements CompatibilityReader {
+
+  @Override
+  public int readBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+    int remaining = readBuf.remaining();
+    while (readBuf.hasRemaining()) {
+      int readCount = f.read(readBuf);
+      if (readCount == -1) {
+        // this is probably a bug in the ParquetReader. We shouldn't have called readBuf with a buffer
+        // that has more remaining than the amount of data in the stream.
+        throw new EOFException("Reached the end of stream. Still have: " + readBuf.remaining() + " bytes left");
+      }
+    }
+
+    return remaining;
+  }
+}
diff --git a/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java b/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java
new file mode 100644
index 0000000000..fdcbcb7a19
--- /dev/null
+++ b/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java
@@ -0,0 +1,107 @@
+/*
+ * 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.parquet.hadoop.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import org.junit.Test;
+
+import junit.framework.Assert;
+
+public class TestCompatibilityReaderV2 {
+
+  private static final byte [] TEST_ARRAY = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
+
+  private static class MockInputStream extends ByteArrayInputStream
+    implements Seekable, PositionedReadable, ByteBufferReadable {
+    public MockInputStream(byte[] buf) {
+      super(buf);
+    }
+
+    // empty implementation for unused methods
+    public int read(long position, byte[] buffer, int offset, int length) { return -1; }
+    public void readFully(long position, byte[] buffer, int offset, int length) {}
+    public void readFully(long position, byte[] buffer) {}
+    public void seek(long position) {}
+    public long getPos() { return 0; }
+    public boolean seekToNewSource(long targetPos) { return false; }
+
+    @Override
+    public int read(ByteBuffer buf) throws IOException {
+      int remaining = buf.remaining();
+      while (buf.hasRemaining()) {
+        int data = read();
+        if (data == -1) {
+          // similar to the pattern used in some Hadoop classes that implement ByteBufferReadable,
+          // we return -1 if we're at EOF
+          return -1;
+        }
+        buf.put((byte) data);
+      }
+
+      return remaining;
+    }
+  }
+
+  // confirm writer version is v2 when flag = true and we're in Hadoop V2
+  @Test
+  public void testReaderFlagOn() {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(true);
+    Assert.assertEquals("Incorrect CompatibilityReader instantiated", CompatibilityReaderV2.class, reader.getClass());
+  }
+
+  @Test
+  public void testReadBuf() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocate(10);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    Assert.assertEquals("Mismatching no of chars read", 10, readCount);
+    Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
+  }
+
+  @Test
+  public void testReadBufWithSmallerBuffer() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocate(5);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    Assert.assertEquals("Mismatching no of chars read", 5, readCount);
+    Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
+  }
+
+  @Test(expected = EOFException.class)
+  public void testReadBufWithLargerBuffer() throws Exception {
+    CompatibilityReader reader = CompatibilityUtil.getHadoopReader(true);
+    ByteBuffer byteBuffer = ByteBuffer.allocate(50);
+    FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
+
+    // if we're trying to read 50 chars and have only 10, we end up with an EOFException
+    reader.readBuf(fsDataInputStream, byteBuffer);
+  }
+}

From 1a3f7a2567d6152ee41efc387e364fe57f0fabb6 Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Fri, 24 Jun 2016 13:03:48 -0700
Subject: [PATCH 14/15] Rename to readFully

---
 .../java/org/apache/parquet/hadoop/ParquetFileReader.java | 2 +-
 .../apache/parquet/hadoop/util/CompatibilityReader.java   | 4 ++--
 .../apache/parquet/hadoop/util/CompatibilityReaderV1.java | 2 +-
 .../parquet/hadoop/util/TestCompatibilityReaderV1.java    | 8 ++++----
 .../apache/parquet/hadoop/util/CompatibilityReaderV2.java | 4 ++--
 .../parquet/hadoop/util/TestCompatibilityReaderV2.java    | 6 +++---
 6 files changed, 13 insertions(+), 13 deletions(-)

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 8e779e732b..a6e1969e8d 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -1054,7 +1054,7 @@ public List<Chunk> readAll(FSDataInputStream f) throws IOException {
 
       //Allocate the bytebuffer based on whether the FS can support it.
       ByteBuffer chunksByteBuffer = allocator.allocate(length);
-      compatibilityReader.readBuf(f, chunksByteBuffer);
+      compatibilityReader.readFully(f, chunksByteBuffer);
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java
index 8ad4ca4d48..e093c53939 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReader.java
@@ -34,9 +34,9 @@ public interface CompatibilityReader {
 
   /**
    * This method attempts to read into the provided readBuffer, readBuffer.remaining() bytes.
-   * @return Number of bytes read - should be readBuf.remaining()
+   * @return Number of bytes read - should be readBuffer.remaining()
    * @throws EOFException if readBuf.remaining() is greater than the number of bytes available to
    * read on the FSDataInputStream f.
    */
-  int readBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException;
+  int readFully(FSDataInputStream f, ByteBuffer readBuffer) throws IOException;
 }
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java
index 8bed0c034b..c54f24fe46 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV1.java
@@ -29,7 +29,7 @@
 public class CompatibilityReaderV1 implements CompatibilityReader {
 
   @Override
-  public int readBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+  public int readFully(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int res;
     if (readBuf.hasArray()) {
       res = readWithExistingArray(f, readBuf);
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java
index 6eeb1064f2..71fdf7a625 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV1.java
@@ -68,7 +68,7 @@ public void testReadBufWithArray() throws Exception {
     ByteBuffer byteBuffer = ByteBuffer.allocate(10);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readFully(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 10, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
@@ -79,7 +79,7 @@ public void testReadBufWithoutArray() throws Exception {
     ByteBuffer byteBuffer = ByteBuffer.allocateDirect(10);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readFully(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 10, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
@@ -90,7 +90,7 @@ public void testReadBufWithSmallerBuffer() throws Exception {
     ByteBuffer byteBuffer = ByteBuffer.allocate(5);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readFully(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 5, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
@@ -102,7 +102,7 @@ public void testReadBufWithLargerBuffer() throws Exception {
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
     // this throws an exception as we are trying to read 50 chars and have only 10
-    reader.readBuf(fsDataInputStream, byteBuffer);
+    reader.readFully(fsDataInputStream, byteBuffer);
   }
 
 }
diff --git a/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java b/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
index bcd28f4c50..4c788edda6 100644
--- a/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
+++ b/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
@@ -30,12 +30,12 @@
 public class CompatibilityReaderV2 implements CompatibilityReader {
 
   @Override
-  public int readBuf(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
+  public int readFully(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int remaining = readBuf.remaining();
     while (readBuf.hasRemaining()) {
       int readCount = f.read(readBuf);
       if (readCount == -1) {
-        // this is probably a bug in the ParquetReader. We shouldn't have called readBuf with a buffer
+        // this is probably a bug in the ParquetReader. We shouldn't have called readFully with a buffer
         // that has more remaining than the amount of data in the stream.
         throw new EOFException("Reached the end of stream. Still have: " + readBuf.remaining() + " bytes left");
       }
diff --git a/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java b/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java
index fdcbcb7a19..4ad02cf6a4 100644
--- a/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java
+++ b/parquet-hadoop2/src/test/java/org/apache/parquet/hadoop/util/TestCompatibilityReaderV2.java
@@ -79,7 +79,7 @@ public void testReadBuf() throws Exception {
     ByteBuffer byteBuffer = ByteBuffer.allocate(10);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readFully(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 10, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
@@ -90,7 +90,7 @@ public void testReadBufWithSmallerBuffer() throws Exception {
     ByteBuffer byteBuffer = ByteBuffer.allocate(5);
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
-    int readCount = reader.readBuf(fsDataInputStream, byteBuffer);
+    int readCount = reader.readFully(fsDataInputStream, byteBuffer);
     Assert.assertEquals("Mismatching no of chars read", 5, readCount);
     Assert.assertFalse("Byte buffer not full", byteBuffer.hasRemaining());
   }
@@ -102,6 +102,6 @@ public void testReadBufWithLargerBuffer() throws Exception {
     FSDataInputStream fsDataInputStream = new FSDataInputStream(new MockInputStream(TEST_ARRAY));
 
     // if we're trying to read 50 chars and have only 10, we end up with an EOFException
-    reader.readBuf(fsDataInputStream, byteBuffer);
+    reader.readFully(fsDataInputStream, byteBuffer);
   }
 }

From 9c76555d21004d711939cd3e90d8ffa22edcc79d Mon Sep 17 00:00:00 2001
From: Piyush Narang <pnarang@twitter.com>
Date: Fri, 24 Jun 2016 13:46:06 -0700
Subject: [PATCH 15/15] Add some comments

---
 .../org/apache/parquet/hadoop/util/CompatibilityReaderV2.java  | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java b/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
index 4c788edda6..70214d1920 100644
--- a/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
+++ b/parquet-hadoop2/src/main/java/org/apache/parquet/hadoop/util/CompatibilityReaderV2.java
@@ -32,6 +32,9 @@ public class CompatibilityReaderV2 implements CompatibilityReader {
   @Override
   public int readFully(FSDataInputStream f, ByteBuffer readBuf) throws IOException {
     int remaining = readBuf.remaining();
+    // unfortunately the Hadoop APIs seem to not have a 'readFully' equivalent for the byteBuffer read
+    // calls. The read(ByteBuffer) call might read fewer than byteBuffer.hasRemaining() bytes. Thus we
+    // have to loop to ensure we read them.
     while (readBuf.hasRemaining()) {
       int readCount = f.read(readBuf);
       if (readCount == -1) {