From dbb097a7410638b8808bb4650615ca5759284a87 Mon Sep 17 00:00:00 2001 From: Jason Altekruse Date: Wed, 6 Jan 2016 15:35:11 -0600 Subject: [PATCH 1/5] 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 readAll(FSDataInputStream f) throws IOException { List result = new ArrayList(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 Date: Wed, 27 Jan 2016 13:16:43 -0800 Subject: [PATCH 2/5] 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 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 readAll(FSDataInputStream f) throws IOException { List result = new ArrayList(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 Date: Wed, 20 Apr 2016 16:36:38 -0700 Subject: [PATCH 3/5] 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 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 Date: Thu, 21 Apr 2016 09:54:09 -0700 Subject: [PATCH 4/5] 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 Date: Mon, 23 May 2016 15:31:47 -0700 Subject: [PATCH 5/5] 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);