This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/parquet-java.git
The following commit(s) were added to refs/heads/master by this push:
new 0756c3d99 GH-3080: HadoopStreams to support
ByteBufferPositionedReadable (#3096)
0756c3d99 is described below
commit 0756c3d99ed024633616ec3cbceece57ac7c42fd
Author: Steve Loughran <[email protected]>
AuthorDate: Wed Sep 23 17:09:04 2026 +0100
GH-3080: HadoopStreams to support ByteBufferPositionedReadable (#3096)
### Rationale for this change
If a stream declares in its StreamCapabilities that it supports
`ByteBufferPositionedReadable`, then use that API for
`readFully(ByteBuffer)`
```
ByteBufferPositionedReadable.readFully(long position, ByteBuffer buf)
```
Adding support for Hadoop `ByteBufferPositionedReadable` streams may
improve performance
by pushing retry/recovery logic into the filesystem client library.
This interface is implemented by the HDFS input stream; we are considering
adding
it elsewhere.
### What changes are included in this PR?
* New SeekableInputStream implementation: `H3ByteBufferInputStream`
* Instantiated in `HadoopStreams` if the `FSDataInputStream` is considered
suitable.
* Tests for the new behavior and that no regressions are caused.
#### Class `H3ByteBufferInputStream`
The reading is done in a new class, `H3ByteBufferInputStream`, which
subclasses `H2ByteBufferInputStream`. This reduces the amount of duplicate
code, it just makes it a bit unclean.
The purist way to do it would be to create an abstract superclass
`HadoopInputStream` to hold all commonality between the the three input streams.
I'm happy to do this, just didn't want to doing some larger refactoring
without (a) showing the core design worked and (b) getting permission to do
this. Should I do this?
#### `HadoopStreams` changes
Selection of the new input stream is done if and only if the stream
declares the capability `in:preadbytebuffer`.
There is no equivalent of `isWrappedStreamByteBufferReadable()` which
recurses through
a chain of wrapped streams looking for the API.
If a stream doesn't declare its support for the API, it won't get picked up.
This is done knowing that the sole production implemenation which currently
exists,
the HDFS input stream, does declare this capability.
### Are these changes tested?
There is new test suite, for new behavior and ensuring that the integration
with
HadoopStreams still retains the correct behavior for existing streams.
Suite is parameterized on heap and direct buffers.
### Are there any user-facing changes?
No
Closes GH-3080
---
.../parquet/hadoop/util/H2SeekableInputStream.java | 4 +
.../hadoop/util/H3ByteBufferInputStream.java | 80 ++++
.../apache/parquet/hadoop/util/HadoopStreams.java | 43 +-
.../parquet/hadoop/util/MockHadoopInputStream.java | 25 +
.../hadoop/util/TestHadoop2ByteBufferReads.java | 13 +-
.../util/TestHadoop3ByteBufferReadFully.java | 525 +++++++++++++++++++++
6 files changed, 652 insertions(+), 38 deletions(-)
diff --git
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H2SeekableInputStream.java
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H2SeekableInputStream.java
index 6b65bdbaf..63565645d 100644
---
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H2SeekableInputStream.java
+++
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H2SeekableInputStream.java
@@ -100,6 +100,10 @@ class H2SeekableInputStream extends
DelegatingSeekableInputStream {
VectorIoBridge.instance().readVectoredRanges(stream, ranges, allocator);
}
+ protected Reader getReader() {
+ return reader;
+ }
+
public static void readFully(Reader reader, ByteBuffer buf) throws
IOException {
// unfortunately the Hadoop 2 APIs do not have a 'readFully' equivalent
for the byteBuffer read
// calls. The read(ByteBuffer) call might read fewer than
byteBuffer.hasRemaining() bytes. Thus we
diff --git
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H3ByteBufferInputStream.java
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H3ByteBufferInputStream.java
new file mode 100644
index 000000000..796e4c23a
--- /dev/null
+++
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H3ByteBufferInputStream.java
@@ -0,0 +1,80 @@
+/*
+ * 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;
+
+/**
+ * Class which implements {@link #readFully(ByteBuffer)} through
+ * {@code ByteBufferPositionedReadable.readFully()}.
+ * <p>This is implemented by HDFS.
+ */
+class H3ByteBufferInputStream extends H2SeekableInputStream {
+ public H3ByteBufferInputStream(final FSDataInputStream stream) {
+ super(stream);
+ }
+
+ @Override
+ public FSDataInputStream getStream() {
+ return (FSDataInputStream) super.getStream();
+ }
+
+ /**
+ * Read the buffer fully through use of {@code
ByteBufferPositionedReadable.readFully()}
+ * at the current location.
+ * <p>That operation is designed to not use the current reading position,
rather
+ * an absolute position is passed in.
+ * In the use here the original read position is saved, and
+ * after the read is finished a {@code seek()} call made to move the
+ * cursor on.
+ *
+ * @param buf a byte buffer to fill with data from the stream
+ *
+ * @throws EOFException the buffer length is greater than the file length
+ * @throws IOException other IO problems.
+ */
+ @Override
+ public void readFully(final ByteBuffer buf) throws EOFException, IOException
{
+ readBufferFully(getStream(), buf);
+ }
+
+ /**
+ * Read the buffer fully through use of {@code
ByteBufferPositionedReadable.readFully()}
+ * from the current location.
+ * That is it reads from {@code stream[pos]} to {@code stream[pos +
buf.remaining() -1]}
+ *
+ * @param buf a byte buffer to fill with data from the stream
+ * @return number of bytes read.
+ *
+ * @throws EOFException the buffer length is greater than the file length
+ * @throws IOException other IO problems.
+ */
+ static int readBufferFully(final FSDataInputStream stream, final ByteBuffer
buf) throws IOException {
+ // remember the current position
+ final long pos = stream.getPos();
+ final int size = buf.remaining();
+ stream.readFully(pos, buf);
+ // then move the read position on.
+ stream.seek(pos + size);
+ return size;
+ }
+}
diff --git
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopStreams.java
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopStreams.java
index fc84729c7..78bba1f91 100644
---
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopStreams.java
+++
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopStreams.java
@@ -21,7 +21,6 @@ package org.apache.parquet.hadoop.util;
import java.io.InputStream;
import java.util.Objects;
-import java.util.function.Function;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -47,42 +46,14 @@ public class HadoopStreams {
public static SeekableInputStream wrap(FSDataInputStream stream) {
Objects.requireNonNull(stream, "Cannot wrap a null input stream");
- // Try to check using hasCapabilities(str)
- Boolean hasCapabilitiesResult = isWrappedStreamByteBufferReadable(stream);
-
- // If it is null, then fall back to the old method
- if (hasCapabilitiesResult != null) {
- if (hasCapabilitiesResult) {
- return new H2SeekableInputStream(stream);
- } else {
- return new H1SeekableInputStream(stream);
- }
- }
-
- return unwrapByteBufferReadableLegacy(stream).apply(stream);
- }
-
- /**
- * Is the inner stream byte buffer readable?
- * The test is 'the stream is not FSDataInputStream
- * and implements ByteBufferReadable'
- * <p>
- * This logic is only used for Hadoop <2.9.x, and <3.x.x
- *
- * @param stream stream to probe
- * @return A H2SeekableInputStream to access, or H1SeekableInputStream if
the stream is not seekable
- */
- private static Function<FSDataInputStream, SeekableInputStream>
unwrapByteBufferReadableLegacy(
- FSDataInputStream stream) {
- InputStream wrapped = stream.getWrappedStream();
- if (wrapped instanceof FSDataInputStream) {
- LOG.debug("Checking on wrapped stream {} of {} whether is
ByteBufferReadable", wrapped, stream);
- return unwrapByteBufferReadableLegacy(((FSDataInputStream) wrapped));
- }
- if (stream.getWrappedStream() instanceof ByteBufferReadable) {
- return H2SeekableInputStream::new;
+ // Check using hasCapabilities(str)
+ if (stream.hasCapability("in:preadbytebuffer")) {
+ LOG.debug("Using ByteBufferPositionedReadable to read {}", stream);
+ return new H3ByteBufferInputStream(stream);
+ } else if (isWrappedStreamByteBufferReadable(stream)) {
+ return new H2SeekableInputStream(stream);
} else {
- return H1SeekableInputStream::new;
+ return new H1SeekableInputStream(stream);
}
}
diff --git
a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/MockHadoopInputStream.java
b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/MockHadoopInputStream.java
index 0e0c8f0db..58d494a20 100644
---
a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/MockHadoopInputStream.java
+++
b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/MockHadoopInputStream.java
@@ -20,6 +20,7 @@
package org.apache.parquet.hadoop.util;
import java.io.ByteArrayInputStream;
+import java.io.EOFException;
import java.io.IOException;
import org.apache.hadoop.fs.PositionedReadable;
import org.apache.hadoop.fs.Seekable;
@@ -71,6 +72,7 @@ class MockHadoopInputStream extends ByteArrayInputStream
implements Seekable, Po
@Override
public void seek(long pos) throws IOException {
+ rejectNegativePosition(pos);
this.pos = (int) pos;
}
@@ -84,4 +86,27 @@ class MockHadoopInputStream extends ByteArrayInputStream
implements Seekable, Po
seek(targetPos);
return true;
}
+
+ /**
+ * How long is the actual test data.
+ * @return the test data
+ */
+ int length() {
+ return TEST_ARRAY.length;
+ }
+
+ byte[] data() {
+ return TEST_ARRAY;
+ }
+
+ /**
+ * For consistency with real Hadoop streams: reject negative positions.
+ * @param pos position to read/seek to.
+ * @throws EOFException if pos is negative
+ */
+ static void rejectNegativePosition(final long pos) throws EOFException {
+ if (pos < 0) {
+ throw new EOFException("Seek before file start: " + pos);
+ }
+ }
}
diff --git
a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop2ByteBufferReads.java
b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop2ByteBufferReads.java
index a3955af6f..9dd30e1cc 100644
---
a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop2ByteBufferReads.java
+++
b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop2ByteBufferReads.java
@@ -29,6 +29,8 @@ 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.StreamCapabilities;
+import org.apache.hadoop.util.StringUtils;
import org.apache.parquet.io.SeekableInputStream;
import org.junit.jupiter.api.Test;
@@ -403,13 +405,20 @@ public class TestHadoop2ByteBufferReads {
}
/**
- * Input stream which claims to implement ByteBufferReadable.
+ * Input stream which claims to implement ByteBufferReadable in both
interfaces and
+ * in {@code hasCapability()}.
*/
- private static final class MockByteBufferInputStream extends
MockHadoopInputStream implements ByteBufferReadable {
+ private static final class MockByteBufferInputStream extends
MockHadoopInputStream
+ implements ByteBufferReadable, StreamCapabilities {
@Override
public int read(final ByteBuffer buf) {
return 0;
}
+
+ @Override
+ public boolean hasCapability(final String capability) {
+ return StringUtils.toLowerCase(capability).equals(READBYTEBUFFER);
+ }
}
}
diff --git
a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop3ByteBufferReadFully.java
b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop3ByteBufferReadFully.java
new file mode 100644
index 000000000..4cf9827cc
--- /dev/null
+++
b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/util/TestHadoop3ByteBufferReadFully.java
@@ -0,0 +1,525 @@
+/*
+ * 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 static org.apache.hadoop.fs.StreamCapabilities.READBYTEBUFFER;
+import static
org.apache.parquet.hadoop.util.H3ByteBufferInputStream.readBufferFully;
+import static org.apache.parquet.hadoop.util.HadoopStreams.wrap;
+import static org.apache.parquet.hadoop.util.MockHadoopInputStream.TEST_ARRAY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.parquet.io.SeekableInputStream;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * Test {@code ByteBufferPositionedReadable.readFully()} reads.
+ * Parameterized on heap vs. direct buffers.
+ */
+@ParameterizedClass(name = "heap={0}")
+@ValueSource(booleans = {true, false})
+public class TestHadoop3ByteBufferReadFully {
+
+ /**
+ * The size of the stream.
+ */
+ private static final int LEN = TEST_ARRAY.length;
+
+ /**
+ * Use a heap buffer?
+ */
+ private final boolean useHeap;
+
+ /**
+ * Instantiate test suite.
+ *
+ * @param useHeap use a heap buffer?
+ */
+ public TestHadoop3ByteBufferReadFully(final boolean useHeap) {
+ this.useHeap = useHeap;
+ }
+
+ /**
+ * Allocate a buffer; choice of on/off heap depends on test suite options.
+ *
+ * @param capacity buffer capacity.
+ *
+ * @return the buffer.
+ */
+ private ByteBuffer allocate(int capacity) {
+ return useHeap ? ByteBuffer.allocate(capacity) :
ByteBuffer.allocateDirect(capacity);
+ }
+
+ /**
+ * Read a buffer smaller than the source file.
+ */
+ @Test
+ public void testReadFullySmallBuffer() throws Exception {
+ ByteBuffer readBuffer = allocate(8);
+
+ FSDataInputStream hadoopStream = stream();
+
+ assertBufferRead(hadoopStream, readBuffer, 8, 8);
+ assertPositionAndLimit(readBuffer, 8, 8);
+ // buffer is full so no more data is read.
+ assertBufferRead(hadoopStream, readBuffer, 8, 8);
+ verifyBufferMatches(readBuffer, 0);
+ }
+
+ /**
+ * Read more than the file size, require an EOF exception to be raised.
+ */
+ @Test
+ public void testReadFullyLargeBuffer() {
+ final ByteBuffer readBuffer = allocate(20);
+
+ FSDataInputStream hadoopStream = stream();
+
+ assertThrowsEOFException(hadoopStream, readBuffer);
+
+ // EOF check happened before the read -at least with this test stream.
+ assertPositionAndLimit(readBuffer, 0, 20);
+ }
+
+ /**
+ * Seek to the file, try to read a buffer more than allowed.
+ * This fails because readFully() requires the whole buffer to be filled.
+ * When the buffer limit is reduced it will work.
+ */
+ @Test
+ public void testReadFullyFromOffset() throws Exception {
+ final int size = 5;
+ final ByteBuffer readBuffer = allocate(size);
+
+ FSDataInputStream hadoopStream = stream();
+ hadoopStream.seek(6);
+
+ // read past EOF is a failure
+ assertThrowsEOFException(hadoopStream, readBuffer);
+ // stream does not change position
+ assertStreamAt(hadoopStream, 6);
+
+ // reduce buffer limit
+ readBuffer.limit(4);
+ // now the read works.
+ assertBufferRead(hadoopStream, readBuffer, 4, 4);
+ }
+
+ /**
+ * Create a data input stream wrapping an {@link
MockByteBufferReadFullyInputStream}.
+ *
+ * @return in input stream.
+ */
+ private static FSDataInputStream stream() {
+ return new FSDataInputStream(new MockByteBufferReadFullyInputStream());
+ }
+
+ /**
+ * Read exactly the size of the file.
+ */
+ @Test
+ public void testReadFullyJustRight() throws Exception {
+ ByteBuffer readBuffer = allocate(LEN);
+
+ FSDataInputStream hadoopStream = stream();
+
+ // reads all of the bytes available without EOFException
+ assertBufferRead(hadoopStream, readBuffer, LEN, LEN);
+
+ // trying to read 0 more bytes doesn't result in EOFException
+ hadoopStream.readFully(11, readBuffer);
+
+ // buffer unchanged
+ verifyBufferMatches(readBuffer, 0);
+ }
+
+ /**
+ * Read with the buffer position set to a value within the buffer.
+ */
+ @Test
+ public void testReadFullyPosition() throws Exception {
+ ByteBuffer readBuffer = allocate(LEN);
+ readBuffer.position(3);
+ readBuffer.mark();
+
+ FSDataInputStream hadoopStream = stream();
+ assertBufferRead(hadoopStream, readBuffer, LEN, LEN);
+ assertBufferRead(hadoopStream, readBuffer, LEN, LEN);
+
+ // reset to where the mark is.
+ readBuffer.reset();
+ assertThat(readBuffer).as("Buffer contents should
match").isEqualTo(ByteBuffer.wrap(TEST_ARRAY, 0, 7));
+ }
+
+ /**
+ * Limit the buffer size, read it.
+ */
+ @Test
+ public void testReadFullyLimit() throws Exception {
+ ByteBuffer readBuffer = allocate(LEN);
+ final int smallLimit = 7;
+ readBuffer.limit(smallLimit);
+
+ // read up to the limit, twice.
+ FSDataInputStream hadoopStream = stream();
+ assertBufferRead(hadoopStream, readBuffer, smallLimit, smallLimit);
+ hadoopStream.seek(0);
+ // the buffer is now full, so no bytes are read.
+ // the the position and the limit are unchanged.
+ assertBufferRead(hadoopStream, readBuffer, smallLimit, smallLimit);
+ // and the stream is still at position zero.
+ assertStreamAt(hadoopStream, 0);
+
+ verifyBufferMatches(readBuffer, 0);
+
+ // recycle the buffer with a larger value and continue
+ // reading from the end of the last read.
+ readBuffer.position(smallLimit);
+ readBuffer.limit(LEN);
+ hadoopStream.seek(smallLimit);
+
+ assertStreamAt(hadoopStream, smallLimit);
+ assertBufferRead(hadoopStream, readBuffer, LEN, LEN);
+ verifyBufferMatches(readBuffer, 0);
+ }
+
+ @Test
+ public void testReadFullyPositionAndLimit() throws Exception {
+ ByteBuffer readBuffer = allocate(LEN);
+ readBuffer.position(3);
+ final int smallLimit = 7;
+ readBuffer.limit(smallLimit);
+ readBuffer.mark();
+
+ FSDataInputStream hadoopStream = stream();
+
+ assertBufferRead(hadoopStream, readBuffer, smallLimit, smallLimit);
+ assertBufferRead(hadoopStream, readBuffer, smallLimit, smallLimit);
+
+ readBuffer.reset();
+ assertThat(readBuffer).as("Buffer contents should
match").isEqualTo(ByteBuffer.wrap(TEST_ARRAY, 0, 4));
+
+ readBuffer.position(smallLimit);
+ readBuffer.limit(LEN);
+
+ assertBufferRead(hadoopStream, readBuffer, LEN, LEN);
+ readBuffer.reset();
+ assertThat(readBuffer).as("Buffer contents should
match").isEqualTo(ByteBuffer.wrap(TEST_ARRAY, 0, smallLimit));
+ }
+
+ /**
+ * Assert that a buffer read raises EOFException.
+ *
+ * @param hadoopStream stream to read
+ * @param readBuffer target buffer.
+ */
+ private static void assertThrowsEOFException(final FSDataInputStream
hadoopStream, final ByteBuffer readBuffer) {
+ assertThatThrownBy(() -> readBufferFully(hadoopStream,
readBuffer)).isInstanceOf(EOFException.class);
+ }
+
+ /**
+ * Regression test: verify that creating a stream for {@link
MockHadoopInputStream}
+ * still generates an {@link H1SeekableInputStream}.
+ */
+ @Test
+ public void testCreateH1Stream() {
+ assertStreamClass(H1SeekableInputStream.class, wrap(new
FSDataInputStream(new MockHadoopInputStream())));
+ }
+
+ /**
+ * Regression test: verify that creating a stream which implements
+ * ByteBufferReadable but doesn't declare the capability generates {@link
H2SeekableInputStream}.
+ */
+ @Test
+ public void testDoubleWrapByteBufferStream() {
+ assertStreamClass(
+ H2SeekableInputStream.class,
+ wrap(new FSDataInputStream(new FSDataInputStream(new
MockByteBufferInputStream()))));
+ }
+
+ /**
+ * Regression test: verify that creating a stream which implements
+ * ByteBufferReadable generates {@link H2SeekableInputStream}.
+ */
+ @Test
+ public void testDoubleWrapByteBufferStreamWithCapability() {
+ assertStreamClass(
+ H2SeekableInputStream.class,
+ wrap(new FSDataInputStream(new FSDataInputStream(new
MockByteBufferInputStream(READBYTEBUFFER)))));
+ }
+
+ /**
+ * Assert that an instantiated stream class matches the expected class.
+ * @param expected expected class
+ * @param stream stream to validate
+ */
+ private static void assertStreamClass(
+ final Class<? extends SeekableInputStream> expected, final
SeekableInputStream stream) {
+ assertThat(stream.getClass()).as("Wrong stream class: " +
stream).isEqualTo(expected);
+ }
+
+ /**
+ * If a stream implements "in:preadbytebuffer" it gets bound to a
H3ByteBufferInputStream.
+ */
+ @Test
+ public void testCreateStreamWithByteBufferPositionedReadable() {
+ assertStreamClass(H3ByteBufferInputStream.class, wrap(stream()));
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testDoubleWrapByteBufferPositionedReadable() {
+ assertStreamClass(
+ H3ByteBufferInputStream.class,
+ wrap(new FSDataInputStream(new FSDataInputStream(new
MockByteBufferReadFullyInputStream()))));
+ }
+
+ /**
+ * The buffer reading stream is only selected if the stream declares support;
+ * implementing the interface is not enough.
+ */
+ @Test
+ public void testPositionedReadableNoCapability() {
+ assertStreamClass(
+ H2SeekableInputStream.class,
+ wrap(new FSDataInputStream(new
MockByteBufferReadFullyInputStream(READBYTEBUFFER))));
+ }
+
+ /**
+ * What happens if a stream declares support for the interface,
+ * but doesn't actually do it?
+ * The check is based on trust: if the stream lied -it doesn't work.
+ */
+ @Test
+ public void testCapabilityWithoutInterface() {
+ class InconsistentStream extends MockHadoopInputStream
+ implements ByteBufferPositionedReadable, StreamCapabilities {
+
+ @Override
+ public boolean hasCapability(final String capability) {
+ return StringUtils.toLowerCase(capability).equals(PREADBYTEBUFFER);
+ }
+
+ @Override
+ public int read(final long position, final ByteBuffer buf) {
+ return 0;
+ }
+
+ @Override
+ public void readFully(final long position, final ByteBuffer buf) {}
+ }
+
+ assertStreamClass(H3ByteBufferInputStream.class, wrap(new
FSDataInputStream(new InconsistentStream())));
+ }
+
+ /**
+ * Assert that the buffer contents match those of the input data from
+ * the offset filePosition.
+ * This operation reads the buffer data, so must be used after any other
+ * assertions about buffer, size, position etc.
+ *
+ * @param readBuffer buffer to examine
+ * @param filePosition file position.
+ */
+ public static void verifyBufferMatches(ByteBuffer readBuffer, int
filePosition) {
+ readBuffer.flip();
+ final int remaining = readBuffer.remaining();
+ byte[] actual = getBytes(readBuffer);
+ byte[] expected = Arrays.copyOfRange(TEST_ARRAY, filePosition, remaining);
+ assertThat(stringify(actual))
+ .as("Buffer contents from data offset " + filePosition + " with length
" + remaining)
+ .isEqualTo(stringify(expected));
+ }
+
+ /**
+ * Gets the bytes of the buffer. This sets the buffer.remaining()
+ * value to 0.
+ *
+ * @param buffer buffer.
+ *
+ * @return buffer contents as bytes.
+ */
+ public static byte[] getBytes(ByteBuffer buffer) {
+ byte[] byteArray = new byte[buffer.remaining()];
+ buffer.get(byteArray);
+ return byteArray;
+ }
+
+ /**
+ * Map a byte array to hex values.
+ * Of limited value once the byte value is greater than 15
+ * as the string is hard to read.
+ *
+ * @param array source data
+ *
+ * @return string list.
+ */
+ private static String stringify(byte[] array) {
+ // convert to offset of lower case A, to make those assertions meaningful
+ final int l = array.length;
+ StringBuilder chars = new StringBuilder(l);
+ for (byte b : array) {
+ chars.append(Integer.toHexString(b));
+ }
+ return chars.toString();
+ }
+
+ /**
+ * Assert the current buffer position and limit are as expected
+ *
+ * @param readBuffer buffer
+ * @param bufferPosition buffer position.
+ * @param limit buffer limit
+ */
+ private static void assertPositionAndLimit(ByteBuffer readBuffer, int
bufferPosition, int limit) {
+ assertThat(readBuffer.position()).as("Buffer
Position").isEqualTo(bufferPosition);
+ assertThat(readBuffer.limit()).as("Buffer Limit").isEqualTo(limit);
+ }
+
+ /**
+ * Assert the stream position is at the expected value.
+ *
+ * @param hadoopStream stream
+ * @param pos expected position
+ *
+ * @throws IOException exception raised on getPos()
+ */
+ private static void assertStreamAt(final FSDataInputStream hadoopStream,
long pos) throws IOException {
+ assertThat(hadoopStream.getPos()).as("Read position of
stream").isEqualTo(pos);
+ }
+
+ /**
+ * Read a buffer at the current position through {@link
H3ByteBufferInputStream#readBufferFully(FSDataInputStream, ByteBuffer)}.
+ * Assert that the stream buffer position and limit are as expected.
+ * That is: the stream position has been moved forwards by the
+ * size of the buffer.
+ *
+ * @param hadoopStream stream
+ * @param readBuffer buffer to fill
+ * @param expectedBufferPosition final buffer position
+ * @param expectedLimit final buffer limit
+ *
+ * @throws IOException read failure
+ */
+ private static void assertBufferRead(
+ final FSDataInputStream hadoopStream,
+ final ByteBuffer readBuffer,
+ final int expectedBufferPosition,
+ final int expectedLimit)
+ throws IOException {
+ final long pos = hadoopStream.getPos();
+ final int remaining = readBuffer.remaining();
+ final int read = readBufferFully(hadoopStream, readBuffer);
+ // the bytes read MUST match the buffer size, as this is a full buffer
read.
+ assertThat(read).as("bytes read from stream").isEqualTo(remaining);
+ // the buffer position and limit match what was expected.
+ assertPositionAndLimit(readBuffer, expectedBufferPosition, expectedLimit);
+ // the stream has moved forwards.
+ assertStreamAt(hadoopStream, pos + remaining);
+ }
+
+ /**
+ * Input stream which claims to implement ByteBufferReadable in both
interfaces and, optionally,
+ * in {@code hasCapability()}.
+ */
+ private static class MockByteBufferInputStream extends MockHadoopInputStream
+ implements ByteBufferReadable, StreamCapabilities {
+
+ private final String[] capabilities;
+
+ /**
+ * Constructor.
+ * @param capabilities an array of capabilities to declare support for.
+ */
+ private MockByteBufferInputStream(String... capabilities) {
+ this.capabilities = capabilities;
+ }
+
+ @Override
+ public int read(final ByteBuffer buf) {
+ return 0;
+ }
+
+ /**
+ * Does a stream have the
+ * @param capability string to query the stream support for.
+ * @return true if there is an entry in the capability list matching the
argument.
+ */
+ @Override
+ public boolean hasCapability(final String capability) {
+ return Arrays.asList(capabilities).contains(capability);
+ }
+ }
+
+ /**
+ * Input stream which claims to implement ByteBufferPositionedReadable,
+ * unless constructed with a capability list that excludes it.
+ */
+ private static class MockByteBufferReadFullyInputStream extends
MockByteBufferInputStream
+ implements ByteBufferPositionedReadable, StreamCapabilities {
+
+ public MockByteBufferReadFullyInputStream() {
+ this(READBYTEBUFFER, PREADBYTEBUFFER);
+ }
+
+ public MockByteBufferReadFullyInputStream(final String... capabilites) {
+ super(capabilites);
+ }
+
+ @Override
+ public int read(final long position, final ByteBuffer buf) throws
IOException {
+ rejectNegativePosition(position);
+ return 0;
+ }
+
+ @Override
+ public void readFully(final long position, final ByteBuffer buf) throws
IOException {
+
+ // validation
+ rejectNegativePosition(position);
+ final int toRead = buf.remaining();
+ if (toRead == 0) {
+ return;
+ }
+ if (toRead + position > length()) {
+ throw new EOFException("ByteBuffer.readFully(" + position
+ + ") buffer size: " + toRead
+ + " reads past file length: " + length());
+ }
+ // return the subset of the data
+ byte[] result = new byte[toRead];
+ System.arraycopy(data(), (int) position, result, 0, toRead);
+ buf.put(result);
+ }
+ }
+}
