This is an automated email from the ASF dual-hosted git repository.
omalley pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/orc.git
The following commit(s) were added to refs/heads/branch-1.5 by this push:
new cafcb41 ORC-435: Ability to read stripes that are greater than 2GB.
cafcb41 is described below
commit cafcb410a67581821213e54a02f6bf2f39ddd5e3
Author: Prasanth Jayachandran <[email protected]>
AuthorDate: Fri Nov 9 16:19:39 2018 -0800
ORC-435: Ability to read stripes that are greater than 2GB.
Fixes #338
Signed-off-by: Owen O'Malley <[email protected]>
---
java/core/src/java/org/apache/orc/OrcConf.java | 19 ++
.../org/apache/orc/impl/DataReaderProperties.java | 13 ++
.../java/org/apache/orc/impl/RecordReaderImpl.java | 8 +-
.../org/apache/orc/impl/RecordReaderUtils.java | 73 ++++----
.../org/apache/orc/impl/TestOrcLargeStripe.java | 199 +++++++++++++++++++++
5 files changed, 280 insertions(+), 32 deletions(-)
diff --git a/java/core/src/java/org/apache/orc/OrcConf.java
b/java/core/src/java/org/apache/orc/OrcConf.java
index bdf8c0d..23e301e 100644
--- a/java/core/src/java/org/apache/orc/OrcConf.java
+++ b/java/core/src/java/org/apache/orc/OrcConf.java
@@ -158,6 +158,9 @@ public enum OrcConf {
+ "HDFS blocks."),
DIRECT_ENCODING_COLUMNS("orc.column.encoding.direct",
"orc.column.encoding.direct", "",
"Comma-separated list of columns for which dictionary encoding is to be
skipped."),
+ // some JVM doesn't allow array creation of size Integer.MAX_VALUE, so chunk
size is slightly less than max int
+ ORC_MAX_DISK_RANGE_CHUNK_LIMIT("orc.max.disk.range.chunk.limit",
"hive.exec.orc.max.disk.range.chunk.limit",
+ Integer.MAX_VALUE - 1024, "When reading stripes >2GB, specify max limit
for the chunk size.")
;
private final String attribute;
@@ -205,6 +208,22 @@ public enum OrcConf {
return result;
}
+ public int getInt(Properties tbl, Configuration conf) {
+ String value = lookupValue(tbl, conf);
+ if (value != null) {
+ return Integer.parseInt(value);
+ }
+ return ((Number) defaultValue).intValue();
+ }
+
+ public int getInt(Configuration conf) {
+ return getInt(null, conf);
+ }
+
+ public void getInt(Configuration conf, int value) {
+ conf.setInt(attribute, value);
+ }
+
public long getLong(Properties tbl, Configuration conf) {
String value = lookupValue(tbl, conf);
if (value != null) {
diff --git a/java/core/src/java/org/apache/orc/impl/DataReaderProperties.java
b/java/core/src/java/org/apache/orc/impl/DataReaderProperties.java
index fbdc145..8420149 100644
--- a/java/core/src/java/org/apache/orc/impl/DataReaderProperties.java
+++ b/java/core/src/java/org/apache/orc/impl/DataReaderProperties.java
@@ -20,6 +20,7 @@ package org.apache.orc.impl;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.orc.CompressionKind;
+import org.apache.orc.OrcConf;
public final class DataReaderProperties {
@@ -29,6 +30,7 @@ public final class DataReaderProperties {
private final boolean zeroCopy;
private final int typeCount;
private final int bufferSize;
+ private final int maxDiskRangeChunkLimit;
private DataReaderProperties(Builder builder) {
this.fileSystem = builder.fileSystem;
@@ -37,6 +39,7 @@ public final class DataReaderProperties {
this.zeroCopy = builder.zeroCopy;
this.typeCount = builder.typeCount;
this.bufferSize = builder.bufferSize;
+ this.maxDiskRangeChunkLimit = builder.maxDiskRangeChunkLimit;
}
public FileSystem getFileSystem() {
@@ -63,6 +66,10 @@ public final class DataReaderProperties {
return bufferSize;
}
+ public int getMaxDiskRangeChunkLimit() {
+ return maxDiskRangeChunkLimit;
+ }
+
public static Builder builder() {
return new Builder();
}
@@ -75,6 +82,7 @@ public final class DataReaderProperties {
private boolean zeroCopy;
private int typeCount;
private int bufferSize;
+ private int maxDiskRangeChunkLimit = (int)
OrcConf.ORC_MAX_DISK_RANGE_CHUNK_LIMIT.getDefaultValue();
private Builder() {
@@ -110,6 +118,11 @@ public final class DataReaderProperties {
return this;
}
+ public Builder withMaxDiskRangeChunkLimit(int value) {
+ this.maxDiskRangeChunkLimit = value;
+ return this;
+ }
+
public DataReaderProperties build() {
if (fileSystem == null || path == null) {
throw new NullPointerException("Filesystem = " + fileSystem +
diff --git a/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
b/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
index 6d9fca8..dbf7e91 100644
--- a/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
+++ b/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
@@ -99,6 +99,7 @@ public class RecordReaderImpl implements RecordReader {
private final DataReader dataReader;
private final boolean ignoreNonUtf8BloomFilter;
private final OrcFile.WriterVersion writerVersion;
+ private final int maxDiskRangeChunkLimit;
/**
* Given a list of column names, find the given column and return the index.
@@ -231,7 +232,7 @@ public class RecordReaderImpl implements RecordReader {
rows += stripe.getNumberOfRows();
}
}
-
+ this.maxDiskRangeChunkLimit =
OrcConf.ORC_MAX_DISK_RANGE_CHUNK_LIMIT.getInt(fileReader.conf);
Boolean zeroCopy = options.getUseZeroCopy();
if (zeroCopy == null) {
zeroCopy = OrcConf.USE_ZEROCOPY.getBoolean(fileReader.conf);
@@ -247,6 +248,7 @@ public class RecordReaderImpl implements RecordReader {
.withPath(fileReader.path)
.withTypeCount(types.size())
.withZeroCopy(zeroCopy)
+ .withMaxDiskRangeChunkLimit(maxDiskRangeChunkLimit)
.build());
}
firstRow = skippedRows;
@@ -1454,4 +1456,8 @@ public class RecordReaderImpl implements RecordReader {
public CompressionCodec getCompressionCodec() {
return dataReader.getCompressionCodec();
}
+
+ public int getMaxDiskRangeChunkLimit() {
+ return maxDiskRangeChunkLimit;
+ }
}
diff --git a/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java
b/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java
index 705e768..3c43ba4 100644
--- a/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java
+++ b/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java
@@ -154,6 +154,7 @@ public class RecordReaderUtils {
private final int bufferSize;
private final int typeCount;
private CompressionKind compressionKind;
+ private final int maxDiskRangeChunkLimit;
private DefaultDataReader(DataReaderProperties properties) {
this.fs = properties.getFileSystem();
@@ -163,6 +164,7 @@ public class RecordReaderUtils {
this.codec = OrcCodecPool.getCodec(compressionKind);
this.bufferSize = properties.getBufferSize();
this.typeCount = properties.getTypeCount();
+ this.maxDiskRangeChunkLimit = properties.getMaxDiskRangeChunkLimit();
}
@Override
@@ -207,7 +209,7 @@ public class RecordReaderUtils {
DiskRangeList ranges = planIndexReading(fileSchema, footer,
ignoreNonUtf8BloomFilter, included, sargColumns, version,
bloomFilterKinds);
- ranges = readDiskRanges(file, zcr, stripe.getOffset(), ranges, false);
+ ranges = readDiskRanges(file, zcr, stripe.getOffset(), ranges, false,
maxDiskRangeChunkLimit);
long offset = 0;
DiskRangeList range = ranges;
for(OrcProto.Stream stream: footer.getStreamsList()) {
@@ -273,7 +275,7 @@ public class RecordReaderUtils {
@Override
public DiskRangeList readFileData(
DiskRangeList range, long baseOffset, boolean doForceDirect) throws
IOException {
- return RecordReaderUtils.readDiskRanges(file, zcr, baseOffset, range,
doForceDirect);
+ return RecordReaderUtils.readDiskRanges(file, zcr, baseOffset, range,
doForceDirect, maxDiskRangeChunkLimit);
}
@Override
@@ -521,8 +523,9 @@ public class RecordReaderUtils {
HadoopShims.ZeroCopyReaderShim zcr,
long base,
DiskRangeList range,
- boolean doForceDirect) throws IOException {
- if (range == null) return null;
+ boolean doForceDirect, int maxChunkLimit)
throws IOException {
+ if (range == null)
+ return null;
DiskRangeList prev = range.prev;
if (prev == null) {
prev = new DiskRangeList.MutateHelper(range);
@@ -532,39 +535,47 @@ public class RecordReaderUtils {
range = range.next;
continue;
}
- int len = (int) (range.getEnd() - range.getOffset());
+ boolean firstRead = true;
+ long len = range.getEnd() - range.getOffset();
long off = range.getOffset();
- if (zcr != null) {
- file.seek(base + off);
- boolean hasReplaced = false;
- while (len > 0) {
- ByteBuffer partial = zcr.readBuffer(len, false);
- BufferChunk bc = new BufferChunk(partial, off);
- if (!hasReplaced) {
- range.replaceSelfWith(bc);
- hasReplaced = true;
+ while (len > 0) {
+ // Stripe could be too large to read fully into a single buffer and
+ // will need to be chunked
+ int readSize = (len >= maxChunkLimit) ? maxChunkLimit : (int) len;
+ ByteBuffer partial;
+
+ // create chunk
+ if (zcr != null) {
+ if (firstRead) {
+ file.seek(base + off);
+ }
+
+ partial = zcr.readBuffer(readSize, false);
+ readSize = partial.remaining();
+ } else {
+ // Don't use HDFS ByteBuffer API because it has no readFully, and is
+ // buggy and pointless.
+ byte[] buffer = new byte[readSize];
+ file.readFully((base + off), buffer, 0, buffer.length);
+ if (doForceDirect) {
+ partial = ByteBuffer.allocateDirect(readSize);
+ partial.put(buffer);
+ partial.position(0);
+ partial.limit(readSize);
} else {
- range.insertAfter(bc);
+ partial = ByteBuffer.wrap(buffer);
}
- range = bc;
- int read = partial.remaining();
- len -= read;
- off += read;
}
- } else {
- // Don't use HDFS ByteBuffer API because it has no readFully, and is
buggy and pointless.
- byte[] buffer = new byte[len];
- file.readFully((base + off), buffer, 0, buffer.length);
- ByteBuffer bb = null;
- if (doForceDirect) {
- bb = ByteBuffer.allocateDirect(len);
- bb.put(buffer);
- bb.position(0);
- bb.limit(len);
+ BufferChunk bc = new BufferChunk(partial, off);
+ if (firstRead) {
+ range.replaceSelfWith(bc);
} else {
- bb = ByteBuffer.wrap(buffer);
+ range.insertAfter(bc);
}
- range = range.replaceSelfWith(new BufferChunk(bb, range.getOffset()));
+ firstRead = false;
+ range = bc;
+ len -= readSize;
+ off += readSize;
}
range = range.next;
}
diff --git a/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java
b/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java
new file mode 100644
index 0000000..37f25db
--- /dev/null
+++ b/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java
@@ -0,0 +1,199 @@
+/*
+ * Copyright 2015 The Apache Software Foundation.
+ *
+ * Licensed 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.orc.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+import static org.mockito.Mockito.anyBoolean;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.io.DiskRangeList;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.CompressionKind;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.RecordReader;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestOrcLargeStripe {
+
+ private static final long MB = 1024 * 1024;
+ private Path workDir = new Path(System.getProperty("test.tmp.dir", "target"
+ File.separator + "test"
+ + File.separator + "tmp"));
+
+ Configuration conf;
+ FileSystem fs;
+ private Path testFilePath;
+
+ @Rule
+ public TestName testCaseName = new TestName();
+
+ @Before
+ public void openFileSystem() throws Exception {
+ conf = new Configuration();
+ fs = FileSystem.getLocal(conf);
+ testFilePath = new Path(workDir, "TestOrcFile." +
testCaseName.getMethodName() + ".orc");
+ fs.delete(testFilePath, false);
+ }
+
+ @Mock
+ private FSDataInputStream mockDataInput;
+
+ private DiskRangeList createRangeList(long... stripeSizes) {
+ DiskRangeList.CreateHelper list = new DiskRangeList.CreateHelper();
+ long prev = 0;
+ for (long stripe : stripeSizes) {
+ list.addOrMerge(prev, stripe, true, true);
+ prev = stripe;
+ }
+ return list.extract();
+ }
+
+ private void verifyDiskRanges(long stripeLength, int expectedChunks) throws
Exception {
+
+ DiskRangeList rangeList = createRangeList(stripeLength);
+
+ DiskRangeList newList = RecordReaderUtils.readDiskRanges(mockDataInput,
null, 0, rangeList, false, (int) (2 * MB));
+ assertEquals(expectedChunks, newList.listSize());
+
+ newList = RecordReaderUtils.readDiskRanges(mockDataInput, null, 0,
rangeList, true, (int) (2 * MB));
+ assertEquals(expectedChunks, newList.listSize());
+
+ HadoopShims.ZeroCopyReaderShim mockZcr =
mock(HadoopShims.ZeroCopyReaderShim.class);
+ when(mockZcr.readBuffer(anyInt(),
anyBoolean())).thenReturn(ByteBuffer.allocate((int) (2 * MB)));
+ newList = RecordReaderUtils.readDiskRanges(mockDataInput, mockZcr, 0,
rangeList, true, (int) (2 * MB));
+ assertEquals(expectedChunks, newList.listSize());
+ }
+
+ @Test
+ public void testStripeSizesBelowAndGreaterThanLimit() throws Exception {
+ verifyDiskRanges(MB, 1);
+ verifyDiskRanges(5 * MB, 3);
+ }
+
+
+ @Test
+ public void testConfigMaxChunkLimit() throws IOException {
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.getLocal(conf);
+ TypeDescription schema = TypeDescription.createTimestamp();
+ testFilePath = new Path(workDir, "TestOrcLargeStripe." +
+ testCaseName.getMethodName() + ".orc");
+ Writer writer = OrcFile.createWriter(testFilePath,
+
OrcFile.writerOptions(conf).setSchema(schema).stripeSize(100000).bufferSize(10000)
+ .version(OrcFile.Version.V_0_11).fileSystem(fs));
+ writer.close();
+
+ try {
+ OrcFile.ReaderOptions opts = OrcFile.readerOptions(conf);
+ Reader reader = OrcFile.createReader(testFilePath, opts);
+ RecordReader recordReader = reader.rows(new Reader.Options().range(0L,
Long.MAX_VALUE));
+ assertTrue(recordReader instanceof RecordReaderImpl);
+ assertEquals(Integer.MAX_VALUE - 1024, ((RecordReaderImpl)
recordReader).getMaxDiskRangeChunkLimit());
+
+ conf = new Configuration();
+ conf.setInt(OrcConf.ORC_MAX_DISK_RANGE_CHUNK_LIMIT.getHiveConfName(),
1000);
+ opts = OrcFile.readerOptions(conf);
+ reader = OrcFile.createReader(testFilePath, opts);
+ recordReader = reader.rows(new Reader.Options().range(0L,
Long.MAX_VALUE));
+ assertTrue(recordReader instanceof RecordReaderImpl);
+ assertEquals(1000, ((RecordReaderImpl)
recordReader).getMaxDiskRangeChunkLimit());
+ } finally {
+ fs.delete(testFilePath, false);
+ }
+ }
+
+ @Test
+ public void testStringDirectGreaterThan2GB() throws IOException {
+ final Runtime rt = Runtime.getRuntime();
+ assumeTrue(rt.maxMemory() > 4_000_000_000L);
+ TypeDescription schema = TypeDescription.createString();
+
+ conf.setDouble("hive.exec.orc.dictionary.key.size.threshold", 0.0);
+ Writer writer = OrcFile.createWriter(
+ testFilePath,
+ OrcFile.writerOptions(conf).setSchema(schema)
+ .compress(CompressionKind.NONE));
+ // 5000 is the lower bound for a stripe
+ int size = 5000;
+ int width = 500_000;
+
+ // generate a random string that is width characters long
+ Random random = new Random(123);
+ char[] randomChars= new char[width];
+ int posn = 0;
+ for(int length = 0; length < width && posn < randomChars.length; ++posn) {
+ char cp = (char) random.nextInt(Character.MIN_SUPPLEMENTARY_CODE_POINT);
+ // make sure we get a valid, non-surrogate
+ while (Character.isSurrogate(cp)) {
+ cp = (char) random.nextInt(Character.MIN_SUPPLEMENTARY_CODE_POINT);
+ }
+ // compute the length of the utf8
+ length += cp < 0x80 ? 1 : (cp < 0x800 ? 2 : 3);
+ randomChars[posn] = cp;
+ }
+
+ // put the random characters in as a repeating value.
+ VectorizedRowBatch batch = schema.createRowBatch();
+ BytesColumnVector string = (BytesColumnVector) batch.cols[0];
+ string.setVal(0, new String(randomChars, 0,
posn).getBytes(StandardCharsets.UTF_8));
+ string.isRepeating = true;
+ for(int rows=size; rows > 0; rows -= batch.size) {
+ batch.size = Math.min(rows, batch.getMaxSize());
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+
+ try {
+ Reader reader = OrcFile.createReader(testFilePath,
+ OrcFile.readerOptions(conf).filesystem(fs));
+ RecordReader rows = reader.rows();
+ batch = reader.getSchema().createRowBatch();
+ int rowsRead = 0;
+ while (rows.nextBatch(batch)) {
+ rowsRead += batch.size;
+ }
+ assertEquals(size, rowsRead);
+ } finally {
+ fs.delete(testFilePath, false);
+ }
+ }
+}
\ No newline at end of file
