ORC-296 : work around HADOOP-15171; also fix stream contract (Sergey Shelukhin, reviewed by Gopal Vijayaraghavan)
Project: http://git-wip-us.apache.org/repos/asf/orc/repo Commit: http://git-wip-us.apache.org/repos/asf/orc/commit/b5a30bbb Tree: http://git-wip-us.apache.org/repos/asf/orc/tree/b5a30bbb Diff: http://git-wip-us.apache.org/repos/asf/orc/diff/b5a30bbb Branch: refs/heads/branch-1.4 Commit: b5a30bbbd443d537e5c48427a0e042f82fbc26e0 Parents: 9d3434f Author: sergey <[email protected]> Authored: Fri Feb 2 16:23:28 2018 -0800 Committer: Owen O'Malley <[email protected]> Committed: Tue Feb 6 12:48:33 2018 -0800 ---------------------------------------------------------------------- .../org/apache/orc/impl/HadoopShimsCurrent.java | 12 ++++++++ .../src/java/org/apache/orc/impl/InStream.java | 31 ++++++++++---------- 2 files changed, 28 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/orc/blob/b5a30bbb/java/core/src/java/org/apache/orc/impl/HadoopShimsCurrent.java ---------------------------------------------------------------------- diff --git a/java/core/src/java/org/apache/orc/impl/HadoopShimsCurrent.java b/java/core/src/java/org/apache/orc/impl/HadoopShimsCurrent.java index 9f40272..3d4875c 100644 --- a/java/core/src/java/org/apache/orc/impl/HadoopShimsCurrent.java +++ b/java/core/src/java/org/apache/orc/impl/HadoopShimsCurrent.java @@ -36,12 +36,18 @@ public class HadoopShimsCurrent implements HadoopShims { private static class SnappyDirectDecompressWrapper implements DirectDecompressor { private final SnappyDirectDecompressor root; + private boolean isFirstCall = true; SnappyDirectDecompressWrapper(SnappyDirectDecompressor root) { this.root = root; } public void decompress(ByteBuffer input, ByteBuffer output) throws IOException { + if (!isFirstCall) { + root.reset(); + } else { + isFirstCall = false; + } root.decompress(input, output); } @@ -58,12 +64,18 @@ public class HadoopShimsCurrent implements HadoopShims { private static class ZlibDirectDecompressWrapper implements DirectDecompressor { private final ZlibDecompressor.ZlibDirectDecompressor root; + private boolean isFirstCall = true; ZlibDirectDecompressWrapper(ZlibDecompressor.ZlibDirectDecompressor root) { this.root = root; } public void decompress(ByteBuffer input, ByteBuffer output) throws IOException { + if (!isFirstCall) { + root.reset(); + } else { + isFirstCall = false; + } root.decompress(input, output); } http://git-wip-us.apache.org/repos/asf/orc/blob/b5a30bbb/java/core/src/java/org/apache/orc/impl/InStream.java ---------------------------------------------------------------------- diff --git a/java/core/src/java/org/apache/orc/impl/InStream.java b/java/core/src/java/org/apache/orc/impl/InStream.java index c85aee5..94e9232 100644 --- a/java/core/src/java/org/apache/orc/impl/InStream.java +++ b/java/core/src/java/org/apache/orc/impl/InStream.java @@ -239,36 +239,37 @@ public abstract class InStream extends InputStream { @Override public int read() throws IOException { - if (uncompressed == null || uncompressed.remaining() == 0) { - if (currentOffset == length) { - return -1; - } - readHeader(); + if (!ensureUncompressed()) { + return -1; } return 0xff & uncompressed.get(); } @Override public int read(byte[] data, int offset, int length) throws IOException { - if (uncompressed == null || uncompressed.remaining() == 0) { - if (currentOffset == this.length) { - return -1; - } - readHeader(); + if (!ensureUncompressed()) { + return -1; } int actualLength = Math.min(length, uncompressed.remaining()); uncompressed.get(data, offset, actualLength); return actualLength; } - @Override - public int available() throws IOException { - if (uncompressed == null || uncompressed.remaining() == 0) { - if (currentOffset == length) { - return 0; + private boolean ensureUncompressed() throws IOException { + while (uncompressed == null || uncompressed.remaining() == 0) { + if (currentOffset == this.length) { + return false; } readHeader(); } + return true; + } + + @Override + public int available() throws IOException { + if (!ensureUncompressed()) { + return 0; + } return uncompressed.remaining(); }
