PARQUET-400: Fix review items.

This commit:
* Makes SeekableInputStream implementations package private
* Adds tests for the H2 stream wrapper
* Adds javadoc to SeekableInputStream
* Simplifies the wrap method by checking the underlying stream

Author: rdblue

parquet-common/src/main/java/org/apache/parquet/io/SeekableInputStream.java

@@ -19,22 +19,88 @@
 
 package org.apache.parquet.io;
 
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 
+/**
+ * {@code SeekableInputStream} is an interface with the methods needed by
+ * Parquet to read data from a file or Hadoop data stream.
+ */
 public abstract class SeekableInputStream extends InputStream {
 
+  /**
+   * Return the current position in the InputStream.
+   *
+   * @return current position in bytes from the start of the stream
+   * @throws IOException If the underlying stream throws IOException
+   */
   public abstract long getPos() throws IOException;
 
+  /**
+   * Seek to a new position in the InputStream.
+   *
+   * @param newPos the new position to seek to
+   * @throws IOException If the underlying stream throws IOException
+   */
   public abstract void seek(long newPos) throws IOException;
 
+  /**
+   * Read a byte array of data, from position 0 to the end of the array.
+   * <p>
+   * This method is equivalent to {@code read(bytes, 0, bytes.length)}.
+   * <p>
+   * This method will block until len bytes are available to copy into the
+   * array, or will throw {@link EOFException} if the stream ends before the
+   * array is full.
+   *
+   * @param bytes a byte array to fill with data from the stream
+   * @throws IOException If the underlying stream throws IOException
+   * @throws EOFException If the stream has fewer bytes left than are needed to
+   *                      fill the array, {@code bytes.length}
+   */
   public abstract void readFully(byte[] bytes) throws IOException;
 
+  /**
+   * Read {@code len} bytes of data into an array, at position {@code start}.
+   * <p>
+   * This method will block until len bytes are available to copy into the
+   * array, or will throw {@link EOFException} if the stream ends before the
+   * array is full.
+   *
+   * @param bytes a byte array to fill with data from the stream
+   * @throws IOException If the underlying stream throws IOException
+   * @throws EOFException If the stream has fewer than {@code len} bytes left
+   */
   public abstract void readFully(byte[] bytes, int start, int len) throws IOException;
 
+  /**
+   * Read {@code buf.remaining()} bytes of data into a {@link ByteBuffer}.
+   * <p>
+   * This method will copy available bytes into the buffer, reading at most
+   * {@code buf.remaining()} bytes. The number of bytes actually copied is
+   * returned by the method, or -1 is returned to signal that the end of the
+   * underlying stream has been reached.
+   *
+   * @param buf a byte array to fill with data from the stream
+   * @return the number of bytes read or -1 if the stream ended
+   * @throws IOException If the underlying stream throws IOException
+   */
   public abstract int read(ByteBuffer buf) throws IOException;
 
+  /**
+   * Read {@code buf.remaining()} bytes of data into a {@link ByteBuffer}.
+   * <p>
+   * This method will block until {@code buf.remaining()} bytes are available
+   * to copy into the buffer, or will throw {@link EOFException} if the stream
+   * ends before the buffer is full.
+   *
+   * @param buf a byte array to fill with data from the stream
+   * @throws IOException If the underlying stream throws IOException
+   * @throws EOFException If the stream has fewer bytes left than are needed to
+   *                      fill the buffer, {@code buf.remaining()}
+   */
   public abstract void readFully(ByteBuffer buf) throws IOException;
 
 }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H1SeekableInputStream.java

