You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by pr...@apache.org on 2015/02/19 22:01:39 UTC
svn commit: r1661003 - in /hive/branches/llap:
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/read...
Author: prasanthj
Date: Thu Feb 19 21:01:39 2015
New Revision: 1661003
URL: http://svn.apache.org/r1661003
Log:
HIVE-9419p6: LLAP: ORC decoding of row-groups (Partial patch6)
Modified:
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/StreamUtils.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BinaryStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BooleanStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ByteStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/CharacterStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DateStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DecimalStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DoubleStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/FloatStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/IntStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/LongStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ShortStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/StringStreamReader.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/TimestampStreamReader.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java Thu Feb 19 21:01:39 2015
@@ -27,6 +27,7 @@ import org.apache.hadoop.hive.llap.Consu
import org.apache.hadoop.hive.llap.io.api.EncodedColumnBatch;
import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch;
import org.apache.hadoop.hive.llap.io.api.orc.OrcBatchKey;
+import org.apache.hadoop.hive.llap.io.decode.orc.stream.StreamUtils;
import org.apache.hadoop.hive.llap.io.decode.orc.stream.readers.BinaryStreamReader;
import org.apache.hadoop.hive.llap.io.decode.orc.stream.readers.BooleanStreamReader;
import org.apache.hadoop.hive.llap.io.decode.orc.stream.readers.ByteStreamReader;
@@ -98,6 +99,7 @@ public class OrcColumnVectorProducer ext
int numCols = batch.columnIxs.length;
RecordReaderImpl.TreeReader[] columnStreams = createTreeReaders(numCols, batch, fileMetadata,
stripeMetadata);
+
for (int i = 0; i < maxBatchesRG; i++) {
if (i == maxBatchesRG - 1) {
batchSize = (int) (nonNullRowCount % VectorizedRowBatch.DEFAULT_SIZE);
@@ -128,7 +130,7 @@ public class OrcColumnVectorProducer ext
for (int i = 0; i < numCols; i++) {
int colIx = batch.columnIxs[i];
int rgIdx = batch.batchKey.rgIx;
- OrcProto.RowIndexEntry rowIndex = stripeMetadata.getRowIndexes()[colIx].getEntry(rgIdx);
+
EncodedColumnBatch.StreamBuffer[] streamBuffers = batch.columnData[i];
OrcProto.Type colType = fileMetadata.getTypes().get(colIx);
// TODO: EncodedColumnBatch is already decompressed, we don't really need to pass codec.
@@ -141,24 +143,40 @@ public class OrcColumnVectorProducer ext
OrcProto.ColumnEncoding columnEncoding = stripeMetadata.getEncodings().get(colIx);
ColumnVector cv = null;
+ OrcProto.RowIndex rowIndex = stripeMetadata.getRowIndexes()[colIx];
+ OrcProto.RowIndexEntry rowIndexEntry = rowIndex.getEntry(rgIdx);
+
EncodedColumnBatch.StreamBuffer present = null;
EncodedColumnBatch.StreamBuffer data = null;
EncodedColumnBatch.StreamBuffer dictionary = null;
EncodedColumnBatch.StreamBuffer lengths = null;
EncodedColumnBatch.StreamBuffer secondary = null;
+
+ int presentCBIdx = -1;
+ int dataCBIdx = -1;
+ int lengthsCBIdx = -1;
+ int secondaryCBIdx = -1;
+
for (EncodedColumnBatch.StreamBuffer streamBuffer : streamBuffers) {
+ int[] cbIndices = StreamUtils.getCompressionBufferIndex(rgIdx, rowIndex, columnEncoding,
+ colType, OrcProto.Stream.Kind.valueOf(streamBuffer.streamKind),
+ present != null, codec != null);
+
switch(streamBuffer.streamKind) {
case 0:
// PRESENT stream
present = streamBuffer;
+ presentCBIdx = cbIndices[0];
break;
case 1:
// DATA stream
data = streamBuffer;
+ dataCBIdx = cbIndices[0];
break;
case 2:
// LENGTH stream
lengths = streamBuffer;
+ lengthsCBIdx = cbIndices[0];
break;
case 3:
// DICTIONARY_DATA stream
@@ -167,6 +185,7 @@ public class OrcColumnVectorProducer ext
case 5:
// SECONDARY stream
secondary = streamBuffer;
+ secondaryCBIdx = cbIndices[0];
break;
default:
throw new IOException("Unexpected stream kind: " + streamBuffer.streamKind);
@@ -179,11 +198,14 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setLengthStream(lengths)
+ .setLengthCompressionBufferIndex(lengthsCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
@@ -192,10 +214,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.build();
break;
case BYTE:
@@ -203,10 +227,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.build();
break;
case SHORT:
@@ -214,10 +240,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
@@ -226,10 +254,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
@@ -238,10 +268,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.skipCorrupt(skipCorrupt)
.build();
@@ -251,10 +283,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.build();
break;
case DOUBLE:
@@ -262,10 +296,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.build();
break;
case CHAR:
@@ -276,12 +312,15 @@ public class OrcColumnVectorProducer ext
.setMaxLength(colType.getMaximumLength())
.setCharacterType(colType)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setLengthStream(lengths)
+ .setLengthCompressionBufferIndex(lengthsCBIdx)
.setDictionaryStream(dictionary)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
@@ -290,12 +329,15 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setLengthStream(lengths)
+ .setLengthCompressionBufferIndex(lengthsCBIdx)
.setDictionaryStream(dictionary)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
@@ -306,11 +348,14 @@ public class OrcColumnVectorProducer ext
.setPrecision(colType.getPrecision())
.setScale(colType.getScale())
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setValueStream(data)
+ .setValueCompressionBufferIndex(dataCBIdx)
.setScaleStream(secondary)
+ .setScaleCompressionBufferIndex(secondaryCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
@@ -319,11 +364,14 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setSecondsStream(data)
+ .setSecondsCompressionBufferIndex(dataCBIdx)
.setNanosStream(secondary)
+ .setNanosCompressionBufferIndex(secondaryCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.skipCorrupt(skipCorrupt)
.build();
@@ -333,10 +381,12 @@ public class OrcColumnVectorProducer ext
.setFileName(file)
.setColumnIndex(colIx)
.setPresentStream(present)
+ .setPresentCompressionBufferIndex(presentCBIdx)
.setDataStream(data)
+ .setDataCompressionBufferIndex(dataCBIdx)
.setCompressionCodec(codec)
.setBufferSize(bufferSize)
- .setRowIndex(rowIndex)
+ .setRowIndex(rowIndexEntry)
.setColumnEncoding(columnEncoding)
.build();
break;
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/StreamUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/StreamUtils.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/StreamUtils.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/StreamUtils.java Thu Feb 19 21:01:39 2015
@@ -19,6 +19,7 @@ package org.apache.hadoop.hive.llap.io.d
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import org.apache.hadoop.hive.common.DiskRange;
@@ -29,6 +30,7 @@ import org.apache.hadoop.hive.ql.io.orc.
import org.apache.hadoop.hive.ql.io.orc.OrcProto;
import org.apache.hadoop.hive.ql.io.orc.PositionProvider;
import org.apache.hadoop.hive.ql.io.orc.RecordReaderImpl;
+import org.apache.hadoop.hive.ql.io.orc.RecordReaderUtils;
/**
*
@@ -36,7 +38,39 @@ import org.apache.hadoop.hive.ql.io.orc.
public class StreamUtils {
public static InStream createInStream(String streamName, String fileName, CompressionCodec codec,
+ int bufferSize, EncodedColumnBatch.StreamBuffer streamBuffer, int compressionBufferIdx)
+ throws IOException {
+ if (streamBuffer == null) {
+ return null;
+ }
+
+ int numBuffers = streamBuffer.cacheBuffers.size();
+ List<DiskRange> input = new ArrayList<>(numBuffers);
+ int totalLength = 0;
+
+ // add the stream buffer (uncompressed) corresponding to the specified compression buffer index
+ LlapMemoryBuffer lmb = streamBuffer.cacheBuffers.get(compressionBufferIdx);
+ input.add(new RecordReaderImpl.CacheChunk(lmb, lmb.byteBuffer.position(),
+ lmb.byteBuffer.limit()));
+ totalLength += lmb.byteBuffer.remaining();
+
+ // also add the next stream buffer as row group groups may span compression buffers
+ if (compressionBufferIdx != streamBuffer.cacheBuffers.size() - 1) {
+ lmb = streamBuffer.cacheBuffers.get(compressionBufferIdx + 1);
+ input.add(new RecordReaderImpl.CacheChunk(lmb, lmb.byteBuffer.position(),
+ lmb.byteBuffer.limit()));
+ totalLength += lmb.byteBuffer.remaining();
+ }
+
+ return InStream.create(fileName, streamName, input, totalLength, codec, bufferSize, null);
+ }
+
+ public static InStream createInStream(String streamName, String fileName, CompressionCodec codec,
int bufferSize, EncodedColumnBatch.StreamBuffer streamBuffer) throws IOException {
+ if (streamBuffer == null) {
+ return null;
+ }
+
int numBuffers = streamBuffer.cacheBuffers.size();
List<DiskRange> input = new ArrayList<>(numBuffers);
int totalLength = 0;
@@ -48,9 +82,104 @@ public class StreamUtils {
return InStream.create(fileName, streamName, input, totalLength, codec, bufferSize, null);
}
-
public static PositionProvider getPositionProvider(OrcProto.RowIndexEntry rowIndex) {
PositionProvider positionProvider = new RecordReaderImpl.PositionProviderImpl(rowIndex);
return positionProvider;
}
+
+ /**
+ * Returns compression buffer index within stream for the specified row group index. In other
+ * words, it tells which compression buffer the specified row group belong. If row group spans
+ * compression buffers then index of both compression buffers are return. If row group is in last
+ * compression buffer then it never return two indices. For example:
+ * |---------------CB0-------------|----------------CB1---------------|------------CB2-----------|
+ * |--RG0--|--RG1--|--RG2--|--RG3--|--RG4--|--RG5--|--RG6--|--RG7--|--RG8--|--RG9--|--RG10--|RG11|
+ *
+ * Input: RG1 Output: [0, -1]
+ * Input: RG3 Output: [0, -1]
+ * Input: RG8 Output: [1, 2]
+ * Input: RG11 Output: [2, -1]
+ *
+ * @param rgIdx - row group index
+ * @param rowIndex - row index entries
+ * @param columnEncoding - column encoding
+ * @param colType - column type
+ * @param streamKind - stream kind
+ * @param hasNull - if present stream is present
+ * @param isCompressed - if compressed
+ * @return return int array with 2 elements. Both values will be non-negative only when row group
+ * spans compression buffer.
+ */
+ public static int[] getCompressionBufferIndex(int rgIdx, OrcProto.RowIndex rowIndex,
+ OrcProto.ColumnEncoding columnEncoding, OrcProto.Type colType,
+ OrcProto.Stream.Kind streamKind, boolean hasNull, boolean isCompressed) {
+ int[] result = new int[2];
+ Arrays.fill(result, -1);
+
+ // get total number of row group entries and see if the specified rgIdx is the last entry
+ int numRgs = rowIndex.getEntryCount();
+ int nextRgIdx = rgIdx == numRgs - 1 ? rgIdx : rgIdx + 1;
+ OrcProto.RowIndexEntry rowIndexEntry = rowIndex.getEntry(rgIdx);
+ OrcProto.RowIndexEntry nextRowIndexEntry = rowIndex.getEntry(nextRgIdx);
+
+ // get the start index in position list for the specified stream
+ int indexIx = RecordReaderUtils.getIndexPosition(columnEncoding.getKind(),
+ colType.getKind(), streamKind, isCompressed, hasNull);
+
+ // start offset of compression buffer corresponding to current row index
+ long cbStartOffset = rowIndexEntry.getPositions(indexIx);
+
+ // start offset of compression buffer corresponding to next row index
+ long nextCbStartOffset = nextRowIndexEntry.getPositions(indexIx);
+
+ // start offset of compression buffer corresponding to last row index
+ long lastCbStartOffset = rowIndex.getEntry(numRgs - 1).getPositions(indexIx);
+
+ boolean lastCompressionBuffer = false;
+ int cbIdx = 0;
+ long previosStartOffset = -1;
+ boolean foundCompressionBufferIdx = false;
+
+ for (OrcProto.RowIndexEntry entry : rowIndex.getEntryList()) {
+ // initialze previous compression buffer start offset
+ if (previosStartOffset == -1) {
+ previosStartOffset = entry.getPositions(indexIx);
+ }
+
+ // get start offset of current row group index
+ long currentStartOffset = entry.getPositions(indexIx);
+
+ // if the start offset of compression buffer corresponding to current row group index does
+ // not match with the one specified then move on to next compression buffer
+ if (currentStartOffset != previosStartOffset) {
+ cbIdx++;
+ previosStartOffset = currentStartOffset;
+ }
+
+ // if it matches, assign the compression buffer index within the stream to result
+ if (currentStartOffset == cbStartOffset) {
+ result[0] = cbIdx;
+ foundCompressionBufferIdx = true;
+ }
+
+ // check if the current compression buffer is the last one in the stream
+ if (currentStartOffset == lastCbStartOffset) {
+ lastCompressionBuffer = true;
+ }
+
+ // if the current compression buffer offset does not match the next compression buffer offset
+ // then it means the current row group is the last one in the compression buffer. In this
+ // case, the current row group may span compression buffers, so include the next compression
+ // buffer as well
+ if ((cbStartOffset != nextCbStartOffset) && !lastCompressionBuffer) {
+ result[1] = cbIdx + 1;
+ }
+
+ // we are done finding the compression buffer index
+ if (foundCompressionBufferIdx) {
+ break;
+ }
+ }
+ return result;
+ }
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BinaryStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BinaryStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BinaryStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BinaryStreamReader.java Thu Feb 19 21:01:39 2015
@@ -73,6 +73,9 @@ public class BinaryStreamReader extends
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int dataCBIdx;
+ private int lengthCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -119,29 +122,34 @@ public class BinaryStreamReader extends
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setLengthCompressionBufferIndex(int lengthsCBIdx) {
+ this.lengthCBIdx = lengthsCBIdx;
+ return this;
+ }
+
public BinaryStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
-
- InStream length = null;
- if (lengthStream != null) {
- length = StreamUtils
- .createInStream(OrcProto.Stream.Kind.LENGTH.name(), fileName, null, bufferSize,
- lengthStream);
- }
- return new BinaryStreamReader(columnIndex, present, data, length,
- compressionCodec != null, columnEncoding, rowIndex);
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
+
+ InStream length = StreamUtils.createInStream(OrcProto.Stream.Kind.LENGTH.name(), fileName,
+ null, bufferSize, lengthStream, lengthCBIdx);
+
+ boolean isFileCompressed = compressionCodec != null;
+ return new BinaryStreamReader(columnIndex, present, data, length, isFileCompressed,
+ columnEncoding, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BooleanStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BooleanStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BooleanStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/BooleanStreamReader.java Thu Feb 19 21:01:39 2015
@@ -66,6 +66,8 @@ public class BooleanStreamReader extends
private CompressionCodec compressionCodec;
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -102,23 +104,25 @@ public class BooleanStreamReader extends
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public BooleanStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new BooleanStreamReader(columnIndex, present, data,
- compressionCodec != null, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new BooleanStreamReader(columnIndex, present, data, isFileCompressed, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ByteStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ByteStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ByteStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ByteStreamReader.java Thu Feb 19 21:01:39 2015
@@ -66,6 +66,8 @@ public class ByteStreamReader extends Re
private CompressionCodec compressionCodec;
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -102,22 +104,25 @@ public class ByteStreamReader extends Re
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public ByteStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new ByteStreamReader(columnIndex, present, data, compressionCodec != null, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new ByteStreamReader(columnIndex, present, data, isFileCompressed, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/CharacterStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/CharacterStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/CharacterStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/CharacterStreamReader.java Thu Feb 19 21:01:39 2015
@@ -98,6 +98,9 @@ public class CharacterStreamReader exten
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int dataCBIdx;
+ private int lengthCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -159,36 +162,37 @@ public class CharacterStreamReader exten
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setLengthCompressionBufferIndex(int lengthsCBIdx) {
+ this.lengthCBIdx = lengthsCBIdx;
+ return this;
+ }
+
public CharacterStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
-
- InStream length = null;
- if (lengthStream != null) {
- length = StreamUtils
- .createInStream(OrcProto.Stream.Kind.LENGTH.name(), fileName, null, bufferSize,
- lengthStream);
- }
-
- InStream dictionary = null;
- if (dictionaryStream != null) {
- dictionary = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DICTIONARY_DATA.name(), fileName, null, bufferSize,
- dictionaryStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
+
+ InStream length = StreamUtils.createInStream(OrcProto.Stream.Kind.LENGTH.name(), fileName,
+ null, bufferSize, lengthStream, lengthCBIdx);
+
+ InStream dictionary = StreamUtils.createInStream(OrcProto.Stream.Kind.DICTIONARY_DATA.name(),
+ fileName, null, bufferSize, dictionaryStream);
+
+ boolean isFileCompressed = compressionCodec != null;
return new CharacterStreamReader(columnIndex, maxLength, charType, present, data, length,
- dictionary, compressionCodec != null, columnEncoding, rowIndex);
+ dictionary, isFileCompressed, columnEncoding, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DateStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DateStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DateStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DateStreamReader.java Thu Feb 19 21:01:39 2015
@@ -68,6 +68,8 @@ public class DateStreamReader extends Re
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -109,23 +111,27 @@ public class DateStreamReader extends Re
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public DateStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, presentCBIdx);
- return new DateStreamReader(columnIndex, present, data,
- compressionCodec != null, columnEncoding, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new DateStreamReader(columnIndex, present, data, isFileCompressed,
+ columnEncoding, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DecimalStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DecimalStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DecimalStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DecimalStreamReader.java Thu Feb 19 21:01:39 2015
@@ -75,6 +75,9 @@ public class DecimalStreamReader extends
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int valueCBIdx;
+ private int scaleCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -131,30 +134,34 @@ public class DecimalStreamReader extends
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setValueCompressionBufferIndex(int valueCBIdx) {
+ this.valueCBIdx = valueCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setScaleCompressionBufferIndex(int scaleCBIdx) {
+ this.scaleCBIdx = scaleCBIdx;
+ return this;
+ }
+
public DecimalStreamReader build() throws IOException {
- InStream presentInStream = null;
- if (presentStream != null) {
- presentInStream = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream valueInStream = null;
- if (valueStream != null) {
- valueInStream = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- valueStream);
- }
-
- InStream scaleInStream = null;
- if (scaleStream != null) {
- scaleInStream = StreamUtils
- .createInStream(OrcProto.Stream.Kind.SECONDARY.name(), fileName, null, bufferSize,
- scaleStream);
- }
+ InStream presentInStream = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(),
+ fileName, null, bufferSize, presentStream, presentCBIdx);
+
+ InStream valueInStream = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(),
+ fileName, null, bufferSize, valueStream, valueCBIdx);
+
+ InStream scaleInStream = StreamUtils.createInStream(OrcProto.Stream.Kind.SECONDARY.name(),
+ fileName, null, bufferSize, scaleStream, scaleCBIdx);
+ boolean isFileCompressed = compressionCodec != null;
return new DecimalStreamReader(columnIndex, precision, scale, presentInStream, valueInStream,
- scaleInStream, compressionCodec != null, rowIndex, columnEncoding);
+ scaleInStream, isFileCompressed, rowIndex, columnEncoding);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DoubleStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DoubleStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DoubleStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/DoubleStreamReader.java Thu Feb 19 21:01:39 2015
@@ -66,6 +66,8 @@ public class DoubleStreamReader extends
private CompressionCodec compressionCodec;
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -102,22 +104,25 @@ public class DoubleStreamReader extends
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public DoubleStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new DoubleStreamReader(columnIndex, present, data, compressionCodec != null, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new DoubleStreamReader(columnIndex, present, data, isFileCompressed, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/FloatStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/FloatStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/FloatStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/FloatStreamReader.java Thu Feb 19 21:01:39 2015
@@ -66,6 +66,8 @@ public class FloatStreamReader extends R
private CompressionCodec compressionCodec;
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -102,22 +104,25 @@ public class FloatStreamReader extends R
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public FloatStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new FloatStreamReader(columnIndex, present, data, compressionCodec != null, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new FloatStreamReader(columnIndex, present, data, isFileCompressed, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/IntStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/IntStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/IntStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/IntStreamReader.java Thu Feb 19 21:01:39 2015
@@ -68,6 +68,8 @@ public class IntStreamReader extends Rec
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -109,23 +111,26 @@ public class IntStreamReader extends Rec
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public IntStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new IntStreamReader(columnIndex, present, data,
- compressionCodec != null, columnEncoding, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new IntStreamReader(columnIndex, present, data, isFileCompressed,
+ columnEncoding, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/LongStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/LongStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/LongStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/LongStreamReader.java Thu Feb 19 21:01:39 2015
@@ -69,6 +69,8 @@ public class LongStreamReader extends Re
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
private boolean skipCorrupt;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -115,23 +117,26 @@ public class LongStreamReader extends Re
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public LongStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new LongStreamReader(columnIndex, present, data,
- compressionCodec != null, columnEncoding, skipCorrupt, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new LongStreamReader(columnIndex, present, data, isFileCompressed,
+ columnEncoding, skipCorrupt, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ShortStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ShortStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ShortStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/ShortStreamReader.java Thu Feb 19 21:01:39 2015
@@ -68,6 +68,8 @@ public class ShortStreamReader extends R
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int dataCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -109,23 +111,26 @@ public class ShortStreamReader extends R
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
public ShortStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
- return new ShortStreamReader(columnIndex, present, data,
- compressionCodec != null, columnEncoding, rowIndex);
+ boolean isFileCompressed = compressionCodec != null;
+ return new ShortStreamReader(columnIndex, present, data, isFileCompressed,
+ columnEncoding, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/StringStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/StringStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/StringStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/StringStreamReader.java Thu Feb 19 21:01:39 2015
@@ -85,6 +85,9 @@ public class StringStreamReader extends
private int bufferSize;
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
+ private int presentCBIdx;
+ private int dataCBIdx;
+ private int lengthCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -136,36 +139,37 @@ public class StringStreamReader extends
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setDataCompressionBufferIndex(int dataCBIdx) {
+ this.dataCBIdx = dataCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setLengthCompressionBufferIndex(int lengthsCBIdx) {
+ this.lengthCBIdx = lengthsCBIdx;
+ return this;
+ }
+
public StringStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
-
- InStream length = null;
- if (lengthStream != null) {
- length = StreamUtils
- .createInStream(OrcProto.Stream.Kind.LENGTH.name(), fileName, null, bufferSize,
- lengthStream);
- }
-
- InStream dictionary = null;
- if (dictionaryStream != null) {
- dictionary = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DICTIONARY_DATA.name(), fileName, null, bufferSize,
- dictionaryStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, dataCBIdx);
+
+ InStream length = StreamUtils.createInStream(OrcProto.Stream.Kind.LENGTH.name(), fileName,
+ null, bufferSize, lengthStream, lengthCBIdx);
+
+ InStream dictionary = StreamUtils.createInStream(OrcProto.Stream.Kind.DICTIONARY_DATA.name(),
+ fileName, null, bufferSize, dictionaryStream);
+
+ boolean isFileCompressed = compressionCodec != null;
return new StringStreamReader(columnIndex, present, data, length, dictionary,
- compressionCodec != null, columnEncoding, rowIndex);
+ isFileCompressed, columnEncoding, rowIndex);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/TimestampStreamReader.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/TimestampStreamReader.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/TimestampStreamReader.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/orc/stream/readers/TimestampStreamReader.java Thu Feb 19 21:01:39 2015
@@ -75,6 +75,9 @@ public class TimestampStreamReader exten
private OrcProto.RowIndexEntry rowIndex;
private OrcProto.ColumnEncoding columnEncoding;
private boolean skipCorrupt;
+ private int presentCBIdx;
+ private int secondsCBIdx;
+ private int nanosCBIdx;
public StreamReaderBuilder setFileName(String fileName) {
this.fileName = fileName;
@@ -126,29 +129,34 @@ public class TimestampStreamReader exten
return this;
}
+ public StreamReaderBuilder setPresentCompressionBufferIndex(int presentCBIdx) {
+ this.presentCBIdx = presentCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setSecondsCompressionBufferIndex(int secondsCBIdx) {
+ this.secondsCBIdx = secondsCBIdx;
+ return this;
+ }
+
+ public StreamReaderBuilder setNanosCompressionBufferIndex(int nanosCBIdx) {
+ this.nanosCBIdx = nanosCBIdx;
+ return this;
+ }
+
public TimestampStreamReader build() throws IOException {
- InStream present = null;
- if (presentStream != null) {
- present = StreamUtils
- .createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName, null, bufferSize,
- presentStream);
- }
-
- InStream data = null;
- if (dataStream != null) {
- data = StreamUtils
- .createInStream(OrcProto.Stream.Kind.DATA.name(), fileName, null, bufferSize,
- dataStream);
- }
-
- InStream nanos = null;
- if (nanosStream != null) {
- nanos = StreamUtils
- .createInStream(OrcProto.Stream.Kind.SECONDARY.name(), fileName, null, bufferSize,
- nanosStream);
- }
+ InStream present = StreamUtils.createInStream(OrcProto.Stream.Kind.PRESENT.name(), fileName,
+ null, bufferSize, presentStream, presentCBIdx);
+
+ InStream data = StreamUtils.createInStream(OrcProto.Stream.Kind.DATA.name(), fileName,
+ null, bufferSize, dataStream, secondsCBIdx);
+
+ InStream nanos = StreamUtils.createInStream(OrcProto.Stream.Kind.SECONDARY.name(), fileName,
+ null, bufferSize, nanosStream, nanosCBIdx);
+
+ boolean isFileCompressed = compressionCodec != null;
return new TimestampStreamReader(columnIndex, present, data, nanos,
- compressionCodec != null, columnEncoding, skipCorrupt, rowIndex);
+ isFileCompressed, columnEncoding, skipCorrupt, rowIndex);
}
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java?rev=1661003&r1=1661002&r2=1661003&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java Thu Feb 19 21:01:39 2015
@@ -41,7 +41,7 @@ import com.google.common.collect.Compari
/**
* Stateless methods shared between RecordReaderImpl and EncodedReaderImpl.
*/
-class RecordReaderUtils {
+public class RecordReaderUtils {
static boolean[] findPresentStreamsByColumn(
List<OrcProto.Stream> streamList, List<OrcProto.Type> types) {
boolean[] hasNull = new boolean[types.size()];