You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2017/09/15 15:08:36 UTC
[2/3] beam git commit: [BEAM-2790] Use byte[] instead of ByteBuffer
to read from HadoopFilesystem
[BEAM-2790] Use byte[] instead of ByteBuffer to read from HadoopFilesystem
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/c3297526
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/c3297526
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/c3297526
Branch: refs/heads/master
Commit: c3297526bc97fda202952f38156be9788910c8fc
Parents: c09886a
Author: Etienne Chauchot <ec...@gmail.com>
Authored: Thu Sep 14 10:28:12 2017 +0200
Committer: Luke Cwik <lc...@google.com>
Committed: Fri Sep 15 08:04:11 2017 -0700
----------------------------------------------------------------------
.../beam/sdk/io/hdfs/HadoopFileSystem.java | 16 ++++-
.../beam/sdk/io/hdfs/HadoopFileSystemTest.java | 68 ++++++++++++++------
2 files changed, 63 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/c3297526/sdks/java/io/hadoop-file-system/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/hadoop-file-system/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystem.java b/sdks/java/io/hadoop-file-system/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystem.java
index 803ddb6..ec71a53 100644
--- a/sdks/java/io/hadoop-file-system/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystem.java
+++ b/sdks/java/io/hadoop-file-system/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystem.java
@@ -189,7 +189,21 @@ class HadoopFileSystem extends FileSystem<HadoopResourceId> {
if (closed) {
throw new IOException("Channel is closed");
}
- return inputStream.read(dst);
+ // O length read must be supported
+ int read = 0;
+ // We avoid using the ByteBuffer based read for Hadoop because some FSDataInputStream
+ // implementations are not ByteBufferReadable,
+ // See https://issues.apache.org/jira/browse/HADOOP-14603
+ if (dst.hasArray()){
+ // does the same as inputStream.read(dst):
+ // stores up to dst.remaining() bytes into dst.array() starting at dst.position().
+ // But dst can have an offset with its backing array hence the + dst.arrayOffset()
+ read = inputStream.read(dst.array(), dst.position() + dst.arrayOffset(), dst.remaining());
+ }
+ if (read > 0) {
+ dst.position(dst.position() + read);
+ }
+ return read;
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/c3297526/sdks/java/io/hadoop-file-system/src/test/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystemTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/hadoop-file-system/src/test/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystemTest.java b/sdks/java/io/hadoop-file-system/src/test/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystemTest.java
index d06142d..18d78c9 100644
--- a/sdks/java/io/hadoop-file-system/src/test/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystemTest.java
+++ b/sdks/java/io/hadoop-file-system/src/test/java/org/apache/beam/sdk/io/hdfs/HadoopFileSystemTest.java
@@ -28,11 +28,13 @@ import static org.junit.Assert.assertThat;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.io.ByteStreams;
+import java.io.InputStream;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
+import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
@@ -86,8 +88,28 @@ public class HadoopFileSystemTest {
@Test
public void testCreateAndReadFile() throws Exception {
- create("testFile", "testData".getBytes());
- assertArrayEquals("testData".getBytes(), read("testFile"));
+ byte[] bytes = "testData".getBytes();
+ create("testFile", bytes);
+ assertArrayEquals(bytes, read("testFile", 0));
+ }
+
+ @Test
+ public void testCreateAndReadFileWithShift() throws Exception {
+ byte[] bytes = "testData".getBytes();
+ create("testFile", bytes);
+ int bytesToSkip = 3;
+ byte[] expected = Arrays.copyOfRange(bytes, bytesToSkip, bytes.length);
+ byte[] actual = read("testFile", bytesToSkip);
+ assertArrayEquals(expected, actual);
+ }
+
+ @Test
+ public void testCreateAndReadFileWithShiftToEnd() throws Exception {
+ byte[] bytes = "testData".getBytes();
+ create("testFile", bytes);
+ int bytesToSkip = bytes.length;
+ byte[] expected = Arrays.copyOfRange(bytes, bytesToSkip, bytes.length);
+ assertArrayEquals(expected, read("testFile", bytesToSkip));
}
@Test
@@ -101,10 +123,10 @@ public class HadoopFileSystemTest {
ImmutableList.of(
testPath("copyTestFileA"),
testPath("copyTestFileB")));
- assertArrayEquals("testDataA".getBytes(), read("testFileA"));
- assertArrayEquals("testDataB".getBytes(), read("testFileB"));
- assertArrayEquals("testDataA".getBytes(), read("copyTestFileA"));
- assertArrayEquals("testDataB".getBytes(), read("copyTestFileB"));
+ assertArrayEquals("testDataA".getBytes(), read("testFileA", 0));
+ assertArrayEquals("testDataB".getBytes(), read("testFileB", 0));
+ assertArrayEquals("testDataA".getBytes(), read("copyTestFileA", 0));
+ assertArrayEquals("testDataB".getBytes(), read("copyTestFileB", 0));
}
@Test
@@ -114,9 +136,9 @@ public class HadoopFileSystemTest {
create("testFileC", "testDataC".getBytes());
// ensure files exist
- assertArrayEquals("testDataA".getBytes(), read("testFileA"));
- assertArrayEquals("testDataB".getBytes(), read("testFileB"));
- assertArrayEquals("testDataC".getBytes(), read("testFileC"));
+ assertArrayEquals("testDataA".getBytes(), read("testFileA", 0));
+ assertArrayEquals("testDataB".getBytes(), read("testFileB", 0));
+ assertArrayEquals("testDataC".getBytes(), read("testFileC", 0));
fileSystem.delete(ImmutableList.of(
testPath("testFileA"),
@@ -139,9 +161,9 @@ public class HadoopFileSystemTest {
create("testFileB", "testDataB".getBytes());
// ensure files exist
- assertArrayEquals("testDataAA".getBytes(), read("testFileAA"));
- assertArrayEquals("testDataA".getBytes(), read("testFileA"));
- assertArrayEquals("testDataB".getBytes(), read("testFileB"));
+ assertArrayEquals("testDataAA".getBytes(), read("testFileAA", 0));
+ assertArrayEquals("testDataA".getBytes(), read("testFileA", 0));
+ assertArrayEquals("testDataB".getBytes(), read("testFileB", 0));
List<MatchResult> results =
fileSystem.match(ImmutableList.of(testPath("testFileA*").toString()));
@@ -165,8 +187,8 @@ public class HadoopFileSystemTest {
create("testFileBB", "testDataBB".getBytes());
// ensure files exist
- assertArrayEquals("testDataAA".getBytes(), read("testFileAA"));
- assertArrayEquals("testDataBB".getBytes(), read("testFileBB"));
+ assertArrayEquals("testDataAA".getBytes(), read("testFileAA", 0));
+ assertArrayEquals("testDataBB".getBytes(), read("testFileBB", 0));
List<MatchResult> matchResults = fileSystem.match(ImmutableList.of(
testPath("testFileAA").toString(),
@@ -196,8 +218,8 @@ public class HadoopFileSystemTest {
create("testFileB", "testDataB".getBytes());
// ensure files exist
- assertArrayEquals("testDataA".getBytes(), read("testFileA"));
- assertArrayEquals("testDataB".getBytes(), read("testFileB"));
+ assertArrayEquals("testDataA".getBytes(), read("testFileA", 0));
+ assertArrayEquals("testDataB".getBytes(), read("testFileB", 0));
fileSystem.rename(
ImmutableList.of(
@@ -221,8 +243,8 @@ public class HadoopFileSystemTest {
.build()));
// ensure files exist
- assertArrayEquals("testDataA".getBytes(), read("renameFileA"));
- assertArrayEquals("testDataB".getBytes(), read("renameFileB"));
+ assertArrayEquals("testDataA".getBytes(), read("renameFileA", 0));
+ assertArrayEquals("testDataB".getBytes(), read("renameFileB", 0));
}
@Test
@@ -267,9 +289,15 @@ public class HadoopFileSystemTest {
}
}
- private byte[] read(String relativePath) throws Exception {
+ private byte[] read(String relativePath, long bytesToSkip) throws Exception {
try (ReadableByteChannel channel = fileSystem.open(testPath(relativePath))) {
- return ByteStreams.toByteArray(Channels.newInputStream(channel));
+ InputStream inputStream = Channels.newInputStream(channel);
+ if (bytesToSkip > 0) {
+ long bytesSkip = inputStream.skip(bytesToSkip);
+ // might skip less than bytesToSkip
+ assertEquals(bytesSkip, bytesToSkip);
+ }
+ return ByteStreams.toByteArray(inputStream);
}
}