You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@orc.apache.org by om...@apache.org on 2019/07/03 16:35:38 UTC
[orc] branch master updated: ORC-487. Refactor stripe planning to a
new class StripePlanner.
This is an automated email from the ASF dual-hosted git repository.
omalley pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/orc.git
The following commit(s) were added to refs/heads/master by this push:
new 184c0a9 ORC-487. Refactor stripe planning to a new class StripePlanner.
184c0a9 is described below
commit 184c0a9e27b95a361c9e3243da483d81e331a9fd
Author: Owen O'Malley <om...@apache.org>
AuthorDate: Mon Feb 20 14:10:14 2017 -0800
ORC-487. Refactor stripe planning to a new class StripePlanner.
Fixes #409
Signed-off-by: Owen O'Malley <om...@apache.org>
---
java/core/src/java/org/apache/orc/DataReader.java | 32 +-
.../apache/orc/impl/ConvertTreeReaderFactory.java | 8 +-
.../org/apache/orc/impl/DataReaderProperties.java | 40 +-
.../java/org/apache/orc/impl/RecordReaderImpl.java | 287 ++-----
.../org/apache/orc/impl/RecordReaderUtils.java | 524 +++++-------
.../org/apache/orc/impl/TreeReaderFactory.java | 232 +++---
.../org/apache/orc/impl/reader/StripePlanner.java | 529 ++++++++++++
.../org/apache/orc/TestNewIntegerEncoding.java | 6 +-
.../src/test/org/apache/orc/TestVectorOrcFile.java | 340 +++++++-
.../test/org/apache/orc/impl/MockDataReader.java | 132 +++
.../src/test/org/apache/orc/impl/MockStream.java | 56 ++
.../src/test/org/apache/orc/impl/MockStripe.java | 168 ++++
.../apache/orc/impl/TestDataReaderProperties.java | 11 +-
.../org/apache/orc/impl/TestOrcLargeStripe.java | 109 ++-
.../org/apache/orc/impl/TestRecordReaderImpl.java | 884 +++++++++++----------
.../org/apache/orc/impl/TestSchemaEvolution.java | 28 +-
.../src/java/org/apache/orc/tools/FileDump.java | 3 +-
17 files changed, 2109 insertions(+), 1280 deletions(-)
diff --git a/java/core/src/java/org/apache/orc/DataReader.java b/java/core/src/java/org/apache/orc/DataReader.java
index ed3acbb..7728cdc 100644
--- a/java/core/src/java/org/apache/orc/DataReader.java
+++ b/java/core/src/java/org/apache/orc/DataReader.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.hadoop.hive.common.io.DiskRangeList;
+import org.apache.orc.impl.BufferChunkList;
import org.apache.orc.impl.OrcIndex;
/** An abstract data reader that IO formats can use to read bytes from underlying storage. */
@@ -30,32 +31,21 @@ public interface DataReader extends AutoCloseable, Cloneable {
/** Opens the DataReader, making it ready to use. */
void open() throws IOException;
- OrcIndex readRowIndex(StripeInformation stripe,
- TypeDescription fileSchema,
- OrcProto.StripeFooter footer,
- boolean ignoreNonUtf8BloomFilter,
- boolean[] included,
- OrcProto.RowIndex[] indexes,
- boolean[] sargColumns,
- OrcFile.WriterVersion version,
- OrcProto.Stream.Kind[] bloomFilterKinds,
- OrcProto.BloomFilterIndex[] bloomFilterIndices
- ) throws IOException;
-
OrcProto.StripeFooter readStripeFooter(StripeInformation stripe) throws IOException;
- /** Reads the data.
+ /**
+ * Reads the data from the file.
*
- * Note that for the cases such as zero-copy read, caller must release the disk ranges
- * produced after being done with them. Call isTrackingDiskRanges to find out if this is needed.
- * @param range List if disk ranges to read. Ranges with data will be ignored.
- * @param baseOffset Base offset from the start of the file of the ranges in disk range list.
+ * Note that for the cases such as zero-copy read, caller must release the disk
+ * ranges produced after being done with them. Call isTrackingDiskRanges to
+ * find out if this is needed.
+ *
+ * @param range List of disk ranges to read. Ranges with data will be ignored.
* @param doForceDirect Whether the data should be read into direct buffers.
* @return New or modified list of DiskRange-s, where all the ranges are filled with data.
*/
- DiskRangeList readFileData(
- DiskRangeList range, long baseOffset, boolean doForceDirect) throws IOException;
-
+ BufferChunkList readFileData(BufferChunkList range,
+ boolean doForceDirect) throws IOException;
/**
* Whether the user should release buffers created by readFileData. See readFileData javadoc.
@@ -77,7 +67,7 @@ public interface DataReader extends AutoCloseable, Cloneable {
DataReader clone();
@Override
- public void close() throws IOException;
+ void close() throws IOException;
/**
* Returns the compression codec used by this datareader.
diff --git a/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java b/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
index b9fb48d..bd2e24c 100644
--- a/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
+++ b/java/core/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
@@ -37,8 +37,10 @@ import org.apache.hadoop.hive.ql.util.TimestampUtils;
import org.apache.hadoop.hive.serde2.io.DateWritable;
import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
import org.apache.orc.OrcProto;
+import org.apache.orc.StripeInformation;
import org.apache.orc.TypeDescription;
import org.apache.orc.TypeDescription.Category;
+import org.apache.orc.impl.reader.StripePlanner;
/**
* Convert ORC tree readers.
@@ -285,11 +287,9 @@ public class ConvertTreeReaderFactory extends TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
+ void startStripe(StripePlanner planner) throws IOException {
// Pass-thru.
- convertTreeReader.startStripe(streams, stripeFooter);
+ convertTreeReader.startStripe(planner);
}
@Override
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 386324c..763c001 100644
--- a/java/core/src/java/org/apache/orc/impl/DataReaderProperties.java
+++ b/java/core/src/java/org/apache/orc/impl/DataReaderProperties.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -20,18 +20,14 @@ package org.apache.orc.impl;
import org.apache.hadoop.fs.FSDataInputStream;
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 {
private final FileSystem fileSystem;
private final Path path;
private final FSDataInputStream file;
- private final CompressionKind compression;
+ private final InStream.StreamOptions compression;
private final boolean zeroCopy;
- private final int typeCount;
- private final int bufferSize;
private final int maxDiskRangeChunkLimit;
private DataReaderProperties(Builder builder) {
@@ -40,8 +36,6 @@ public final class DataReaderProperties {
this.file = builder.file;
this.compression = builder.compression;
this.zeroCopy = builder.zeroCopy;
- this.typeCount = builder.typeCount;
- this.bufferSize = builder.bufferSize;
this.maxDiskRangeChunkLimit = builder.maxDiskRangeChunkLimit;
}
@@ -57,7 +51,7 @@ public final class DataReaderProperties {
return file;
}
- public CompressionKind getCompression() {
+ public InStream.StreamOptions getCompression() {
return compression;
}
@@ -65,14 +59,6 @@ public final class DataReaderProperties {
return zeroCopy;
}
- public int getTypeCount() {
- return typeCount;
- }
-
- public int getBufferSize() {
- return bufferSize;
- }
-
public int getMaxDiskRangeChunkLimit() {
return maxDiskRangeChunkLimit;
}
@@ -86,11 +72,9 @@ public final class DataReaderProperties {
private FileSystem fileSystem;
private Path path;
private FSDataInputStream file;
- private CompressionKind compression;
+ private InStream.StreamOptions compression;
private boolean zeroCopy;
- private int typeCount;
- private int bufferSize;
- private int maxDiskRangeChunkLimit = (int) OrcConf.ORC_MAX_DISK_RANGE_CHUNK_LIMIT.getDefaultValue();
+ private int maxDiskRangeChunkLimit;
private Builder() {
@@ -111,7 +95,7 @@ public final class DataReaderProperties {
return this;
}
- public Builder withCompression(CompressionKind value) {
+ public Builder withCompression(InStream.StreamOptions value) {
this.compression = value;
return this;
}
@@ -121,18 +105,8 @@ public final class DataReaderProperties {
return this;
}
- public Builder withTypeCount(int value) {
- this.typeCount = value;
- return this;
- }
-
- public Builder withBufferSize(int value) {
- this.bufferSize = value;
- return this;
- }
-
public Builder withMaxDiskRangeChunkLimit(int value) {
- this.maxDiskRangeChunkLimit = value;
+ maxDiskRangeChunkLimit = value;
return this;
}
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 bf72982..d6ae7d5 100644
--- a/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
+++ b/java/core/src/java/org/apache/orc/impl/RecordReaderImpl.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -19,8 +19,6 @@ package org.apache.orc.impl;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.common.io.DiskRangeList;
-import org.apache.hadoop.hive.common.io.DiskRangeList.CreateHelper;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
@@ -48,6 +46,8 @@ import org.apache.orc.StringColumnStatistics;
import org.apache.orc.StripeInformation;
import org.apache.orc.TimestampColumnStatistics;
import org.apache.orc.TypeDescription;
+import org.apache.orc.impl.reader.ReaderEncryption;
+import org.apache.orc.impl.reader.StripePlanner;
import org.apache.orc.util.BloomFilter;
import org.apache.orc.util.BloomFilterIO;
import org.slf4j.Logger;
@@ -59,9 +59,7 @@ import java.sql.Date;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.TimeZone;
public class RecordReaderImpl implements RecordReader {
@@ -69,14 +67,10 @@ public class RecordReaderImpl implements RecordReader {
private static final boolean isLogDebugEnabled = LOG.isDebugEnabled();
protected final Path path;
private final long firstRow;
- private final List<StripeInformation> stripes =
- new ArrayList<StripeInformation>();
+ private final List<StripeInformation> stripes = new ArrayList<>();
private OrcProto.StripeFooter stripeFooter;
private final long totalRowCount;
protected final TypeDescription schema;
- private final List<OrcProto.Type> types;
- private final int bufferSize;
- private final SchemaEvolution evolution;
// the file included columns indexed by the file's column ids.
private final boolean[] fileIncluded;
private final long rowIndexStride;
@@ -84,20 +78,14 @@ public class RecordReaderImpl implements RecordReader {
private int currentStripe = -1;
private long rowBaseInStripe = 0;
private long rowCountInStripe = 0;
- private final Map<StreamName, InStream> streams =
- new HashMap<StreamName, InStream>();
- DiskRangeList bufferChunks = null;
private final TreeReaderFactory.TreeReader reader;
- private final OrcProto.RowIndex[] indexes;
- private final OrcProto.BloomFilterIndex[] bloomFilterIndices;
- private final OrcProto.Stream.Kind[] bloomFilterKind;
+ private final OrcIndex indexes;
private final SargApplier sargApp;
// an array about which row groups aren't skipped
private boolean[] includedRowGroups = null;
private final DataReader dataReader;
- private final boolean ignoreNonUtf8BloomFilter;
- private final OrcFile.WriterVersion writerVersion;
private final int maxDiskRangeChunkLimit;
+ private final StripePlanner planner;
/**
* Given a list of column names, find the given column and return the index.
@@ -134,50 +122,10 @@ public class RecordReaderImpl implements RecordReader {
return result;
}
- /**
- * Given a list of column names, find the given column and return the index.
- *
- * @param columnNames the list of potential column names
- * @param columnName the column name to look for
- * @param rootColumn offset the result with the rootColumn
- * @return the column number or -1 if the column wasn't found
- */
- private static int findColumns(String[] columnNames,
- String columnName,
- int rootColumn) {
- for(int i=0; i < columnNames.length; ++i) {
- if (columnName.equals(columnNames[i])) {
- return i + rootColumn;
- }
- }
- return -1;
- }
-
- /**
- * Find the mapping from predicate leaves to columns.
- * @param sargLeaves the search argument that we need to map
- * @param columnNames the names of the columns
- * @param rootColumn the offset of the top level row, which offsets the
- * result
- * @return an array mapping the sarg leaves to concrete column numbers
- * @deprecated Use #mapSargColumnsToOrcInternalColIdx(List, SchemaEvolution)
- */
- @Deprecated
- public static int[] mapSargColumnsToOrcInternalColIdx(List<PredicateLeaf> sargLeaves,
- String[] columnNames,
- int rootColumn) {
- int[] result = new int[sargLeaves.size()];
- Arrays.fill(result, -1);
- for(int i=0; i < result.length; ++i) {
- String colName = sargLeaves.get(i).getColumnName();
- result[i] = findColumns(columnNames, colName, rootColumn);
- }
- return result;
- }
-
protected RecordReaderImpl(ReaderImpl fileReader,
Reader.Options options) throws IOException {
- this.writerVersion = fileReader.getWriterVersion();
+ OrcFile.WriterVersion writerVersion = fileReader.getWriterVersion();
+ SchemaEvolution evolution;
if (options.getSchema() == null) {
if (LOG.isInfoEnabled()) {
LOG.info("Reader schema not provided -- using file schema " +
@@ -201,11 +149,14 @@ public class RecordReaderImpl implements RecordReader {
}
this.schema = evolution.getReaderSchema();
this.path = fileReader.path;
- this.types = fileReader.types;
- this.bufferSize = fileReader.bufferSize;
+ List<OrcProto.Type> types = fileReader.types;
+ InStream.StreamOptions unencryptedOptions = InStream.options()
+ .withCodec(OrcCodecPool.getCodec(fileReader.getCompressionKind()))
+ .withBufferSize(fileReader.getCompressionSize());
this.rowIndexStride = fileReader.rowIndexStride;
- this.ignoreNonUtf8BloomFilter =
- OrcConf.IGNORE_NON_UTF8_BLOOM_FILTERS.getBoolean(fileReader.conf);
+ boolean ignoreNonUtf8BloomFilter = OrcConf.IGNORE_NON_UTF8_BLOOM_FILTERS.getBoolean(fileReader.conf);
+ ReaderEncryption encryption = fileReader.getEncryption();
+ this.fileIncluded = evolution.getFileIncluded();
SearchArgument sarg = options.getSearchArgument();
if (sarg != null && rowIndexStride != 0) {
sargApp = new SargApplier(sarg,
@@ -240,11 +191,9 @@ public class RecordReaderImpl implements RecordReader {
} else {
DataReaderProperties.Builder builder =
DataReaderProperties.builder()
- .withBufferSize(bufferSize)
- .withCompression(fileReader.compressionKind)
+ .withCompression(unencryptedOptions)
.withFileSystem(fileReader.getFileSystem())
.withPath(fileReader.path)
- .withTypeCount(types.size())
.withMaxDiskRangeChunkLimit(maxDiskRangeChunkLimit)
.withZeroCopy(zeroCopy);
FSDataInputStream file = fileReader.takeFile();
@@ -266,14 +215,18 @@ public class RecordReaderImpl implements RecordReader {
.setSchemaEvolution(evolution)
.skipCorrupt(skipCorrupt)
.fileFormat(fileReader.getFileVersion())
- .useUTCTimestamp(fileReader.useUTCTimestamp);
+ .useUTCTimestamp(fileReader.useUTCTimestamp)
+ .setEncryption(encryption);
reader = TreeReaderFactory.createTreeReader(evolution.getReaderSchema(),
readerContext);
- this.fileIncluded = evolution.getFileIncluded();
- indexes = new OrcProto.RowIndex[types.size()];
- bloomFilterIndices = new OrcProto.BloomFilterIndex[types.size()];
- bloomFilterKind = new OrcProto.Stream.Kind[types.size()];
+ indexes = new OrcIndex(new OrcProto.RowIndex[types.size()],
+ new OrcProto.Stream.Kind[types.size()],
+ new OrcProto.BloomFilterIndex[types.size()]);
+
+ planner = new StripePlanner(evolution.getFileSchema(), encryption,
+ unencryptedOptions, dataReader, writerVersion,
+ ignoreNonUtf8BloomFilter, maxDiskRangeChunkLimit);
try {
advanceToNextRow(reader, 0L, true);
@@ -1052,27 +1005,14 @@ public class RecordReaderImpl implements RecordReader {
return null;
}
readRowIndex(currentStripe, fileIncluded, sargApp.sargColumns);
- return sargApp.pickRowGroups(stripes.get(currentStripe), indexes,
- bloomFilterKind, stripeFooter.getColumnsList(), bloomFilterIndices, false);
+ return sargApp.pickRowGroups(stripes.get(currentStripe),
+ indexes.getRowGroupIndex(),
+ indexes.getBloomFilterKinds(), stripeFooter.getColumnsList(),
+ indexes.getBloomFilterIndex(), false);
}
private void clearStreams() {
- // explicit close of all streams to de-ref ByteBuffers
- for (InStream is : streams.values()) {
- is.close();
- }
- if (bufferChunks != null) {
- if (dataReader.isTrackingDiskRanges()) {
- for (DiskRangeList range = bufferChunks; range != null; range = range.next) {
- if (!(range instanceof BufferChunk)) {
- continue;
- }
- dataReader.releaseBuffer(range.getData());
- }
- }
- }
- bufferChunks = null;
- streams.clear();
+ planner.clearStreams();
}
/**
@@ -1082,6 +1022,7 @@ public class RecordReaderImpl implements RecordReader {
*/
private void readStripe() throws IOException {
StripeInformation stripe = beginReadStripe();
+ planner.parseStripe(stripe, fileIncluded);
includedRowGroups = pickRowGroups();
// move forward to the first unskipped row
@@ -1094,13 +1035,8 @@ public class RecordReaderImpl implements RecordReader {
// if we haven't skipped the whole stripe, read the data
if (rowInStripe < rowCountInStripe) {
- // if we aren't projecting columns or filtering rows, just read it all
- if (isFullRead() && includedRowGroups == null) {
- readAllDataStreams(stripe);
- } else {
- readPartialDataStreams(stripe);
- }
- reader.startStripe(streams, stripeFooter);
+ planner.readData(indexes, includedRowGroups, false);
+ reader.startStripe(planner);
// if we skipped the first row group, move the pointers forward
if (rowInStripe != 0) {
seekToRowEntry(reader, (int) (rowInStripe / rowIndexStride));
@@ -1108,15 +1044,6 @@ public class RecordReaderImpl implements RecordReader {
}
}
- private boolean isFullRead() {
- for (boolean isColumnPresent : fileIncluded){
- if (!isColumnPresent){
- return false;
- }
- }
- return true;
- }
-
private StripeInformation beginReadStripe() throws IOException {
StripeInformation stripe = stripes.get(currentStripe);
stripeFooter = readStripeFooter(stripe);
@@ -1129,118 +1056,13 @@ public class RecordReaderImpl implements RecordReader {
rowBaseInStripe += stripes.get(i).getNumberOfRows();
}
// reset all of the indexes
- for (int i = 0; i < indexes.length; ++i) {
- indexes[i] = null;
+ OrcProto.RowIndex[] rowIndex = indexes.getRowGroupIndex();
+ for (int i = 0; i < rowIndex.length; ++i) {
+ rowIndex[i] = null;
}
return stripe;
}
- private void readAllDataStreams(StripeInformation stripe) throws IOException {
- long start = stripe.getIndexLength();
- long end = start + stripe.getDataLength();
- // explicitly trigger 1 big read
- DiskRangeList toRead = new DiskRangeList(start, end);
- bufferChunks = dataReader.readFileData(toRead, stripe.getOffset(), false);
- List<OrcProto.Stream> streamDescriptions = stripeFooter.getStreamsList();
- createStreams(streamDescriptions, bufferChunks, null,
- dataReader.getCompressionCodec(), bufferSize, streams);
- }
-
- /**
- * Plan the ranges of the file that we need to read given the list of
- * columns and row groups.
- *
- * @param streamList the list of streams available
- * @param indexes the indexes that have been loaded
- * @param includedColumns which columns are needed
- * @param includedRowGroups which row groups are needed
- * @param isCompressed does the file have generic compression
- * @param encodings the encodings for each column
- * @param types the types of the columns
- * @param compressionSize the compression block size
- * @return the list of disk ranges that will be loaded
- */
- static DiskRangeList planReadPartialDataStreams
- (List<OrcProto.Stream> streamList,
- OrcProto.RowIndex[] indexes,
- boolean[] includedColumns,
- boolean[] includedRowGroups,
- boolean isCompressed,
- List<OrcProto.ColumnEncoding> encodings,
- List<OrcProto.Type> types,
- int compressionSize,
- boolean doMergeBuffers) {
- long offset = 0;
- // figure out which columns have a present stream
- boolean[] hasNull = RecordReaderUtils.findPresentStreamsByColumn(streamList, types);
- CreateHelper list = new CreateHelper();
- for (OrcProto.Stream stream : streamList) {
- long length = stream.getLength();
- int column = stream.getColumn();
- OrcProto.Stream.Kind streamKind = stream.getKind();
- // since stream kind is optional, first check if it exists
- if (stream.hasKind() &&
- (StreamName.getArea(streamKind) == StreamName.Area.DATA) &&
- (column < includedColumns.length && includedColumns[column])) {
- // if we aren't filtering or it is a dictionary, load it.
- if (includedRowGroups == null
- || RecordReaderUtils.isDictionary(streamKind, encodings.get(column))) {
- RecordReaderUtils.addEntireStreamToRanges(offset, length, list, doMergeBuffers);
- } else {
- RecordReaderUtils.addRgFilteredStreamToRanges(stream, includedRowGroups,
- isCompressed, indexes[column], encodings.get(column), types.get(column),
- compressionSize, hasNull[column], offset, length, list, doMergeBuffers);
- }
- }
- offset += length;
- }
- return list.extract();
- }
-
- void createStreams(List<OrcProto.Stream> streamDescriptions,
- DiskRangeList ranges,
- boolean[] includeColumn,
- CompressionCodec codec,
- int bufferSize,
- Map<StreamName, InStream> streams) throws IOException {
- long streamOffset = 0;
- InStream.StreamOptions options = InStream.options().withCodec(codec)
- .withBufferSize(bufferSize);
- for (OrcProto.Stream streamDesc : streamDescriptions) {
- int column = streamDesc.getColumn();
- if ((includeColumn != null &&
- (column < includeColumn.length && !includeColumn[column])) ||
- streamDesc.hasKind() &&
- (StreamName.getArea(streamDesc.getKind()) != StreamName.Area.DATA)) {
- streamOffset += streamDesc.getLength();
- continue;
- }
- DiskRangeList buffers = RecordReaderUtils.getStreamBuffers(
- ranges, streamOffset, streamDesc.getLength());
- StreamName name = new StreamName(column, streamDesc.getKind());
- streams.put(name, InStream.create(name.toString(), buffers,
- 0, streamDesc.getLength(), options));
- streamOffset += streamDesc.getLength();
- }
- }
-
- private void readPartialDataStreams(StripeInformation stripe) throws IOException {
- List<OrcProto.Stream> streamList = stripeFooter.getStreamsList();
- DiskRangeList toRead = planReadPartialDataStreams(streamList,
- indexes, fileIncluded, includedRowGroups, dataReader.getCompressionCodec() != null,
- stripeFooter.getColumnsList(), types, bufferSize, true);
- if (LOG.isDebugEnabled()) {
- LOG.debug("chunks = " + RecordReaderUtils.stringifyDiskRanges(toRead));
- }
- bufferChunks = dataReader.readFileData(toRead, stripe.getOffset(), false);
- if (LOG.isDebugEnabled()) {
- LOG.debug("merge = " + RecordReaderUtils.stringifyDiskRanges(bufferChunks));
- }
-
- createStreams(streamList, bufferChunks, fileIncluded,
- dataReader.getCompressionCodec(), bufferSize, streams);
- }
-
/**
* Read the next stripe until we find a row that we don't skip.
*
@@ -1399,34 +1221,29 @@ public class RecordReaderImpl implements RecordReader {
public OrcIndex readRowIndex(int stripeIndex, boolean[] included,
boolean[] sargColumns) throws IOException {
- return readRowIndex(stripeIndex, included, null, null, sargColumns);
- }
-
- public OrcIndex readRowIndex(int stripeIndex, boolean[] included,
- OrcProto.RowIndex[] indexes,
- OrcProto.BloomFilterIndex[] bloomFilterIndex,
- boolean[] sargColumns) throws IOException {
- StripeInformation stripe = stripes.get(stripeIndex);
- OrcProto.StripeFooter stripeFooter = null;
// if this is the current stripe, use the cached objects.
if (stripeIndex == currentStripe) {
- stripeFooter = this.stripeFooter;
- indexes = indexes == null ? this.indexes : indexes;
- bloomFilterIndex = bloomFilterIndex == null ? this.bloomFilterIndices : bloomFilterIndex;
- sargColumns = sargColumns == null ?
- (sargApp == null ? null : sargApp.sargColumns) : sargColumns;
+ return planner.readRowIndex(
+ sargColumns != null || sargApp == null
+ ? sargColumns : sargApp.sargColumns, indexes);
+ } else {
+ StripePlanner copy = new StripePlanner(planner);
+ if (included == null) {
+ included = new boolean[schema.getMaximumId() + 1];
+ Arrays.fill(included, true);
+ }
+ copy.parseStripe(stripes.get(stripeIndex), included);
+ return copy.readRowIndex(sargColumns, null);
}
- return dataReader.readRowIndex(stripe, evolution.getFileType(0), stripeFooter,
- ignoreNonUtf8BloomFilter, included, indexes, sargColumns, writerVersion,
- bloomFilterKind, bloomFilterIndex);
}
private void seekToRowEntry(TreeReaderFactory.TreeReader reader, int rowEntry)
throws IOException {
- PositionProvider[] index = new PositionProvider[indexes.length];
- for (int i = 0; i < indexes.length; ++i) {
- if (indexes[i] != null) {
- index[i] = new PositionProviderImpl(indexes[i].getEntry(rowEntry));
+ OrcProto.RowIndex[] rowIndices = indexes.getRowGroupIndex();
+ PositionProvider[] index = new PositionProvider[rowIndices.length];
+ for (int i = 0; i < index.length; ++i) {
+ if (rowIndices[i] != null) {
+ index[i] = new PositionProviderImpl(rowIndices[i].getEntry(rowEntry));
}
}
reader.seek(index);
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 aeb33d4..7ed62ef 100644
--- a/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java
+++ b/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java
@@ -22,6 +22,8 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@@ -31,9 +33,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.orc.CompressionCodec;
-import org.apache.orc.CompressionKind;
import org.apache.orc.DataReader;
-import org.apache.orc.OrcFile;
import org.apache.orc.OrcProto;
import org.apache.orc.StripeInformation;
import org.apache.orc.TypeDescription;
@@ -45,102 +45,6 @@ public class RecordReaderUtils {
private static final HadoopShims SHIMS = HadoopShimsFactory.get();
private static final Logger LOG = LoggerFactory.getLogger(RecordReaderUtils.class);
- static boolean hadBadBloomFilters(TypeDescription.Category category,
- OrcFile.WriterVersion version) {
- switch(category) {
- case STRING:
- case CHAR:
- case VARCHAR:
- return !version.includes(OrcFile.WriterVersion.HIVE_12055);
- case DECIMAL:
- return true;
- case TIMESTAMP:
- return !version.includes(OrcFile.WriterVersion.ORC_135);
- default:
- return false;
- }
- }
-
- /**
- * Plans the list of disk ranges that the given stripe needs to read the
- * indexes. All of the positions are relative to the start of the stripe.
- * @param fileSchema the schema for the file
- * @param footer the stripe footer
- * @param ignoreNonUtf8BloomFilter should the reader ignore non-utf8
- * encoded bloom filters
- * @param fileIncluded the columns (indexed by file columns) that should be
- * read
- * @param sargColumns true for the columns (indexed by file columns) that
- * we need bloom filters for
- * @param version the version of the software that wrote the file
- * @param bloomFilterKinds (output) the stream kind of the bloom filters
- * @return a list of merged disk ranges to read
- */
- public static DiskRangeList planIndexReading(TypeDescription fileSchema,
- OrcProto.StripeFooter footer,
- boolean ignoreNonUtf8BloomFilter,
- boolean[] fileIncluded,
- boolean[] sargColumns,
- OrcFile.WriterVersion version,
- OrcProto.Stream.Kind[] bloomFilterKinds) {
- DiskRangeList.CreateHelper result = new DiskRangeList.CreateHelper();
- List<OrcProto.Stream> streams = footer.getStreamsList();
- // figure out which kind of bloom filter we want for each column
- // picks bloom_filter_utf8 if its available, otherwise bloom_filter
- if (sargColumns != null) {
- for (OrcProto.Stream stream : streams) {
- if (stream.hasKind() && stream.hasColumn()) {
- int column = stream.getColumn();
- if (sargColumns[column]) {
- switch (stream.getKind()) {
- case BLOOM_FILTER:
- if (bloomFilterKinds[column] == null &&
- !(ignoreNonUtf8BloomFilter &&
- hadBadBloomFilters(fileSchema.findSubtype(column)
- .getCategory(), version))) {
- bloomFilterKinds[column] = OrcProto.Stream.Kind.BLOOM_FILTER;
- }
- break;
- case BLOOM_FILTER_UTF8:
- bloomFilterKinds[column] = OrcProto.Stream.Kind.BLOOM_FILTER_UTF8;
- break;
- default:
- break;
- }
- }
- }
- }
- }
- long offset = 0;
- for(OrcProto.Stream stream: footer.getStreamsList()) {
- if (stream.hasKind() && stream.hasColumn()) {
- int column = stream.getColumn();
- if (fileIncluded == null || fileIncluded[column]) {
- boolean needStream = false;
- switch (stream.getKind()) {
- case ROW_INDEX:
- needStream = true;
- break;
- case BLOOM_FILTER:
- needStream = bloomFilterKinds[column] == OrcProto.Stream.Kind.BLOOM_FILTER;
- break;
- case BLOOM_FILTER_UTF8:
- needStream = bloomFilterKinds[column] == OrcProto.Stream.Kind.BLOOM_FILTER_UTF8;
- break;
- default:
- // PASS
- break;
- }
- if (needStream) {
- result.addOrMerge(offset, offset + stream.getLength(), true, false);
- }
- }
- }
- offset += stream.getLength();
- }
- return result.get();
- }
-
private static class DefaultDataReader implements DataReader {
private FSDataInputStream file;
private ByteBufferAllocatorPool pool;
@@ -148,10 +52,7 @@ public class RecordReaderUtils {
private final FileSystem fs;
private final Path path;
private final boolean useZeroCopy;
- private InStream.StreamOptions options = InStream.options();
- private final int typeCount;
- private CompressionKind compressionKind;
- private final int maxDiskRangeChunkLimit;
+ private InStream.StreamOptions options;
private boolean isOpen = false;
private DefaultDataReader(DataReaderProperties properties) {
@@ -159,11 +60,7 @@ public class RecordReaderUtils {
this.path = properties.getPath();
this.file = properties.getFile();
this.useZeroCopy = properties.getZeroCopy();
- this.compressionKind = properties.getCompression();
- options.withCodec(OrcCodecPool.getCodec(compressionKind))
- .withBufferSize(properties.getBufferSize());
- this.typeCount = properties.getTypeCount();
- this.maxDiskRangeChunkLimit = properties.getMaxDiskRangeChunkLimit();
+ this.options = properties.getCompression();
}
@Override
@@ -182,83 +79,6 @@ public class RecordReaderUtils {
}
@Override
- public OrcIndex readRowIndex(StripeInformation stripe,
- TypeDescription fileSchema,
- OrcProto.StripeFooter footer,
- boolean ignoreNonUtf8BloomFilter,
- boolean[] included,
- OrcProto.RowIndex[] indexes,
- boolean[] sargColumns,
- OrcFile.WriterVersion version,
- OrcProto.Stream.Kind[] bloomFilterKinds,
- OrcProto.BloomFilterIndex[] bloomFilterIndices
- ) throws IOException {
- if (!isOpen) {
- open();
- }
- if (footer == null) {
- footer = readStripeFooter(stripe);
- }
- if (indexes == null) {
- indexes = new OrcProto.RowIndex[typeCount];
- }
- if (bloomFilterKinds == null) {
- bloomFilterKinds = new OrcProto.Stream.Kind[typeCount];
- }
- if (bloomFilterIndices == null) {
- bloomFilterIndices = new OrcProto.BloomFilterIndex[typeCount];
- }
- DiskRangeList ranges = planIndexReading(fileSchema, footer,
- ignoreNonUtf8BloomFilter, included, sargColumns, version,
- bloomFilterKinds);
- ranges = readDiskRanges(file, zcr, stripe.getOffset(), ranges, false, maxDiskRangeChunkLimit);
- long offset = 0;
- DiskRangeList range = ranges;
- for(OrcProto.Stream stream: footer.getStreamsList()) {
- // advance to find the next range
- while (range != null && range.getEnd() <= offset) {
- range = range.next;
- }
- // no more ranges, so we are done
- if (range == null) {
- break;
- }
- int column = stream.getColumn();
- if (stream.hasKind() && range.getOffset() <= offset) {
- switch (stream.getKind()) {
- case ROW_INDEX:
- if (included == null || included[column]) {
- ByteBuffer bb = range.getData().duplicate();
- bb.position((int) (offset - range.getOffset()));
- bb.limit((int) (bb.position() + stream.getLength()));
- indexes[column] = OrcProto.RowIndex.parseFrom(
- InStream.createCodedInputStream(InStream.create("index",
- new BufferChunk(bb, 0),
- 0, stream.getLength(), options)));
- }
- break;
- case BLOOM_FILTER:
- case BLOOM_FILTER_UTF8:
- if (sargColumns != null && sargColumns[column]) {
- ByteBuffer bb = range.getData().duplicate();
- bb.position((int) (offset - range.getOffset()));
- bb.limit((int) (bb.position() + stream.getLength()));
- bloomFilterIndices[column] = OrcProto.BloomFilterIndex.parseFrom
- (InStream.createCodedInputStream(InStream.create(
- "bloom_filter", new BufferChunk(bb, 0),
- 0, stream.getLength(), options)));
- }
- break;
- default:
- break;
- }
- }
- offset += stream.getLength();
- }
- return new OrcIndex(indexes, bloomFilterKinds, bloomFilterIndices);
- }
-
- @Override
public OrcProto.StripeFooter readStripeFooter(StripeInformation stripe) throws IOException {
if (!isOpen) {
open();
@@ -275,15 +95,17 @@ public class RecordReaderUtils {
}
@Override
- public DiskRangeList readFileData(
- DiskRangeList range, long baseOffset, boolean doForceDirect) throws IOException {
- return RecordReaderUtils.readDiskRanges(file, zcr, baseOffset, range, doForceDirect, maxDiskRangeChunkLimit);
+ public BufferChunkList readFileData(BufferChunkList range,
+ boolean doForceDirect
+ ) throws IOException {
+ RecordReaderUtils.readDiskRanges(file, zcr, range, doForceDirect);
+ return range;
}
@Override
public void close() throws IOException {
if (options.getCodec() != null) {
- OrcCodecPool.returnCodec(compressionKind, options.getCodec());
+ OrcCodecPool.returnCodec(options.getCodec().getKind(), options.getCodec());
options.withCodec(null);
}
if (pool != null) {
@@ -337,17 +159,6 @@ public class RecordReaderUtils {
return new DefaultDataReader(properties);
}
- public static boolean[] findPresentStreamsByColumn(
- List<OrcProto.Stream> streamList, List<OrcProto.Type> types) {
- boolean[] hasNull = new boolean[types.size()];
- for(OrcProto.Stream stream: streamList) {
- if (stream.hasKind() && (stream.getKind() == OrcProto.Stream.Kind.PRESENT)) {
- hasNull[stream.getColumn()] = true;
- }
- }
- return hasNull;
- }
-
/**
* Does region A overlap region B? The end points are inclusive on both sides.
* @param leftA A's left point
@@ -363,38 +174,17 @@ public class RecordReaderUtils {
return rightB >= leftA;
}
- public static void addEntireStreamToRanges(
- long offset, long length, DiskRangeList.CreateHelper list, boolean doMergeBuffers) {
- list.addOrMerge(offset, offset + length, doMergeBuffers, false);
- }
-
- public static void addRgFilteredStreamToRanges(OrcProto.Stream stream,
- boolean[] includedRowGroups, boolean isCompressed, OrcProto.RowIndex index,
- OrcProto.ColumnEncoding encoding, OrcProto.Type type, int compressionSize, boolean hasNull,
- long offset, long length, DiskRangeList.CreateHelper list, boolean doMergeBuffers) {
- for (int group = 0; group < includedRowGroups.length; ++group) {
- if (!includedRowGroups[group]) continue;
- int posn = getIndexPosition(
- encoding.getKind(), type.getKind(), stream.getKind(), isCompressed, hasNull);
- long start = index.getEntry(group).getPositions(posn);
- final long nextGroupOffset;
- boolean isLast = group == (includedRowGroups.length - 1);
- nextGroupOffset = isLast ? length : index.getEntry(group + 1).getPositions(posn);
-
- start += offset;
- long end = offset + estimateRgEndOffset(
- isCompressed, isLast, nextGroupOffset, length, compressionSize);
- list.addOrMerge(start, end, doMergeBuffers, true);
- }
- }
-
- public static long estimateRgEndOffset(boolean isCompressed, boolean isLast,
- long nextGroupOffset, long streamLength, int bufferSize) {
+ public static long estimateRgEndOffset(InStream.StreamOptions unencryptedOptions,
+ boolean isLast,
+ long nextGroupOffset,
+ long streamLength) {
// figure out the worst case last location
// if adjacent groups have the same compressed block offset then stretch the slop
// by factor of 2 to safely accommodate the next compression block.
// One for the current compression block and another for the next compression block.
- long slop = isCompressed ? 2 * (OutStream.HEADER_SIZE + bufferSize) : WORST_UNCOMPRESSED_SLOP;
+ long slop = unencryptedOptions.isCompressed()
+ ? 2 * (OutStream.HEADER_SIZE + unencryptedOptions.getBufferSize())
+ : WORST_UNCOMPRESSED_SLOP;
return isLast ? streamLength : Math.min(streamLength, nextGroupOffset + slop);
}
@@ -409,12 +199,12 @@ public class RecordReaderUtils {
* @param columnEncoding the encoding of the column
* @param columnType the type of the column
* @param streamType the kind of the stream
- * @param isCompressed is the file compressed
+ * @param isCompressed is the stream compressed?
* @param hasNulls does the column have a PRESENT stream?
* @return the number of positions that will be used for that stream
*/
public static int getIndexPosition(OrcProto.ColumnEncoding.Kind columnEncoding,
- OrcProto.Type.Kind columnType,
+ TypeDescription.Category columnType,
OrcProto.Stream.Kind streamType,
boolean isCompressed,
boolean hasNulls) {
@@ -451,10 +241,6 @@ public class RecordReaderUtils {
}
}
case BINARY:
- if (streamType == OrcProto.Stream.Kind.DATA) {
- return base;
- }
- return base + BYTE_STREAM_POSITIONS + compressionValue;
case DECIMAL:
if (streamType == OrcProto.Stream.Kind.DATA) {
return base;
@@ -512,122 +298,190 @@ public class RecordReaderUtils {
return buffer.toString();
}
+ static long computeEnd(BufferChunk first, BufferChunk last) {
+ long end = 0;
+ for(BufferChunk ptr=first; ptr != last.next; ptr = (BufferChunk) ptr.next) {
+ end = Math.max(ptr.getEnd(), end);
+ }
+ return end;
+ }
+
/**
- * Read the list of ranges from the file.
- * @param file the file to read
- * @param base the base of the stripe
- * @param range the disk ranges within the stripe to read
- * @return the bytes read for each disk range, which is the same length as
- * ranges
+ * Zero-copy tead the data from the file based on a list of ranges in a
+ * single read.
+ *
+ * As a side note, the HDFS zero copy API really sucks from a user's point of
+ * view.
+ *
+ * @param file the file we're reading from
+ * @param zcr the zero copy shim
+ * @param first the first range to read
+ * @param last the last range to read
+ * @param allocateDirect if we need to allocate buffers, should we use direct
* @throws IOException
*/
- static DiskRangeList readDiskRanges(FSDataInputStream file,
- HadoopShims.ZeroCopyReaderShim zcr,
- long base,
- DiskRangeList range,
- boolean doForceDirect, int maxChunkLimit) throws IOException {
- if (range == null)
- return null;
- DiskRangeList prev = range.prev;
- if (prev == null) {
- prev = new DiskRangeList.MutateHelper(range);
+ static void zeroCopyReadRanges(FSDataInputStream file,
+ HadoopShims.ZeroCopyReaderShim zcr,
+ BufferChunk first,
+ BufferChunk last,
+ boolean allocateDirect) throws IOException {
+ // read all of the bytes that we need
+ final long offset = first.getOffset();
+ int length = (int)(computeEnd(first, last) - offset);
+ file.seek(offset);
+ List<ByteBuffer> bytes = new ArrayList<>();
+ while (length > 0) {
+ ByteBuffer read= zcr.readBuffer(length, false);
+ bytes.add(read);
+ length -= read.remaining();
}
- while (range != null) {
- if (range.hasData()) {
- range = range.next;
- continue;
+ long currentOffset = offset;
+
+ // iterate and fill each range
+ BufferChunk current = first;
+ Iterator<ByteBuffer> buffers = bytes.iterator();
+ ByteBuffer currentBuffer = buffers.next();
+ while (current != last.next) {
+
+ // if we are past the start of the range, restart the iterator
+ if (current.getOffset() < offset) {
+ buffers = bytes.iterator();
+ currentBuffer = buffers.next();
+ currentOffset = offset;
}
- boolean firstRead = true;
- long len = range.getEnd() - range.getOffset();
- long off = range.getOffset();
- 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);
+ // walk through the buffers to find the start of the buffer
+ while (currentOffset + currentBuffer.remaining() <= current.getOffset()) {
+ currentOffset += currentBuffer.remaining();
+ // We assume that buffers.hasNext is true because we know we read
+ // enough data to cover the last range.
+ currentBuffer = buffers.next();
+ }
+
+ // did we get the current range in a single read?
+ if (currentOffset + currentBuffer.remaining() >= current.getEnd()) {
+ ByteBuffer copy = currentBuffer.duplicate();
+ copy.position((int) (current.getOffset() - currentOffset));
+ copy.limit(copy.position() + current.getLength());
+ current.setChunk(copy);
+
+ } else {
+ // otherwise, build a single buffer that holds the entire range
+ ByteBuffer result = allocateDirect
+ ? ByteBuffer.allocateDirect(current.getLength())
+ : ByteBuffer.allocate(current.getLength());
+ // we know that the range spans buffers
+ ByteBuffer copy = currentBuffer.duplicate();
+ // skip over the front matter
+ copy.position((int) (current.getOffset() - currentOffset));
+ result.put(copy);
+ // advance the buffer
+ currentOffset += currentBuffer.remaining();
+ currentBuffer = buffers.next();
+ while (result.hasRemaining()) {
+ if (result.remaining() > currentBuffer.remaining()) {
+ result.put(currentBuffer.duplicate());
+ currentOffset += currentBuffer.remaining();
+ currentBuffer = buffers.next();
} else {
- partial = ByteBuffer.wrap(buffer);
+ copy = currentBuffer.duplicate();
+ copy.limit(result.remaining());
+ result.put(copy);
}
}
- BufferChunk bc = new BufferChunk(partial, off);
- if (firstRead) {
- range.replaceSelfWith(bc);
- } else {
- range.insertAfter(bc);
- }
- firstRead = false;
- range = bc;
- len -= readSize;
- off += readSize;
+ result.flip();
+ current.setChunk(result);
}
- range = range.next;
+ current = (BufferChunk) current.next;
}
- return prev.next;
}
+ /**
+ * Read the data from the file based on a list of ranges in a single read.
+ * @param file the file to read from
+ * @param first the first range to read
+ * @param last the last range to read
+ * @param allocateDirect should we use direct buffers
+ */
+ static void readRanges(FSDataInputStream file,
+ BufferChunk first,
+ BufferChunk last,
+ boolean allocateDirect) throws IOException {
+ // assume that the chunks are sorted by offset
+ long offset = first.getOffset();
+ int readSize = (int) (computeEnd(first, last) - offset);
+ byte[] buffer = new byte[readSize];
+ file.readFully(offset, buffer, 0, buffer.length);
+
+ // get the data into a ByteBuffer
+ ByteBuffer bytes;
+ if (allocateDirect) {
+ bytes = ByteBuffer.allocateDirect(readSize);
+ bytes.put(buffer);
+ bytes.flip();
+ } else {
+ bytes = ByteBuffer.wrap(buffer);
+ }
- static DiskRangeList getStreamBuffers(DiskRangeList range, long offset,
- long length) {
- // This assumes sorted ranges (as do many other parts of ORC code.
- BufferChunkList result = new BufferChunkList();
- if (length != 0) {
- long streamEnd = offset + length;
- boolean inRange = false;
- while (range != null) {
- if (!inRange) {
- if (range.getEnd() <= offset) {
- range = range.next;
- continue; // Skip until we are in range.
- }
- inRange = true;
- if (range.getOffset() < offset) {
- // Partial first buffer, add a slice of it.
- result.add((BufferChunk) range.sliceAndShift(offset,
- Math.min(streamEnd, range.getEnd()), -offset));
- if (range.getEnd() >= streamEnd)
- break; // Partial first buffer is also partial last buffer.
- range = range.next;
- continue;
- }
- } else if (range.getOffset() >= streamEnd) {
- break;
- }
- if (range.getEnd() > streamEnd) {
- // Partial last buffer (may also be the first buffer), add a slice of it.
- result.add((BufferChunk) range.sliceAndShift(range.getOffset(),
- streamEnd, -offset));
- break;
- }
- // Buffer that belongs entirely to one stream.
- // TODO: ideally we would want to reuse the object and remove it from
- // the list, but we cannot because bufferChunks is also used by
- // clearStreams for zcr. Create a useless dup.
- result.add((BufferChunk) range.sliceAndShift(range.getOffset(),
- range.getEnd(), -offset));
- if (range.getEnd() == streamEnd) break;
- range = range.next;
+ // populate each BufferChunks with the data
+ BufferChunk current = first;
+ while (current != last.next) {
+ ByteBuffer currentBytes = current == last ? bytes : bytes.duplicate();
+ currentBytes.position((int) (current.getOffset() - offset));
+ currentBytes.limit((int) (current.getEnd() - offset));
+ current.setChunk(currentBytes);
+ current = (BufferChunk) current.next;
+ }
+ }
+
+ /**
+ * Find the list of ranges that should be read in a single read.
+ * The read will stop when there is a gap, one of the ranges already has data,
+ * or we have reached the maximum read size of 2^31.
+ * @param first the first range to read
+ * @return the last range to read
+ */
+ static BufferChunk findSingleRead(BufferChunk first) {
+ BufferChunk last = first;
+ long currentEnd = first.getEnd();
+ while (last.next != null &&
+ !last.next.hasData() &&
+ last.next.getOffset() <= currentEnd &&
+ last.next.getEnd() - first.getOffset() < Integer.MAX_VALUE) {
+ last = (BufferChunk) last.next;
+ currentEnd = Math.max(currentEnd, last.getEnd());
+ }
+ return last;
+ }
+
+ /**
+ * Read the list of ranges from the file by updating each range in the list
+ * with a buffer that has the bytes from the file.
+ *
+ * The ranges must be sorted, but may overlap or include holes.
+ *
+ * @param file the file to read
+ * @param zcr the zero copy shim
+ * @param list the disk ranges within the file to read
+ * @param doForceDirect allocate direct buffers
+ */
+ static void readDiskRanges(FSDataInputStream file,
+ HadoopShims.ZeroCopyReaderShim zcr,
+ BufferChunkList list,
+ boolean doForceDirect) throws IOException {
+ BufferChunk current = list == null ? null : list.get();
+ while (current != null) {
+ while (current.hasData()) {
+ current = (BufferChunk) current.next;
+ }
+ BufferChunk last = findSingleRead(current);
+ if (zcr != null) {
+ zeroCopyReadRanges(file, zcr, current, last, doForceDirect);
+ } else {
+ readRanges(file, current, last, doForceDirect);
}
+ current = (BufferChunk) last.next;
}
- return result.get();
}
static HadoopShims.ZeroCopyReaderShim createZeroCopyShim(FSDataInputStream file,
@@ -681,9 +535,9 @@ public class RecordReaderUtils {
}
}
- private final TreeMap<Key, ByteBuffer> buffers = new TreeMap<Key, ByteBuffer>();
+ private final TreeMap<Key, ByteBuffer> buffers = new TreeMap<>();
- private final TreeMap<Key, ByteBuffer> directBuffers = new TreeMap<Key, ByteBuffer>();
+ private final TreeMap<Key, ByteBuffer> directBuffers = new TreeMap<>();
private long currentGeneration = 0;
diff --git a/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java b/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java
index b0fd5a7..dae736a 100644
--- a/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java
+++ b/java/core/src/java/org/apache/orc/impl/TreeReaderFactory.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -45,6 +45,8 @@ import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
import org.apache.orc.OrcFile;
import org.apache.orc.TypeDescription;
import org.apache.orc.OrcProto;
+import org.apache.orc.impl.reader.ReaderEncryption;
+import org.apache.orc.impl.reader.StripePlanner;
import org.apache.orc.impl.writer.TimestampTreeWriter;
/**
@@ -61,6 +63,8 @@ public class TreeReaderFactory {
String getWriterTimezone();
OrcFile.Version getFileFormat();
+
+ ReaderEncryption getEncryption();
}
public static class ReaderContext implements Context {
@@ -69,12 +73,18 @@ public class TreeReaderFactory {
private boolean useUTCTimestamp = false;
private String writerTimezone;
private OrcFile.Version fileFormat;
+ private ReaderEncryption encryption;
public ReaderContext setSchemaEvolution(SchemaEvolution evolution) {
this.evolution = evolution;
return this;
}
+ public ReaderContext setEncryption(ReaderEncryption value) {
+ encryption = value;
+ return this;
+ }
+
public ReaderContext skipCorrupt(boolean skipCorrupt) {
this.skipCorrupt = skipCorrupt;
return this;
@@ -119,6 +129,11 @@ public class TreeReaderFactory {
public OrcFile.Version getFileFormat() {
return fileFormat;
}
+
+ @Override
+ public ReaderEncryption getEncryption() {
+ return encryption;
+ }
}
public abstract static class TreeReader {
@@ -169,11 +184,9 @@ public class TreeReaderFactory {
}
}
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- checkEncoding(stripeFooter.getColumnsList().get(columnId));
- InStream in = streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ checkEncoding(planner.getEncoding(columnId));
+ InStream in = planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.PRESENT));
if (in == null) {
present = null;
@@ -287,8 +300,7 @@ public class TreeReaderFactory {
}
@Override
- public void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter footer) {
+ public void startStripe(StripePlanner planner) {
// PASS
}
@@ -330,11 +342,9 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- reader = new BitFieldReader(streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ reader = new BitFieldReader(planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.DATA)));
}
@@ -381,11 +391,9 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- reader = new RunLengthByteReader(streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ reader = new RunLengthByteReader(planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.DATA)));
}
@@ -446,14 +454,12 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- reader = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(name), true, context);
+ reader = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(name), true, context);
}
@Override
@@ -513,14 +519,12 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- reader = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(name), true, context);
+ reader = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(name), true, context);
}
@Override
@@ -581,14 +585,12 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- reader = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(name), true, context);
+ reader = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(name), true, context);
}
@Override
@@ -636,13 +638,11 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- stream = streams.get(name);
+ stream = planner.getStream(name);
}
@Override
@@ -731,14 +731,12 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name =
new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- stream = streams.get(name);
+ stream = planner.getStream(name);
}
@Override
@@ -842,15 +840,13 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- stream = streams.get(name);
- lengths = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId, OrcProto.Stream.Kind.LENGTH)), false, context);
+ stream = planner.getStream(name);
+ lengths = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(new StreamName(columnId, OrcProto.Stream.Kind.LENGTH)), false, context);
}
@Override
@@ -948,17 +944,16 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- data = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ OrcProto.ColumnEncoding.Kind kind = planner.getEncoding(columnId).getKind();
+ data = createIntegerReader(kind,
+ planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.DATA)), true, context);
- nanos = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId,
+ nanos = createIntegerReader(kind,
+ planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.SECONDARY)), false, context);
- base_timestamp = getBaseTimestamp(stripeFooter.getWriterTimezone());
+ base_timestamp = getBaseTimestamp(planner.getWriterTimezone());
}
protected long getBaseTimestamp(String timeZoneId) throws IOException {
@@ -1088,14 +1083,12 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- reader = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(name), true, context);
+ reader = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(name), true, context);
}
@Override
@@ -1173,14 +1166,12 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- valueStream = streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ valueStream = planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.DATA));
- scaleReader = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId, OrcProto.Stream.Kind.SECONDARY)), true, context);
+ scaleReader = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(new StreamName(columnId, OrcProto.Stream.Kind.SECONDARY)), true, context);
}
@Override
@@ -1328,11 +1319,9 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- InStream stream = streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ InStream stream = planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.DATA));
valueReader = new RunLengthIntegerReaderV2(stream, true, skipCorrupt);
}
@@ -1432,12 +1421,10 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
+ void startStripe(StripePlanner planner) throws IOException {
// For each stripe, checks the encoding and initializes the appropriate
// reader
- switch (stripeFooter.getColumnsList().get(columnId).getKind()) {
+ switch (planner.getEncoding(columnId).getKind()) {
case DIRECT:
case DIRECT_V2:
reader = new StringDirectTreeReader(columnId);
@@ -1448,9 +1435,9 @@ public class TreeReaderFactory {
break;
default:
throw new IllegalArgumentException("Unsupported encoding " +
- stripeFooter.getColumnsList().get(columnId).getKind());
+ planner.getEncoding(columnId).getKind());
}
- reader.startStripe(streams, stripeFooter);
+ reader.startStripe(planner);
}
@Override
@@ -1588,15 +1575,13 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DATA);
- stream = streams.get(name);
- lengths = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId, OrcProto.Stream.Kind.LENGTH)),
+ stream = planner.getStream(name);
+ lengths = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(new StreamName(columnId, OrcProto.Stream.Kind.LENGTH)),
false, context);
}
@@ -1694,26 +1679,25 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
// read the dictionary blob
StreamName name = new StreamName(columnId,
OrcProto.Stream.Kind.DICTIONARY_DATA);
- InStream in = streams.get(name);
+ InStream in = planner.getStream(name);
readDictionaryStream(in);
// read the lengths
name = new StreamName(columnId, OrcProto.Stream.Kind.LENGTH);
- in = streams.get(name);
- readDictionaryLengthStream(in, stripeFooter.getColumnsList().get(columnId));
+ in = planner.getStream(name);
+ OrcProto.ColumnEncoding encoding = planner.getEncoding(columnId);
+ readDictionaryLengthStream(in, encoding);
// set up the row reader
name = new StreamName(columnId, OrcProto.Stream.Kind.DATA);
- reader = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(name), false, context);
+ reader = createIntegerReader(encoding.getKind(),
+ planner.getStream(name), false, context);
}
private void readDictionaryLengthStream(InStream in, OrcProto.ColumnEncoding encoding)
@@ -2034,13 +2018,11 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
for (TreeReader field : fields) {
if (field != null) {
- field.startStripe(streams, stripeFooter);
+ field.startStripe(planner);
}
}
}
@@ -2116,15 +2098,13 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- tags = new RunLengthByteReader(streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ tags = new RunLengthByteReader(planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.DATA)));
for (TreeReader field : fields) {
if (field != null) {
- field.startStripe(streams, stripeFooter);
+ field.startStripe(planner);
}
}
}
@@ -2209,15 +2189,13 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- lengths = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ lengths = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.LENGTH)), false, context);
if (elementReader != null) {
- elementReader.startStripe(streams, stripeFooter);
+ elementReader.startStripe(planner);
}
}
@@ -2306,18 +2284,16 @@ public class TreeReaderFactory {
}
@Override
- void startStripe(Map<StreamName, InStream> streams,
- OrcProto.StripeFooter stripeFooter
- ) throws IOException {
- super.startStripe(streams, stripeFooter);
- lengths = createIntegerReader(stripeFooter.getColumnsList().get(columnId).getKind(),
- streams.get(new StreamName(columnId,
+ void startStripe(StripePlanner planner) throws IOException {
+ super.startStripe(planner);
+ lengths = createIntegerReader(planner.getEncoding(columnId).getKind(),
+ planner.getStream(new StreamName(columnId,
OrcProto.Stream.Kind.LENGTH)), false, context);
if (keyReader != null) {
- keyReader.startStripe(streams, stripeFooter);
+ keyReader.startStripe(planner);
}
if (valueReader != null) {
- valueReader.startStripe(streams, stripeFooter);
+ valueReader.startStripe(planner);
}
}
diff --git a/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java b/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java
new file mode 100644
index 0000000..b25654f
--- /dev/null
+++ b/java/core/src/java/org/apache/orc/impl/reader/StripePlanner.java
@@ -0,0 +1,529 @@
+/*
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.reader;
+
+import com.google.protobuf.CodedInputStream;
+import org.apache.orc.DataReader;
+import org.apache.orc.EncryptionAlgorithm;
+import org.apache.orc.OrcFile;
+import org.apache.orc.OrcProto;
+import org.apache.orc.StripeInformation;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.impl.BufferChunk;
+import org.apache.orc.impl.BufferChunkList;
+import org.apache.orc.impl.CryptoUtils;
+import org.apache.orc.impl.InStream;
+import org.apache.orc.impl.OrcIndex;
+import org.apache.orc.impl.RecordReaderUtils;
+import org.apache.orc.impl.StreamName;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.Key;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class handles parsing the stripe information and handling the necessary
+ * filtering and selection.
+ *
+ * It supports:
+ * <ul>
+ * <li>column projection</li>
+ * <li>row group selection</li>
+ * <li>encryption</li>
+ * </ul>
+ */
+public class StripePlanner {
+ // global information
+ private final TypeDescription schema;
+ private final OrcFile.WriterVersion version;
+ private final OrcProto.ColumnEncoding[] encodings;
+ private final ReaderEncryption encryption;
+ private final DataReader dataReader;
+ private final InStream.StreamOptions compression;
+ private final boolean ignoreNonUtf8BloomFilter;
+ private final long maxBufferSize;
+
+ // specific to the current stripe
+ private String writerTimezone;
+ private long currentStripeId;
+ private long originalStripeId;
+ private Map<StreamName, StreamInformation> streams = new HashMap<>();
+ // the index streams sorted by offset
+ private final List<StreamInformation> indexStreams = new ArrayList<>();
+ // the data streams sorted by offset
+ private final List<StreamInformation> dataStreams = new ArrayList<>();
+ private final boolean[] usedVariants;
+ private final OrcProto.Stream.Kind[] bloomFilterKinds;
+ // does each column have a null stream?
+ private final boolean[] hasNull;
+
+ /**
+ * Create a stripe parser.
+ * @param schema the file schema
+ * @param encryption the encryption information
+ * @param compression the options to handle the compression
+ * @param dataReader the underlying data reader
+ * @param version the file writer version
+ * @param ignoreNonUtf8BloomFilter ignore old non-utf8 bloom filters
+ * @param maxBufferSize the largest single buffer to use
+ */
+ public StripePlanner(TypeDescription schema,
+ ReaderEncryption encryption,
+ InStream.StreamOptions compression,
+ DataReader dataReader,
+ OrcFile.WriterVersion version,
+ boolean ignoreNonUtf8BloomFilter,
+ long maxBufferSize) {
+ this.schema = schema;
+ this.version = version;
+ encodings = new OrcProto.ColumnEncoding[schema.getMaximumId()+1];
+ usedVariants = new boolean[encryption.getVariants().length];
+ this.encryption = encryption;
+ this.compression = compression;
+ this.dataReader = dataReader;
+ this.ignoreNonUtf8BloomFilter = ignoreNonUtf8BloomFilter;
+ bloomFilterKinds = new OrcProto.Stream.Kind[schema.getMaximumId() + 1];
+ hasNull = new boolean[schema.getMaximumId() + 1];
+ this.maxBufferSize = maxBufferSize;
+ }
+
+ public StripePlanner(StripePlanner old) {
+ this(old.schema, old.encryption, old.compression, old.dataReader,
+ old.version, old.ignoreNonUtf8BloomFilter, old.maxBufferSize);
+ }
+
+ /**
+ * Parse a new stripe. Resets the current stripe state.
+ * @param stripe the new stripe
+ * @param columnInclude an array with true for each column to read
+ * @return this for method chaining
+ */
+ public StripePlanner parseStripe(StripeInformation stripe,
+ boolean[] columnInclude) throws IOException {
+ OrcProto.StripeFooter footer = dataReader.readStripeFooter(stripe);
+ currentStripeId = stripe.getStripeId();
+ originalStripeId = stripe.getEncryptionStripeId();
+ writerTimezone = footer.getWriterTimezone();
+ streams.clear();
+ dataStreams.clear();
+ indexStreams.clear();
+ buildEncodings(footer, columnInclude);
+ findStreams(stripe.getOffset(), footer, columnInclude);
+ // figure out whether each column has null values in this stripe
+ Arrays.fill(hasNull, false);
+ for(StreamInformation stream: dataStreams) {
+ if (stream.kind == OrcProto.Stream.Kind.PRESENT) {
+ hasNull[stream.column] = true;
+ }
+ }
+ return this;
+ }
+
+ /**
+ * Read the stripe data from the file.
+ * @param index null for no row filters or the index for filtering
+ * @param rowGroupInclude null for all of the rows or an array with boolean
+ * for each row group in the current stripe.
+ * @param forceDirect should direct buffers be created?
+ * @return the buffers that were read
+ */
+ public BufferChunkList readData(OrcIndex index,
+ boolean[] rowGroupInclude,
+ boolean forceDirect) throws IOException {
+ BufferChunkList chunks = (index == null || rowGroupInclude == null)
+ ? planDataReading() : planPartialDataReading(index, rowGroupInclude);
+ dataReader.readFileData(chunks, forceDirect);
+ return chunks;
+ }
+
+ public String getWriterTimezone() {
+ return writerTimezone;
+ }
+
+ /**
+ * Get the stream for the given name.
+ * It is assumed that the name does <b>not</b> have the encryption set,
+ * because the TreeReader's don't know if they are reading encrypted data.
+ * Assumes that readData has already been called on this stripe.
+ * @param name the column/kind of the stream
+ * @return a new stream with the options set correctly
+ */
+ public InStream getStream(StreamName name) throws IOException {
+ StreamInformation stream = streams.get(name);
+ return stream == null ? null
+ : InStream.create(name, stream.firstChunk, stream.offset,
+ stream.length, getStreamOptions(stream.column, stream.kind));
+ }
+
+ /**
+ * Release all of the buffers for the current stripe.
+ */
+ public void clearStreams() {
+ if (dataReader.isTrackingDiskRanges()) {
+ for (StreamInformation stream : indexStreams) {
+ stream.releaseBuffers(dataReader);
+ }
+ for (StreamInformation stream : dataStreams) {
+ stream.releaseBuffers(dataReader);
+ }
+ }
+ indexStreams.clear();
+ dataStreams.clear();
+ streams.clear();
+ }
+
+ /**
+ * Get the stream options for a stream in a stripe.
+ * @param column the column we are reading
+ * @param kind the stream kind we are reading
+ * @return a new stream options to read the given column
+ */
+ private InStream.StreamOptions getStreamOptions(int column,
+ OrcProto.Stream.Kind kind
+ ) throws IOException {
+ ReaderEncryptionVariant variant = encryption.getVariant(column);
+ if (variant == null) {
+ return compression;
+ } else {
+ EncryptionAlgorithm algorithm = variant.getKeyDescription().getAlgorithm();
+ byte[] iv = new byte[algorithm.getIvLength()];
+ Key key = variant.getStripeKey(currentStripeId);
+ CryptoUtils.modifyIvForStream(column, kind, originalStripeId).accept(iv);
+ return new InStream.StreamOptions(compression)
+ .withEncryption(algorithm, key, iv);
+ }
+ }
+
+ public OrcProto.ColumnEncoding getEncoding(int column) {
+ return encodings[column];
+ }
+
+ private void buildEncodings(OrcProto.StripeFooter footer,
+ boolean[] columnInclude) throws IOException {
+ for(int c=0; c < encodings.length; ++c) {
+ if (columnInclude == null || columnInclude[c]) {
+ ReaderEncryptionVariant variant = encryption.getVariant(c);
+ if (variant == null) {
+ encodings[c] = footer.getColumns(c);
+ } else {
+ usedVariants[variant.getVariantId()] = true;
+ int subColumn = c - variant.getRoot().getId();
+ encodings[c] = footer.getEncryption(variant.getVariantId())
+ .getEncoding(subColumn);
+ }
+ }
+ }
+ }
+
+ /**
+ * For each stream, decide whether to include it in the list of streams.
+ * @param offset the position in the file for this stream
+ * @param columnInclude which columns are being read
+ * @param stream the stream to consider
+ * @param variant the variant being read
+ * @return the offset for the next stream
+ */
+ private long handleStream(long offset,
+ boolean[] columnInclude,
+ OrcProto.Stream stream,
+ ReaderEncryptionVariant variant) throws IOException {
+ int column = stream.getColumn();
+ if (stream.hasKind()) {
+ OrcProto.Stream.Kind kind = stream.getKind();
+
+ if (kind == OrcProto.Stream.Kind.ENCRYPTED_INDEX ||
+ kind == OrcProto.Stream.Kind.ENCRYPTED_DATA) {
+ // Ignore the placeholders that shouldn't count toward moving the
+ // offsets.
+ return 0;
+ }
+
+ if (columnInclude[column] && encryption.getVariant(column) == variant) {
+ // Ignore any broken bloom filters unless the user forced us to use
+ // them.
+ if (kind != OrcProto.Stream.Kind.BLOOM_FILTER ||
+ !ignoreNonUtf8BloomFilter ||
+ !hadBadBloomFilters(schema.findSubtype(column).getCategory(),
+ version)) {
+ // record what kind of bloom filters we are using
+ if (kind == OrcProto.Stream.Kind.BLOOM_FILTER_UTF8 ||
+ kind == OrcProto.Stream.Kind.BLOOM_FILTER) {
+ bloomFilterKinds[column] = kind;
+ }
+ StreamInformation info =
+ new StreamInformation(kind, column, offset, stream.getLength());
+ switch (StreamName.getArea(kind)) {
+ case DATA:
+ dataStreams.add(info);
+ break;
+ case INDEX:
+ indexStreams.add(info);
+ break;
+ default:
+ }
+ streams.put(new StreamName(column, kind), info);
+ }
+ }
+ }
+ return stream.getLength();
+ }
+
+ /**
+ * Find the complete list of streams.
+ * @param streamStart the starting offset of streams in the file
+ * @param footer the footer for the stripe
+ * @param columnInclude which columns are being read
+ */
+ private void findStreams(long streamStart,
+ OrcProto.StripeFooter footer,
+ boolean[] columnInclude) throws IOException {
+ long currentOffset = streamStart;
+ Arrays.fill(bloomFilterKinds, null);
+ for(OrcProto.Stream stream: footer.getStreamsList()) {
+ currentOffset += handleStream(currentOffset, columnInclude, stream, null);
+ }
+
+ // Add the encrypted streams that we are using
+ for(ReaderEncryptionVariant variant: encryption.getVariants()) {
+ int variantId = variant.getVariantId();
+ OrcProto.StripeEncryptionVariant stripeVariant =
+ footer.getEncryption(variantId);
+ for(OrcProto.Stream stream: stripeVariant.getStreamsList()) {
+ currentOffset += handleStream(currentOffset, columnInclude, stream, variant);
+ }
+ }
+ }
+
+ /**
+ * Read and parse the indexes for the current stripe.
+ * @param sargColumns the columns we can use bloom filters for
+ * @param output an OrcIndex to reuse
+ * @return the indexes for the required columns
+ */
+ public OrcIndex readRowIndex(boolean[] sargColumns,
+ OrcIndex output) throws IOException {
+ int typeCount = schema.getMaximumId() + 1;
+ if (output == null) {
+ output = new OrcIndex(new OrcProto.RowIndex[typeCount],
+ new OrcProto.Stream.Kind[typeCount],
+ new OrcProto.BloomFilterIndex[typeCount]);
+ }
+ System.arraycopy(bloomFilterKinds, 0, output.getBloomFilterKinds(), 0,
+ bloomFilterKinds.length);
+ BufferChunkList ranges = planIndexReading(sargColumns);
+ dataReader.readFileData(ranges, false);
+ OrcProto.RowIndex[] indexes = output.getRowGroupIndex();
+ OrcProto.BloomFilterIndex[] blooms = output.getBloomFilterIndex();
+ for(StreamInformation stream: indexStreams) {
+ int column = stream.column;
+ if (stream.firstChunk != null) {
+ CodedInputStream data = InStream.createCodedInputStream(InStream.create(
+ "index", stream.firstChunk, stream.offset,
+ stream.length, getStreamOptions(column, stream.kind)));
+ switch (stream.kind) {
+ case ROW_INDEX:
+ indexes[column] = OrcProto.RowIndex.parseFrom(data);
+ break;
+ case BLOOM_FILTER:
+ case BLOOM_FILTER_UTF8:
+ if (sargColumns != null && sargColumns[column]) {
+ blooms[column] = OrcProto.BloomFilterIndex.parseFrom(data);
+ }
+ break;
+ default:
+ break;
+ }
+ }
+ }
+ return output;
+ }
+
+ private void addChunk(BufferChunkList list, StreamInformation stream,
+ long offset, long length) {
+ while (length > 0) {
+ long thisLen = Math.min(length, maxBufferSize);
+ BufferChunk chunk = new BufferChunk(offset, (int) thisLen);
+ if (stream.firstChunk == null) {
+ stream.firstChunk = chunk;
+ }
+ list.add(chunk);
+ offset += thisLen;
+ length -= thisLen;
+ }
+ }
+
+ /**
+ * Plans the list of disk ranges that the given stripe needs to read the
+ * indexes. All of the positions are relative to the start of the stripe.
+ * @param bloomFilterColumns true for the columns (indexed by file columns) that
+ * we need bloom filters for
+ * @return a list of merged disk ranges to read
+ */
+ private BufferChunkList planIndexReading(boolean[] bloomFilterColumns) {
+ BufferChunkList result = new BufferChunkList();
+ for(StreamInformation stream: indexStreams) {
+ switch (stream.kind) {
+ case ROW_INDEX:
+ addChunk(result, stream, stream.offset, stream.length);
+ break;
+ case BLOOM_FILTER:
+ case BLOOM_FILTER_UTF8:
+ if (bloomFilterColumns[stream.column] &&
+ bloomFilterKinds[stream.column] == stream.kind) {
+ addChunk(result, stream, stream.offset, stream.length);
+ }
+ break;
+ default:
+ // PASS
+ break;
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Plans the list of disk ranges that the given stripe needs to read the
+ * data.
+ * @return a list of merged disk ranges to read
+ */
+ private BufferChunkList planDataReading() {
+ BufferChunkList result = new BufferChunkList();
+ for(StreamInformation stream: dataStreams) {
+ addChunk(result, stream, stream.offset, stream.length);
+ }
+ return result;
+ }
+
+ static boolean hadBadBloomFilters(TypeDescription.Category category,
+ OrcFile.WriterVersion version) {
+ switch(category) {
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return !version.includes(OrcFile.WriterVersion.HIVE_12055);
+ case DECIMAL:
+ // fixed by ORC-101, but ORC-101 changed stream kind to BLOOM_FILTER_UTF8
+ return true;
+ case TIMESTAMP:
+ return !version.includes(OrcFile.WriterVersion.ORC_135);
+ default:
+ return false;
+ }
+ }
+
+ private static boolean hasSomeRowGroups(boolean[] includedRowGroups) {
+ for(boolean include: includedRowGroups) {
+ if (include) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Plan the ranges of the file that we need to read given the list of
+ * columns and row groups.
+ *
+ * @param index the index to use for offsets
+ * @param includedRowGroups which row groups are needed
+ * @return the list of disk ranges that will be loaded
+ */
+ private BufferChunkList planPartialDataReading(OrcIndex index,
+ @NotNull boolean[] includedRowGroups) {
+ BufferChunkList result = new BufferChunkList();
+ if (hasSomeRowGroups(includedRowGroups)) {
+ OrcProto.RowIndex[] rowIndex = index.getRowGroupIndex();
+ boolean isCompressed = compression.isCompressed();
+ for (StreamInformation stream : dataStreams) {
+ if (RecordReaderUtils.isDictionary(stream.kind, encodings[stream.column])) {
+ addChunk(result, stream, stream.offset, stream.length);
+ } else {
+ int column = stream.column;
+ OrcProto.RowIndex ri = rowIndex[column];
+ TypeDescription.Category kind = schema.findSubtype(column).getCategory();
+ long alreadyRead = 0;
+ for (int group = 0; group < includedRowGroups.length; ++group) {
+ if (includedRowGroups[group]) {
+ // find the last group that is selected
+ int endGroup = group;
+ while (endGroup < includedRowGroups.length - 1 &&
+ includedRowGroups[endGroup + 1]) {
+ endGroup += 1;
+ }
+ int posn = RecordReaderUtils.getIndexPosition(
+ encodings[stream.column].getKind(), kind, stream.kind,
+ isCompressed, hasNull[column]);
+ long start = Math.max(alreadyRead,
+ stream.offset + ri.getEntry(group).getPositions(posn));
+ long end = stream.offset;
+ if (endGroup == includedRowGroups.length - 1) {
+ end += stream.length;
+ } else {
+ long nextGroupOffset = ri.getEntry(endGroup + 1).getPositions(posn);
+ end += RecordReaderUtils.estimateRgEndOffset(
+ compression, false, nextGroupOffset, stream.length);
+ }
+ if (alreadyRead < end) {
+ addChunk(result, stream, start, end - start);
+ alreadyRead = end;
+ }
+ group = endGroup;
+ }
+ }
+ }
+ }
+ }
+ return result;
+ }
+
+ private static class StreamInformation {
+ final OrcProto.Stream.Kind kind;
+ final int column;
+ final long offset;
+ final long length;
+ BufferChunk firstChunk;
+
+ StreamInformation(OrcProto.Stream.Kind kind,
+ int column, long offset, long length) {
+ this.kind = kind;
+ this.column = column;
+ this.offset = offset;
+ this.length = length;
+ }
+
+ void releaseBuffers(DataReader reader) {
+ long end = offset + length;
+ BufferChunk ptr = firstChunk;
+ while (ptr != null && ptr.getOffset() < end) {
+ ByteBuffer buffer = ptr.getData();
+ if (buffer != null) {
+ reader.releaseBuffer(buffer);
+ ptr.setChunk(null);
+ }
+ ptr = (BufferChunk) ptr.next;
+ }
+ }
+ }
+}
diff --git a/java/core/src/test/org/apache/orc/TestNewIntegerEncoding.java b/java/core/src/test/org/apache/orc/TestNewIntegerEncoding.java
index 526dd81..3dce13c 100644
--- a/java/core/src/test/org/apache/orc/TestNewIntegerEncoding.java
+++ b/java/core/src/test/org/apache/orc/TestNewIntegerEncoding.java
@@ -207,7 +207,7 @@ public class TestNewIntegerEncoding {
}
}
}
-
+
@Test
public void testBasicDelta1() throws Exception {
TypeDescription schema = TypeDescription.createLong();
@@ -1276,8 +1276,9 @@ public class TestNewIntegerEncoding {
tslist.add(Timestamp.valueOf("1974-01-01 00:00:00"));
int idx = 0;
for (Timestamp ts : tslist) {
- ((TimestampColumnVector) batch.cols[0]).set(idx, ts);
+ ((TimestampColumnVector) batch.cols[0]).set(idx++, ts);
}
+ batch.size = tslist.size();
writer.addRowBatch(batch);
writer.close();
@@ -1287,6 +1288,7 @@ public class TestNewIntegerEncoding {
batch = reader.getSchema().createRowBatch();
idx = 0;
while (rows.nextBatch(batch)) {
+ assertEquals(tslist.size(), batch.size);
for(int r=0; r < batch.size; ++r) {
assertEquals(tslist.get(idx++),
((TimestampColumnVector) batch.cols[0]).asScratchTimestamp(r));
diff --git a/java/core/src/test/org/apache/orc/TestVectorOrcFile.java b/java/core/src/test/org/apache/orc/TestVectorOrcFile.java
index 95f6458..2eb2d23 100644
--- a/java/core/src/test/org/apache/orc/TestVectorOrcFile.java
+++ b/java/core/src/test/org/apache/orc/TestVectorOrcFile.java
@@ -18,10 +18,9 @@
package org.apache.orc;
-import org.apache.hadoop.hive.ql.exec.vector.Decimal64ColumnVector;
+import org.apache.orc.impl.InStream;
import org.apache.orc.impl.OrcCodecPool;
-import org.apache.orc.impl.PhysicalFsWriter;
import org.apache.orc.impl.WriterImpl;
import org.apache.orc.OrcFile.WriterOptions;
@@ -29,12 +28,15 @@ import org.apache.orc.OrcFile.WriterOptions;
import com.google.common.collect.Lists;
import org.apache.orc.impl.ReaderImpl;
+import org.apache.orc.impl.reader.ReaderEncryption;
+import org.apache.orc.impl.reader.StripePlanner;
import org.junit.Assert;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.Decimal64ColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
@@ -80,6 +82,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.function.IntFunction;
import static junit.framework.TestCase.assertNotNull;
import static org.junit.Assert.*;
@@ -549,9 +552,8 @@ public class TestVectorOrcFile {
@Test
public void testStripeLevelStats() throws Exception {
- TypeDescription schema = TypeDescription.createStruct()
- .addField("int1", TypeDescription.createInt())
- .addField("string1", TypeDescription.createString());
+ TypeDescription schema =
+ TypeDescription.fromString("struct<int1:int,string1:string>");
Writer writer = OrcFile.createWriter(testFilePath,
OrcFile.writerOptions(conf)
.setSchema(schema)
@@ -956,19 +958,13 @@ public class TestVectorOrcFile {
}
private static TypeDescription createInnerSchema() {
- return TypeDescription.createStruct()
- .addField("int1", TypeDescription.createInt())
- .addField("string1", TypeDescription.createString());
+ return TypeDescription.fromString("struct<int1:int,string1:string>");
}
private static TypeDescription createComplexInnerSchema()
{
- return TypeDescription.createStruct()
- .addField("int1", TypeDescription.createInt())
- .addField("complex",
- TypeDescription.createStruct()
- .addField("int2", TypeDescription.createInt())
- .addField("String1",TypeDescription.createString()));
+ return TypeDescription.fromString("struct<int1:int,"
+ + "complex:struct<int2:int,String1:string>>");
}
private static TypeDescription createBigRowSchema() {
@@ -1680,14 +1676,9 @@ public class TestVectorOrcFile {
*/
@Test
public void testUnionAndTimestamp() throws Exception {
- TypeDescription schema = TypeDescription.createStruct()
- .addField("time", TypeDescription.createTimestamp())
- .addField("union", TypeDescription.createUnion()
- .addUnionChild(TypeDescription.createInt())
- .addUnionChild(TypeDescription.createString()))
- .addField("decimal", TypeDescription.createDecimal()
- .withPrecision(38)
- .withScale(18));
+ TypeDescription schema = TypeDescription.fromString(
+ "struct<time:timestamp,union:uniontype<int,string>," +
+ "decimal:decimal(38,18)>");
HiveDecimal maxValue = HiveDecimal.create("10000000000000000000");
Writer writer = OrcFile.createWriter(testFilePath,
OrcFile.writerOptions(conf)
@@ -2098,10 +2089,8 @@ public class TestVectorOrcFile {
VectorizedRowBatch batch
) throws IOException {
fs.delete(testFilePath, false);
- PhysicalWriter physical = new PhysicalFsWriter(fs, testFilePath, opts);
- CompressionCodec codec = physical.getStreamOptions().getCodec();
- Writer writer = OrcFile.createWriter(testFilePath,
- opts.physicalWriter(physical));
+ Writer writer = OrcFile.createWriter(testFilePath, opts);
+ CompressionCodec codec = ((WriterImpl) writer).getCompressionCodec();
writeRandomIntBytesBatches(writer, batch, count, size);
writer.close();
return codec;
@@ -2203,7 +2192,8 @@ public class TestVectorOrcFile {
.setSchema(schema)
.stripeSize(200000)
.bufferSize(65536)
- .rowIndexStride(1000));
+ .rowIndexStride(1000)
+ .version(fileFormat));
VectorizedRowBatch batch = schema.createRowBatch();
Random rand = new Random(42);
final int COUNT=32768;
@@ -2244,18 +2234,24 @@ public class TestVectorOrcFile {
Assert.assertEquals(COUNT, reader.getNumberOfRows());
RecordReader rows = reader.rows();
// get the row index
+ InStream.StreamOptions options = InStream.options();
+ if (reader.getCompressionKind() != CompressionKind.NONE) {
+ options.withCodec(OrcCodecPool.getCodec(reader.getCompressionKind()))
+ .withBufferSize(reader.getCompressionSize());
+ }
DataReader meta = RecordReaderUtils.createDefaultDataReader(
DataReaderProperties.builder()
- .withBufferSize(reader.getCompressionSize())
.withFileSystem(fs)
.withPath(testFilePath)
- .withCompression(reader.getCompressionKind())
- .withTypeCount(reader.getSchema().getMaximumId() + 1)
+ .withCompression(options)
.withZeroCopy(false)
.build());
- OrcIndex index =
- meta.readRowIndex(reader.getStripes().get(0), null, null, false, null, null,
- null, OrcFile.WriterVersion.ORC_101, null, null);
+ StripePlanner planner = new StripePlanner(schema, new ReaderEncryption(),
+ options, meta, reader.getWriterVersion(), true, Integer.MAX_VALUE);
+ boolean[] columns = new boolean[schema.getMaximumId() + 1];
+ Arrays.fill(columns, true);
+ OrcIndex index = planner.parseStripe(reader.getStripes().get(0), columns)
+ .readRowIndex(null, null);
// check the primitive columns to make sure they have the right number of
// items in the first row group
for(int c=1; c < 9; ++c) {
@@ -2293,7 +2289,7 @@ public class TestVectorOrcFile {
offsetOfStripe4 = stripe.getOffset();
}
}
- boolean[] columns = new boolean[reader.getStatistics().length];
+ Arrays.fill(columns, false);
columns[5] = true; // long colulmn
columns[9] = true; // text column
rows = reader.rows(reader.options()
@@ -3580,7 +3576,8 @@ public class TestVectorOrcFile {
.stripeSize(400000L)
.compress(CompressionKind.NONE)
.bufferSize(500)
- .rowIndexStride(1000));
+ .rowIndexStride(1000)
+ .version(fileFormat));
VectorizedRowBatch batch = schema.createRowBatch();
batch.ensureSize(3500);
batch.size = 3500;
@@ -3675,4 +3672,277 @@ public class TestVectorOrcFile {
Assert.assertEquals(3500, rows.getRowNumber());
}
+ @Test
+ public void testColumnEncryption() throws Exception {
+ Assume.assumeTrue(fileFormat != OrcFile.Version.V_0_11);
+ final int ROWS = 1000;
+ final int SEED = 2;
+ final Random random = new Random(SEED);
+
+ TypeDescription schema =
+ TypeDescription.fromString("struct<i:int,norm:int,x:array<string>>");
+
+ byte[] piiKey = new byte[16];
+ random.nextBytes(piiKey);
+ byte[] creditKey = new byte[32];
+ random.nextBytes(creditKey);
+ InMemoryKeystore keys = new InMemoryKeystore()
+ .addKey("pii", EncryptionAlgorithm.AES_CTR_128, piiKey)
+ .addKey("credit", EncryptionAlgorithm.AES_CTR_256, creditKey);
+
+ Writer writer = OrcFile.createWriter(testFilePath,
+ OrcFile.writerOptions(conf)
+ .setSchema(schema)
+ .version(fileFormat)
+ .setKeyProvider(keys)
+ .encryptColumn("i", "pii")
+ .encryptColumn("x", "credit"));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ batch.size = ROWS;
+ LongColumnVector i = (LongColumnVector) batch.cols[0];
+ LongColumnVector norm = (LongColumnVector) batch.cols[1];
+ ListColumnVector x = (ListColumnVector) batch.cols[2];
+ BytesColumnVector xElem = (BytesColumnVector) x.child;
+ xElem.ensureSize(3 * ROWS, false);
+ for(int r=0; r < ROWS; ++r) {
+ i.vector[r] = r * 3;
+ norm.vector[r] = r * 5;
+ int start = x.childCount;
+ x.offsets[r] = start;
+ x.lengths[r] = 3;
+ x.childCount += x.lengths[r];
+ for(int child=0; child < x.lengths[r]; ++child) {
+ xElem.setVal(start + child,
+ String.format("%d.%d", r, child).getBytes(StandardCharsets.UTF_8));
+ }
+ }
+ writer.addRowBatch(batch);
+ writer.close();
+
+ // Read without any keys
+ Reader reader = OrcFile.createReader(testFilePath,
+ OrcFile.readerOptions(conf)
+ .setKeyProvider(new InMemoryKeystore()));
+ ColumnStatistics[] stats = reader.getStatistics();
+ assertEquals(ROWS, stats[0].getNumberOfValues());
+
+ assertEquals(0, stats[1].getNumberOfValues());
+ assertEquals(true, stats[1].hasNull());
+ assertEquals(ROWS, stats[2].getNumberOfValues());
+ assertEquals(0, ((IntegerColumnStatistics) stats[2]).getMinimum());
+ assertEquals(ROWS * 5 - 5, ((IntegerColumnStatistics) stats[2]).getMaximum());
+ assertEquals(0, stats[3].getNumberOfValues());
+ assertEquals(0, stats[4].getNumberOfValues());
+
+ RecordReader rows = reader.rows();
+ batch = reader.getSchema().createRowBatch();
+ i = (LongColumnVector) batch.cols[0];
+ norm = (LongColumnVector) batch.cols[1];
+ x = (ListColumnVector) batch.cols[2];
+
+ // ensure that we get the right number of rows with all nulls
+ assertTrue(rows.nextBatch(batch));
+ assertEquals(ROWS, batch.size);
+ assertEquals(true, i.isRepeating);
+ assertEquals(false, i.noNulls);
+ assertEquals(true, i.isNull[0]);
+ assertEquals(true, x.isRepeating);
+ assertEquals(false, x.noNulls);
+ assertEquals(true, x.isNull[0]);
+ for(int r=0; r < ROWS; ++r) {
+ assertEquals("row " + r, r * 5, norm.vector[r]);
+ }
+ assertFalse(rows.nextBatch(batch));
+ rows.close();
+
+ // Add a new version of the pii key
+ random.nextBytes(piiKey);
+ keys.addKey("pii", 1, EncryptionAlgorithm.AES_CTR_128, piiKey);
+
+ // Read with the keys
+ reader = OrcFile.createReader(testFilePath,
+ OrcFile.readerOptions(conf)
+ .setKeyProvider(keys));
+ stats = reader.getStatistics();
+ assertEquals(ROWS, stats[0].getNumberOfValues());
+ assertEquals(ROWS, stats[1].getNumberOfValues());
+ assertEquals(0, ((IntegerColumnStatistics) stats[1]).getMinimum());
+ assertEquals(3 * (ROWS - 1), ((IntegerColumnStatistics) stats[1]).getMaximum());
+ assertEquals(0, ((IntegerColumnStatistics) stats[2]).getMinimum());
+ assertEquals(5 * (ROWS - 1), ((IntegerColumnStatistics) stats[2]).getMaximum());
+ assertEquals(ROWS, stats[3].getNumberOfValues());
+ assertEquals(3 * ROWS, stats[4].getNumberOfValues());
+ assertEquals("0.0", ((StringColumnStatistics)stats[4]).getMinimum());
+ assertEquals("999.2", ((StringColumnStatistics)stats[4]).getMaximum());
+
+ rows = reader.rows();
+ batch = reader.getSchema().createRowBatch();
+ i = (LongColumnVector) batch.cols[0];
+ norm = (LongColumnVector) batch.cols[1];
+ x = (ListColumnVector) batch.cols[2];
+ xElem = (BytesColumnVector) x.child;
+ assertTrue(rows.nextBatch(batch));
+ assertEquals(ROWS, batch.size);
+ assertEquals(false, i.isRepeating);
+ assertEquals(false, x.isRepeating);
+ assertEquals(false, xElem.isRepeating);
+ assertEquals(true, i.noNulls);
+ assertEquals(true, x.noNulls);
+ assertEquals(true, xElem.noNulls);
+ for(int r=0; r < ROWS; ++r) {
+ assertEquals("row " + r, r * 3, i.vector[r]);
+ assertEquals("row " + r, r * 5, norm.vector[r]);
+ assertEquals("row " + r, r * 3, x.offsets[r]);
+ assertEquals("row " + r, 3, x.lengths[r]);
+ for(int child=0; child < x.lengths[r]; ++child) {
+ assertEquals("row " + r + "." + child, String.format("%d.%d", r, child),
+ xElem.toString((int) x.offsets[r] + child));
+ }
+ }
+ assertFalse(rows.nextBatch(batch));
+ rows.close();
+ }
+
+ @Test
+ public void testMultiStripeColumnEncryption() throws Exception {
+ Assume.assumeTrue(fileFormat != OrcFile.Version.V_0_11);
+ final EncryptionAlgorithm algorithm = EncryptionAlgorithm.AES_CTR_128;
+ final int BATCHES = 100;
+ final int SEED = 3;
+ final Random random = new Random(SEED);
+
+ TypeDescription schema = TypeDescription.fromString(
+ "struct<dec:decimal(20,4)," +
+ "dt:date," +
+ "time:timestamp," +
+ "dbl:double," +
+ "bool:boolean," +
+ "bin:binary>");
+
+ InMemoryKeystore allKeys = new InMemoryKeystore();
+ byte[][] masterKeys = new byte[6][];
+ for(int k=0; k < masterKeys.length; ++k) {
+ masterKeys[k] = new byte[algorithm.keyLength()];
+ random.nextBytes(masterKeys[k]);
+ allKeys.addKey("key_" + k, algorithm, masterKeys[k]);
+ }
+
+ Writer writer = OrcFile.createWriter(testFilePath,
+ OrcFile.writerOptions(conf)
+ .setSchema(schema)
+ .version(fileFormat)
+ .stripeSize(10000)
+ .setKeyProvider(allKeys)
+ .encryptColumn("dec", "key_0")
+ .encryptColumn("dt", "key_1")
+ .encryptColumn("time", "key_2")
+ .encryptColumn("dbl", "key_3")
+ .encryptColumn("bool", "key_4")
+ .encryptColumn("bin", "key_5"));
+ // Set size to 1000 precisely so that stripes are exactly 5000 rows long.
+ VectorizedRowBatch batch = schema.createRowBatch(1000);
+ DecimalColumnVector dec = (DecimalColumnVector) batch.cols[0];
+ LongColumnVector dt = (LongColumnVector) batch.cols[1];
+ TimestampColumnVector time = (TimestampColumnVector) batch.cols[2];
+ DoubleColumnVector dbl = (DoubleColumnVector) batch.cols[3];
+ LongColumnVector bool = (LongColumnVector) batch.cols[4];
+ BytesColumnVector bin = (BytesColumnVector) batch.cols[5];
+ // Generate 100 batches of 1,000 rows each
+ batch.size = 1000;
+ dec.isRepeating = true;
+ dt.isRepeating = true;
+ time.isRepeating = true;
+ dbl.isRepeating = true;
+ bool.isRepeating = true;
+ bin.isRepeating = true;
+ for(int b=0; b < BATCHES; ++b) {
+ dec.set(0, new HiveDecimalWritable(String.format("%d.%03d", b, b)));
+ dt.vector[0] = new DateWritable(new Date(96 + b, 12, 11)).getDays();
+ time.set(0, Timestamp.valueOf(String.format("2014-12-14 12:00:00.%04d", b)));
+ dbl.vector[0] = b + 0.5;
+ bool.vector[0] = b % 2;
+ bin.setVal(0, Integer.toString(b).getBytes(StandardCharsets.UTF_8));
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+
+ // Read without any keys
+ Reader reader = OrcFile.createReader(testFilePath,
+ OrcFile.readerOptions(conf)
+ .setKeyProvider(new InMemoryKeystore()));
+ checkHasData(reader.rows(), batch, BATCHES,
+ false, false, false, false, false, false);
+
+ // read with all of the keys
+ reader = OrcFile.createReader(testFilePath,
+ OrcFile.readerOptions(conf)
+ .setKeyProvider(allKeys));
+ checkHasData(reader.rows(), batch, BATCHES,
+ true, true, true, true, true, true);
+
+ // try enabling each key by itself
+ for(int c=0; c < 6; ++c) {
+ InMemoryKeystore single = new InMemoryKeystore();
+ single.addKey("key_" + c, algorithm, masterKeys[c]);
+ reader = OrcFile.createReader(testFilePath,
+ OrcFile.readerOptions(conf).setKeyProvider(single));
+ boolean[] hasData = new boolean[6];
+ hasData[c] = true;
+ checkHasData(reader.rows(), batch, BATCHES, hasData);
+ }
+ }
+
+ private void checkHasData(RecordReader reader, VectorizedRowBatch batch,
+ int BATCHES, boolean... hasData) throws IOException {
+ for(int b=0; b < BATCHES; ++b) {
+ assertEquals("batch " + b, true, reader.nextBatch(batch));
+ for(int c=0; c < hasData.length; c++) {
+ if (hasData[c]) {
+ // the expected value
+ String expected = null;
+ // a function from the row to the value as a string
+ IntFunction<String> actual = row -> null;
+ switch (c) {
+ case 0:
+ expected = new HiveDecimalWritable(String.format("%d.%03d", b, b)).toString();
+ actual = row -> ((DecimalColumnVector) batch.cols[0]).vector[row].toString();
+ break;
+ case 1:
+ expected = Long.toString(new DateWritable(new Date(96 + b, 12, 11)).getDays());
+ actual = row -> Long.toString(((LongColumnVector) batch.cols[1]).vector[row]);
+ break;
+ case 2:
+ expected = Timestamp.valueOf(String.format("2014-12-14 12:00:00.%04d", b)).toString();
+ actual = row -> ((TimestampColumnVector) batch.cols[2]).asScratchTimestamp(row).toString();
+ break;
+ case 3:
+ expected = Double.toString(b + 0.5);
+ actual = row -> Double.toString(((DoubleColumnVector) batch.cols[3]).vector[row]);
+ break;
+ case 4:
+ expected = Long.toString(b % 2);
+ actual = row -> Long.toString(((LongColumnVector) batch.cols[4]).vector[row]);
+ break;
+ case 5:
+ expected = Integer.toString(b);
+ actual = row -> ((BytesColumnVector) batch.cols[5]).toString(row);
+ break;
+ }
+ assertEquals("batch " + b + " column " + c, true, batch.cols[c].noNulls);
+ assertEquals("batch " + b + " column " + c + " row 0", expected, actual.apply(0));
+ // Not all of the readers set isRepeating, so if it isn't set, check the values.
+ if (!batch.cols[c].isRepeating) {
+ for(int r=1; r < batch.size; ++r) {
+ assertEquals("batch " + b + " column " + c + " row " + r, expected, actual.apply(r));
+ }
+ }
+ } else {
+ assertEquals("batch " + b + " column " + c, true, batch.cols[c].isRepeating);
+ assertEquals("batch " + b + " column " + c, true, batch.cols[c].isNull[0]);
+ }
+ }
+ }
+ assertEquals("end", false, reader.nextBatch(batch));
+ reader.close();
+ }
}
diff --git a/java/core/src/test/org/apache/orc/impl/MockDataReader.java b/java/core/src/test/org/apache/orc/impl/MockDataReader.java
new file mode 100644
index 0000000..ccf533d
--- /dev/null
+++ b/java/core/src/test/org/apache/orc/impl/MockDataReader.java
@@ -0,0 +1,132 @@
+/*
+ * 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.orc.impl;
+
+import org.apache.orc.CompressionCodec;
+import org.apache.orc.DataReader;
+import org.apache.orc.OrcProto;
+import org.apache.orc.StripeInformation;
+import org.apache.orc.TypeDescription;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Set;
+
+public class MockDataReader implements DataReader {
+ private final List<MockStripe> stripes = new ArrayList<>();
+ private final int numColumns;
+ private long offset = 3;
+ private MockStripe next;
+ private Set<ByteBuffer> outBuffers =
+ Collections.newSetFromMap(new IdentityHashMap<>());
+
+ public MockDataReader(TypeDescription schema) {
+ numColumns = schema.getMaximumId() + 1;
+ next = new MockStripe(numColumns, stripes.size(), offset);
+ }
+
+ public MockDataReader addStream(int column, OrcProto.Stream.Kind kind,
+ ByteBuffer bytes) {
+ next.addStream(column, kind, offset, bytes);
+ offset += bytes.remaining();
+ return this;
+ }
+
+ public MockDataReader addEncoding(OrcProto.ColumnEncoding.Kind kind) {
+ next.addEncoding(kind);
+ return this;
+ }
+
+ public MockDataReader addStripeFooter(int rows, String timezone) {
+ next.close(rows, timezone);
+ stripes.add(next);
+ offset += next.getFooterLength();
+ next = new MockStripe(numColumns, stripes.size(), offset);
+ return this;
+ }
+
+ @Override
+ public void open() { }
+
+ @Override
+ public OrcProto.StripeFooter readStripeFooter(StripeInformation stripe) {
+ return stripes.get((int) stripe.getStripeId()).getFooter();
+ }
+
+ @Override
+ public BufferChunkList readFileData(BufferChunkList list,
+ boolean doForceDirect) {
+ for(BufferChunk buffer = list.get(); buffer != null;
+ buffer = (BufferChunk) buffer.next) {
+ if (!buffer.hasData()) {
+ MockStripe stripe = getStripeByOffset(buffer.getOffset());
+ ByteBuffer data = stripe.getData(buffer.getOffset(), buffer.getLength());
+ outBuffers.add(data);
+ buffer.setChunk(data);
+ }
+ }
+ return list;
+ }
+
+ @Override
+ public boolean isTrackingDiskRanges() {
+ return true;
+ }
+
+ @Override
+ public void releaseBuffer(ByteBuffer toRelease) {
+ outBuffers.remove(toRelease);
+ }
+
+ @Override
+ public DataReader clone() {
+ throw new UnsupportedOperationException("Clone not supported.");
+ }
+
+ @Override
+ public void close() { }
+
+ @Override
+ public CompressionCodec getCompressionCodec() {
+ return null;
+ }
+
+ public MockStripe getStripe(int id) {
+ return stripes.get(id);
+ }
+
+ private MockStripe getStripeByOffset(long offset) {
+ for(MockStripe stripe: stripes) {
+ if (stripe.getOffset() <= offset &&
+ (offset - stripe.getOffset() < stripe.getLength())) {
+ return stripe;
+ }
+ }
+ throw new IllegalArgumentException("Can't find stripe at " + offset);
+ }
+
+ void resetCounts() {
+ for(MockStripe stripe: stripes) {
+ stripe.resetCounts();
+ }
+ }
+}
diff --git a/java/core/src/test/org/apache/orc/impl/MockStream.java b/java/core/src/test/org/apache/orc/impl/MockStream.java
new file mode 100644
index 0000000..cf45ff7
--- /dev/null
+++ b/java/core/src/test/org/apache/orc/impl/MockStream.java
@@ -0,0 +1,56 @@
+/*
+ * 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.orc.impl;
+
+import org.apache.orc.OrcProto;
+
+import java.nio.ByteBuffer;
+
+class MockStream {
+ final int column;
+ final OrcProto.Stream.Kind kind;
+ private final ByteBuffer bytes;
+ final long offset;
+ final int length;
+ int readCount = 0;
+
+ MockStream(int column, OrcProto.Stream.Kind kind, long offset,
+ ByteBuffer bytes) {
+ this.column = column;
+ this.kind = kind;
+ this.bytes = bytes;
+ this.offset = offset;
+ this.length = bytes.remaining();
+ }
+
+ ByteBuffer getData(long offset, int length) {
+ if (offset < this.offset ||
+ offset + length > this.offset + this.length) {
+ throw new IllegalArgumentException("Bad getData [" + offset + ", " +
+ (offset + length) + ") from [" +
+ this.offset + ", " +
+ (this.offset + this.length) + ")");
+ }
+ ByteBuffer copy = bytes.duplicate();
+ int posn = (int) (offset - this.offset);
+ copy.position(posn);
+ copy.limit(posn + length);
+ return copy;
+ }
+}
diff --git a/java/core/src/test/org/apache/orc/impl/MockStripe.java b/java/core/src/test/org/apache/orc/impl/MockStripe.java
new file mode 100644
index 0000000..0e1e3f8
--- /dev/null
+++ b/java/core/src/test/org/apache/orc/impl/MockStripe.java
@@ -0,0 +1,168 @@
+/*
+ * 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.orc.impl;
+
+import org.apache.orc.OrcProto;
+import org.apache.orc.StripeInformation;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MockStripe implements StripeInformation {
+ private final int numColumns;
+ private final List<MockStream> streams = new ArrayList<>();
+ private final List<OrcProto.ColumnEncoding> encodings = new ArrayList<>();
+ private final long startOffset;
+ private final int stripeId;
+ private int indexLength = 0;
+ private int dataLength = 0;
+ private int footerLength = 0;
+ private OrcProto.StripeFooter footer;
+ private int rows = 0;
+
+ public MockStripe(int numColumns, int stripeId, long offset) {
+ this.numColumns = numColumns;
+ this.startOffset = offset;
+ this.stripeId = stripeId;
+ }
+
+ public MockStream addStream(int column, OrcProto.Stream.Kind kind,
+ long offset, ByteBuffer bytes) {
+ MockStream result = new MockStream(column, kind, offset, bytes);
+ streams.add(result);
+ int length = bytes.remaining();
+ if (StreamName.getArea(kind) == StreamName.Area.INDEX) {
+ indexLength += length;
+ } else {
+ dataLength += length;
+ }
+ return result;
+ }
+
+ public void addEncoding(OrcProto.ColumnEncoding.Kind kind) {
+ OrcProto.ColumnEncoding.Builder result =
+ OrcProto.ColumnEncoding.newBuilder();
+ result.setKind(kind);
+ encodings.add(result.build());
+ }
+
+ public void close(int rows, String timezone) {
+ this.rows = rows;
+ // make sure we have enough encodings
+ while (encodings.size() < numColumns) {
+ addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT);
+ }
+ OrcProto.StripeFooter.Builder foot = OrcProto.StripeFooter.newBuilder();
+ for(MockStream stream: streams) {
+ foot.addStreams(OrcProto.Stream.newBuilder()
+ .setKind(stream.kind)
+ .setLength(stream.length)
+ .setColumn(stream.column).build());
+ }
+ for(OrcProto.ColumnEncoding encoding: encodings) {
+ foot.addColumns(encoding);
+ }
+ if (timezone != null) {
+ foot.setWriterTimezone(timezone);
+ }
+ footer = foot.build();
+ footerLength = footer.getSerializedSize();
+ }
+
+ @Override
+ public long getOffset() {
+ return startOffset;
+ }
+
+ @Override
+ public long getLength() {
+ return indexLength + dataLength + footerLength;
+ }
+
+ @Override
+ public long getIndexLength() {
+ return indexLength;
+ }
+
+ @Override
+ public long getDataLength() {
+ return dataLength;
+ }
+
+ @Override
+ public long getFooterLength() {
+ return footerLength;
+ }
+
+ @Override
+ public long getNumberOfRows() {
+ return rows;
+ }
+
+ @Override
+ public long getStripeId() {
+ return stripeId;
+ }
+
+ @Override
+ public long getEncryptionStripeId() {
+ return stripeId;
+ }
+
+ @Override
+ public byte[][] getEncryptedLocalKeys() {
+ return new byte[0][];
+ }
+
+ public MockStream getStream(int column, OrcProto.Stream.Kind kind) {
+ for (MockStream stream: streams) {
+ if (stream.column == column && stream.kind == kind) {
+ return stream;
+ }
+ }
+ throw new IllegalArgumentException("Can't find stream column " + column
+ + " kind " + kind);
+ }
+
+ private MockStream getStream(long offset) {
+ for (MockStream stream: streams) {
+ if (stream.offset <= offset && offset < stream.offset + stream.length) {
+ return stream;
+ }
+ }
+ throw new IllegalArgumentException("Can't find stream at offset " + offset);
+ }
+
+ public ByteBuffer getData(long offset, int length) {
+ MockStream stream = getStream(offset);
+ stream.readCount += 1;
+ return stream.getData(offset, length);
+ }
+
+ public void resetCounts() {
+ for(MockStream stream: streams) {
+ stream.readCount = 0;
+ }
+ }
+
+ public OrcProto.StripeFooter getFooter() {
+ return footer;
+ }
+}
diff --git a/java/core/src/test/org/apache/orc/impl/TestDataReaderProperties.java b/java/core/src/test/org/apache/orc/impl/TestDataReaderProperties.java
index 46546b0..8fcf98d 100644
--- a/java/core/src/test/org/apache/orc/impl/TestDataReaderProperties.java
+++ b/java/core/src/test/org/apache/orc/impl/TestDataReaderProperties.java
@@ -36,15 +36,18 @@ public class TestDataReaderProperties {
@Test
public void testCompleteBuild() {
+ InStream.StreamOptions options = InStream.options()
+ .withCodec(OrcCodecPool.getCodec(CompressionKind.ZLIB));
DataReaderProperties properties = DataReaderProperties.builder()
.withFileSystem(mockedFileSystem)
.withPath(mockedPath)
- .withCompression(CompressionKind.ZLIB)
+ .withCompression(options)
.withZeroCopy(mockedZeroCopy)
.build();
assertEquals(mockedFileSystem, properties.getFileSystem());
assertEquals(mockedPath, properties.getPath());
- assertEquals(CompressionKind.ZLIB, properties.getCompression());
+ assertEquals(CompressionKind.ZLIB,
+ properties.getCompression().getCodec().getKind());
assertEquals(mockedZeroCopy, properties.getZeroCopy());
}
@@ -69,7 +72,7 @@ public class TestDataReaderProperties {
public void testMissingPath() {
DataReaderProperties.builder()
.withFileSystem(mockedFileSystem)
- .withCompression(CompressionKind.NONE)
+ .withCompression(InStream.options())
.withZeroCopy(mockedZeroCopy)
.build();
}
@@ -78,7 +81,7 @@ public class TestDataReaderProperties {
public void testMissingFileSystem() {
DataReaderProperties.builder()
.withPath(mockedPath)
- .withCompression(CompressionKind.NONE)
+ .withCompression(InStream.options())
.withZeroCopy(mockedZeroCopy)
.build();
}
diff --git a/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java b/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java
index 37f25db..a861494 100644
--- a/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java
+++ b/java/core/src/test/org/apache/orc/impl/TestOrcLargeStripe.java
@@ -18,9 +18,14 @@ 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.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.File;
@@ -33,7 +38,6 @@ 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;
@@ -43,7 +47,6 @@ 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;
@@ -55,7 +58,6 @@ 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"));
@@ -77,38 +79,62 @@ public class TestOrcLargeStripe {
@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 {
+ static class RangeBuilder {
+ BufferChunkList result = new BufferChunkList();
- DiskRangeList rangeList = createRangeList(stripeLength);
-
- DiskRangeList newList = RecordReaderUtils.readDiskRanges(mockDataInput, null, 0, rangeList, false, (int) (2 * MB));
- assertEquals(expectedChunks, newList.listSize());
+ RangeBuilder range(long offset, int length) {
+ result.add(new BufferChunk(offset, length));
+ return this;
+ }
- newList = RecordReaderUtils.readDiskRanges(mockDataInput, null, 0, rangeList, true, (int) (2 * MB));
- assertEquals(expectedChunks, newList.listSize());
+ BufferChunkList build() {
+ return result;
+ }
+ }
+ @Test
+ public void testZeroCopy() throws Exception {
+ BufferChunkList ranges = new RangeBuilder().range(1000, 3000).build();
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());
+ when(mockZcr.readBuffer(anyInt(), anyBoolean()))
+ .thenAnswer(invocation -> ByteBuffer.allocate(1000));
+ RecordReaderUtils.readDiskRanges(mockDataInput, mockZcr, ranges, true);
+ verify(mockDataInput).seek(1000);
+ verify(mockZcr).readBuffer(3000, false);
+ verify(mockZcr).readBuffer(2000, false);
+ verify(mockZcr).readBuffer(1000, false);
+ }
+
+ @Test
+ public void testRangeMerge() throws Exception {
+ BufferChunkList rangeList = new RangeBuilder()
+ .range(100, 1000)
+ .range(1000, 10000)
+ .range(3000, 30000).build();
+ RecordReaderUtils.readDiskRanges(mockDataInput, null, rangeList, false);
+ verify(mockDataInput).readFully(eq(100L), any(), eq(0), eq(32900));
}
@Test
- public void testStripeSizesBelowAndGreaterThanLimit() throws Exception {
- verifyDiskRanges(MB, 1);
- verifyDiskRanges(5 * MB, 3);
+ public void testRangeSkip() throws Exception {
+ BufferChunkList rangeList = new RangeBuilder()
+ .range(1000, 1000)
+ .range(2000, 1000)
+ .range(4000, 1000)
+ .range(4100, 100)
+ .range(8000, 1000).build();
+ RecordReaderUtils.readDiskRanges(mockDataInput, null, rangeList, false);
+ verify(mockDataInput).readFully(eq(1000L), any(), eq(0), eq(2000));
+ verify(mockDataInput).readFully(eq(4000L), any(), eq(0), eq(1000));
+ verify(mockDataInput).readFully(eq(8000L), any(), eq(0), eq(1000));
}
+ @Test
+ public void testEmpty() throws Exception {
+ BufferChunkList rangeList = new RangeBuilder().build();
+ RecordReaderUtils.readDiskRanges(mockDataInput, null, rangeList, false);
+ verify(mockDataInput, never()).readFully(anyLong(), any(), anyInt(), anyInt());
+ }
@Test
public void testConfigMaxChunkLimit() throws IOException {
@@ -117,28 +143,25 @@ public class TestOrcLargeStripe {
TypeDescription schema = TypeDescription.createTimestamp();
testFilePath = new Path(workDir, "TestOrcLargeStripe." +
testCaseName.getMethodName() + ".orc");
+ fs.delete(testFilePath, false);
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);
- }
+ 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());
}
@Test
diff --git a/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java b/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java
index 9eaaa61..0763363 100644
--- a/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java
+++ b/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -18,13 +18,11 @@
package org.apache.orc.impl;
-import static junit.framework.Assert.assertEquals;
-import static junit.framework.TestCase.fail;
-import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
@@ -36,7 +34,8 @@ import static org.mockito.Mockito.when;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
+import java.nio.ByteBuffer;
+import java.nio.IntBuffer;
import java.sql.Date;
import java.sql.Timestamp;
import java.text.DateFormat;
@@ -47,7 +46,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.TimeZone;
-import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
@@ -62,6 +60,11 @@ import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentImpl;
+import org.apache.orc.CompressionCodec;
+import org.apache.orc.CompressionKind;
+import org.apache.orc.impl.reader.ReaderEncryption;
+import org.apache.orc.impl.reader.StripePlanner;
+import org.apache.orc.impl.writer.StreamOptions;
import org.apache.orc.util.BloomFilter;
import org.apache.orc.DataReader;
import org.apache.orc.RecordReader;
@@ -80,7 +83,6 @@ import org.apache.orc.OrcProto;
import org.apache.orc.util.BloomFilterIO;
import org.apache.orc.util.BloomFilterUtf8;
-import org.junit.Assert;
import org.junit.Test;
import org.mockito.MockSettings;
import org.mockito.Mockito;
@@ -112,9 +114,6 @@ public class TestRecordReaderImpl {
literal, literalList);
}
- // can add .verboseLogging() to cause Mockito to log invocations
- private final MockSettings settings = Mockito.withSettings().verboseLogging();
-
static class BufferInStream
extends InputStream implements PositionedReadable, Seekable {
private final byte[] buffer;
@@ -205,18 +204,19 @@ public class TestRecordReaderImpl {
footer.writeTo(buffer);
ps.writeTo(buffer);
buffer.write(ps.getSerializedSize());
- FileSystem fs = mock(FileSystem.class, settings);
+ FileSystem fs = mock(FileSystem.class);
FSDataInputStream file =
new FSDataInputStream(new BufferInStream(buffer.getData(),
buffer.getLength()));
Path p = new Path("/dir/file.orc");
- when(fs.open(p)).thenReturn(file);
+ when(fs.open(eq(p))).thenReturn(file);
OrcFile.ReaderOptions options = OrcFile.readerOptions(conf);
options.filesystem(fs);
options.maxLength(buffer.getLength());
- when(fs.getFileStatus(p))
+ when(fs.getFileStatus(eq(p)))
.thenReturn(new FileStatus(10, false, 3, 3000, 0, p));
Reader reader = OrcFile.createReader(p, options);
+ assertEquals(0, reader.getNumberOfRows());
}
static class StubPredicate implements PredicateLeaf {
@@ -953,7 +953,7 @@ public class TestRecordReaderImpl {
}
@Test
- public void testBetween() throws Exception {
+ public void testBetween() {
List<Object> args = new ArrayList<Object>();
args.add(10L);
args.add(20L);
@@ -1001,7 +1001,7 @@ public class TestRecordReaderImpl {
@Test
- public void testEqualsWithNullInStats() throws Exception {
+ public void testEqualsWithNullInStats() {
PredicateLeaf pred = createPredicateLeaf
(PredicateLeaf.Operator.EQUALS, PredicateLeaf.Type.STRING,
"x", "c", null);
@@ -1020,7 +1020,7 @@ public class TestRecordReaderImpl {
}
@Test
- public void testNullSafeEqualsWithNullInStats() throws Exception {
+ public void testNullSafeEqualsWithNullInStats() {
PredicateLeaf pred = createPredicateLeaf
(PredicateLeaf.Operator.NULL_SAFE_EQUALS, PredicateLeaf.Type.STRING,
"x", "c", null);
@@ -1039,7 +1039,7 @@ public class TestRecordReaderImpl {
}
@Test
- public void testLessThanWithNullInStats() throws Exception {
+ public void testLessThanWithNullInStats() {
PredicateLeaf pred = createPredicateLeaf
(PredicateLeaf.Operator.LESS_THAN, PredicateLeaf.Type.STRING,
"x", "c", null);
@@ -1236,341 +1236,277 @@ public class TestRecordReaderImpl {
@Test
public void testGetIndexPosition() throws Exception {
+ boolean uncompressed = false;
+ boolean compressed = true;
assertEquals(0, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.INT,
- OrcProto.Stream.Kind.PRESENT, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.INT,
+ OrcProto.Stream.Kind.PRESENT, compressed, true));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.INT,
- OrcProto.Stream.Kind.DATA, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.INT,
+ OrcProto.Stream.Kind.DATA, compressed, true));
assertEquals(3, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.INT,
- OrcProto.Stream.Kind.DATA, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.INT,
+ OrcProto.Stream.Kind.DATA, uncompressed, true));
assertEquals(0, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.INT,
- OrcProto.Stream.Kind.DATA, true, false));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.INT,
+ OrcProto.Stream.Kind.DATA, compressed, false));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DICTIONARY, OrcProto.Type.Kind.STRING,
- OrcProto.Stream.Kind.DATA, true, true));
+ (OrcProto.ColumnEncoding.Kind.DICTIONARY, TypeDescription.Category.STRING,
+ OrcProto.Stream.Kind.DATA, compressed, true));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.BINARY,
- OrcProto.Stream.Kind.DATA, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.BINARY,
+ OrcProto.Stream.Kind.DATA, compressed, true));
assertEquals(3, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.BINARY,
- OrcProto.Stream.Kind.DATA, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.BINARY,
+ OrcProto.Stream.Kind.DATA, uncompressed, true));
assertEquals(6, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.BINARY,
- OrcProto.Stream.Kind.LENGTH, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.BINARY,
+ OrcProto.Stream.Kind.LENGTH, compressed, true));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.BINARY,
- OrcProto.Stream.Kind.LENGTH, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.BINARY,
+ OrcProto.Stream.Kind.LENGTH, uncompressed, true));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.DECIMAL,
- OrcProto.Stream.Kind.DATA, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.DECIMAL,
+ OrcProto.Stream.Kind.DATA, compressed, true));
assertEquals(3, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.DECIMAL,
- OrcProto.Stream.Kind.DATA, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.DECIMAL,
+ OrcProto.Stream.Kind.DATA, uncompressed, true));
assertEquals(6, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.DECIMAL,
- OrcProto.Stream.Kind.SECONDARY, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.DECIMAL,
+ OrcProto.Stream.Kind.SECONDARY, compressed, true));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.DECIMAL,
- OrcProto.Stream.Kind.SECONDARY, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.DECIMAL,
+ OrcProto.Stream.Kind.SECONDARY, uncompressed, true));
assertEquals(4, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.TIMESTAMP,
- OrcProto.Stream.Kind.DATA, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.TIMESTAMP,
+ OrcProto.Stream.Kind.DATA, compressed, true));
assertEquals(3, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.TIMESTAMP,
- OrcProto.Stream.Kind.DATA, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.TIMESTAMP,
+ OrcProto.Stream.Kind.DATA, uncompressed, true));
assertEquals(7, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.TIMESTAMP,
- OrcProto.Stream.Kind.SECONDARY, true, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.TIMESTAMP,
+ OrcProto.Stream.Kind.SECONDARY, compressed, true));
assertEquals(5, RecordReaderUtils.getIndexPosition
- (OrcProto.ColumnEncoding.Kind.DIRECT, OrcProto.Type.Kind.TIMESTAMP,
- OrcProto.Stream.Kind.SECONDARY, false, true));
+ (OrcProto.ColumnEncoding.Kind.DIRECT, TypeDescription.Category.TIMESTAMP,
+ OrcProto.Stream.Kind.SECONDARY, uncompressed, true));
}
@Test
public void testPartialPlan() throws Exception {
- DiskRangeList result;
-
- // set the streams
- List<OrcProto.Stream> streams = new ArrayList<OrcProto.Stream>();
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.PRESENT)
- .setColumn(1).setLength(1000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DATA)
- .setColumn(1).setLength(99000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.PRESENT)
- .setColumn(2).setLength(2000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DATA)
- .setColumn(2).setLength(98000).build());
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:int>");
+ MockDataReader dataReader = new MockDataReader(schema)
+ .addStream(1, OrcProto.Stream.Kind.ROW_INDEX,
+ createRowIndex(entry(0, -1, -1, 0),
+ entry(100, -1, -1, 10000),
+ entry(200, -1, -1, 20000),
+ entry(300, -1, -1, 30000),
+ entry(400, -1, -1, 40000),
+ entry(500, -1, -1, 50000)))
+ .addStream(2, OrcProto.Stream.Kind.ROW_INDEX,
+ createRowIndex(entry(0, -1, -1, 0),
+ entry(200, -1, -1, 20000),
+ entry(400, -1, -1, 40000),
+ entry(600, -1, -1, 60000),
+ entry(800, -1, -1, 80000),
+ entry(1000, -1, -1, 100000)))
+ .addStream(1, OrcProto.Stream.Kind.PRESENT, createDataStream(1, 1000))
+ .addStream(1, OrcProto.Stream.Kind.DATA, createDataStream(2, 99000))
+ .addStream(2, OrcProto.Stream.Kind.PRESENT, createDataStream(3, 2000))
+ .addStream(2, OrcProto.Stream.Kind.DATA, createDataStream(4, 198000))
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addStripeFooter(1000, null);
+ MockStripe stripe = dataReader.getStripe(0);
+ // get the start of the data streams
+ final long START = stripe.getStream(1, OrcProto.Stream.Kind.PRESENT).offset;
boolean[] columns = new boolean[]{true, true, false};
boolean[] rowGroups = new boolean[]{true, true, false, false, true, false};
- // set the index
- OrcProto.RowIndex[] indexes = new OrcProto.RowIndex[columns.length];
- indexes[1] = OrcProto.RowIndex.newBuilder()
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(0).addPositions(-1).addPositions(-1)
- .addPositions(0)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(100).addPositions(-1).addPositions(-1)
- .addPositions(10000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(200).addPositions(-1).addPositions(-1)
- .addPositions(20000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(300).addPositions(-1).addPositions(-1)
- .addPositions(30000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(400).addPositions(-1).addPositions(-1)
- .addPositions(40000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(500).addPositions(-1).addPositions(-1)
- .addPositions(50000)
- .build())
- .build();
-
- // set encodings
- List<OrcProto.ColumnEncoding> encodings =
- new ArrayList<OrcProto.ColumnEncoding>();
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
-
- // set types struct{x: int, y: int}
- List<OrcProto.Type> types = new ArrayList<OrcProto.Type>();
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.STRUCT)
- .addSubtypes(1).addSubtypes(2).addFieldNames("x")
- .addFieldNames("y").build());
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.INT).build());
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.INT).build());
-
// filter by rows and groups
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(0, 21000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP,
- 41000, 51000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP)));
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(0, 21000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP,
- 41000, 51000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP)));
+ StripePlanner planner = new StripePlanner(schema, new ReaderEncryption(),
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.ORC_14, false, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, columns);
+ OrcIndex index = planner.readRowIndex(null, null);
+ BufferChunkList result = planner.readData(index, rowGroups, false);
+
+ assertEquals(START, result.get(0).getOffset());
+ assertEquals(1000, result.get(0).getLength());
+ assertEquals(START + 1000, result.get(1).getOffset());
+ assertEquals(20000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP, result.get(1).getLength());
+ assertEquals(START + 41000, result.get(2).getOffset());
+ assertEquals(10000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP, result.get(2).getLength());
+ assertEquals(null, result.get(3));
// if we read no rows, don't read any bytes
rowGroups = new boolean[]{false, false, false, false, false, false};
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertNull(result);
+ result = planner.readData(index, rowGroups, false);
+ assertEquals(null, result.get(0));
// all rows, but only columns 0 and 2.
rowGroups = null;
columns = new boolean[]{true, false, true};
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, null, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(100000, 200000)));
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, null, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(100000, 200000)));
+ planner.parseStripe(stripe, columns).readRowIndex(null, index);
+ result = planner.readData(index, rowGroups, false);
+ assertEquals(START + 100000, result.get(0).getOffset());
+ assertEquals(2000, result.get(0).getLength());
+ assertEquals(START + 102000, result.get(1).getOffset());
+ assertEquals(198000, result.get(1).getLength());
+ assertEquals(null, result.get(2));
rowGroups = new boolean[]{false, true, false, false, false, false};
- indexes[2] = indexes[1];
- indexes[1] = null;
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(100100, 102000,
- 112000, 122000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP)));
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(100100, 102000,
- 112000, 122000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP)));
+ result = planner.readData(index, rowGroups, false);
+ assertEquals(START + 100200, result.get(0).getOffset());
+ assertEquals(1800, result.get(0).getLength());
+ assertEquals(START + 122000, result.get(1).getOffset());
+ assertEquals(20000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP,
+ result.get(1).getLength());
+ assertEquals(null, result.get(2));
rowGroups = new boolean[]{false, false, false, false, false, true};
- indexes[1] = indexes[2];
columns = new boolean[]{true, true, true};
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(500, 1000, 51000, 100000, 100500, 102000,
- 152000, 200000)));
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(500, 1000, 51000, 100000, 100500, 102000,
- 152000, 200000)));
+ planner.parseStripe(stripe, columns).readRowIndex(null, index);
+ result = planner.readData(index, rowGroups, false);
+ assertEquals(START + 500, result.get(0).getOffset());
+ assertEquals(500, result.get(0).getLength());
+ assertEquals(START + 51000, result.get(1).getOffset());
+ assertEquals(49000, result.get(1).getLength());
+ assertEquals(START + 101000, result.get(2).getOffset());
+ assertEquals(1000, result.get(2).getLength());
+ assertEquals(START + 202000, result.get(3).getOffset());
+ assertEquals(98000, result.get(3).getLength());
+ assertEquals(null, result.get(4));
}
@Test
public void testPartialPlanCompressed() throws Exception {
- DiskRangeList result;
-
- // set the streams
- List<OrcProto.Stream> streams = new ArrayList<OrcProto.Stream>();
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.PRESENT)
- .setColumn(1).setLength(1000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DATA)
- .setColumn(1).setLength(99000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.PRESENT)
- .setColumn(2).setLength(2000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DATA)
- .setColumn(2).setLength(98000).build());
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:int>");
+ InStream.StreamOptions options =
+ new InStream.StreamOptions()
+ .withCodec(OrcCodecPool.getCodec(CompressionKind.ZLIB))
+ .withBufferSize(1024);
+ final int SLOP = 2 * (OutStream.HEADER_SIZE + options.getBufferSize());
+ MockDataReader dataReader = new MockDataReader(schema)
+ .addStream(1, OrcProto.Stream.Kind.ROW_INDEX,
+ createRowIndex(options,
+ entry(0, -1, -1, -1, 0),
+ entry(100, -1, -1, -1, 10000),
+ entry(200, -1, -1, -1, 20000),
+ entry(300, -1, -1, -1, 30000),
+ entry(400, -1, -1, -1, 40000),
+ entry(500, -1, -1, -1, 50000)))
+ .addStream(2, OrcProto.Stream.Kind.ROW_INDEX,
+ createRowIndex(options,
+ entry(0, -1, -1, -1, 0),
+ entry(200, -1, -1, -1, 20000),
+ entry(400, -1, -1, -1, 40000),
+ entry(600, -1, -1, -1, 60000),
+ entry(800, -1, -1, -1, 80000),
+ entry(1000, -1, -1, -1, 100000)))
+ .addStream(1, OrcProto.Stream.Kind.PRESENT, createDataStream(1, 1000))
+ .addStream(1, OrcProto.Stream.Kind.DATA, createDataStream(2, 99000))
+ .addStream(2, OrcProto.Stream.Kind.PRESENT, createDataStream(3, 2000))
+ .addStream(2, OrcProto.Stream.Kind.DATA, createDataStream(4, 198000))
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addStripeFooter(1000, null);
+ MockStripe stripe = dataReader.getStripe(0);
+ // get the start of the data streams
+ final long START = stripe.getStream(1, OrcProto.Stream.Kind.PRESENT).offset;
+
+ StripePlanner planner = new StripePlanner(schema, new ReaderEncryption(),
+ options, dataReader, OrcFile.WriterVersion.ORC_14, false,
+ Integer.MAX_VALUE);
+ // filter by rows and groups
boolean[] columns = new boolean[]{true, true, false};
boolean[] rowGroups = new boolean[]{true, true, false, false, true, false};
-
- // set the index
- OrcProto.RowIndex[] indexes = new OrcProto.RowIndex[columns.length];
- indexes[1] = OrcProto.RowIndex.newBuilder()
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(0).addPositions(-1).addPositions(-1).addPositions(-1)
- .addPositions(0)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(100).addPositions(-1).addPositions(-1).addPositions(-1)
- .addPositions(10000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(200).addPositions(-1).addPositions(-1).addPositions(-1)
- .addPositions(20000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(300).addPositions(-1).addPositions(-1).addPositions(-1)
- .addPositions(30000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(400).addPositions(-1).addPositions(-1).addPositions(-1)
- .addPositions(40000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(500).addPositions(-1).addPositions(-1).addPositions(-1)
- .addPositions(50000)
- .build())
- .build();
-
- // set encodings
- List<OrcProto.ColumnEncoding> encodings =
- new ArrayList<OrcProto.ColumnEncoding>();
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
-
- // set types struct{x: int, y: int}
- List<OrcProto.Type> types = new ArrayList<OrcProto.Type>();
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.STRUCT)
- .addSubtypes(1).addSubtypes(2).addFieldNames("x")
- .addFieldNames("y").build());
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.INT).build());
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.INT).build());
-
- // filter by rows and groups
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, true, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(0, 100000)));
+ planner.parseStripe(stripe, columns);
+ OrcIndex index = planner.readRowIndex(null, null);
+ BufferChunkList result = planner.readData(index, rowGroups, false);
+ assertEquals(START, result.get(0).getOffset());
+ assertEquals(1000, result.get(0).getLength());
+ assertEquals(START + 1000, result.get(1).getOffset());
+ assertEquals(20000 + SLOP, result.get(1).getLength());
+ assertEquals(START + 41000, result.get(2).getOffset());
+ assertEquals(10000 + SLOP, result.get(2).getLength());
+ assertEquals(null, result.get(3));
rowGroups = new boolean[]{false, false, false, false, false, true};
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, true, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(500, 1000, 51000, 100000)));
+ result = planner.readData(index, rowGroups, false);
+ assertEquals(START + 500, result.get(0).getOffset());
+ assertEquals(500, result.get(0).getLength());
+ assertEquals(START + 51000, result.get(1).getOffset());
+ assertEquals(49000, result.get(1).getLength());
+ assertEquals(null, result.get(2));
}
@Test
public void testPartialPlanString() throws Exception {
- DiskRangeList result;
-
- // set the streams
- List<OrcProto.Stream> streams = new ArrayList<OrcProto.Stream>();
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.PRESENT)
- .setColumn(1).setLength(1000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DATA)
- .setColumn(1).setLength(94000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.LENGTH)
- .setColumn(1).setLength(2000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DICTIONARY_DATA)
- .setColumn(1).setLength(3000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.PRESENT)
- .setColumn(2).setLength(2000).build());
- streams.add(OrcProto.Stream.newBuilder()
- .setKind(OrcProto.Stream.Kind.DATA)
- .setColumn(2).setLength(98000).build());
-
- boolean[] columns = new boolean[]{true, true, false};
- boolean[] rowGroups = new boolean[]{false, true, false, false, true, true};
-
- // set the index
- OrcProto.RowIndex[] indexes = new OrcProto.RowIndex[columns.length];
- indexes[1] = OrcProto.RowIndex.newBuilder()
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(0).addPositions(-1).addPositions(-1)
- .addPositions(0)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(100).addPositions(-1).addPositions(-1)
- .addPositions(10000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(200).addPositions(-1).addPositions(-1)
- .addPositions(20000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(300).addPositions(-1).addPositions(-1)
- .addPositions(30000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(400).addPositions(-1).addPositions(-1)
- .addPositions(40000)
- .build())
- .addEntry(OrcProto.RowIndexEntry.newBuilder()
- .addPositions(500).addPositions(-1).addPositions(-1)
- .addPositions(50000)
- .build())
- .build();
+ TypeDescription schema = TypeDescription.fromString("struct<x:string,y:int>");
+ MockDataReader dataReader =
+ new MockDataReader(schema)
+ .addStream(1, OrcProto.Stream.Kind.ROW_INDEX,
+ createRowIndex(entry(0, -1, -1, 0),
+ entry(100, -1, -1, 10000),
+ entry(200, -1, -1, 20000),
+ entry(300, -1, -1, 30000),
+ entry(400, -1, -1, 40000),
+ entry(500, -1, -1, 50000)))
+ .addStream(2, OrcProto.Stream.Kind.ROW_INDEX,
+ createRowIndex(entry(0, -1, -1, 0),
+ entry(200, -1, -1, 20000),
+ entry(400, -1, -1, 40000),
+ entry(600, -1, -1, 60000),
+ entry(800, -1, -1, 80000),
+ entry(1000, -1, -1, 100000)))
+ .addStream(1, OrcProto.Stream.Kind.PRESENT, createDataStream(1, 1000))
+ .addStream(1, OrcProto.Stream.Kind.DATA, createDataStream(2, 94000))
+ .addStream(1, OrcProto.Stream.Kind.LENGTH, createDataStream(3, 2000))
+ .addStream(1, OrcProto.Stream.Kind.DICTIONARY_DATA, createDataStream(4, 3000))
+ .addStream(2, OrcProto.Stream.Kind.PRESENT, createDataStream(5, 2000))
+ .addStream(2, OrcProto.Stream.Kind.DATA, createDataStream(6, 198000))
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DICTIONARY)
+ .addEncoding(OrcProto.ColumnEncoding.Kind.DIRECT)
+ .addStripeFooter(1000, null);
+ MockStripe stripe = dataReader.getStripe(0);
+ // get the start of the data streams
+ final long START = stripe.getStream(1, OrcProto.Stream.Kind.PRESENT).offset;
- // set encodings
- List<OrcProto.ColumnEncoding> encodings =
- new ArrayList<OrcProto.ColumnEncoding>();
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DICTIONARY).build());
- encodings.add(OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build());
-
- // set types struct{x: string, y: int}
- List<OrcProto.Type> types = new ArrayList<OrcProto.Type>();
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.STRUCT)
- .addSubtypes(1).addSubtypes(2).addFieldNames("x")
- .addFieldNames("y").build());
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.STRING).build());
- types.add(OrcProto.Type.newBuilder().setKind(OrcProto.Type.Kind.INT).build());
+ // filter by rows and groups
+ StripePlanner planner = new StripePlanner(schema, new ReaderEncryption(),
+ new InStream.StreamOptions(),
+ dataReader, OrcFile.WriterVersion.ORC_14, false, Integer.MAX_VALUE);
// filter by rows and groups
- result = RecordReaderImpl.planReadPartialDataStreams(streams, indexes,
- columns, rowGroups, false, encodings, types, 32768, true);
- assertThat(result, is(diskRanges(100, 1000,
- 11000, 21000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP,
- 41000, 100000)));
+ boolean[] columns = new boolean[]{true, true, false};
+ boolean[] rowGroups = new boolean[]{false, true, false, false, true, true};
+ planner.parseStripe(stripe, columns);
+ OrcIndex index = planner.readRowIndex(null, null);
+ BufferChunkList result = planner.readData(index, rowGroups, false);
+
+ assertEquals(START + 100, result.get(0).getOffset());
+ assertEquals(900, result.get(0).getLength());
+ assertEquals(START + 11000, result.get(1).getOffset());
+ assertEquals(10000 + RecordReaderUtils.WORST_UNCOMPRESSED_SLOP,
+ result.get(1).getLength());
+ assertEquals(START + 41000, result.get(2).getOffset());
+ assertEquals(54000, result.get(2).getLength());
+ assertEquals(START + 95000, result.get(3).getOffset());
+ assertEquals(2000, result.get(3).getLength());
+ assertEquals(START + 97000, result.get(4).getOffset());
+ assertEquals(3000, result.get(4).getLength());
+ assertEquals(null, result.get(5));
+
+ // Don't read anything if no groups are selected
+ rowGroups = new boolean[6];
+ result = planner.readData(index, rowGroups, false);
+ assertEquals(null, result.get(0));
}
@Test
@@ -1895,171 +1831,267 @@ public class TestRecordReaderImpl {
recordReader.close();
}
+ static ByteBuffer createDataStream(int tag, int bytes) {
+ ByteBuffer result = ByteBuffer.allocate(bytes);
+ IntBuffer iBuf = result.asIntBuffer();
+ for(int i= 0; i < bytes; i += 4) {
+ iBuf.put((tag << 24) + i);
+ }
+ result.limit(bytes);
+ return result;
+ }
+
+ static OrcProto.RowIndexEntry entry(int... values) {
+ OrcProto.RowIndexEntry.Builder builder = OrcProto.RowIndexEntry.newBuilder();
+ for(int v: values) {
+ builder.addPositions(v);
+ }
+ return builder.build();
+ }
+
+ static ByteBuffer createRowIndex(InStream.StreamOptions options,
+ OrcProto.RowIndexEntry... entries
+ ) throws IOException {
+ ByteBuffer uncompressed = createRowIndex(entries);
+ if (options.getCodec() != null) {
+ CompressionCodec codec = options.getCodec();
+ PhysicalFsWriter.BufferedStream buffer =
+ new PhysicalFsWriter.BufferedStream();
+ StreamOptions writerOptions = new StreamOptions(options.getBufferSize())
+ .withCodec(codec, codec.getDefaultOptions());
+ OutStream out = new OutStream("row index", writerOptions, buffer);
+ out.write(uncompressed.array(),
+ uncompressed.arrayOffset() + uncompressed.position(),
+ uncompressed.remaining());
+ out.flush();
+ return buffer.getByteBuffer();
+ } else {
+ return uncompressed;
+ }
+ }
+
+ static ByteBuffer createRowIndex(OrcProto.RowIndexEntry... entries) {
+ OrcProto.RowIndex.Builder builder = OrcProto.RowIndex.newBuilder();
+ for(OrcProto.RowIndexEntry entry: entries) {
+ builder.addEntry(entry);
+ }
+ return ByteBuffer.wrap(builder.build().toByteArray());
+ }
+
+ static ByteBuffer createRowIndex(int value) {
+ OrcProto.RowIndexEntry entry =
+ OrcProto.RowIndexEntry.newBuilder().addPositions(value).build();
+ return ByteBuffer.wrap(OrcProto.RowIndex.newBuilder().addEntry(entry)
+ .build().toByteArray());
+ }
+
+ static ByteBuffer createBloomFilter(int value) {
+ OrcProto.BloomFilter entry =
+ OrcProto.BloomFilter.newBuilder().setNumHashFunctions(value).build();
+ return ByteBuffer.wrap(OrcProto.BloomFilterIndex.newBuilder()
+ .addBloomFilter(entry).build().toByteArray());
+ }
+
+ static MockDataReader createOldBlooms(TypeDescription schema) {
+ return new MockDataReader(schema)
+ .addStream(1, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(10))
+ .addStream(1, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(11))
+ .addStream(2, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(20))
+ .addStream(2, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(21))
+ .addStream(3, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(30))
+ .addStream(3, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(31))
+ .addStripeFooter(1000, null);
+ }
+
+ static MockDataReader createMixedBlooms(TypeDescription schema) {
+ return new MockDataReader(schema)
+ .addStream(1, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(10))
+ .addStream(1, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(11))
+ .addStream(2, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(20))
+ .addStream(2, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(21))
+ .addStream(2, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, createBloomFilter(22))
+ .addStream(3, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(30))
+ .addStream(3, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(31))
+ .addStream(3, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, createBloomFilter(32))
+ .addStripeFooter(1000, null);
+ }
+
+ static MockDataReader createNewBlooms(TypeDescription schema) {
+ return new MockDataReader(schema)
+ .addStream(1, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(10))
+ .addStream(1, OrcProto.Stream.Kind.BLOOM_FILTER, createBloomFilter(11))
+ .addStream(2, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(20))
+ .addStream(2, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, createBloomFilter(22))
+ .addStream(3, OrcProto.Stream.Kind.ROW_INDEX, createRowIndex(30))
+ .addStream(3, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, createBloomFilter(32))
+ .addStripeFooter(1000, null);
+ }
+
@Test
public void TestOldBloomFilters() throws Exception {
- OrcProto.StripeFooter footer =
- OrcProto.StripeFooter.newBuilder()
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(1).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(1).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .build();
TypeDescription schema = TypeDescription.fromString("struct<x:int,y:decimal(10,2),z:string>");
- OrcProto.Stream.Kind[] bloomFilterKinds = new OrcProto.Stream.Kind[4];
-
- // normal read
- DiskRangeList ranges = RecordReaderUtils.planIndexReading(schema, footer,
- false, new boolean[]{true, true, false, true},
- new boolean[]{false, true, false, true},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ MockDataReader dataReader = createOldBlooms(schema);
+ MockStripe stripe = dataReader.getStripe(0);
+
+ // use old blooms
+ ReaderEncryption encryption = new ReaderEncryption();
+ StripePlanner planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, false, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, false, true});
+ OrcIndex index =
+ planner.readRowIndex(new boolean[]{false, true, false, true}, null);
+ OrcProto.Stream.Kind[] bloomFilterKinds = index.getBloomFilterKinds();
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[3]);
- assertEquals("range start: 0 end: 2000", ranges.toString());
- assertEquals("range start: 4000 end: 6000", ranges.next.toString());
- assertEquals(null, ranges.next.next);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
// ignore non-utf8 bloom filter
+ dataReader.resetCounts();
Arrays.fill(bloomFilterKinds, null);
- ranges = RecordReaderUtils.planIndexReading(schema, footer,
- true, new boolean[]{true, true, false, true},
- new boolean[]{false, true, false, true},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, true, false});
+ planner.readRowIndex(new boolean[]{false, true, true, false}, index);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
- assertEquals(null, bloomFilterKinds[3]);
- assertEquals("range start: 0 end: 2000", ranges.toString());
- assertEquals("range start: 4000 end: 5000", ranges.next.toString());
- assertEquals(null, ranges.next.next);
+ assertEquals(null, bloomFilterKinds[2]);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
// check that we are handling the post hive-12055 strings correctly
+ dataReader.resetCounts();
Arrays.fill(bloomFilterKinds, null);
- ranges = RecordReaderUtils.planIndexReading(schema, footer,
- true, null, new boolean[]{false, true, true, true},
- OrcFile.WriterVersion.HIVE_12055, bloomFilterKinds);
+ planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_12055, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, true, true});
+ planner.readRowIndex(new boolean[]{false, true, true, true}, index);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
assertEquals(null, bloomFilterKinds[2]);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[3]);
- assertEquals("range start: 0 end: 3000", ranges.toString());
- assertEquals("range start: 4000 end: 6000", ranges.next.toString());
- assertEquals(null, ranges.next.next);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
// ignore non-utf8 bloom filter on decimal
+ dataReader.resetCounts();
Arrays.fill(bloomFilterKinds, null);
- ranges = RecordReaderUtils.planIndexReading(schema, footer,
- true, null,
- new boolean[]{false, false, true, false},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, false, true, false});
+ planner.readRowIndex(new boolean[]{false, false, true, false}, index);
+ assertEquals(null, bloomFilterKinds[1]);
assertEquals(null, bloomFilterKinds[2]);
- assertEquals("range start: 0 end: 1000", ranges.toString());
- assertEquals("range start: 2000 end: 3000", ranges.next.toString());
- assertEquals("range start: 4000 end: 5000", ranges.next.next.toString());
- assertEquals(null, ranges.next.next.next);
+ assertEquals(0, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
}
@Test
public void TestCompatibleBloomFilters() throws Exception {
- OrcProto.StripeFooter footer =
- OrcProto.StripeFooter.newBuilder()
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(1).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(1).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).setLength(1000).build())
- .build();
TypeDescription schema = TypeDescription.fromString("struct<x:int,y:decimal(10,2),z:string>");
- OrcProto.Stream.Kind[] bloomFilterKinds = new OrcProto.Stream.Kind[4];
-
- // normal read
- DiskRangeList ranges = RecordReaderUtils.planIndexReading(schema, footer,
- false, new boolean[]{true, true, false, true},
- new boolean[]{false, true, false, true},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ MockDataReader dataReader = createMixedBlooms(schema);
+ MockStripe stripe = dataReader.getStripe(0);
+
+ // use old bloom filters
+ ReaderEncryption encryption = new ReaderEncryption();
+ StripePlanner planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, false, true});
+ OrcIndex index =
+ planner.readRowIndex(new boolean[]{false, true, false, true}, null);
+
+ OrcProto.Stream.Kind[] bloomFilterKinds = index.getBloomFilterKinds();
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, bloomFilterKinds[3]);
- assertEquals("range start: 0 end: 2000", ranges.toString());
- assertEquals("range start: 5000 end: 6000", ranges.next.toString());
- assertEquals("range start: 7000 end: 8000", ranges.next.next.toString());
- assertEquals(null, ranges.next.next.next);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
- //
+ // ignore non-utf8 bloom filter
Arrays.fill(bloomFilterKinds, null);
- ranges = RecordReaderUtils.planIndexReading(schema, footer,
- true, null,
- new boolean[]{false, true, true, false},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ dataReader.resetCounts();
+ planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, true, false});
+ planner.readRowIndex(new boolean[]{false, true, true, false}, index);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, bloomFilterKinds[2]);
- assertEquals("range start: 0 end: 3000", ranges.toString());
- assertEquals("range start: 4000 end: 6000", ranges.next.toString());
- assertEquals(null, ranges.next.next);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
}
@Test
public void TestNewBloomFilters() throws Exception {
- OrcProto.StripeFooter footer =
- OrcProto.StripeFooter.newBuilder()
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(1).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(1).setKind(OrcProto.Stream.Kind.BLOOM_FILTER).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(2).setKind(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.ROW_INDEX).setLength(1000).build())
- .addStreams(OrcProto.Stream.newBuilder()
- .setColumn(3).setKind(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).setLength(1000).build())
- .build();
TypeDescription schema = TypeDescription.fromString("struct<x:int,y:decimal(10,2),z:string>");
- OrcProto.Stream.Kind[] bloomFilterKinds = new OrcProto.Stream.Kind[4];
-
- // normal read
- DiskRangeList ranges = RecordReaderUtils.planIndexReading(schema, footer,
- false, new boolean[]{true, true, false, true},
- new boolean[]{false, true, false, true},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ MockDataReader dataReader = createNewBlooms(schema);
+ MockStripe stripe = dataReader.getStripe(0);
+
+ // use old bloom filters
+ ReaderEncryption encryption = new ReaderEncryption();
+ StripePlanner planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, false, true});
+ OrcIndex index =
+ planner.readRowIndex(new boolean[]{false, true, false, true}, null);
+
+ OrcProto.Stream.Kind[] bloomFilterKinds = index.getBloomFilterKinds();
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, bloomFilterKinds[3]);
- assertEquals("range start: 0 end: 2000", ranges.toString());
- assertEquals("range start: 4000 end: 6000", ranges.next.toString());
- assertEquals(null, ranges.next.next);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
- //
+ // ignore non-utf8 bloom filter
Arrays.fill(bloomFilterKinds, null);
- ranges = RecordReaderUtils.planIndexReading(schema, footer,
- true, null,
- new boolean[]{false, true, true, false},
- OrcFile.WriterVersion.HIVE_4243,
- bloomFilterKinds);
+ dataReader.resetCounts();
+ planner = new StripePlanner(schema, encryption,
+ new InStream.StreamOptions(), dataReader,
+ OrcFile.WriterVersion.HIVE_4243, true, Integer.MAX_VALUE);
+ planner.parseStripe(stripe, new boolean[]{true, true, true, false});
+ planner.readRowIndex(new boolean[]{false, true, true, false}, index);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER, bloomFilterKinds[1]);
assertEquals(OrcProto.Stream.Kind.BLOOM_FILTER_UTF8, bloomFilterKinds[2]);
- assertEquals("range start: 0 end: 5000", ranges.toString());
- assertEquals(null, ranges.next);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(1, OrcProto.Stream.Kind.BLOOM_FILTER).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(1, stripe.getStream(2, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.ROW_INDEX).readCount);
+ assertEquals(0, stripe.getStream(3, OrcProto.Stream.Kind.BLOOM_FILTER_UTF8).readCount);
}
static OrcProto.RowIndexEntry createIndexEntry(Long min, Long max) {
@@ -2104,7 +2136,8 @@ public class TestRecordReaderImpl {
.setKind(OrcProto.ColumnEncoding.Kind.DIRECT_V2).build());
encodings.add(OrcProto.ColumnEncoding.newBuilder()
.setKind(OrcProto.ColumnEncoding.Kind.DIRECT_V2).build());
- boolean[] rows = applier.pickRowGroups(new ReaderImpl.StripeInformationImpl(stripe, 0, 1, null),
+ boolean[] rows = applier.pickRowGroups(
+ new ReaderImpl.StripeInformationImpl(stripe, 1, -1, null),
indexes, null, encodings, null, false);
assertEquals(4, rows.length);
assertEquals(false, rows[0]);
@@ -2150,7 +2183,8 @@ public class TestRecordReaderImpl {
.setKind(OrcProto.ColumnEncoding.Kind.DIRECT_V2).build());
encodings.add(OrcProto.ColumnEncoding.newBuilder()
.setKind(OrcProto.ColumnEncoding.Kind.DIRECT_V2).build());
- boolean[] rows = applier.pickRowGroups(new ReaderImpl.StripeInformationImpl(stripe, 0, 1, null),
+ boolean[] rows = applier.pickRowGroups(
+ new ReaderImpl.StripeInformationImpl(stripe, 1, -1, null),
indexes, null, encodings, null, false);
assertEquals(3, rows.length);
assertEquals(false, rows[0]);
diff --git a/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java b/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java
index 87478d4..736c719 100644
--- a/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java
+++ b/java/core/src/test/org/apache/orc/impl/TestSchemaEvolution.java
@@ -44,6 +44,8 @@ import org.apache.orc.Reader;
import org.apache.orc.RecordReader;
import org.apache.orc.TypeDescription;
import org.apache.orc.Writer;
+import org.apache.orc.impl.reader.ReaderEncryption;
+import org.apache.orc.impl.reader.StripePlanner;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -1618,19 +1620,19 @@ public class TestSchemaEvolution {
children[2].getClass());
// check to make sure the data is read correctly
- OrcProto.StripeFooter.Builder footer = OrcProto.StripeFooter.newBuilder();
- OrcProto.ColumnEncoding DIRECT =
- OrcProto.ColumnEncoding.newBuilder()
- .setKind(OrcProto.ColumnEncoding.Kind.DIRECT).build();
- footer.addColumns(DIRECT);
- footer.addColumns(DIRECT);
- footer.addColumns(DIRECT);
- Map<StreamName, InStream> streams = new HashMap<>();
- createStream(streams, 1, OrcProto.Stream.Kind.DATA, 7, 1, 0);
- createStream(streams, 2, OrcProto.Stream.Kind.DATA,
- 65, 66, 67, 68, 69, 70, 71, 72, 73, 74);
- createStream(streams, 2, OrcProto.Stream.Kind.LENGTH, 7, 0, 1);
- reader.startStripe(streams, footer.build());
+ MockDataReader dataReader = new MockDataReader(fileType)
+ .addStream(1, OrcProto.Stream.Kind.DATA, createBuffer(7, 1, 0))
+ .addStream(2, OrcProto.Stream.Kind.DATA, createBuffer(65, 66, 67, 68,
+ 69, 70, 71, 72, 73, 74))
+ .addStream(2, OrcProto.Stream.Kind.LENGTH, createBuffer(7, 0, 1))
+ .addStripeFooter(100, null);
+ StripePlanner planner = new StripePlanner(fileType, new ReaderEncryption(),
+ InStream.options(), dataReader,
+ OrcFile.WriterVersion.ORC_14, true, Integer.MAX_VALUE);
+ boolean[] columns = new boolean[]{true, true, true};
+ planner.parseStripe(dataReader.getStripe(0), columns)
+ .readData(null, null, false);
+ reader.startStripe(planner);
VectorizedRowBatch batch = readType.createRowBatch();
reader.nextBatch(batch, 10);
final String EXPECTED = "ABCDEFGHIJ";
diff --git a/java/tools/src/java/org/apache/orc/tools/FileDump.java b/java/tools/src/java/org/apache/orc/tools/FileDump.java
index 3eae30d..297361b 100644
--- a/java/tools/src/java/org/apache/orc/tools/FileDump.java
+++ b/java/tools/src/java/org/apache/orc/tools/FileDump.java
@@ -380,8 +380,7 @@ public final class FileDump {
for (int colIdx : rowIndexCols) {
sargColumns[colIdx] = true;
}
- OrcIndex indices = rows
- .readRowIndex(stripeIx, null, null, null, sargColumns);
+ OrcIndex indices = rows.readRowIndex(stripeIx, null, sargColumns);
for (int col : rowIndexCols) {
StringBuilder buf = new StringBuilder();
String rowIdxString = getFormattedRowIndices(col,