You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2016/11/22 02:46:28 UTC
[19/35] hive git commit: HIVE-14089 : complex type support in LLAP IO
is broken (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
HIVE-14089 : complex type support in LLAP IO is broken (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/6efa869f
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/6efa869f
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/6efa869f
Branch: refs/heads/hive-14535
Commit: 6efa869f71ec1283dc3b4d09562f24a8bdcb0aa3
Parents: aebc906
Author: Sergey Shelukhin <se...@apache.org>
Authored: Fri Nov 18 10:55:48 2016 -0800
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Fri Nov 18 11:06:49 2016 -0800
----------------------------------------------------------------------
.../hive/llap/io/api/impl/LlapInputFormat.java | 5 -
.../llap/io/decode/OrcEncodedDataConsumer.java | 197 ++--
.../llap/io/encoded/OrcEncodedDataReader.java | 89 +-
.../hive/llap/io/metadata/OrcFileMetadata.java | 6 +
.../orc/impl/ConvertTreeReaderFactory.java | 2 +-
.../org/apache/orc/impl/TreeReaderFactory.java | 90 +-
.../apache/hadoop/hive/ql/exec/Utilities.java | 6 +-
.../hadoop/hive/ql/io/orc/OrcInputFormat.java | 2 +-
.../hive/ql/io/orc/encoded/EncodedReader.java | 2 +-
.../ql/io/orc/encoded/EncodedReaderImpl.java | 80 +-
.../orc/encoded/EncodedTreeReaderFactory.java | 985 +++++++++++++++----
.../hadoop/hive/ql/io/orc/encoded/Reader.java | 6 +-
.../org/apache/hadoop/hive/ql/plan/MapWork.java | 41 +-
.../queries/clientpositive/vector_complex_all.q | 52 +-
.../llap/vector_complex_all.q.out | 315 +++---
.../llap/vector_complex_join.q.out | 4 +-
.../clientpositive/vector_complex_all.q.out | 302 +++---
.../common/io/encoded/EncodedColumnBatch.java | 52 +-
18 files changed, 1464 insertions(+), 772 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapInputFormat.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapInputFormat.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapInputFormat.java
index a27266f..e803125 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapInputFormat.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapInputFormat.java
@@ -115,11 +115,6 @@ public class LlapInputFormat implements InputFormat<NullWritable, VectorizedRowB
}
boolean isVectorized = Utilities.getUseVectorizedInputFileFormat(job);
- // validate for supported types. Until we fix HIVE-14089 we need this check.
- if (useLlapIo) {
- useLlapIo = Utilities.checkLlapIOSupportedTypes(job);
- }
-
if (!useLlapIo) {
LlapIoImpl.LOG.warn("Not using LLAP IO for an unsupported split: " + split);
return sourceInputFormat.getRecordReader(split, job, reporter);
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java
index 2fa68a9..8bd985c 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java
@@ -6,9 +6,7 @@
* 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
- *
+ * 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.
@@ -18,6 +16,7 @@
package org.apache.hadoop.hive.llap.io.decode;
import java.io.IOException;
+import java.util.Arrays;
import java.util.List;
import org.apache.hadoop.hive.common.io.encoded.EncodedColumnBatch;
@@ -40,6 +39,8 @@ import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.UnionColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.orc.CompressionCodec;
+import org.apache.orc.impl.PositionProvider;
+import org.apache.orc.OrcProto.RowIndexEntry;
import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer;
import org.apache.hadoop.hive.ql.io.orc.encoded.EncodedTreeReaderFactory;
import org.apache.hadoop.hive.ql.io.orc.encoded.EncodedTreeReaderFactory.SettableTreeReader;
@@ -49,12 +50,19 @@ import org.apache.hadoop.hive.ql.io.orc.RecordReaderImpl;
import org.apache.orc.OrcUtils;
import org.apache.orc.TypeDescription;
import org.apache.orc.impl.TreeReaderFactory;
+import org.apache.orc.impl.TreeReaderFactory.StructTreeReader;
+import org.apache.orc.impl.TreeReaderFactory.TreeReader;
import org.apache.hadoop.hive.ql.io.orc.WriterImpl;
import org.apache.orc.OrcProto;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
public class OrcEncodedDataConsumer
extends EncodedDataConsumer<OrcBatchKey, OrcEncodedColumnBatch> {
+ public static final Logger LOG = LoggerFactory.getLogger(OrcEncodedDataConsumer.class);
private TreeReaderFactory.TreeReader[] columnReaders;
+ private int[] columnMapping; // Mapping from columnReaders (by index) to columns in file schema.
private int previousStripeIndex = -1;
private OrcFileMetadata fileMetadata; // We assume one request is only for one file.
private CompressionCodec codec;
@@ -84,58 +92,6 @@ public class OrcEncodedDataConsumer
stripes[m.getStripeIx()] = m;
}
- private static ColumnVector createColumn(List<OrcProto.Type> types,
- final int columnId, int batchSize) {
- OrcProto.Type type = types.get(columnId);
- switch (type.getKind()) {
- case BOOLEAN:
- case BYTE:
- case SHORT:
- case INT:
- case LONG:
- case DATE:
- return new LongColumnVector(batchSize);
- case FLOAT:
- case DOUBLE:
- return new DoubleColumnVector(batchSize);
- case BINARY:
- case STRING:
- case CHAR:
- case VARCHAR:
- return new BytesColumnVector(batchSize);
- case TIMESTAMP:
- return new TimestampColumnVector(batchSize);
- case DECIMAL:
- return new DecimalColumnVector(batchSize, type.getPrecision(),
- type.getScale());
- case STRUCT: {
- List<Integer> subtypeIdxs = type.getSubtypesList();
- ColumnVector[] fieldVector = new ColumnVector[subtypeIdxs.size()];
- for(int i=0; i < fieldVector.length; ++i) {
- fieldVector[i] = createColumn(types, subtypeIdxs.get(i), batchSize);
- }
- return new StructColumnVector(batchSize, fieldVector);
- }
- case UNION: {
- List<Integer> subtypeIdxs = type.getSubtypesList();
- ColumnVector[] fieldVector = new ColumnVector[subtypeIdxs.size()];
- for(int i=0; i < fieldVector.length; ++i) {
- fieldVector[i] = createColumn(types, subtypeIdxs.get(i), batchSize);
- }
- return new UnionColumnVector(batchSize, fieldVector);
- }
- case LIST:
- return new ListColumnVector(batchSize, createColumn(types, type.getSubtypes(0), batchSize));
- case MAP:
- return new MapColumnVector(batchSize,
- createColumn(types, type.getSubtypes(0), batchSize),
- createColumn(types, type.getSubtypes(1), batchSize));
- default:
- throw new IllegalArgumentException("LLAP does not support " +
- type.getKind());
- }
- }
-
@Override
protected void decodeBatch(OrcEncodedColumnBatch batch,
Consumer<ColumnVectorBatch> downstreamConsumer) {
@@ -157,14 +113,18 @@ public class OrcEncodedDataConsumer
}
int maxBatchesRG = (int) ((nonNullRowCount / VectorizedRowBatch.DEFAULT_SIZE) + 1);
int batchSize = VectorizedRowBatch.DEFAULT_SIZE;
- int numCols = batch.getColumnIxs().length;
+ TypeDescription schema = fileMetadata.getSchema();
+
if (columnReaders == null || !sameStripe) {
- this.columnReaders = EncodedTreeReaderFactory.createEncodedTreeReader(numCols,
- fileMetadata.getTypes(), stripeMetadata.getEncodings(), batch, codec, skipCorrupt,
- stripeMetadata.getWriterTimezone());
- positionInStreams(columnReaders, batch, numCols, stripeMetadata);
+ int[] columnMapping = new int[schema.getChildren().size()];
+ StructTreeReader treeReader = EncodedTreeReaderFactory.createRootTreeReader(
+ schema, stripeMetadata.getEncodings(), batch, codec, skipCorrupt,
+ stripeMetadata.getWriterTimezone(), columnMapping);
+ this.columnReaders = treeReader.getChildReaders();
+ this.columnMapping = Arrays.copyOf(columnMapping, columnReaders.length);
+ positionInStreams(columnReaders, batch, stripeMetadata);
} else {
- repositionInStreams(this.columnReaders, batch, sameStripe, numCols, stripeMetadata);
+ repositionInStreams(this.columnReaders, batch, sameStripe, stripeMetadata);
}
previousStripeIndex = currentStripeIndex;
@@ -176,18 +136,17 @@ public class OrcEncodedDataConsumer
}
ColumnVectorBatch cvb = cvbPool.take();
- assert cvb.cols.length == batch.getColumnIxs().length; // Must be constant per split.
+ // assert cvb.cols.length == batch.getColumnIxs().length; // Must be constant per split.
cvb.size = batchSize;
- List<OrcProto.Type> types = fileMetadata.getTypes();
- int[] columnMapping = batch.getColumnIxs();
- for (int idx = 0; idx < batch.getColumnIxs().length; idx++) {
+ for (int idx = 0; idx < columnReaders.length; ++idx) {
+ TreeReader reader = columnReaders[idx];
if (cvb.cols[idx] == null) {
// Orc store rows inside a root struct (hive writes it this way).
// When we populate column vectors we skip over the root struct.
- cvb.cols[idx] = createColumn(types, columnMapping[idx], batchSize);
+ cvb.cols[idx] = createColumn(schema.getChildren().get(columnMapping[idx]), batchSize);
}
cvb.cols[idx].ensureSize(batchSize, false);
- columnReaders[idx].nextVector(cvb.cols[idx], null, batchSize);
+ reader.nextVector(cvb.cols[idx], null, batchSize);
}
// we are done reading a batch, send it to consumer for processing
@@ -203,36 +162,98 @@ public class OrcEncodedDataConsumer
}
}
+ private ColumnVector createColumn(TypeDescription type, int batchSize) {
+ switch (type.getCategory()) {
+ case BOOLEAN:
+ case BYTE:
+ case SHORT:
+ case INT:
+ case LONG:
+ case DATE:
+ return new LongColumnVector(batchSize);
+ case FLOAT:
+ case DOUBLE:
+ return new DoubleColumnVector(batchSize);
+ case BINARY:
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return new BytesColumnVector(batchSize);
+ case TIMESTAMP:
+ return new TimestampColumnVector(batchSize);
+ case DECIMAL:
+ return new DecimalColumnVector(batchSize, type.getPrecision(),
+ type.getScale());
+ case STRUCT: {
+ List<TypeDescription> subtypeIdxs = type.getChildren();
+ ColumnVector[] fieldVector = new ColumnVector[subtypeIdxs.size()];
+ for(int i = 0; i < fieldVector.length; ++i) {
+ fieldVector[i] = createColumn(subtypeIdxs.get(i), batchSize);
+ }
+ return new StructColumnVector(batchSize, fieldVector);
+ }
+ case UNION: {
+ List<TypeDescription> subtypeIdxs = type.getChildren();
+ ColumnVector[] fieldVector = new ColumnVector[subtypeIdxs.size()];
+ for(int i=0; i < fieldVector.length; ++i) {
+ fieldVector[i] = createColumn(subtypeIdxs.get(i), batchSize);
+ }
+ return new UnionColumnVector(batchSize, fieldVector);
+ }
+ case LIST:
+ return new ListColumnVector(batchSize, createColumn(type.getChildren().get(0), batchSize));
+ case MAP:
+ List<TypeDescription> subtypeIdxs = type.getChildren();
+ return new MapColumnVector(batchSize, createColumn(subtypeIdxs.get(0), batchSize),
+ createColumn(subtypeIdxs.get(1), batchSize));
+ default:
+ throw new IllegalArgumentException("LLAP does not support " + type.getCategory());
+ }
+ }
+
private void positionInStreams(TreeReaderFactory.TreeReader[] columnReaders,
- EncodedColumnBatch<OrcBatchKey> batch, int numCols,
- OrcStripeMetadata stripeMetadata) throws IOException {
- for (int i = 0; i < numCols; i++) {
- int columnIndex = batch.getColumnIxs()[i];
- int rowGroupIndex = batch.getBatchKey().rgIx;
- OrcProto.RowIndex rowIndex = stripeMetadata.getRowIndexes()[columnIndex];
- OrcProto.RowIndexEntry rowIndexEntry = rowIndex.getEntry(rowGroupIndex);
- columnReaders[i].seek(new RecordReaderImpl.PositionProviderImpl(rowIndexEntry));
+ EncodedColumnBatch<OrcBatchKey> batch, OrcStripeMetadata stripeMetadata) throws IOException {
+ PositionProvider[] pps = createPositionProviders(columnReaders, batch, stripeMetadata);
+ if (pps == null) return;
+ for (int i = 0; i < columnReaders.length; i++) {
+ columnReaders[i].seek(pps);
}
}
private void repositionInStreams(TreeReaderFactory.TreeReader[] columnReaders,
- EncodedColumnBatch<OrcBatchKey> batch, boolean sameStripe, int numCols,
+ EncodedColumnBatch<OrcBatchKey> batch, boolean sameStripe,
OrcStripeMetadata stripeMetadata) throws IOException {
- for (int i = 0; i < numCols; i++) {
- int columnIndex = batch.getColumnIxs()[i];
- int rowGroupIndex = batch.getBatchKey().rgIx;
- ColumnStreamData[] streamBuffers = batch.getColumnData()[i];
- OrcProto.RowIndex rowIndex = stripeMetadata.getRowIndexes()[columnIndex];
- OrcProto.RowIndexEntry rowIndexEntry = rowIndex.getEntry(rowGroupIndex);
- ((SettableTreeReader)columnReaders[i]).setBuffers(streamBuffers, sameStripe);
- // TODO: When hive moves to java8, make updateTimezone() as default method in SettableTreeReader so that we can
- // avoid this check
- if (columnReaders[i] instanceof EncodedTreeReaderFactory.TimestampStreamReader && !sameStripe) {
- ((EncodedTreeReaderFactory.TimestampStreamReader) columnReaders[i])
+ PositionProvider[] pps = createPositionProviders(columnReaders, batch, stripeMetadata);
+ if (pps == null) return;
+ for (int i = 0; i < columnReaders.length; i++) {
+ TreeReader reader = columnReaders[i];
+ ((SettableTreeReader) reader).setBuffers(batch, sameStripe);
+ // TODO: When hive moves to java8, make updateTimezone() as default method in
+ // SettableTreeReader so that we can avoid this check.
+ if (reader instanceof EncodedTreeReaderFactory.TimestampStreamReader && !sameStripe) {
+ ((EncodedTreeReaderFactory.TimestampStreamReader) reader)
.updateTimezone(stripeMetadata.getWriterTimezone());
}
- columnReaders[i].seek(new RecordReaderImpl.PositionProviderImpl(rowIndexEntry));
+ reader.seek(pps);
+ }
+ }
+
+ private PositionProvider[] createPositionProviders(TreeReaderFactory.TreeReader[] columnReaders,
+ EncodedColumnBatch<OrcBatchKey> batch, OrcStripeMetadata stripeMetadata) throws IOException {
+ if (columnReaders.length == 0) return null;
+ int rowGroupIndex = batch.getBatchKey().rgIx;
+ if (rowGroupIndex == OrcEncodedColumnBatch.ALL_RGS) {
+ throw new IOException("Cannot position readers without RG information");
+ }
+ // TODO: this assumes indexes in getRowIndexes would match column IDs
+ OrcProto.RowIndex[] ris = stripeMetadata.getRowIndexes();
+ PositionProvider[] pps = new PositionProvider[ris.length];
+ for (int i = 0; i < ris.length; ++i) {
+ OrcProto.RowIndex ri = ris[i];
+ if (ri == null) continue;
+ pps[i] = new RecordReaderImpl.PositionProviderImpl(ri.getEntry(rowGroupIndex));
}
+ return pps;
}
private long getRowCount(OrcProto.RowIndexEntry rowIndexEntry) {
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
index 72f733a..39f4112 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
@@ -138,7 +138,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
private final BufferUsageManager bufferManager;
private final Configuration conf;
private final FileSplit split;
- private List<Integer> columnIds;
+ private List<Integer> includedColumnIds;
private final SearchArgument sarg;
private final String[] columnNames;
private final OrcEncodedDataConsumer consumer;
@@ -174,9 +174,9 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
this.bufferManager = bufferManager;
this.conf = conf;
this.split = split;
- this.columnIds = columnIds;
- if (this.columnIds != null) {
- Collections.sort(this.columnIds);
+ this.includedColumnIds = columnIds;
+ if (this.includedColumnIds != null) {
+ Collections.sort(this.includedColumnIds);
}
this.sarg = sarg;
this.columnNames = columnNames;
@@ -197,7 +197,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
fileKey = determineFileId(fs, split,
HiveConf.getBoolVar(conf, ConfVars.LLAP_CACHE_ALLOW_SYNTHETIC_FILEID));
fileMetadata = getOrReadFileMetadata();
- globalIncludes = OrcInputFormat.genIncludedColumns(fileMetadata.getTypes(), columnIds, true);
+ globalIncludes = OrcInputFormat.genIncludedColumns(fileMetadata.getTypes(), includedColumnIds, true);
consumer.setFileMetadata(fileMetadata);
consumer.setIncludedColumns(globalIncludes);
}
@@ -242,8 +242,8 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
+ (fileKey == null ? "" : " (" + fileKey + ")"));
try {
validateFileMetadata();
- if (columnIds == null) {
- columnIds = createColumnIds(fileMetadata);
+ if (includedColumnIds == null) {
+ includedColumnIds = getAllColumnIds(fileMetadata);
}
// 2. Determine which stripes to read based on the split.
@@ -303,13 +303,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
return null;
}
- // 4. Get data from high-level cache.
- // If some cols are fully in cache, this will also give us the modified list of columns to
- // read for every stripe (null means read all of them - the usual path). In any case,
- // readState will be modified for column x rgs that were fetched from high-level cache.
- List<Integer>[] stripeColsToRead = null;
-
- // 5. Create encoded data reader.
+ // 4. Create encoded data reader.
try {
ensureOrcReader();
// Reader creating updates HDFS counters, don't do it here.
@@ -335,17 +329,17 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
int stripeIx = stripeIxFrom + stripeIxMod;
boolean[][] colRgs = null;
- boolean[] stripeIncludes = null;
OrcStripeMetadata stripeMetadata = null;
StripeInformation stripe;
try {
- List<Integer> cols = stripeColsToRead == null ? null : stripeColsToRead[stripeIxMod];
- if (cols != null && cols.isEmpty()) continue; // No need to read this stripe.
stripe = fileMetadata.getStripes().get(stripeIx);
LlapIoImpl.ORC_LOGGER.trace("Reading stripe {}: {}, {}", stripeIx, stripe.getOffset(),
stripe.getLength());
colRgs = readState[stripeIxMod];
+ if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.ORC_LOGGER.trace("readState[{}]: {}", stripeIxMod, Arrays.toString(colRgs));
+ }
// We assume that NO_RGS value is only set from SARG filter and for all columns;
// intermediate changes for individual columns will unset values in the array.
// Skip this case for 0-column read. We could probably special-case it just like we do
@@ -353,17 +347,6 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
if (colRgs.length > 0 && colRgs[0] ==
RecordReaderImpl.SargApplier.READ_NO_RGS) continue;
- // 6.1. Determine the columns to read (usually the same as requested).
- if (cols == null || cols.size() == colRgs.length) {
- cols = columnIds;
- stripeIncludes = globalIncludes;
- } else {
- // We are reading subset of the original columns, remove unnecessary bitmasks/etc.
- // This will never happen w/o high-level cache.
- stripeIncludes = OrcInputFormat.genIncludedColumns(fileMetadata.getTypes(), cols, true);
- colRgs = genStripeColRgs(cols, colRgs);
- }
-
// 6.2. Ensure we have stripe metadata. We might have read it before for RG filtering.
boolean isFoundInCache = false;
if (stripeMetadatas != null) {
@@ -379,27 +362,27 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
ensureMetadataReader();
long startTimeHdfs = counters.startTimeCounter();
stripeMetadata = new OrcStripeMetadata(new OrcBatchKey(fileKey, stripeIx, 0),
- metadataReader, stripe, stripeIncludes, sargColumns);
+ metadataReader, stripe, globalIncludes, sargColumns);
counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTimeHdfs);
if (hasFileId && metadataCache != null) {
stripeMetadata = metadataCache.putStripeMetadata(stripeMetadata);
if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
LlapIoImpl.ORC_LOGGER.trace("Caching stripe {} metadata with includes: {}",
- stripeKey.stripeIx, DebugUtils.toString(stripeIncludes));
+ stripeKey.stripeIx, DebugUtils.toString(globalIncludes));
}
}
}
consumer.setStripeMetadata(stripeMetadata);
}
- if (!stripeMetadata.hasAllIndexes(stripeIncludes)) {
+ if (!stripeMetadata.hasAllIndexes(globalIncludes)) {
if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
LlapIoImpl.ORC_LOGGER.trace("Updating indexes in stripe {} metadata for includes: {}",
- stripeKey.stripeIx, DebugUtils.toString(stripeIncludes));
+ stripeKey.stripeIx, DebugUtils.toString(globalIncludes));
}
assert isFoundInCache;
counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
ensureMetadataReader();
- updateLoadedIndexes(stripeMetadata, stripe, stripeIncludes, sargColumns);
+ updateLoadedIndexes(stripeMetadata, stripe, globalIncludes, sargColumns);
} else if (isFoundInCache) {
counters.incrCounter(LlapIOCounters.METADATA_CACHE_HIT);
}
@@ -415,7 +398,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
return null;
}
- // 6.3. Finally, hand off to the stripe reader to produce the data.
+ // 5.2. Finally, hand off to the stripe reader to produce the data.
// This is a sync call that will feed data to the consumer.
try {
// TODO: readEncodedColumns is not supposed to throw; errors should be propagated thru
@@ -423,7 +406,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
// Also, currently readEncodedColumns is not stoppable. The consumer will discard the
// data it receives for one stripe. We could probably interrupt it, if it checked that.
stripeReader.readEncodedColumns(stripeIx, stripe, stripeMetadata.getRowIndexes(),
- stripeMetadata.getEncodings(), stripeMetadata.getStreams(), stripeIncludes,
+ stripeMetadata.getEncodings(), stripeMetadata.getStreams(), globalIncludes,
colRgs, consumer);
} catch (Throwable t) {
consumer.setError(t);
@@ -521,22 +504,14 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
return HdfsUtils.getFileId(fs, split.getPath(), allowSynthetic);
}
- private boolean[][] genStripeColRgs(List<Integer> stripeCols, boolean[][] globalColRgs) {
- boolean[][] stripeColRgs = new boolean[stripeCols.size()][];
- for (int i = 0, i2 = -1; i < globalColRgs.length; ++i) {
- if (globalColRgs[i] == null) continue;
- stripeColRgs[i2] = globalColRgs[i];
- ++i2;
- }
- return stripeColRgs;
- }
-
/**
* Puts all column indexes from metadata to make a column list to read all column.
*/
- private static List<Integer> createColumnIds(OrcFileMetadata metadata) {
- List<Integer> columnIds = new ArrayList<Integer>(metadata.getTypes().size());
- for (int i = 1; i < metadata.getTypes().size(); ++i) {
+ private static List<Integer> getAllColumnIds(OrcFileMetadata metadata) {
+ int rootColumn = OrcInputFormat.getRootColumn(true);
+ List<Integer> types = metadata.getTypes().get(rootColumn).getSubtypesList();
+ List<Integer> columnIds = new ArrayList<Integer>(types.size());
+ for (int i = 0; i < types.size(); ++i) {
columnIds.add(i);
}
return columnIds;
@@ -690,8 +665,9 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
@Override
public void returnData(OrcEncodedColumnBatch ecb) {
- for (ColumnStreamData[] datas : ecb.getColumnData()) {
- if (datas == null) continue;
+ for (int colIx = 0; colIx < ecb.getTotalColCount(); ++colIx) {
+ if (!ecb.hasData(colIx)) continue;
+ ColumnStreamData[] datas = ecb.getColumnData(colIx);
for (ColumnStreamData data : datas) {
if (data == null || data.decRef() != 0) continue;
if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
@@ -749,12 +725,17 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
}
assert isAll || isNone || rgsToRead.length == rgCount;
- readState[stripeIxMod] = new boolean[columnIds.size()][];
- for (int j = 0; j < columnIds.size(); ++j) {
- readState[stripeIxMod][j] = (isAll || isNone) ? rgsToRead :
+ int fileIncludesCount = 0;
+ // TODO: hacky for now - skip the root 0-s column.
+ // We don't need separate readState w/o HL cache, should get rid of that instead.
+ for (int includeIx = 1; includeIx < globalIncludes.length; ++includeIx) {
+ fileIncludesCount += (globalIncludes[includeIx] ? 1 : 0);
+ }
+ readState[stripeIxMod] = new boolean[fileIncludesCount][];
+ for (int includeIx = 0; includeIx < fileIncludesCount; ++includeIx) {
+ readState[stripeIxMod][includeIx] = (isAll || isNone) ? rgsToRead :
Arrays.copyOf(rgsToRead, rgsToRead.length);
}
-
adjustRgMetric(rgCount, rgsToRead, isNone, isAll);
}
return hasAnyData;
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
index c9b0a4d..70cba05 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
@@ -32,7 +32,9 @@ import org.apache.hadoop.hive.ql.io.orc.Reader;
import org.apache.orc.CompressionKind;
import org.apache.orc.FileMetadata;
import org.apache.orc.OrcProto;
+import org.apache.orc.OrcUtils;
import org.apache.orc.StripeInformation;
+import org.apache.orc.TypeDescription;
import org.apache.orc.impl.ReaderImpl;
/** ORC file metadata. Currently contains some duplicate info due to how different parts
@@ -222,4 +224,8 @@ public final class OrcFileMetadata extends LlapCacheableBuffer implements FileMe
public List<OrcProto.ColumnStatistics> getFileStats() {
return fileStats;
}
+
+ public TypeDescription getSchema() {
+ return OrcUtils.convertTypeFromProtobuf(this.types, 0);
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/orc/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
----------------------------------------------------------------------
diff --git a/orc/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java b/orc/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
index c347181..5d5f991 100644
--- a/orc/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
+++ b/orc/src/java/org/apache/orc/impl/ConvertTreeReaderFactory.java
@@ -292,7 +292,7 @@ public class ConvertTreeReaderFactory extends TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
// Pass-thru.
convertTreeReader.seek(index);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/orc/src/java/org/apache/orc/impl/TreeReaderFactory.java
----------------------------------------------------------------------
diff --git a/orc/src/java/org/apache/orc/impl/TreeReaderFactory.java b/orc/src/java/org/apache/orc/impl/TreeReaderFactory.java
index 245e3ce..e46a0a4 100644
--- a/orc/src/java/org/apache/orc/impl/TreeReaderFactory.java
+++ b/orc/src/java/org/apache/orc/impl/TreeReaderFactory.java
@@ -80,7 +80,7 @@ public class TreeReaderFactory {
}
}
- static IntegerReader createIntegerReader(OrcProto.ColumnEncoding.Kind kind,
+ protected static IntegerReader createIntegerReader(OrcProto.ColumnEncoding.Kind kind,
InStream in,
boolean signed, boolean skipCorrupt) throws IOException {
switch (kind) {
@@ -115,7 +115,7 @@ public class TreeReaderFactory {
* @param index the indexes loaded from the file
* @throws IOException
*/
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -201,6 +201,10 @@ public class TreeReaderFactory {
public BitFieldReader getPresent() {
return present;
}
+
+ public int getColumnId() {
+ return columnId;
+ }
}
public static class NullTreeReader extends TreeReader {
@@ -262,7 +266,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -313,7 +317,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -380,7 +384,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -447,7 +451,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -515,7 +519,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -569,7 +573,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -663,7 +667,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -774,7 +778,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -898,7 +902,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -1007,7 +1011,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -1083,7 +1087,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -1197,7 +1201,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
reader.seek(index);
}
@@ -1346,7 +1350,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -1495,7 +1499,7 @@ public class TreeReaderFactory {
}
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
seek(index[columnId]);
}
@@ -1701,7 +1705,7 @@ public class TreeReaderFactory {
}
}
- protected static class StructTreeReader extends TreeReader {
+ public static class StructTreeReader extends TreeReader {
protected final TreeReader[] fields;
protected StructTreeReader(int columnId,
@@ -1719,8 +1723,21 @@ public class TreeReaderFactory {
}
}
+ public TreeReader[] getChildReaders() {
+ return fields;
+ }
+
+ protected StructTreeReader(int columnId, InStream present,
+ OrcProto.ColumnEncoding encoding, TreeReader[] childReaders) throws IOException {
+ super(columnId, present);
+ if (encoding != null) {
+ checkEncoding(encoding);
+ }
+ this.fields = childReaders;
+ }
+
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
super.seek(index);
for (TreeReader kid : fields) {
if (kid != null) {
@@ -1804,8 +1821,17 @@ public class TreeReaderFactory {
}
}
+ protected UnionTreeReader(int columnId, InStream present,
+ OrcProto.ColumnEncoding encoding, TreeReader[] childReaders) throws IOException {
+ super(columnId, present);
+ if (encoding != null) {
+ checkEncoding(encoding);
+ }
+ this.fields = childReaders;
+ }
+
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
super.seek(index);
tags.seek(index[columnId]);
for (TreeReader kid : fields) {
@@ -1877,8 +1903,18 @@ public class TreeReaderFactory {
skipCorrupt);
}
+ protected ListTreeReader(int columnId, InStream present, InStream data,
+ OrcProto.ColumnEncoding encoding, TreeReader elementReader) throws IOException {
+ super(columnId, present);
+ if (data != null && encoding != null) {
+ checkEncoding(encoding);
+ this.lengths = createIntegerReader(encoding.getKind(), data, false, false);
+ }
+ this.elementReader = elementReader;
+ }
+
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
super.seek(index);
lengths.seek(index[columnId]);
elementReader.seek(index);
@@ -1958,8 +1994,20 @@ public class TreeReaderFactory {
valueReader = createTreeReader(valueType, evolution, included, skipCorrupt);
}
+ protected MapTreeReader(int columnId, InStream present, InStream data,
+ OrcProto.ColumnEncoding encoding, TreeReader keyReader, TreeReader valueReader)
+ throws IOException {
+ super(columnId, present);
+ if (data != null && encoding != null) {
+ checkEncoding(encoding);
+ this.lengths = createIntegerReader(encoding.getKind(), data, false, false);
+ }
+ this.keyReader = keyReader;
+ this.valueReader = valueReader;
+ }
+
@Override
- void seek(PositionProvider[] index) throws IOException {
+ public void seek(PositionProvider[] index) throws IOException {
super.seek(index);
lengths.seek(index[columnId]);
keyReader.seek(index);
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
index a9dbc3e..2b1d1ce 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
@@ -3636,7 +3636,7 @@ public final class Utilities {
* @param conf - configuration
* @return false for types not supported by vectorization, true otherwise
*/
- public static boolean checkLlapIOSupportedTypes(final Configuration conf) {
+ public static boolean checkVectorizerSupportedTypes(final Configuration conf) {
final String[] readColumnNames = ColumnProjectionUtils.getReadColumnNames(conf);
final String columnNames = conf.get(serdeConstants.LIST_COLUMNS);
final String columnTypes = conf.get(serdeConstants.LIST_COLUMN_TYPES);
@@ -3649,7 +3649,7 @@ public final class Utilities {
final List<String> allColumnNames = Lists.newArrayList(columnNames.split(","));
final List<TypeInfo> typeInfos = TypeInfoUtils.getTypeInfosFromTypeString(columnTypes);
final List<String> allColumnTypes = TypeInfoUtils.getTypeStringsFromTypeInfo(typeInfos);
- return checkLlapIOSupportedTypes(Lists.newArrayList(readColumnNames), allColumnNames,
+ return checkVectorizerSupportedTypes(Lists.newArrayList(readColumnNames), allColumnNames,
allColumnTypes);
}
@@ -3660,7 +3660,7 @@ public final class Utilities {
* @param allColumnTypes - all column types
* @return false for types not supported by vectorization, true otherwise
*/
- public static boolean checkLlapIOSupportedTypes(final List<String> readColumnNames,
+ public static boolean checkVectorizerSupportedTypes(final List<String> readColumnNames,
final List<String> allColumnNames, final List<String> allColumnTypes) {
final String[] readColumnTypes = getReadColumnTypes(readColumnNames, allColumnNames,
allColumnTypes);
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
index ccb39da..572953a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
@@ -282,7 +282,7 @@ public class OrcInputFormat implements InputFormat<NullWritable, OrcStruct>,
* @param isOriginal is the file in the original format?
* @return the column number for the root of row.
*/
- static int getRootColumn(boolean isOriginal) {
+ public static int getRootColumn(boolean isOriginal) {
return isOriginal ? 0 : (OrcRecordUpdater.ROW + 1);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReader.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReader.java
index 4d09dcd..ea9904a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReader.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReader.java
@@ -55,4 +55,4 @@ public interface EncodedReader {
* to just checking the constant in the first place.
*/
void setTracing(boolean isEnabled);
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/6efa869f/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
index a4925b9..8f0c237 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hive.ql.io.orc.encoded;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import org.slf4j.Logger;
@@ -42,6 +43,7 @@ import org.apache.orc.impl.RecordReaderUtils;
import org.apache.orc.impl.StreamName;
import org.apache.orc.StripeInformation;
import org.apache.orc.impl.BufferChunk;
+import org.apache.hadoop.hive.llap.DebugUtils;
import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.OrcEncodedColumnBatch;
import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.PoolFactory;
import org.apache.orc.OrcProto;
@@ -127,11 +129,13 @@ class EncodedReaderImpl implements EncodedReader {
/** Helper context for each column being read */
private static final class ColumnReadContext {
+
public ColumnReadContext(int colIx, OrcProto.ColumnEncoding encoding,
- OrcProto.RowIndex rowIndex) {
+ OrcProto.RowIndex rowIndex, int colRgIx) {
this.encoding = encoding;
this.rowIndex = rowIndex;
this.colIx = colIx;
+ this.includedIx = colRgIx;
streamCount = 0;
}
@@ -145,6 +149,8 @@ class EncodedReaderImpl implements EncodedReader {
OrcProto.RowIndex rowIndex;
/** Column index in the file. */
int colIx;
+ /** Column index in the included columns only (for RG masks). */
+ int includedIx;
public void addStream(long offset, OrcProto.Stream stream, int indexIx) {
streams[streamCount++] = new StreamContext(stream, offset, indexIx);
@@ -154,6 +160,7 @@ class EncodedReaderImpl implements EncodedReader {
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(" column_index: ").append(colIx);
+ sb.append(" included_index: ").append(includedIx);
sb.append(" encoding: ").append(encoding);
sb.append(" stream_count: ").append(streamCount);
int i = 0;
@@ -200,6 +207,7 @@ class EncodedReaderImpl implements EncodedReader {
OrcProto.RowIndex[] indexes, List<OrcProto.ColumnEncoding> encodings, List<OrcProto.Stream> streamList,
boolean[] included, boolean[][] colRgs,
Consumer<OrcEncodedColumnBatch> consumer) throws IOException {
+ isTracingEnabled = true;
// Note: for now we don't have to setError here, caller will setError if we throw.
// We are also not supposed to call setDone, since we are only part of the operation.
long stripeOffset = stripe.getOffset();
@@ -208,18 +216,26 @@ class EncodedReaderImpl implements EncodedReader {
// 1.1. Figure out which columns have a present stream
boolean[] hasNull = RecordReaderUtils.findPresentStreamsByColumn(streamList, types);
if (isTracingEnabled) {
- LOG.trace("The following columns have PRESENT streams: " + arrayToString(hasNull));
+ LOG.error("The following columns have PRESENT streams: " + arrayToString(hasNull));
}
// We assume stream list is sorted by column and that non-data
// streams do not interleave data streams for the same column.
// 1.2. With that in mind, determine disk ranges to read/get from cache (not by stream).
- int colRgIx = -1, lastColIx = -1;
- ColumnReadContext[] colCtxs = new ColumnReadContext[colRgs.length];
- boolean[] includedRgs = null;
+ ColumnReadContext[] colCtxs = new ColumnReadContext[included.length];
+ int colRgIx = -1;
+ // Don't create context for the 0-s column.
+ for (int i = 1; i < included.length; ++i) {
+ if (!included[i]) continue;
+ colCtxs[i] = new ColumnReadContext(i, encodings.get(i), indexes[i], ++colRgIx);
+ if (isTracingEnabled) {
+ LOG.error("Creating context: " + colCtxs[i].toString());
+ }
+ }
boolean isCompressed = (codec != null);
CreateHelper listToRead = new CreateHelper();
boolean hasIndexOnlyCols = false;
+ boolean[] includedRgs = null; // Will always be the same for all cols at the moment.
for (OrcProto.Stream stream : streamList) {
long length = stream.getLength();
int colIx = stream.getColumn();
@@ -227,28 +243,17 @@ class EncodedReaderImpl implements EncodedReader {
if (!included[colIx] || StreamName.getArea(streamKind) != StreamName.Area.DATA) {
// We have a stream for included column, but in future it might have no data streams.
// It's more like "has at least one column included that has an index stream".
- hasIndexOnlyCols = hasIndexOnlyCols | included[colIx];
+ hasIndexOnlyCols = hasIndexOnlyCols || included[colIx];
if (isTracingEnabled) {
- LOG.trace("Skipping stream: " + streamKind + " at " + offset + ", " + length);
+ LOG.trace("Skipping stream for column " + colIx + ": "
+ + streamKind + " at " + offset + ", " + length);
}
offset += length;
continue;
}
- ColumnReadContext ctx = null;
- if (lastColIx != colIx) {
- ++colRgIx;
- assert colCtxs[colRgIx] == null;
- lastColIx = colIx;
- includedRgs = colRgs[colRgIx];
- ctx = colCtxs[colRgIx] = new ColumnReadContext(
- colIx, encodings.get(colIx), indexes[colIx]);
- if (isTracingEnabled) {
- LOG.trace("Creating context " + colRgIx + " for column " + colIx + ":" + ctx.toString());
- }
- } else {
- ctx = colCtxs[colRgIx];
- assert ctx != null;
- }
+ ColumnReadContext ctx = colCtxs[colIx];
+ assert ctx != null;
+ includedRgs = colRgs[ctx.includedIx];
int indexIx = RecordReaderUtils.getIndexPosition(ctx.encoding.getKind(),
types.get(colIx).getKind(), streamKind, isCompressed, hasNull[colIx]);
ctx.addStream(offset, stream, indexIx);
@@ -275,7 +280,7 @@ class EncodedReaderImpl implements EncodedReader {
// TODO: there may be a bug here. Could there be partial RG filtering on index-only column?
if (hasIndexOnlyCols && (includedRgs == null)) {
OrcEncodedColumnBatch ecb = POOLS.ecbPool.take();
- ecb.init(fileKey, stripeIx, OrcEncodedColumnBatch.ALL_RGS, colRgs.length);
+ ecb.init(fileKey, stripeIx, OrcEncodedColumnBatch.ALL_RGS, included.length);
consumer.consumeData(ecb);
} else {
LOG.warn("Nothing to read for stripe [" + stripe + "]");
@@ -309,8 +314,9 @@ class EncodedReaderImpl implements EncodedReader {
// 3. For uncompressed case, we need some special processing before read.
DiskRangeList iter = toRead.next; // Keep "toRead" list for future use, don't extract().
if (codec == null) {
- for (int colIxMod = 0; colIxMod < colRgs.length; ++colIxMod) {
- ColumnReadContext ctx = colCtxs[colIxMod];
+ for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
+ ColumnReadContext ctx = colCtxs[colIx];
+ if (ctx == null) continue; // This column is not included.
for (int streamIx = 0; streamIx < ctx.streamCount; ++streamIx) {
StreamContext sctx = ctx.streams[streamIx];
DiskRangeList newIter = preReadUncompressedStream(
@@ -334,19 +340,27 @@ class EncodedReaderImpl implements EncodedReader {
boolean isLastRg = rgIx == rgCount - 1;
// Create the batch we will use to return data for this RG.
OrcEncodedColumnBatch ecb = POOLS.ecbPool.take();
- ecb.init(fileKey, stripeIx, rgIx, colRgs.length);
+ ecb.init(fileKey, stripeIx, rgIx, included.length);
boolean isRGSelected = true;
- for (int colIxMod = 0; colIxMod < colRgs.length; ++colIxMod) {
- // TODO: simplify this now that high-level cache has been removed.
- if (colRgs[colIxMod] != null && !colRgs[colIxMod][rgIx]) {
+ for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
+ ColumnReadContext ctx = colCtxs[colIx];
+ if (ctx == null) continue; // This column is not included.
+ if (isTracingEnabled) {
+ LOG.trace("ctx: {} rgIx: {} isLastRg: {} rgCount: {}", ctx, rgIx, isLastRg, rgCount);
+ }
+ // TODO: simplify this now that high-level cache has been removed. Same RGs for all cols.
+ if (colRgs[ctx.includedIx] != null && !colRgs[ctx.includedIx][rgIx]) {
// RG x col filtered.
isRGSelected = false;
- continue;
+ if (isTracingEnabled) {
+ LOG.trace("colIxMod: {} rgIx: {} colRgs[{}]: {} colRgs[{}][{}]: {}", ctx.includedIx, rgIx, ctx.includedIx,
+ Arrays.toString(colRgs[ctx.includedIx]), ctx.includedIx, rgIx, colRgs[ctx.includedIx][rgIx]);
+ }
+ continue;
}
- ColumnReadContext ctx = colCtxs[colIxMod];
OrcProto.RowIndexEntry index = ctx.rowIndex.getEntry(rgIx),
nextIndex = isLastRg ? null : ctx.rowIndex.getEntry(rgIx + 1);
- ecb.initColumn(colIxMod, ctx.colIx, OrcEncodedColumnBatch.MAX_DATA_STREAMS);
+ ecb.initOrcColumn(ctx.colIx);
for (int streamIx = 0; streamIx < ctx.streamCount; ++streamIx) {
StreamContext sctx = ctx.streams[streamIx];
ColumnStreamData cb = null;
@@ -402,7 +416,7 @@ class EncodedReaderImpl implements EncodedReader {
sctx.bufferIter = iter = lastCached;
}
}
- ecb.setStreamData(colIxMod, sctx.kind.getNumber(), cb);
+ ecb.setStreamData(ctx.colIx, sctx.kind.getNumber(), cb);
} catch (Exception ex) {
DiskRangeList drl = toRead == null ? null : toRead.next;
LOG.error("Error getting stream [" + sctx.kind + ", " + ctx.encoding + "] for"