@@ -25,7 +25,11 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
-public class H1SeekableInputStream extends SeekableInputStream {
+/**
+ * SeekableInputStream implementation that implements read(ByteBuffer) for
+ * Hadoop 1 FSDataInputStream.
+ */
+class H1SeekableInputStream extends SeekableInputStream {
 
   private final int COPY_BUFFER_SIZE = 8192;
   private final byte[] temp = new byte[COPY_BUFFER_SIZE];

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H2SeekableInputStream.java

@@ -27,7 +27,11 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
-public class H2SeekableInputStream extends SeekableInputStream {
+/**
+ * SeekableInputStream implementation for FSDataInputStream that implements
+ * ByteBufferReadable in Hadoop 2.
+ */
+class H2SeekableInputStream extends SeekableInputStream {
 
   private final FSDataInputStream stream;
 
@@ -74,11 +78,15 @@ public int read(ByteBuffer buf) throws IOException {
 
   @Override
   public void readFully(ByteBuffer buf) throws IOException {
+    readFully(stream, buf);
+  }
+
+  public static void readFully(FSDataInputStream stream, ByteBuffer buf) throws IOException {
     // 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 (buf.hasRemaining()) {
-      int readCount = read(buf);
+      int readCount = stream.read(buf);
       if (readCount == -1) {
         // 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.

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopStreams.java

@@ -23,28 +23,29 @@
 import org.apache.parquet.Log;
 import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.SeekableInputStream;
-import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
-import java.nio.ByteBuffer;
 
+/**
+ * Convenience methods to get Parquet abstractions for Hadoop data streams.
+ */
 public class HadoopStreams {
 
   private static final Log LOG = Log.getLog(SeekableInputStream.class);
 
-  private static final Class<?> byteBufferReadableClass;
-
-  private static final Constructor<SeekableInputStream> h2SeekableConstructor;
-
-  static {
-    byteBufferReadableClass = getReadableClass();
-    h2SeekableConstructor = getH2SeekableConstructor();
-  }
+  private static final Class<?> byteBufferReadableClass = getReadableClass();
+  private static final Constructor<SeekableInputStream> h2SeekableConstructor = getH2SeekableConstructor();
 
+  /**
+   * Wraps a {@link FSDataInputStream} in a {@link SeekableInputStream}
+   * implementation for Parquet readers.
+   *
+   * @param stream a Hadoop FSDataInputStream
+   * @return a SeekableInputStream
+   */
   public static SeekableInputStream wrap(FSDataInputStream stream) {
     if (byteBufferReadableClass != null && h2SeekableConstructor != null &&
-        byteBufferReadableClass.isInstance(stream) &&
-        supportsByteBufferReads(stream)) {
+        byteBufferReadableClass.isInstance(stream.getWrappedStream())) {
       try {
         return h2SeekableConstructor.newInstance(stream);
       } catch (InstantiationException e) {
@@ -62,23 +63,6 @@ public static SeekableInputStream wrap(FSDataInputStream stream) {
     }
   }
 
-  private static final ByteBuffer ZERO_LEN_BYTE_BUFFER = ByteBuffer.wrap(new byte[0]);
-
-  private static boolean supportsByteBufferReads(FSDataInputStream stream) {
-    // FSDataInputStream implements ByteBufferReadable, but may throw
-    // UnsupportedOperationException if it isn't actually supported. This tests
-    // whether the method throws by trying to read a 0-length buffer, which has
-    // no effect on the stream when it is supported.
-    try {
-      stream.read(ZERO_LEN_BYTE_BUFFER);
-      return true;
-    } catch (UnsupportedOperationException e) {
-      return false;
-    } catch (IOException e) {
-      return false;
-    }
-  }
-
   private static Class<?> getReadableClass() {
     try {
       return Class.forName("org.apache.hadoop.fs.ByteBufferReadable");

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/H2MockInputStream.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 org.apache.hadoop.fs.ByteBufferReadable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class H2MockInputStream extends MockInputStream implements ByteBufferReadable {
+  public H2MockInputStream(int... actualReadLengths) {
+    super(actualReadLengths);
+  }
+
+  @Override
+  public int read(ByteBuffer buf) throws IOException {
+    // this is inefficient, but simple for correctness tests of
+    // readFully(ByteBuffer)
+    byte[] temp = new byte[buf.remaining()];
+    int bytesRead = read(temp, 0, temp.length);
+    if (bytesRead > 0) {
+      buf.put(temp, 0, bytesRead);
+    }
+    return bytesRead;
+  }
+}

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/MockInputStream.java

@@ -0,0 +1,87 @@
+/*
+ *  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 org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+
+class MockInputStream extends ByteArrayInputStream
+    implements Seekable, PositionedReadable {
+  static final byte[] TEST_ARRAY = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
+
+  private int[] lengths;
+  private int current = 0;
+  MockInputStream(int... actualReadLengths) {
+    super(TEST_ARRAY);
+    this.lengths = actualReadLengths;
+  }
+
+  @Override
+  public synchronized int read(byte[] b, int off, int len) {
+    if (current < lengths.length) {
+      if (len <= lengths[current]) {
+        // when len == lengths[current], the next read will by 0 bytes
+        int bytesRead = super.read(b, off, len);
+        lengths[current] -= bytesRead;
+        return bytesRead;
+      } else {
+        int bytesRead = super.read(b, off, lengths[current]);
+        current += 1;
+        return bytesRead;
+      }
+    } else {
+      return super.read(b, off, len);
+    }
+  }
+
+  @Override
+  public int read(long position, byte[] buffer, int offset, int length) throws IOException {
+    seek(position);
+    return read(buffer, offset, length);
+  }
+
+  @Override
+  public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
+    throw new UnsupportedOperationException("Not actually supported.");
+  }
+
+  @Override
+  public void readFully(long position, byte[] buffer) throws IOException {
+    throw new UnsupportedOperationException("Not actually supported.");
+  }
+
+  @Override
+  public void seek(long pos) throws IOException {
+    this.pos = (int) pos;
+  }
+
+  @Override
+  public long getPos() throws IOException {
+    return this.pos;
+  }
+
+  @Override
+  public boolean seekToNewSource(long targetPos) throws IOException {
+    seek(targetPos);
+    return true;
+  }
+}