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 2017/01/25 03:34:37 UTC
hive git commit: HIVE-15664 : LLAP text cache: improve first query
perf I (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/master 20210dec9 -> 881deac63
HIVE-15664 : LLAP text cache: improve first query perf I (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/881deac6
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/881deac6
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/881deac6
Branch: refs/heads/master
Commit: 881deac63e835c4cfa8b9232b7aa280fe82acea7
Parents: 20210de
Author: Sergey Shelukhin <se...@apache.org>
Authored: Tue Jan 24 19:21:47 2017 -0800
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Tue Jan 24 19:21:47 2017 -0800
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 5 +
.../hive/llap/cache/EvictionDispatcher.java | 4 +-
.../hive/llap/cache/SerDeLowLevelCacheImpl.java | 12 +-
.../hive/llap/io/api/impl/LlapIoImpl.java | 18 +-
.../hive/llap/io/api/impl/LlapRecordReader.java | 6 +-
.../llap/io/decode/ColumnVectorProducer.java | 5 +-
.../io/decode/GenericColumnVectorProducer.java | 21 +-
.../llap/io/decode/OrcColumnVectorProducer.java | 8 +-
.../llap/io/encoded/SerDeEncodedDataReader.java | 195 ++++++++++----
.../io/encoded/VertorDeserializeOrcWriter.java | 261 +++++++++++++++++++
.../java/org/apache/orc/impl/WriterImpl.java | 5 +-
.../hadoop/hive/ql/io/HiveInputFormat.java | 13 +-
.../org/apache/hadoop/hive/ql/plan/MapWork.java | 1 -
.../hive/ql/exec/vector/VectorizedRowBatch.java | 6 +-
14 files changed, 482 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index 8e319c6..2ccb764 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -2930,6 +2930,9 @@ public class HiveConf extends Configuration {
"Whether LLAP should use fileId (inode)-based path to ensure better consistency for the\n" +
"cases of file overwrites. This is supported on HDFS."),
// Restricted to text for now as this is a new feature; only text files can be sliced.
+ LLAP_IO_ENCODE_ENABLED("hive.llap.io.encode.enabled", true,
+ "Whether LLAP should try to re-encode and cache data for non-ORC formats. This is used\n" +
+ "on LLAP Server side to determine if the infrastructure for that is initialized."),
LLAP_IO_ENCODE_FORMATS("hive.llap.io.encode.formats",
"org.apache.hadoop.mapred.TextInputFormat,",
"The table input formats for which LLAP IO should re-encode and cache data.\n" +
@@ -2938,6 +2941,8 @@ public class HiveConf extends Configuration {
"Allocation size for the buffers used to cache encoded data from non-ORC files. Must\n" +
"be a power of two between " + LLAP_ALLOCATOR_MIN_ALLOC + " and\n" +
LLAP_ALLOCATOR_MAX_ALLOC + "."),
+ LLAP_IO_ENCODE_VECTOR_SERDE_ENABLED("hive.llap.io.encode.vector.serde.enabled", true,
+ "Whether LLAP should use vectorized SerDe reader to read text data when re-encoding."),
LLAP_IO_ENCODE_SLICE_ROW_COUNT("hive.llap.io.encode.slice.row.count", 100000,
"Row count to use to separate cache slices when reading encoded data from row-based\n" +
"inputs into LLAP cache, if this feature is enabled."),
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
index 2d3197c..a6b0abd 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
@@ -48,7 +48,9 @@ public final class EvictionDispatcher implements EvictionListener {
// Note: we don't know which cache this is from, so we notify both. They can noop if they
// want to find the buffer in their structures and can't.
dataCache.notifyEvicted(buffer);
- serdeCache.notifyEvicted(buffer);
+ if (serdeCache != null) {
+ serdeCache.notifyEvicted(buffer);
+ }
allocator.deallocateEvicted(buffer);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
index caa60e5..85fae9a 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
@@ -201,17 +201,16 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapOomDebugD
sb.append("null, ");
continue;
}
- sb.append("[");
+ sb.append("colData [");
for (int j = 0; j < colData.length; ++j) {
LlapDataBuffer[] streamData = colData[j];
if (streamData == null) {
sb.append("null, ");
continue;
}
- sb.append("[");
+ sb.append("buffers [");
for (int k = 0; k < streamData.length; ++k) {
- LlapDataBuffer s = streamData[k];
- sb.append(LlapDataBuffer.toDataString(s));
+ sb.append(streamData[k]);
}
sb.append("], ");
}
@@ -396,7 +395,10 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapOomDebugD
for (int colIx = 0; colIx < cached.colCount; ++colIx) {
if (!includes[colIx]) continue;
if (cStripe.encodings[colIx] == null || cStripe.data[colIx] == null) {
- assert cStripe.data[colIx] == null; // No encoding => must have no data.
+ if (cStripe.data[colIx] != null) {
+ assert false : cStripe;
+ // No encoding => must have no data.
+ }
isMissed = true;
if (gotAllData != null) {
gotAllData.value = false;
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
index 7cfd133..7c309a4 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
@@ -117,6 +117,7 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
LowLevelCache cache = null;
SerDeLowLevelCacheImpl serdeCache = null; // TODO: extract interface when needed
BufferUsageManager bufferManager = null;
+ boolean isEncodeEnabled = HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENCODE_ENABLED);
if (useLowLevelCache) {
// Memory manager uses cache policy to trigger evictions, so create the policy first.
boolean useLrfu = HiveConf.getBoolVar(conf, HiveConf.ConfVars.LLAP_USE_LRFU);
@@ -130,15 +131,17 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
this.allocator = allocator;
LowLevelCacheImpl cacheImpl = new LowLevelCacheImpl(
cacheMetrics, cachePolicy, allocator, true);
- SerDeLowLevelCacheImpl serdeCacheImpl = new SerDeLowLevelCacheImpl(
- cacheMetrics, cachePolicy, allocator);
cache = cacheImpl;
- serdeCache = serdeCacheImpl;
+ if (isEncodeEnabled) {
+ SerDeLowLevelCacheImpl serdeCacheImpl = new SerDeLowLevelCacheImpl(
+ cacheMetrics, cachePolicy, allocator);
+ serdeCache = serdeCacheImpl;
+ }
boolean useGapCache = HiveConf.getBoolVar(conf, ConfVars.LLAP_CACHE_ENABLE_ORC_GAP_CACHE);
metadataCache = new OrcMetadataCache(memManager, cachePolicy, useGapCache);
// And finally cache policy uses cache to notify it of eviction. The cycle is complete!
cachePolicy.setEvictionListener(new EvictionDispatcher(
- cache, serdeCacheImpl, metadataCache, allocator));
+ cache, serdeCache, metadataCache, allocator));
cachePolicy.setParentDebugDumper(cacheImpl);
cacheImpl.startThreads(); // Start the cache threads.
bufferManager = cacheImpl; // Cache also serves as buffer manager.
@@ -157,8 +160,8 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
// TODO: this should depends on input format and be in a map, or something.
this.orcCvp = new OrcColumnVectorProducer(
metadataCache, cache, bufferManager, conf, cacheMetrics, ioMetrics);
- this.genericCvp = new GenericColumnVectorProducer(
- serdeCache, bufferManager, conf, cacheMetrics, ioMetrics);
+ this.genericCvp = isEncodeEnabled ? new GenericColumnVectorProducer(
+ serdeCache, bufferManager, conf, cacheMetrics, ioMetrics) : null;
LOG.info("LLAP IO initialized");
registerMXBeans();
@@ -175,6 +178,9 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
ColumnVectorProducer cvp = genericCvp;
if (sourceInputFormat instanceof OrcInputFormat) {
cvp = orcCvp; // Special-case for ORC.
+ } else if (cvp == null) {
+ LOG.warn("LLAP encode is disabled; cannot use for " + sourceInputFormat.getClass());
+ return null;
}
return new LlapInputFormat(sourceInputFormat, sourceSerDe, cvp, executor);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
index 1a76eae..325208f 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
@@ -23,6 +23,8 @@ import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.llap.ConsumerFeedback;
@@ -42,6 +44,7 @@ import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.MapWork;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.io.NullWritable;
@@ -131,7 +134,8 @@ class LlapRecordReader
// Create the consumer of encoded data; it will coordinate decoding to CVBs.
feedback = rp = cvp.createReadPipeline(this, split, columnIds, sarg, columnNames,
- counters, schema, sourceInputFormat, sourceSerDe, reporter, job);
+ counters, schema, sourceInputFormat, sourceSerDe, reporter, job,
+ mapWork.getPathToPartitionInfo());
fileSchema = rp.getFileSchema();
includedColumns = rp.getIncludedColumns();
}
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ColumnVectorProducer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ColumnVectorProducer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ColumnVectorProducer.java
index 2e4f2ba..d08dfbb 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ColumnVectorProducer.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ColumnVectorProducer.java
@@ -20,11 +20,14 @@ package org.apache.hadoop.hive.llap.io.decode;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch;
import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputFormat;
@@ -40,5 +43,5 @@ public interface ColumnVectorProducer {
List<Integer> columnIds, SearchArgument sarg, String[] columnNames,
QueryFragmentCounters counters, TypeDescription readerSchema,
InputFormat<?, ?> sourceInputFormat, Deserializer sourceSerDe, Reporter reporter,
- JobConf job) throws IOException;
+ JobConf job, Map<Path, PartitionDesc> parts) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java
index 5cddae5..61d385e 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java
@@ -21,9 +21,11 @@ package org.apache.hadoop.hive.llap.io.decode;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.TimeZone;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.llap.cache.BufferUsageManager;
import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl;
import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
@@ -34,9 +36,11 @@ import org.apache.hadoop.hive.llap.io.metadata.ConsumerFileMetadata;
import org.apache.hadoop.hive.llap.io.metadata.ConsumerStripeMetadata;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonIOMetrics;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx;
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
@@ -75,13 +79,14 @@ public class GenericColumnVectorProducer implements ColumnVectorProducer {
public ReadPipeline createReadPipeline(Consumer<ColumnVectorBatch> consumer, FileSplit split,
List<Integer> columnIds, SearchArgument sarg, String[] columnNames,
QueryFragmentCounters counters, TypeDescription schema, InputFormat<?, ?> sourceInputFormat,
- Deserializer sourceSerDe, Reporter reporter, JobConf job) throws IOException {
+ Deserializer sourceSerDe, Reporter reporter, JobConf job, Map<Path, PartitionDesc> parts)
+ throws IOException {
cacheMetrics.incrCacheReadRequests();
OrcEncodedDataConsumer edc = new OrcEncodedDataConsumer(
consumer, columnIds.size(), false, counters, ioMetrics);
- TextFileMetadata fm;
+ SerDeFileMetadata fm;
try {
- fm = new TextFileMetadata(sourceSerDe);
+ fm = new SerDeFileMetadata(sourceSerDe);
} catch (SerDeException e) {
throw new IOException(e);
}
@@ -89,7 +94,7 @@ public class GenericColumnVectorProducer implements ColumnVectorProducer {
// Note that we pass job config to the record reader, but use global config for LLAP IO.
SerDeEncodedDataReader reader = new SerDeEncodedDataReader(cache,
bufferManager, conf, split, columnIds, edc, job, reporter, sourceInputFormat,
- sourceSerDe, counters, fm.getSchema());
+ sourceSerDe, counters, fm.getSchema(), parts);
edc.init(reader, reader);
if (LlapIoImpl.LOG.isDebugEnabled()) {
LlapIoImpl.LOG.debug("Ignoring schema: " + schema);
@@ -98,14 +103,14 @@ public class GenericColumnVectorProducer implements ColumnVectorProducer {
}
- public static final class TextStripeMetadata implements ConsumerStripeMetadata {
+ public static final class SerDeStripeMetadata implements ConsumerStripeMetadata {
// The writer is local to the process.
private final String writerTimezone = TimeZone.getDefault().getID();
private List<ColumnEncoding> encodings;
private final int stripeIx;
private long rowCount = -1;
- public TextStripeMetadata(int stripeIx) {
+ public SerDeStripeMetadata(int stripeIx) {
this.stripeIx = stripeIx;
}
@@ -159,10 +164,10 @@ public class GenericColumnVectorProducer implements ColumnVectorProducer {
}
- private static final class TextFileMetadata implements ConsumerFileMetadata {
+ private static final class SerDeFileMetadata implements ConsumerFileMetadata {
private final List<Type> orcTypes = new ArrayList<>();
private final TypeDescription schema;
- public TextFileMetadata(Deserializer sourceSerDe) throws SerDeException {
+ public SerDeFileMetadata(Deserializer sourceSerDe) throws SerDeException {
TypeDescription schema = OrcInputFormat.convertTypeInfo(
TypeInfoUtils.getTypeInfoFromObjectInspector(sourceSerDe.getObjectInspector()));
this.schema = schema;
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java
index 565e3d2..7c89e82 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcColumnVectorProducer.java
@@ -20,8 +20,10 @@ package org.apache.hadoop.hive.llap.io.decode;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.llap.cache.BufferUsageManager;
import org.apache.hadoop.hive.llap.cache.LowLevelCache;
@@ -32,8 +34,10 @@ import org.apache.hadoop.hive.llap.io.encoded.OrcEncodedDataReader;
import org.apache.hadoop.hive.llap.io.metadata.OrcMetadataCache;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonIOMetrics;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx;
import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputFormat;
@@ -69,8 +73,8 @@ public class OrcColumnVectorProducer implements ColumnVectorProducer {
public ReadPipeline createReadPipeline(
Consumer<ColumnVectorBatch> consumer, FileSplit split, List<Integer> columnIds,
SearchArgument sarg, String[] columnNames, QueryFragmentCounters counters,
- TypeDescription readerSchema, InputFormat<?, ?> sourceInputFormat, Deserializer sourceSerDe,
- Reporter reporter, JobConf job) throws IOException {
+ TypeDescription readerSchema, InputFormat<?, ?> unused0, Deserializer unused1,
+ Reporter reporter, JobConf job, Map<Path, PartitionDesc> unused2) throws IOException {
cacheMetrics.incrCacheReadRequests();
OrcEncodedDataConsumer edc = new OrcEncodedDataConsumer(consumer, columnIds.size(),
_skipCorrupt, counters, ioMetrics);
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
index dd189f1..9ab26e6 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
@@ -30,6 +30,7 @@ import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.Pool.PoolObjectHelper;
import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
import org.apache.hadoop.hive.common.io.DiskRangeList;
@@ -48,9 +49,8 @@ import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl.FileData;
import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl.StripeData;
import org.apache.hadoop.hive.llap.counters.LlapIOCounters;
import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
-import org.apache.hadoop.hive.llap.io.api.LlapIo;
import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
-import org.apache.hadoop.hive.llap.io.decode.GenericColumnVectorProducer.TextStripeMetadata;
+import org.apache.hadoop.hive.llap.io.decode.GenericColumnVectorProducer.SerDeStripeMetadata;
import org.apache.hadoop.hive.llap.io.decode.OrcEncodedDataConsumer;
import org.apache.hadoop.hive.ql.io.HdfsUtils;
import org.apache.hadoop.hive.ql.io.orc.OrcFile;
@@ -59,9 +59,11 @@ import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.ql.io.orc.WriterImpl;
import org.apache.hadoop.hive.ql.io.orc.encoded.CacheChunk;
import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.OrcEncodedColumnBatch;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputFormat;
@@ -137,12 +139,13 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private final SerDeLowLevelCacheImpl cache;
private final BufferUsageManager bufferManager;
- private final Configuration conf;
+ private final Configuration daemonConf;
private final FileSplit split;
private List<Integer> columnIds;
private final OrcEncodedDataConsumer consumer;
private final QueryFragmentCounters counters;
private final UserGroupInformation ugi;
+ private final Map<Path, PartitionDesc> parts;
private final Object fileKey;
private final FileSystem fs;
@@ -157,7 +160,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private final boolean isLrrEnabled;
private final boolean[] writerIncludes;
- private WriterImpl orcWriter = null;
+ private EncodingWriter writer = null;
private CacheWriter cacheWriter = null;
/**
* Data from cache currently being processed. We store it here so that we could decref
@@ -165,16 +168,19 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
* the consumer, at which point the consumer is responsible for it.
*/
private FileData cachedData;
-
+
public SerDeEncodedDataReader(SerDeLowLevelCacheImpl cache,
BufferUsageManager bufferManager, Configuration daemonConf, FileSplit split,
- List<Integer> columnIds, OrcEncodedDataConsumer consumer,
- JobConf jobConf, Reporter reporter, InputFormat<?, ?> sourceInputFormat,
- Deserializer sourceSerDe, QueryFragmentCounters counters, TypeDescription schema)
+ List<Integer> columnIds, OrcEncodedDataConsumer consumer, JobConf jobConf, Reporter reporter,
+ InputFormat<?, ?> sourceInputFormat, Deserializer sourceSerDe,
+ QueryFragmentCounters counters, TypeDescription schema, Map<Path, PartitionDesc> parts)
throws IOException {
this.cache = cache;
this.bufferManager = bufferManager;
- this.conf = daemonConf;
+ this.parts = parts;
+ this.daemonConf = new Configuration(daemonConf);
+ // Disable dictionary encoding for the writer.
+ this.daemonConf.setDouble(ConfVars.HIVE_ORC_DICTIONARY_KEY_SIZE_THRESHOLD.varname, 0);
this.split = split;
this.columnIds = columnIds;
this.allocSize = determineAllocSize(bufferManager, daemonConf);
@@ -327,13 +333,16 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
// These are global since ORC reuses objects between stripes.
private final Map<StreamName, OutStream> streams = new HashMap<>();
private final Map<Integer, List<CacheOutStream>> colStreams = new HashMap<>();
+ private final boolean doesSourceHaveIncludes;
- public CacheWriter(BufferUsageManager bufferManager, int bufferSize, List<Integer> columnIds,
- boolean[] writerIncludes) {
+ public CacheWriter(BufferUsageManager bufferManager, int bufferSize,
+ List<Integer> columnIds, boolean[] writerIncludes, boolean doesSourceHaveIncludes) {
this.bufferManager = bufferManager;
this.bufferSize = bufferSize;
this.columnIds = columnIds;
+ assert writerIncludes != null; // Taken care of on higher level.
this.writerIncludes = writerIncludes;
+ this.doesSourceHaveIncludes = doesSourceHaveIncludes;
startStripe();
}
@@ -359,10 +368,10 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
}
public void validateIncludes(OrcProto.Footer footer) throws IOException {
+ if (doesSourceHaveIncludes) return; // Irrelevant.
boolean[] translatedIncludes = columnIds == null ? null : OrcInputFormat.genIncludedColumns(
OrcUtils.convertTypeFromProtobuf(footer.getTypesList(), 0), columnIds);
- if (translatedIncludes == null && writerIncludes == null) return;
- if (translatedIncludes == null || writerIncludes == null) {
+ if (translatedIncludes == null) {
throwIncludesMismatchError(translatedIncludes);
}
int len = Math.min(translatedIncludes.length, writerIncludes.length);
@@ -465,16 +474,28 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
LlapIoImpl.LOG.trace(("Finalizing stripe " + footer.build() + " => " + si)
.replace('\n', ' '));
}
- currentStripe.encodings = new ArrayList<>(allEnc);
- for (int i = 0; i < currentStripe.encodings.size(); ++i) {
- // Don't record encodings for unneeded columns.
- if (writerIncludes == null || writerIncludes[i]) continue;
- currentStripe.encodings.set(i, null);
+ if (doesSourceHaveIncludes) {
+ currentStripe.encodings = new ArrayList<>(writerIncludes.length);
+ for (int i = 0; i < writerIncludes.length; ++i) {
+ currentStripe.encodings.add(null);
+ }
+ currentStripe.encodings.set(0, allEnc.get(0));
+ for (int i = 1; i < allEnc.size(); ++i) {
+ int colIx = getSparseOrcIndexFromDenseDest(i);
+ // LlapIoImpl.LOG.info("Setting enc " + i + "; " + colIx + " to " + allEnc.get(i));
+ currentStripe.encodings.set(colIx, allEnc.get(i));
+ }
+ } else {
+ currentStripe.encodings = new ArrayList<>(allEnc);
+ for (int i = 0; i < currentStripe.encodings.size(); ++i) {
+ // Don't record encodings for unneeded columns.
+ if (writerIncludes[i]) continue;
+ currentStripe.encodings.set(i, null);
+ }
}
currentStripe.rowCount = si.getNumberOfRows();
// ORC writer reuses streams, so we need to clean them here and extract data.
for (Map.Entry<Integer, List<CacheOutStream>> e : colStreams.entrySet()) {
- int colIx = e.getKey();
List<CacheOutStream> streams = e.getValue();
List<CacheStreamData> data = new ArrayList<>(streams.size());
for (CacheOutStream stream : streams) {
@@ -488,11 +509,27 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
buffers == null ? new ArrayList<MemoryBuffer>() : new ArrayList<>(buffers)));
stream.clear();
}
+ int colIx = e.getKey();
+ if (doesSourceHaveIncludes) {
+ int newColIx = getSparseOrcIndexFromDenseDest(colIx);
+ if (LlapIoImpl.LOG.isTraceEnabled()) {
+ LlapIoImpl.LOG.trace("Mapping the ORC writer column " + colIx + " to " + newColIx);
+ }
+ colIx = newColIx;
+ }
currentStripe.colStreams.put(colIx, data);
}
startStripe();
}
+ private int getSparseOrcIndexFromDenseDest(int denseColIx) {
+ // denseColIx is index in ORC writer with includes. We -1 to skip the root column; get the
+ // original text file index; then add the root column again. This makes many assumptions.
+ // Also this only works for primitive types; vectordeserializer only supports these anyway.
+ // The mapping for complex types with sub-cols in ORC would be much more difficult to build.
+ return columnIds.get(denseColIx - 1) + 1;
+ }
+
@Override
public long estimateMemory() {
return 0; // We never ever use any memory.
@@ -506,7 +543,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
}
private boolean isNeeded(StreamName name) {
- return writerIncludes == null || writerIncludes[name.getColumn()];
+ return doesSourceHaveIncludes || writerIncludes[name.getColumn()];
}
@Override
@@ -751,7 +788,8 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
// have happened, someone should have supplied a split that ends inside the last row, i.e.
// a few bytes earlier than the current split, which is pretty unlikely. What is more likely
// is that the split, and the last row, both end at the end of file. Check for this.
- long size = split.getPath().getFileSystem(conf).getFileStatus(split.getPath()).getLen();
+ long size = split.getPath().getFileSystem(
+ daemonConf).getFileStatus(split.getPath()).getLen();
isUnfortunate = size > endOfSplit;
if (isUnfortunate) {
// Log at warn, given how unfortunate this is.
@@ -835,7 +873,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
ColumnEncoding[] cacheEncodings = slice == null ? null : slice.getEncodings();
LlapDataBuffer[][][] cacheData = slice == null ? null : slice.getData();
long cacheRowCount = slice == null ? -1L : slice.getRowCount();
- TextStripeMetadata metadata = new TextStripeMetadata(stripeIx);
+ SerDeStripeMetadata metadata = new SerDeStripeMetadata(stripeIx);
StripeData sliceToCache = null;
boolean hasAllData = csd == null;
if (!hasAllData) {
@@ -979,9 +1017,15 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
offsetReader = createOffsetReader(sourceReader);
maySplitTheSplit = maySplitTheSplit && offsetReader.hasOffsets();
- // Column IDs are only used to verify eventual writer includes.
- cacheWriter = new CacheWriter(bufferManager, allocSize, columnIds, splitIncludes);
- orcWriter = new WriterImpl(cacheWriter, null, createOrcWriterOptions());
+ // writer writes to orcWriter which writes to cacheWriter
+ // TODO: in due course, writer will also propagate row batches if it's capable
+ StructObjectInspector originalOi = (StructObjectInspector)getOiFromSerDe();
+ writer = VertorDeserializeOrcWriter.create(sourceInputFormat, sourceSerDe, parts,
+ daemonConf, jobConf, split.getPath(), originalOi, columnIds);
+ cacheWriter = new CacheWriter(
+ bufferManager, allocSize, columnIds, splitIncludes, writer.hasIncludes());
+ writer.init(new WriterImpl(cacheWriter, null,
+ createOrcWriterOptions(writer.getDestinationOi())));
int rowsPerSlice = 0;
long currentKnownTornStart = split.getStart();
@@ -994,15 +1038,11 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
if (firstStartOffset == Long.MIN_VALUE) {
firstStartOffset = lastStartOffset;
}
- Object row = null;
- try {
- row = sourceSerDe.deserialize(value);
- } catch (SerDeException e) {
- throw new IOException(e);
- }
- orcWriter.addRow(row);
+ writer.writeOneRow(value);
+
if (maySplitTheSplit && ++rowsPerSlice == targetSliceRowCount) {
assert offsetReader.hasOffsets();
+ writer.flushIntermediateData();
long fileOffset = offsetReader.getCurrentRowEndOffset();
// Must support offsets to be able to split.
if (firstStartOffset < 0 || lastStartOffset < 0 || fileOffset < 0) {
@@ -1017,7 +1057,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
lastStartOffset = Long.MIN_VALUE;
firstStartOffset = Long.MIN_VALUE;
rowsPerSlice = 0;
- orcWriter.writeIntermediateFooter();
+ writer.writeIntermediateFooter();
}
}
if (rowsPerSlice > 0 || (!maySplitTheSplit && hasData)) {
@@ -1042,8 +1082,8 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
currentKnownTornStart, firstStartOffset, lastStartOffset, fileOffset);
}
// Close the writer to finalize the metadata. No catch since we cannot go on if this throws.
- orcWriter.close();
- orcWriter = null;
+ writer.close();
+ writer = null;
} finally {
// We don't need the source reader anymore.
if (offsetReader != null) {
@@ -1063,22 +1103,83 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
}
}
- private WriterOptions createOrcWriterOptions() throws IOException {
- ObjectInspector sourceOi;
- try {
- sourceOi = sourceSerDe.getObjectInspector();
- } catch (SerDeException e) {
- throw new IOException(e);
+ interface EncodingWriter {
+ void writeOneRow(Writable row) throws IOException;
+ StructObjectInspector getDestinationOi();
+ void init(WriterImpl orcWriter);
+ boolean hasIncludes();
+ void writeIntermediateFooter() throws IOException;
+ void flushIntermediateData() throws IOException;
+ void close() throws IOException;
+ }
+
+ static class DeserialerOrcWriter implements EncodingWriter {
+ private WriterImpl orcWriter;
+ private final Deserializer sourceSerDe;
+ private final StructObjectInspector sourceOi;
+
+ public DeserialerOrcWriter(Deserializer sourceSerDe, StructObjectInspector sourceOi) {
+ this.sourceSerDe = sourceSerDe;
+ this.sourceOi = sourceOi;
+ }
+
+ @Override
+ public void init(WriterImpl orcWriter) {
+ this.orcWriter = orcWriter;
+ }
+
+ @Override
+ public void close() throws IOException {
+ orcWriter.close();
+ }
+
+ @Override
+ public void writeOneRow(Writable value) throws IOException {
+ Object row = null;
+ try {
+ row = sourceSerDe.deserialize(value);
+ } catch (SerDeException e) {
+ throw new IOException(e);
+ }
+ orcWriter.addRow(row);
+ }
+
+ @Override
+ public void flushIntermediateData() {
+ // No-op.
+ }
+
+ @Override
+ public void writeIntermediateFooter() throws IOException {
+ orcWriter.writeIntermediateFooter();
}
- // TODO: ideally, we want to transform the rows to only have the included columns, and
- // only write those to the writer, with modified schema; then map back to full set later.
- return OrcFile.writerOptions(conf).stripeSize(Long.MAX_VALUE).blockSize(Long.MAX_VALUE)
+ @Override
+ public boolean hasIncludes() {
+ return false; // LazySimpleSerDe doesn't support projection.
+ }
+
+ @Override
+ public StructObjectInspector getDestinationOi() {
+ return sourceOi;
+ }
+ }
+
+ private WriterOptions createOrcWriterOptions(ObjectInspector sourceOi) throws IOException {
+ return OrcFile.writerOptions(daemonConf).stripeSize(Long.MAX_VALUE).blockSize(Long.MAX_VALUE)
.rowIndexStride(Integer.MAX_VALUE) // For now, do not limit this - one RG per split
.blockPadding(false).compress(CompressionKind.NONE).version(Version.CURRENT)
.encodingStrategy(EncodingStrategy.SPEED).bloomFilterColumns(null).inspector(sourceOi);
}
+ private ObjectInspector getOiFromSerDe() throws IOException {
+ try {
+ return sourceSerDe.getObjectInspector();
+ } catch (SerDeException e) {
+ throw new IOException(e);
+ }
+ }
+
private ReaderWithOffsets createOffsetReader(RecordReader sourceReader) {
if (LlapIoImpl.LOG.isDebugEnabled()) {
LlapIoImpl.LOG.debug("Using " + sourceReader.getClass().getSimpleName() + " to read data");
@@ -1125,10 +1226,10 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private void cleanupReaders() {
- if (orcWriter != null) {
+ if (writer != null) {
try {
- orcWriter.close();
- orcWriter = null;
+ writer.close();
+ writer = null;
} catch (Exception ex) {
LlapIoImpl.LOG.error("Failed to close ORC writer", ex);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VertorDeserializeOrcWriter.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VertorDeserializeOrcWriter.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VertorDeserializeOrcWriter.java
new file mode 100644
index 0000000..98fc9df
--- /dev/null
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VertorDeserializeOrcWriter.java
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.llap.io.encoded;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.llap.DebugUtils;
+import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
+import org.apache.hadoop.hive.llap.io.encoded.SerDeEncodedDataReader.DeserialerOrcWriter;
+import org.apache.hadoop.hive.llap.io.encoded.SerDeEncodedDataReader.EncodingWriter;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorDeserializeRow;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx;
+import org.apache.hadoop.hive.ql.io.HiveFileFormatUtils;
+import org.apache.hadoop.hive.ql.io.orc.WriterImpl;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
+import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.hive.serde2.Deserializer;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazy.LazySerDeParameters;
+import org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe;
+import org.apache.hadoop.hive.serde2.lazy.fast.LazySimpleDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazy.objectinspector.LazySimpleStructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
+import org.apache.hadoop.io.BinaryComparable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.TextInputFormat;
+
+/** The class that writes rows from a text reader to an ORC writer using VectorDeserializeRow. */
+class VertorDeserializeOrcWriter implements EncodingWriter {
+ private WriterImpl orcWriter;
+ private final LazySimpleDeserializeRead deserializeRead;
+ private final VectorDeserializeRow<?> vectorDeserializeRow;
+ private final VectorizedRowBatch sourceBatch, destinationBatch;
+ private final boolean hasIncludes;
+ private final StructObjectInspector destinationOi;
+
+ // TODO: if more writers are added, separate out an EncodingWriterFactory
+ public static EncodingWriter create(InputFormat<?, ?> sourceIf, Deserializer serDe,
+ Map<Path, PartitionDesc> parts, Configuration daemonConf,
+ Configuration jobConf, Path splitPath, StructObjectInspector sourceOi,
+ List<Integer> includes) throws IOException {
+ // Vector SerDe can be disabled both on client and server side.
+ if (!HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_IO_ENCODE_VECTOR_SERDE_ENABLED)
+ || !HiveConf.getBoolVar(jobConf, ConfVars.LLAP_IO_ENCODE_VECTOR_SERDE_ENABLED)
+ || !(sourceIf instanceof TextInputFormat) || !(serDe instanceof LazySimpleSerDe)) {
+ return new DeserialerOrcWriter(serDe, sourceOi);
+ }
+ Path path = splitPath.getFileSystem(daemonConf).makeQualified(splitPath);
+ PartitionDesc partDesc = HiveFileFormatUtils.getPartitionDescFromPathRecursively(
+ parts, path, null);
+ if (partDesc == null) {
+ LlapIoImpl.LOG.info("Not using VertorDeserializeOrcWriter: no partition desc for " + path);
+ return new DeserialerOrcWriter(serDe, sourceOi);
+ }
+ Properties tblProps = partDesc.getTableDesc().getProperties();
+ if ("true".equalsIgnoreCase(tblProps.getProperty(
+ serdeConstants.SERIALIZATION_LAST_COLUMN_TAKES_REST))) {
+ LlapIoImpl.LOG.info("Not using VertorDeserializeOrcWriter due to "
+ + serdeConstants.SERIALIZATION_LAST_COLUMN_TAKES_REST);
+ return new DeserialerOrcWriter(serDe, sourceOi);
+ }
+ for (StructField sf : sourceOi.getAllStructFieldRefs()) {
+ Category c = sf.getFieldObjectInspector().getCategory();
+ if (c != Category.PRIMITIVE) {
+ LlapIoImpl.LOG.info("Not using VertorDeserializeOrcWriter: " + c + " is not supported");
+ return new DeserialerOrcWriter(serDe, sourceOi);
+ }
+ }
+ LlapIoImpl.LOG.info("Creating VertorDeserializeOrcWriter for " + path);
+ return new VertorDeserializeOrcWriter(daemonConf, tblProps, sourceOi, includes);
+ }
+
+ private VertorDeserializeOrcWriter(Configuration conf, Properties tblProps,
+ StructObjectInspector sourceOi, List<Integer> columnIds) throws IOException {
+ // See also: the usage of VectorDeserializeType, for binary. For now, we only want text.
+ VectorizedRowBatchCtx vrbCtx = createVrbCtx(sourceOi);
+ this.sourceBatch = vrbCtx.createVectorizedRowBatch();
+ deserializeRead = new LazySimpleDeserializeRead(vrbCtx.getRowColumnTypeInfos(),
+ /* useExternalBuffer */ true, createSerdeParams(conf, tblProps));
+ vectorDeserializeRow = new VectorDeserializeRow<LazySimpleDeserializeRead>(deserializeRead);
+ int colCount = vrbCtx.getRowColumnTypeInfos().length;
+ boolean[] includes = null;
+ this.hasIncludes = columnIds.size() < colCount;
+ if (hasIncludes) {
+ // VectorDeserializeRow produces "sparse" VRB when includes are used; we need to write the
+ // "dense" VRB to ORC. Ideally, we'd use projection columns, but ORC writer doesn't use them.
+ // In any case, we would also need to build a new OI for OrcWriter config.
+ // This is why OrcWriter is created after this writer, by the way.
+ this.destinationBatch = new VectorizedRowBatch(columnIds.size());
+ includes = new boolean[colCount];
+ int inclBatchIx = 0;
+ List<String> childNames = new ArrayList<>(columnIds.size());
+ List<ObjectInspector> childOis = new ArrayList<>(columnIds.size());
+ List<? extends StructField> sourceFields = sourceOi.getAllStructFieldRefs();
+ for (Integer columnId : columnIds) {
+ includes[columnId] = true;
+ assert inclBatchIx <= columnId;
+ // Note that we use the same vectors in both batches. Clever, very clever.
+ destinationBatch.cols[inclBatchIx++] = sourceBatch.cols[columnId];
+ StructField sourceField = sourceFields.get(columnId);
+ childNames.add(sourceField.getFieldName());
+ childOis.add(sourceField.getFieldObjectInspector());
+ }
+ // This is only used by ORC to derive the structure. Most fields are unused.
+ destinationOi = new LazySimpleStructObjectInspector(
+ childNames, childOis, null, (byte)0, null);
+ destinationBatch.setPartitionInfo(columnIds.size(), 0);
+ if (LlapIoImpl.LOG.isDebugEnabled()) {
+ LlapIoImpl.LOG.debug("Includes for deserializer are " + DebugUtils.toString(includes));
+ }
+ try {
+ vectorDeserializeRow.init(includes);
+ } catch (HiveException e) {
+ throw new IOException(e);
+ }
+ } else {
+ // No includes - use the standard batch.
+ this.destinationBatch = sourceBatch;
+ this.destinationOi = sourceOi;
+ try {
+ vectorDeserializeRow.init();
+ } catch (HiveException e) {
+ throw new IOException(e);
+ }
+ }
+ }
+
+ private static VectorizedRowBatchCtx createVrbCtx(StructObjectInspector oi) throws IOException {
+ VectorizedRowBatchCtx vrbCtx = new VectorizedRowBatchCtx();
+ try {
+ vrbCtx.init(oi, new String[0]);
+ } catch (HiveException e) {
+ throw new IOException(e);
+ }
+ return vrbCtx;
+ }
+
+ private static LazySerDeParameters createSerdeParams(
+ Configuration conf, Properties tblProps) throws IOException {
+ try {
+ return new LazySerDeParameters(conf, tblProps, LazySimpleSerDe.class.getName());
+ } catch (SerDeException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public boolean hasIncludes() {
+ return hasIncludes;
+ }
+
+ @Override
+ public StructObjectInspector getDestinationOi() {
+ return destinationOi;
+ }
+
+ @Override
+ public void writeOneRow(Writable row) throws IOException {
+ if (sourceBatch.size == VectorizedRowBatch.DEFAULT_SIZE) {
+ flushBatch();
+ }
+
+ BinaryComparable binComp = (BinaryComparable)row;
+ deserializeRead.set(binComp.getBytes(), 0, binComp.getLength());
+
+ // Deserialize and append new row using the current batch size as the index.
+ try {
+ // TODO: can we use ByRef? Probably not, need to see text record reader.
+ vectorDeserializeRow.deserialize(sourceBatch, sourceBatch.size++);
+ } catch (Exception e) {
+ throw new IOException("DeserializeRead detail: "
+ + vectorDeserializeRow.getDetailedReadPositionString(), e);
+ }
+ }
+
+ private void flushBatch() throws IOException {
+ addBatchToWriter();
+
+ for (int c = 0; c < sourceBatch.cols.length; ++c) {
+ // This resets vectors in both batches.
+ ColumnVector colVector = sourceBatch.cols[c];
+ if (colVector != null) {
+ colVector.reset();
+ colVector.init();
+ }
+ }
+ sourceBatch.selectedInUse = false;
+ sourceBatch.size = 0;
+ sourceBatch.endOfFile = false;
+ propagateSourceBatchFieldsToDest();
+ }
+
+ private void propagateSourceBatchFieldsToDest() {
+ if (destinationBatch == sourceBatch) return;
+ destinationBatch.selectedInUse = sourceBatch.selectedInUse;
+ destinationBatch.size = sourceBatch.size;
+ destinationBatch.endOfFile = sourceBatch.endOfFile;
+ }
+
+ private void addBatchToWriter() throws IOException {
+ propagateSourceBatchFieldsToDest();
+ // LlapIoImpl.LOG.info("Writing includeOnlyBatch " + s + "; data "+ includeOnlyBatch);
+ orcWriter.addRowBatch(destinationBatch);
+ }
+
+ @Override
+ public void flushIntermediateData() throws IOException {
+ if (sourceBatch.size > 0) {
+ flushBatch();
+ }
+ }
+
+ @Override
+ public void writeIntermediateFooter() throws IOException {
+ orcWriter.writeIntermediateFooter();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (sourceBatch.size > 0) {
+ addBatchToWriter();
+ }
+ orcWriter.close();
+ }
+
+ @Override
+ public void init(WriterImpl orcWriter) {
+ this.orcWriter = orcWriter;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/orc/src/java/org/apache/orc/impl/WriterImpl.java
----------------------------------------------------------------------
diff --git a/orc/src/java/org/apache/orc/impl/WriterImpl.java b/orc/src/java/org/apache/orc/impl/WriterImpl.java
index 988d9d8..b1f3cfb 100644
--- a/orc/src/java/org/apache/orc/impl/WriterImpl.java
+++ b/orc/src/java/org/apache/orc/impl/WriterImpl.java
@@ -942,7 +942,7 @@ public class WriterImpl implements Writer, MemoryManager.Callback {
// If the number of keys in a dictionary is greater than this fraction of
//the total number of non-null rows, turn off dictionary encoding
private final double dictionaryKeySizeThreshold;
- protected boolean useDictionaryEncoding = true;
+ protected boolean useDictionaryEncoding;
private boolean isDirectV2 = true;
private boolean doneDictionaryCheck;
private final boolean strideDictionaryCheck;
@@ -970,7 +970,8 @@ public class WriterImpl implements Writer, MemoryManager.Callback {
OrcConf.DICTIONARY_KEY_SIZE_THRESHOLD.getDouble(conf);
strideDictionaryCheck =
OrcConf.ROW_INDEX_STRIDE_DICTIONARY_CHECK.getBoolean(conf);
- doneDictionaryCheck = false;
+ useDictionaryEncoding = dictionaryKeySizeThreshold >= 0.000001; // Epsilon.
+ doneDictionaryCheck = !useDictionaryEncoding;
}
private boolean checkDictionaryEncoding() {
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java b/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java
index 3ee8fdc..51530ac 100755
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java
@@ -217,12 +217,14 @@ public class HiveInputFormat<K extends WritableComparable, V extends Writable>
boolean isSupported = inputFormat instanceof LlapWrappableInputFormatInterface;
boolean isVectorized = Utilities.getUseVectorizedInputFileFormat(conf);
if (!isVectorized) {
- // Pretend it's vectorized.
+ // Pretend it's vectorized if the non-vector wrapped is enabled.
isVectorized = HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_NONVECTOR_WRAPPER_ENABLED)
&& (Utilities.getPlanPath(conf) != null);
}
boolean isSerdeBased = false;
- if (isVectorized && !isSupported) {
+ if (isVectorized && !isSupported
+ && HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENCODE_ENABLED)) {
+ // See if we can use re-encoding to read the format thru IO elevator.
String formatList = HiveConf.getVar(conf, ConfVars.LLAP_IO_ENCODE_FORMATS);
if (LOG.isDebugEnabled()) {
LOG.debug("Checking " + ifName + " against " + formatList);
@@ -247,6 +249,7 @@ public class HiveInputFormat<K extends WritableComparable, V extends Writable>
if (LOG.isDebugEnabled()) {
LOG.debug("Wrapping " + ifName);
}
+
@SuppressWarnings("unchecked")
LlapIo<VectorizedRowBatch> llapIo = LlapProxy.getIo();
if (llapIo == null) {
@@ -275,7 +278,11 @@ public class HiveInputFormat<K extends WritableComparable, V extends Writable>
throw new HiveException("Error creating SerDe for LLAP IO", e);
}
}
- return castInputFormat(llapIo.getInputFormat(inputFormat, serde));
+ InputFormat<?, ?> wrappedIf = llapIo.getInputFormat(inputFormat, serde);
+ if (wrappedIf == null) {
+ return inputFormat; // We cannot wrap; the cause is logged inside.
+ }
+ return castInputFormat(wrappedIf);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java
index e3158f6..601324a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java
@@ -245,7 +245,6 @@ public class MapWork extends BaseWork {
}
public void deriveLlap(Configuration conf, boolean isExecDriver) {
- // TODO# HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_NONVECTOR_WRAPPER_ENABLED)
boolean hasLlap = false, hasNonLlap = false, hasAcid = false;
// Assume the IO is enabled on the daemon by default. We cannot reasonably check it here.
boolean isLlapOn = HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENABLED, llapMode);
http://git-wip-us.apache.org/repos/asf/hive/blob/881deac6/storage-api/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatch.java
----------------------------------------------------------------------
diff --git a/storage-api/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatch.java b/storage-api/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatch.java
index 9c066e0..0235ffc 100644
--- a/storage-api/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatch.java
+++ b/storage-api/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatch.java
@@ -164,7 +164,11 @@ public class VectorizedRowBatch implements Writable {
b.append(", ");
}
if (cv != null) {
- cv.stringifyValue(b, i);
+ try {
+ cv.stringifyValue(b, i);
+ } catch (Exception ex) {
+ b.append("<invalid>");
+ }
}
}
b.append(']');