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/09/15 19:14:09 UTC
[1/3] hive git commit: HIVE-15665 : LLAP: OrcFileMetadata objects in
cache can impact heap usage (Sergey Shelukhin,
reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/master 4a4ae12ca -> 50fb6f3cb
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/ql/src/test/results/clientpositive/llap/orc_ppd_schema_evol_3a.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientpositive/llap/orc_ppd_schema_evol_3a.q.out b/ql/src/test/results/clientpositive/llap/orc_ppd_schema_evol_3a.q.out
index b799527..1157edb 100644
--- a/ql/src/test/results/clientpositive/llap/orc_ppd_schema_evol_3a.q.out
+++ b/ql/src/test/results/clientpositive/llap/orc_ppd_schema_evol_3a.q.out
@@ -215,9 +215,10 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 524288
- ALLOCATED_USED_BYTES: 269
- CACHE_MISS_BYTES: 249
+ ALLOCATED_BYTES: 1048576
+ ALLOCATED_USED_BYTES: 382
+ CACHE_HIT_BYTES: 0
+ CACHE_MISS_BYTES: 359
METADATA_CACHE_MISS: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
@@ -255,7 +256,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -268,9 +269,9 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 0
+ HDFS_BYTES_READ: 720
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 2
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -280,8 +281,10 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
- CACHE_MISS_BYTES: 0
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 2376
+ CACHE_HIT_BYTES: 359
+ CACHE_MISS_BYTES: 720
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 1
@@ -305,7 +308,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -330,7 +333,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -359,7 +362,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -398,7 +401,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -423,7 +426,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -448,7 +451,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -473,7 +476,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -502,7 +505,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -541,7 +544,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -566,7 +569,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -591,7 +594,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -616,7 +619,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -645,7 +648,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -684,7 +687,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -709,7 +712,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -734,7 +737,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 359
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -759,7 +762,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -890,9 +893,9 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 5287
+ HDFS_BYTES_READ: 4912
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 4
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -902,12 +905,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 262144
- ALLOCATED_USED_BYTES: 8400
- CACHE_HIT_BYTES: 0
- CACHE_MISS_BYTES: 4809
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 524288
+ ALLOCATED_USED_BYTES: 8527
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 4912
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
ROWS_EMITTED: 2100
@@ -918,25 +920,27 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 0
+ HDFS_BYTES_READ: 1751
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 2
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
CREATED_FILES: 1
DESERIALIZE_ERRORS: 0
- RECORDS_IN_Map_1: 2100
+ RECORDS_IN_Map_1: 100
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4809
- CACHE_MISS_BYTES: 0
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 2376
+ CACHE_HIT_BYTES: 4936
+ CACHE_MISS_BYTES: 1751
METADATA_CACHE_HIT: 2
- NUM_DECODED_BATCHES: 3
- NUM_VECTOR_BATCHES: 3
- ROWS_EMITTED: 2100
- SELECTED_ROWGROUPS: 3
+ NUM_DECODED_BATCHES: 1
+ NUM_VECTOR_BATCHES: 1
+ ROWS_EMITTED: 100
+ SELECTED_ROWGROUPS: 1
2
PREHOOK: query: alter table orc_ppd change column f f double
PREHOOK: type: ALTERTABLE_RENAMECOL
@@ -1027,9 +1031,9 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 4577
+ HDFS_BYTES_READ: 4099
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 4
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -1039,12 +1043,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 786432
- ALLOCATED_USED_BYTES: 11299
- CACHE_HIT_BYTES: 0
- CACHE_MISS_BYTES: 3980
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 1048576
+ ALLOCATED_USED_BYTES: 11420
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 4099
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
ROWS_EMITTED: 2100
@@ -1055,9 +1058,9 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 0
+ HDFS_BYTES_READ: 1592
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 2
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -1067,8 +1070,10 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
- CACHE_MISS_BYTES: 0
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 2376
+ CACHE_HIT_BYTES: 4123
+ CACHE_MISS_BYTES: 1592
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
@@ -1138,7 +1143,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 4123
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -1163,7 +1168,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -1234,7 +1239,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 4123
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -1259,7 +1264,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -1276,9 +1281,9 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 2780
+ HDFS_BYTES_READ: 2183
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 4
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -1288,12 +1293,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 4
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 524288
- ALLOCATED_USED_BYTES: 4201
- CACHE_HIT_BYTES: 0
- CACHE_MISS_BYTES: 2098
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 786432
+ ALLOCATED_USED_BYTES: 4293
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 2183
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
ROWS_EMITTED: 2100
@@ -1323,25 +1327,27 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 0
+ HDFS_BYTES_READ: 1217
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 2
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
CREATED_FILES: 1
DESERIALIZE_ERRORS: 0
- RECORDS_IN_Map_1: 2100
+ RECORDS_IN_Map_1: 1000
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 4
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 2098
- CACHE_MISS_BYTES: 0
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 2376
+ CACHE_HIT_BYTES: 2207
+ CACHE_MISS_BYTES: 1217
METADATA_CACHE_HIT: 2
- NUM_DECODED_BATCHES: 3
- NUM_VECTOR_BATCHES: 3
- ROWS_EMITTED: 2100
- SELECTED_ROWGROUPS: 3
+ NUM_DECODED_BATCHES: 1
+ NUM_VECTOR_BATCHES: 1
+ ROWS_EMITTED: 1000
+ SELECTED_ROWGROUPS: 1
4
PREHOOK: query: select count(*) from orc_ppd where si = 442 or boo is not null or boo = false
PREHOOK: type: QUERY
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
----------------------------------------------------------------------
diff --git a/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java b/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
index 403c3ad..e684ece 100644
--- a/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
+++ b/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hive.common.io;
+
+import java.nio.ByteBuffer;
import java.io.IOException;
import java.io.InputStream;
@@ -30,8 +32,6 @@ public interface FileMetadataCache {
*/
MemoryBufferOrBuffers getFileMetadata(Object fileKey);
- // TODO: add BB put method(s) when merging with ORC off-heap metadata cache
-
/**
* Puts the metadata for a given file (e.g. a footer buffer into cache).
* @param fileKey The file key.
@@ -43,9 +43,11 @@ public interface FileMetadataCache {
MemoryBufferOrBuffers putFileMetadata(
Object fileKey, int length, InputStream is) throws IOException;
+ MemoryBufferOrBuffers putFileMetadata(Object fileKey, ByteBuffer tailBuffer);
+
/**
* Releases the buffer returned from getFileMetadata or putFileMetadata method.
* @param buffer The buffer to release.
*/
void decRefBuffer(MemoryBufferOrBuffers buffer);
-}
\ No newline at end of file
+}
\ No newline at end of file
[2/3] hive git commit: HIVE-15665 : LLAP: OrcFileMetadata objects in
cache can impact heap usage (Sergey Shelukhin,
reviewed by Prasanth Jayachandran)
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ParquetMetadataCacheImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ParquetMetadataCacheImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ParquetMetadataCacheImpl.java
deleted file mode 100644
index b61a8ca..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ParquetMetadataCacheImpl.java
+++ /dev/null
@@ -1,353 +0,0 @@
-/**
- * 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.metadata;
-
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.hadoop.hive.common.FileUtils;
-import org.apache.hadoop.hive.common.io.FileMetadataCache;
-import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
-import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
-import org.apache.hadoop.hive.llap.cache.EvictionAwareAllocator;
-import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
-import org.apache.hadoop.hive.llap.cache.LlapAllocatorBuffer;
-import org.apache.hadoop.hive.llap.cache.LlapOomDebugDump;
-import org.apache.hadoop.hive.llap.cache.LowLevelCachePolicy;
-import org.apache.hadoop.hive.llap.cache.MemoryManager;
-import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
-import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
-import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
-import org.apache.parquet.io.SeekableInputStream;
-
-// TODO# merge with OrcMetadataCache (and rename) after HIVE-15665. Shares a lot of the code.
-public class ParquetMetadataCacheImpl implements LlapOomDebugDump, FileMetadataCache {
- private final ConcurrentHashMap<Object, LlapBufferOrBuffers> metadata =
- new ConcurrentHashMap<>();
-
- private final MemoryManager memoryManager;
- private final LowLevelCachePolicy policy;
- private final EvictionAwareAllocator allocator;
- private final LlapDaemonCacheMetrics metrics;
-
- public ParquetMetadataCacheImpl(EvictionAwareAllocator allocator, MemoryManager memoryManager,
- LowLevelCachePolicy policy, LlapDaemonCacheMetrics metrics) {
- this.memoryManager = memoryManager;
- this.allocator = allocator;
- this.policy = policy;
- this.metrics = metrics;
- }
-
- public void notifyEvicted(LlapFileMetadataBuffer buffer) {
- LlapBufferOrBuffers removed = metadata.remove(buffer.getFileKey());
- if (removed == null) return;
- if (removed.getSingleBuffer() != null) {
- assert removed.getSingleBuffer() == buffer;
- return;
- }
- discardMultiBuffer(removed);
- }
-
- @Override
- public LlapBufferOrBuffers getFileMetadata(Object fileKey) {
- LlapBufferOrBuffers result = metadata.get(fileKey);
- if (result == null) return null;
- if (!lockBuffer(result, true)) {
- // No need to discard the buffer we cannot lock - eviction takes care of that.
- metadata.remove(fileKey, result);
- return null;
- }
- return result;
- }
-
- @Override
- public LlapBufferOrBuffers putFileMetadata(
- Object fileKey, int length, InputStream is) throws IOException {
- LlapBufferOrBuffers result = null;
- while (true) { // Overwhelmingly executes once, or maybe twice (replacing stale value).
- LlapBufferOrBuffers oldVal = metadata.get(fileKey);
- if (oldVal == null) {
- result = wrapBbForFile(result, fileKey, length, is);
- if (!lockBuffer(result, false)) {
- throw new AssertionError("Cannot lock a newly created value " + result);
- }
- oldVal = metadata.putIfAbsent(fileKey, result);
- if (oldVal == null) {
- cacheInPolicy(result); // Cached successfully, add to policy.
- return result;
- }
- }
- if (lockOldVal(fileKey, result, oldVal)) {
- return oldVal;
- }
- // We found some old value but couldn't incRef it; remove it.
- metadata.remove(fileKey, oldVal);
- }
- }
-
- private void cacheInPolicy(LlapBufferOrBuffers buffers) {
- LlapAllocatorBuffer singleBuffer = buffers.getSingleLlapBuffer();
- if (singleBuffer != null) {
- policy.cache(singleBuffer, Priority.HIGH);
- return;
- }
- for (LlapAllocatorBuffer buffer : buffers.getMultipleLlapBuffers()) {
- policy.cache(buffer, Priority.HIGH);
- }
- }
-
- private <T extends LlapBufferOrBuffers> boolean lockOldVal(Object key, T newVal, T oldVal) {
- if (LlapIoImpl.CACHE_LOGGER.isTraceEnabled()) {
- LlapIoImpl.CACHE_LOGGER.trace("Trying to cache when metadata is already cached for" +
- " {}; old {}, new {}", key, oldVal, newVal);
- }
- if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
- LlapIoImpl.LOCKING_LOGGER.trace("Locking {} due to cache collision", oldVal);
- }
- if (lockBuffer(oldVal, true)) {
- // We found an old, valid block for this key in the cache.
- if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
- LlapIoImpl.LOCKING_LOGGER.trace("Unlocking {} due to cache collision with {}",
- newVal, oldVal);
- }
-
- if (newVal != null) {
- unlockBuffer(newVal, false);
- }
- return true;
- }
- return false;
- }
-
- @Override
- public void decRefBuffer(MemoryBufferOrBuffers buffer) {
- if (!(buffer instanceof LlapBufferOrBuffers)) {
- throw new AssertionError(buffer.getClass());
- }
- unlockBuffer((LlapBufferOrBuffers)buffer, true);
- }
-
- private LlapBufferOrBuffers wrapBbForFile(LlapBufferOrBuffers result,
- Object fileKey, int length, InputStream stream) throws IOException {
- if (result != null) return result;
- int maxAlloc = allocator.getMaxAllocation();
- LlapFileMetadataBuffer[] largeBuffers = null;
- if (maxAlloc < length) {
- largeBuffers = new LlapFileMetadataBuffer[length / maxAlloc];
- for (int i = 0; i < largeBuffers.length; ++i) {
- largeBuffers[i] = new LlapFileMetadataBuffer(fileKey);
- }
- allocator.allocateMultiple(largeBuffers, maxAlloc, null);
- for (int i = 0; i < largeBuffers.length; ++i) {
- readIntoCacheBuffer(stream, maxAlloc, largeBuffers[i]);
- }
- }
- int smallSize = length % maxAlloc;
- if (smallSize == 0) {
- return new LlapFileMetadataBuffers(largeBuffers);
- } else {
- LlapFileMetadataBuffer[] smallBuffer = new LlapFileMetadataBuffer[1];
- smallBuffer[0] = new LlapFileMetadataBuffer(fileKey);
- allocator.allocateMultiple(smallBuffer, length, null);
- readIntoCacheBuffer(stream, smallSize, smallBuffer[0]);
- if (largeBuffers == null) {
- return smallBuffer[0];
- } else {
- LlapFileMetadataBuffer[] cacheData = new LlapFileMetadataBuffer[largeBuffers.length + 1];
- System.arraycopy(largeBuffers, 0, cacheData, 0, largeBuffers.length);
- cacheData[largeBuffers.length] = smallBuffer[0];
- return new LlapFileMetadataBuffers(largeBuffers);
- }
- }
- }
-
- private static void readIntoCacheBuffer(
- InputStream stream, int length, MemoryBuffer dest) throws IOException {
- ByteBuffer bb = dest.getByteBufferRaw();
- int pos = bb.position();
- bb.limit(pos + length);
- // TODO: SeekableInputStream.readFully eventually calls a Hadoop method that used to be
- // buggy in 2.7 and also anyway just does a copy for a direct buffer. Do a copy here.
- // ((SeekableInputStream)stream).readFully(bb);
- FileUtils.readFully(stream, length, bb);
- bb.position(pos);
- }
-
-
- private boolean lockBuffer(LlapBufferOrBuffers buffers, boolean doNotifyPolicy) {
- LlapAllocatorBuffer buffer = buffers.getSingleLlapBuffer();
- if (buffer != null) {
- return lockOneBuffer(buffer, doNotifyPolicy);
- }
- LlapAllocatorBuffer[] bufferArray = buffers.getMultipleLlapBuffers();
- for (int i = 0; i < bufferArray.length; ++i) {
- if (lockOneBuffer(bufferArray[i], doNotifyPolicy)) continue;
- for (int j = 0; j < i; ++j) {
- unlockSingleBuffer(buffer, true);
- }
- discardMultiBuffer(buffers);
- return false;
- }
- return true;
- }
-
- private void discardMultiBuffer(LlapBufferOrBuffers removed) {
- long memoryFreed = 0;
- for (LlapAllocatorBuffer buf : removed.getMultipleLlapBuffers()) {
- long memUsage = buf.getMemoryUsage();
- // We cannot just deallocate the buffer, as it can hypothetically have users.
- int result = buf.invalidate();
- switch (result) {
- case LlapAllocatorBuffer.INVALIDATE_ALREADY_INVALID: continue; // Nothing to do.
- case LlapAllocatorBuffer.INVALIDATE_FAILED: {
- // Someone is using this buffer; eventually, it will be evicted.
- continue;
- }
- case LlapAllocatorBuffer.INVALIDATE_OK: {
- memoryFreed += memUsage;
- allocator.deallocateEvicted(buf);
- break;
- }
- default: throw new AssertionError(result);
- }
- }
- memoryManager.releaseMemory(memoryFreed);
- }
-
- private boolean lockOneBuffer(LlapAllocatorBuffer buffer, boolean doNotifyPolicy) {
- int rc = buffer.incRef();
- if (rc > 0) {
- metrics.incrCacheNumLockedBuffers();
- }
- if (doNotifyPolicy && rc == 1) {
- // We have just locked a buffer that wasn't previously locked.
- policy.notifyLock(buffer);
- }
- return rc > 0;
- }
-
- private void unlockBuffer(LlapBufferOrBuffers buffers, boolean isCached) {
- LlapAllocatorBuffer singleBuffer = buffers.getSingleLlapBuffer();
- if (singleBuffer != null) {
- unlockSingleBuffer(singleBuffer, isCached);
- return;
- }
- for (LlapAllocatorBuffer buffer : buffers.getMultipleLlapBuffers()) {
- unlockSingleBuffer(buffer, isCached);
- }
- }
-
- private void unlockSingleBuffer(LlapAllocatorBuffer buffer, boolean isCached) {
- boolean isLastDecref = (buffer.decRef() == 0);
- if (isLastDecref) {
- if (isCached) {
- policy.notifyUnlock(buffer);
- } else {
- allocator.deallocate(buffer);
- }
- }
- metrics.decrCacheNumLockedBuffers();
- }
-
-
- public static interface LlapBufferOrBuffers extends MemoryBufferOrBuffers {
- LlapAllocatorBuffer getSingleLlapBuffer();
- LlapAllocatorBuffer[] getMultipleLlapBuffers();
- }
-
- public final static class LlapFileMetadataBuffer
- extends LlapAllocatorBuffer implements LlapBufferOrBuffers {
- private final Object fileKey;
-
- public LlapFileMetadataBuffer(Object fileKey) {
- this.fileKey = fileKey;
- }
-
- @Override
- public void notifyEvicted(EvictionDispatcher evictionDispatcher) {
- evictionDispatcher.notifyEvicted(this);
- }
-
- public Object getFileKey() {
- return fileKey;
- }
-
- @Override
- public LlapAllocatorBuffer getSingleLlapBuffer() {
- return this;
- }
-
- @Override
- public LlapAllocatorBuffer[] getMultipleLlapBuffers() {
- return null;
- }
-
- @Override
- public MemoryBuffer getSingleBuffer() {
- return this;
- }
-
- @Override
- public MemoryBuffer[] getMultipleBuffers() {
- return null;
- }
- }
-
- public final static class LlapFileMetadataBuffers implements LlapBufferOrBuffers {
- private final LlapFileMetadataBuffer[] buffers;
-
- public LlapFileMetadataBuffers(LlapFileMetadataBuffer[] buffers) {
- this.buffers = buffers;
- }
-
- @Override
- public LlapAllocatorBuffer getSingleLlapBuffer() {
- return null;
- }
-
- @Override
- public LlapAllocatorBuffer[] getMultipleLlapBuffers() {
- return buffers;
- }
-
- @Override
- public MemoryBuffer getSingleBuffer() {
- return null;
- }
-
- @Override
- public MemoryBuffer[] getMultipleBuffers() {
- return buffers;
- }
- }
-
- @Override
- public String debugDumpForOom() {
- // TODO: nothing, will be merged with ORC cache
- return null;
- }
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- // TODO: nothing, will be merged with ORC cache
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestIncrementalObjectSizeEstimator.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestIncrementalObjectSizeEstimator.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestIncrementalObjectSizeEstimator.java
index 13c7767..1ec7020 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestIncrementalObjectSizeEstimator.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestIncrementalObjectSizeEstimator.java
@@ -160,7 +160,7 @@ public class TestIncrementalObjectSizeEstimator {
public void close() throws IOException {
}
}
-
+/*
@Test
public void testMetadata() throws IOException {
// Mostly tests that it doesn't crash.
@@ -202,7 +202,7 @@ public class TestIncrementalObjectSizeEstimator {
root = map.get(OrcFileMetadata.class);
LOG.info("Estimated " + root.estimate(ofm, map) + " for a dummy OFM");
- }
+ }*/
private static class Struct {
Integer i;
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
index 03a955c..3ebb4b0 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
@@ -19,10 +19,18 @@ package org.apache.hadoop.hive.llap.cache;
import static org.junit.Assert.*;
+import java.nio.ByteBuffer;
+import java.util.Random;
+
+import org.apache.hadoop.hive.common.io.DataCache;
+import org.apache.hadoop.hive.common.io.DiskRange;
+import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
-import org.apache.hadoop.hive.llap.io.metadata.OrcFileMetadata;
-import org.apache.hadoop.hive.llap.io.metadata.OrcMetadataCache;
-import org.apache.hadoop.hive.llap.io.metadata.OrcStripeMetadata;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache.LlapBufferOrBuffers;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache.LlapMetadataBuffer;
+import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
+import org.apache.hadoop.hive.ql.io.orc.encoded.IncompleteCb;
import org.junit.Test;
public class TestOrcMetadataCache {
@@ -66,16 +74,12 @@ public class TestOrcMetadataCache {
}
private static class DummyMemoryManager implements MemoryManager {
- int allocs = 0;
-
@Override
public void reserveMemory(long memoryToReserve) {
- ++allocs;
}
@Override
public void releaseMemory(long memUsage) {
- --allocs;
}
@Override
@@ -93,38 +97,119 @@ public class TestOrcMetadataCache {
}
@Test
- public void testGetPut() throws Exception {
+ public void testBuffers() throws Exception {
+ DummyMemoryManager mm = new DummyMemoryManager();
+ DummyCachePolicy cp = new DummyCachePolicy();
+ final int MAX_ALLOC = 64;
+ LlapDaemonCacheMetrics metrics = LlapDaemonCacheMetrics.create("", "");
+ BuddyAllocator alloc = new BuddyAllocator(
+ false, false, 8, MAX_ALLOC, 1, 4096, 0, null, mm, metrics, null);
+ MetadataCache cache = new MetadataCache(alloc, mm, cp, true, metrics);
+ Object fileKey1 = new Object();
+ Random rdm = new Random();
+
+ ByteBuffer smallBuffer = ByteBuffer.allocate(MAX_ALLOC - 1);
+ rdm.nextBytes(smallBuffer.array());
+ LlapBufferOrBuffers result = cache.putFileMetadata(fileKey1, smallBuffer);
+ cache.decRefBuffer(result);
+ ByteBuffer cacheBuf = result.getSingleBuffer().getByteBufferDup();
+ assertEquals(smallBuffer, cacheBuf);
+ result = cache.putFileMetadata(fileKey1, smallBuffer);
+ cache.decRefBuffer(result);
+ cacheBuf = result.getSingleBuffer().getByteBufferDup();
+ assertEquals(smallBuffer, cacheBuf);
+ result = cache.getFileMetadata(fileKey1);
+ cacheBuf = result.getSingleBuffer().getByteBufferDup();
+ assertEquals(smallBuffer, cacheBuf);
+ cache.decRefBuffer(result);
+ cache.notifyEvicted((LlapMetadataBuffer<?>) result.getSingleBuffer());
+ result = cache.getFileMetadata(fileKey1);
+ assertNull(result);
+
+ ByteBuffer largeBuffer = ByteBuffer.allocate((int)(MAX_ALLOC * 2.5));
+ rdm.nextBytes(largeBuffer.array());
+ result = cache.putFileMetadata(fileKey1, largeBuffer);
+ cache.decRefBuffer(result);
+ assertNull(result.getSingleBuffer());
+ assertEquals(largeBuffer, extractResultBbs(result));
+ result = cache.getFileMetadata(fileKey1);
+ assertNull(result.getSingleBuffer());
+ assertEquals(largeBuffer, extractResultBbs(result));
+ LlapAllocatorBuffer b0 = result.getMultipleLlapBuffers()[0],
+ b1 = result.getMultipleLlapBuffers()[1];
+ cache.decRefBuffer(result);
+ cache.notifyEvicted((LlapMetadataBuffer<?>) b1);
+ result = cache.getFileMetadata(fileKey1);
+ assertNull(result);
+ assertFalse(b0.incRef() > 0); // Should have also been thrown out.
+ }
+
+ public ByteBuffer extractResultBbs(LlapBufferOrBuffers result) {
+ int totalLen = 0;
+ for (LlapAllocatorBuffer buf : result.getMultipleLlapBuffers()) {
+ totalLen += buf.getByteBufferRaw().remaining();
+ }
+ ByteBuffer combinedBb = ByteBuffer.allocate(totalLen);
+ for (LlapAllocatorBuffer buf : result.getMultipleLlapBuffers()) {
+ combinedBb.put(buf.getByteBufferDup());
+ }
+ combinedBb.flip();
+ return combinedBb;
+ }
+
+ @Test
+ public void testIncompleteCbs() throws Exception {
DummyMemoryManager mm = new DummyMemoryManager();
DummyCachePolicy cp = new DummyCachePolicy();
- OrcMetadataCache cache = new OrcMetadataCache(mm, cp, false);
- OrcFileMetadata ofm1 = OrcFileMetadata.createDummy(1), ofm2 = OrcFileMetadata.createDummy(2);
- assertSame(ofm1, cache.putFileMetadata(ofm1));
- assertEquals(1, mm.allocs);
+ final int MAX_ALLOC = 64;
+ LlapDaemonCacheMetrics metrics = LlapDaemonCacheMetrics.create("", "");
+ BuddyAllocator alloc = new BuddyAllocator(
+ false, false, 8, MAX_ALLOC, 1, 4096, 0, null, mm, metrics, null);
+ MetadataCache cache = new MetadataCache(alloc, mm, cp, true, metrics);
+ DataCache.BooleanRef gotAllData = new DataCache.BooleanRef();
+ Object fileKey1 = new Object();
+
+ // Note: incomplete CBs are always an exact match.
+ cache.putIncompleteCbs(fileKey1, new DiskRange[] { new DiskRangeList(0, 3) }, 0);
cp.verifyEquals(1);
- assertSame(ofm2, cache.putFileMetadata(ofm2));
- assertEquals(2, mm.allocs);
- cp.verifyEquals(2);
- assertSame(ofm1, cache.getFileMetadata(1));
- assertSame(ofm2, cache.getFileMetadata(2));
- cp.verifyEquals(4);
- OrcFileMetadata ofm3 = OrcFileMetadata.createDummy(1);
- assertSame(ofm1, cache.putFileMetadata(ofm3));
- assertEquals(2, mm.allocs);
- cp.verifyEquals(5);
- assertSame(ofm1, cache.getFileMetadata(1));
- cp.verifyEquals(6);
-
- OrcStripeMetadata osm1 = OrcStripeMetadata.createDummy(1), osm2 = OrcStripeMetadata.createDummy(2);
- assertSame(osm1, cache.putStripeMetadata(osm1));
- assertEquals(3, mm.allocs);
- assertSame(osm2, cache.putStripeMetadata(osm2));
- assertEquals(4, mm.allocs);
- assertSame(osm1, cache.getStripeMetadata(osm1.getKey()));
- assertSame(osm2, cache.getStripeMetadata(osm2.getKey()));
- OrcStripeMetadata osm3 = OrcStripeMetadata.createDummy(1);
- assertSame(osm1, cache.putStripeMetadata(osm3));
- assertEquals(4, mm.allocs);
- assertSame(osm1, cache.getStripeMetadata(osm3.getKey()));
- cp.verifyEquals(12);
+ DiskRangeList result = cache.getIncompleteCbs(
+ fileKey1, new DiskRangeList(0, 3), 0, gotAllData);
+ assertTrue(gotAllData.value);
+ verifyResult(result, INCOMPLETE, 0, 3);
+ cache.putIncompleteCbs(fileKey1, new DiskRange[] { new DiskRangeList(5, 6) }, 0);
+ cp.verifyEquals(3);
+ DiskRangeList ranges = new DiskRangeList(0, 3);
+ ranges.insertAfter(new DiskRangeList(4, 6));
+ result = cache.getIncompleteCbs(fileKey1, ranges, 0, gotAllData);
+ assertFalse(gotAllData.value);
+ verifyResult(result, INCOMPLETE, 0, 3, DRL, 4, 6);
+ ranges = new DiskRangeList(0, 3);
+ ranges.insertAfter(new DiskRangeList(3, 5)).insertAfter(new DiskRangeList(5, 6));
+ result = cache.getIncompleteCbs(fileKey1, ranges, 0, gotAllData);
+ assertFalse(gotAllData.value);
+ verifyResult(result, INCOMPLETE, 0, 3, DRL, 3, 5, INCOMPLETE, 5, 6);
+ result = cache.getIncompleteCbs(fileKey1, new DiskRangeList(5, 6), 0, gotAllData);
+ assertTrue(gotAllData.value);
+ verifyResult(result, INCOMPLETE, 5, 6);
+ result = cache.getIncompleteCbs(fileKey1, new DiskRangeList(4, 5), 0, gotAllData);
+ assertFalse(gotAllData.value);
+ verifyResult(result, DRL, 4, 5);
+ }
+
+ private static final int INCOMPLETE = 0, DRL = 1;
+ public void verifyResult(DiskRangeList result, long... vals) {
+ for (int i = 0; i < vals.length; i += 3) {
+ switch ((int)vals[i]) {
+ case INCOMPLETE: assertTrue(result instanceof IncompleteCb); break;
+ case DRL: assertFalse(result instanceof IncompleteCb); break;
+ default: fail();
+ }
+ assertEquals(vals[i + 1], result.getOffset());
+ assertEquals(vals[i + 2], result.getEnd());
+ result = result.next;
+ }
+ assertNull(result);
}
+
+
}
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 69a9f9f..c5d9b7b 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
@@ -112,6 +112,7 @@ import org.apache.hadoop.util.StringUtils;
import org.apache.hive.common.util.Ref;
import org.apache.orc.ColumnStatistics;
import org.apache.orc.OrcProto;
+import org.apache.orc.OrcProto.Footer;
import org.apache.orc.OrcUtils;
import org.apache.orc.StripeInformation;
import org.apache.orc.StripeStatistics;
@@ -330,6 +331,16 @@ public class OrcInputFormat implements InputFormat<NullWritable, OrcStruct>,
return !file.hasMetadataValue(OrcRecordUpdater.ACID_KEY_INDEX_NAME);
}
+ public static boolean isOriginal(Footer footer) {
+ for(OrcProto.UserMetadataItem item: footer.getMetadataList()) {
+ if (item.hasName() && item.getName().equals(OrcRecordUpdater.ACID_KEY_INDEX_NAME)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+
public static boolean[] genIncludedColumns(TypeDescription readerSchema,
List<Integer> included) {
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 7540e72..dcfb577 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
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.orc.StripeInformation;
import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.OrcEncodedColumnBatch;
import org.apache.orc.OrcProto;
+import org.apache.orc.impl.OrcIndex;
public interface EncodedReader {
@@ -54,4 +55,17 @@ public interface EncodedReader {
* to just checking the constant in the first place.
*/
void setTracing(boolean isEnabled);
+
+ /**
+ * Read the indexes from ORC file.
+ * @param index The destination with pre-allocated arrays to put index data into.
+ * @param stripe Externally provided metadata (from metadata reader or external cache).
+ * @param streams Externally provided metadata (from metadata reader or external cache).
+ * @param included The array of booleans indicating whether each column should be read.
+ * @param sargColumns The array of booleans indicating whether each column's
+ * bloom filters should be read.
+ */
+ void readIndexStreams(OrcIndex index, StripeInformation stripe,
+ List<OrcProto.Stream> streams, boolean[] included, boolean[] sargColumns)
+ throws IOException;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 690cce7..467bac2 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
@@ -18,6 +18,7 @@
package org.apache.hadoop.hive.ql.io.orc.encoded;
import java.io.IOException;
+import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -35,16 +36,23 @@ import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
import org.apache.hadoop.hive.common.io.DiskRangeList.CreateHelper;
+import org.apache.hadoop.hive.common.io.DiskRangeList.MutateHelper;
import org.apache.hadoop.hive.common.io.encoded.EncodedColumnBatch.ColumnStreamData;
import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
import org.apache.orc.CompressionCodec;
import org.apache.orc.DataReader;
import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile.WriterVersion;
import org.apache.orc.OrcProto.ColumnEncoding;
+import org.apache.orc.OrcProto.Stream;
import org.apache.orc.OrcProto.Stream.Kind;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.impl.InStream;
+import org.apache.orc.impl.OrcIndex;
import org.apache.orc.impl.OutStream;
import org.apache.orc.impl.RecordReaderUtils;
import org.apache.orc.impl.StreamName;
+import org.apache.orc.impl.StreamName.Area;
import org.apache.orc.StripeInformation;
import org.apache.orc.impl.BufferChunk;
import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace.RangesSrc;
@@ -53,6 +61,7 @@ import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.PoolFactory;
import org.apache.orc.OrcProto;
import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.CodedInputStream;
import sun.misc.Cleaner;
@@ -126,13 +135,18 @@ class EncodedReaderImpl implements EncodedReader {
private final DataCache cacheWrapper;
private boolean isTracingEnabled;
private final IoTrace trace;
+ private final TypeDescription fileSchema;
+ private final WriterVersion version;
- public EncodedReaderImpl(Object fileKey, List<OrcProto.Type> types, CompressionCodec codec,
+ public EncodedReaderImpl(Object fileKey, List<OrcProto.Type> types,
+ TypeDescription fileSchema, CompressionCodec codec, WriterVersion version,
int bufferSize, long strideRate, DataCache cacheWrapper, DataReader dataReader,
PoolFactory pf, IoTrace trace) throws IOException {
this.fileKey = fileKey;
this.codec = codec;
this.types = types;
+ this.fileSchema = fileSchema; // Note: this is redundant with types
+ this.version = version;
this.bufferSize = bufferSize;
this.rowIndexStride = strideRate;
this.cacheWrapper = cacheWrapper;
@@ -150,25 +164,60 @@ class EncodedReaderImpl implements EncodedReader {
}
/** Helper context for each column being read */
- private static final class ColumnReadContext {
+ private static final class ColumnReadContext extends ReadContext {
public ColumnReadContext(int colIx, OrcProto.ColumnEncoding encoding,
OrcProto.RowIndex rowIndex, int colRgIx) {
+ super(colIx, colRgIx, MAX_STREAMS);
this.encoding = encoding;
this.rowIndex = rowIndex;
+ }
+
+ public static final int MAX_STREAMS = countMaxStreams(Area.DATA);
+ /** Column encoding. */
+ OrcProto.ColumnEncoding encoding;
+ /** Column rowindex. */
+ OrcProto.RowIndex rowIndex;
+
+ public void addStream(long offset, OrcProto.Stream stream, int indexIx) {
+ streams[streamCount++] = new StreamContext(stream, offset, indexIx);
+ }
+
+ @Override
+ 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;
+ for (StreamContext sc : streams) {
+ if (sc != null) {
+ sb.append(" stream_").append(i).append(":").append(sc.toString());
+ }
+ i++;
+ }
+ return sb.toString();
+ }
+ }
+
+ /** Helper context for each column for which the index is being read */
+ private static class ReadContext {
+ protected ReadContext(int colIx, int colRgIx, int maxStreams) {
this.colIx = colIx;
this.includedIx = colRgIx;
streamCount = 0;
+ streams = new StreamContext[maxStreams];
}
- public static final int MAX_STREAMS = OrcProto.Stream.Kind.ROW_INDEX_VALUE;
+ public ReadContext(int colIx, int colRgIx) {
+ this(colIx, colRgIx, MAX_STREAMS);
+ }
+
+ public static final int MAX_STREAMS = countMaxStreams(Area.INDEX);
/** The number of streams that are part of this column. */
int streamCount = 0;
- final StreamContext[] streams = new StreamContext[MAX_STREAMS];
- /** Column encoding. */
- OrcProto.ColumnEncoding encoding;
- /** Column rowindex. */
- OrcProto.RowIndex rowIndex;
+ final StreamContext[] streams;
/** Column index in the file. */
int colIx;
/** Column index in the included columns only (for RG masks). */
@@ -183,7 +232,6 @@ class EncodedReaderImpl implements EncodedReader {
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;
for (StreamContext sc : streams) {
@@ -313,56 +361,10 @@ class EncodedReaderImpl implements EncodedReader {
}
// 2. Now, read all of the ranges from cache or disk.
- DiskRangeList.MutateHelper toRead = new DiskRangeList.MutateHelper(listToRead.get());
- if (LOG.isDebugEnabled()) {
- LOG.debug("Resulting disk ranges to read (file " + fileKey + "): "
- + RecordReaderUtils.stringifyDiskRanges(toRead.next));
- }
- trace.logRanges(fileKey, stripeOffset, toRead.next, RangesSrc.PLAN);
- BooleanRef isAllInCache = new BooleanRef();
- if (hasFileId) {
- cacheWrapper.getFileData(fileKey, toRead.next, stripeOffset, CC_FACTORY, isAllInCache);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Disk ranges after cache (found everything " + isAllInCache.value + "; file "
- + fileKey + ", base offset " + stripeOffset + "): "
- + RecordReaderUtils.stringifyDiskRanges(toRead.next));
- }
- trace.logRanges(fileKey, stripeOffset, toRead.next, RangesSrc.CACHE);
- }
+ IdentityHashMap<ByteBuffer, Boolean> toRelease = new IdentityHashMap<>();
+ MutateHelper toRead = getDataFromCacheAndDisk(
+ listToRead.get(), stripeOffset, hasFileId, toRelease);
- // TODO: the memory release could be optimized - we could release original buffers after we
- // are fully done with each original buffer from disk. For now release all at the end;
- // it doesn't increase the total amount of memory we hold, just the duration a bit.
- // This is much simpler - we can just remember original ranges after reading them, and
- // release them at the end. In a few cases where it's easy to determine that a buffer
- // can be freed in advance, we remove it from the map.
- IdentityHashMap<ByteBuffer, Boolean> toRelease = null;
- if (!isAllInCache.value) {
- boolean hasError = true;
- try {
- if (!isDataReaderOpen) {
- this.dataReader.open();
- isDataReaderOpen = true;
- }
- dataReader.readFileData(toRead.next, stripeOffset,
- cacheWrapper.getAllocator().isDirectAlloc());
- toRelease = new IdentityHashMap<>();
- DiskRangeList drl = toRead.next;
- while (drl != null) {
- if (drl instanceof BufferChunk) {
- toRelease.put(drl.getData(), true);
- }
- drl = drl.next;
- }
- hasError = false;
- } finally {
- // The FS can be closed from under us if the task is interrupted. Release cache buffers.
- // We are assuming here that toRelease will not be present in such cases.
- if (hasError) {
- releaseInitialRefcounts(toRead.next);
- }
- }
- }
// 3. For uncompressed case, we need some special processing before read.
// Basically, we are trying to create artificial, consistent ranges to cache, as there are
@@ -370,46 +372,7 @@ class EncodedReaderImpl implements EncodedReader {
// either cache buffers, or buffers allocated by us and not cached (if we are only reading
// parts of the data for some ranges and don't want to cache it). Both are represented by
// CacheChunks, so the list is just CacheChunk-s from that point on.
- DiskRangeList iter = toRead.next;
- if (codec == null) {
- boolean hasError = true;
- try {
- 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(
- stripeOffset, iter, sctx.offset, sctx.offset + sctx.length, sctx.kind);
- if (newIter != null) {
- iter = newIter;
- }
- }
- }
- // Release buffers as we are done with all the streams... also see toRelease comment.
- // With uncompressed streams, we know we are done earlier.
- if (toRelease != null) {
- releaseBuffers(toRelease.keySet(), true);
- toRelease = null;
- }
- if (LOG.isInfoEnabled()) {
- LOG.info("Disk ranges after pre-read (file " + fileKey + ", base offset "
- + stripeOffset + "): " + RecordReaderUtils.stringifyDiskRanges(toRead.next));
- }
- iter = toRead.next; // Reset the iter to start.
- hasError = false;
- } finally {
- // At this point, everything in the list is going to have a refcount of one. Unless it
- // failed between the allocation and the incref for a single item, we should be ok.
- if (hasError) {
- releaseInitialRefcounts(toRead.next);
- if (toRelease != null) {
- releaseBuffers(toRelease.keySet(), true);
- toRelease = null;
- }
- }
- }
- }
+ DiskRangeList iter = preReadUncompressedStreams(stripeOffset, colCtxs, toRead, toRelease);
try {
// 4. Finally, decompress data, map per RG, and return to caller.
@@ -542,13 +505,74 @@ class EncodedReaderImpl implements EncodedReader {
releaseInitialRefcounts(toRead.next);
// Release buffers as we are done with all the streams... also see toRelease comment.
- if (toRelease != null) {
- releaseBuffers(toRelease.keySet(), true);
- }
+ releaseBuffers(toRelease.keySet(), true);
}
releaseCacheChunksIntoObjectPool(toRead.next);
}
+ private static int countMaxStreams(Area area) {
+ int count = 0;
+ for (Stream.Kind sk : Stream.Kind.values()) {
+ if (StreamName.getArea(sk) == area) {
+ ++count;
+ }
+ }
+ return count;
+ }
+
+ private DiskRangeList.MutateHelper getDataFromCacheAndDisk(DiskRangeList listToRead,
+ long stripeOffset, boolean hasFileId, IdentityHashMap<ByteBuffer, Boolean> toRelease)
+ throws IOException {
+ DiskRangeList.MutateHelper toRead = new DiskRangeList.MutateHelper(listToRead);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Resulting disk ranges to read (file " + fileKey + "): "
+ + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ }
+ BooleanRef isAllInCache = new BooleanRef();
+ if (hasFileId) {
+ cacheWrapper.getFileData(fileKey, toRead.next, stripeOffset, CC_FACTORY, isAllInCache);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Disk ranges after cache (found everything " + isAllInCache.value + "; file "
+ + fileKey + ", base offset " + stripeOffset + "): "
+ + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ }
+ trace.logRanges(fileKey, stripeOffset, toRead.next, RangesSrc.CACHE);
+ }
+
+ // TODO: the memory release could be optimized - we could release original buffers after we
+ // are fully done with each original buffer from disk. For now release all at the end;
+ // it doesn't increase the total amount of memory we hold, just the duration a bit.
+ // This is much simpler - we can just remember original ranges after reading them, and
+ // release them at the end. In a few cases where it's easy to determine that a buffer
+ // can be freed in advance, we remove it from the map.
+ if (!isAllInCache.value) {
+ boolean hasError = true;
+ try {
+ if (!isDataReaderOpen) {
+ this.dataReader.open();
+ isDataReaderOpen = true;
+ }
+ dataReader.readFileData(toRead.next, stripeOffset,
+ cacheWrapper.getAllocator().isDirectAlloc());
+ toRelease = new IdentityHashMap<>();
+ DiskRangeList drl = toRead.next;
+ while (drl != null) {
+ if (drl instanceof BufferChunk) {
+ toRelease.put(drl.getData(), true);
+ }
+ drl = drl.next;
+ }
+ hasError = false;
+ } finally {
+ // The FS can be closed from under us if the task is interrupted. Release cache buffers.
+ // We are assuming here that toRelease will not be present in such cases.
+ if (hasError) {
+ releaseInitialRefcounts(toRead.next);
+ }
+ }
+ }
+ return toRead;
+ }
private void releaseEcbRefCountsOnError(OrcEncodedColumnBatch ecb) {
if (isTracingEnabled) {
@@ -805,7 +829,8 @@ class EncodedReaderImpl implements EncodedReader {
targetBuffers[ix] = chunk.getBuffer();
++ix;
}
- cacheWrapper.getAllocator().allocateMultiple(targetBuffers, bufferSize);
+ cacheWrapper.getAllocator().allocateMultiple(targetBuffers, bufferSize,
+ cacheWrapper.getDataBufferFactory());
// 4. Now decompress (or copy) the data into cache buffers.
for (ProcCacheChunk chunk : toDecompress) {
@@ -1067,8 +1092,8 @@ class EncodedReaderImpl implements EncodedReader {
cacheKeys[ix] = chunk; // Relies on the fact that cache does not actually store these.
++ix;
}
- cacheWrapper.getAllocator().allocateMultiple(
- targetBuffers, (int)(partCount == 1 ? streamLen : partSize));
+ cacheWrapper.getAllocator().allocateMultiple(targetBuffers,
+ (int)(partCount == 1 ? streamLen : partSize), cacheWrapper.getDataBufferFactory());
// 4. Now copy the data into cache buffers.
ix = 0;
@@ -1120,7 +1145,8 @@ class EncodedReaderImpl implements EncodedReader {
// non-cached. Since we are at the first gap, the previous stuff must be contiguous.
singleAlloc[0] = null;
trace.logPartialUncompressedData(partOffset, candidateEnd, true);
- cacheWrapper.getAllocator().allocateMultiple(singleAlloc, (int)(candidateEnd - partOffset));
+ cacheWrapper.getAllocator().allocateMultiple(
+ singleAlloc, (int)(candidateEnd - partOffset), cacheWrapper.getDataBufferFactory());
MemoryBuffer buffer = singleAlloc[0];
cacheWrapper.reuseBuffer(buffer);
ByteBuffer dest = buffer.getByteBufferRaw();
@@ -1134,7 +1160,8 @@ class EncodedReaderImpl implements EncodedReader {
BufferChunk bc, DataCache cacheWrapper, MemoryBuffer[] singleAlloc) {
singleAlloc[0] = null;
trace.logPartialUncompressedData(bc.getOffset(), bc.getEnd(), false);
- cacheWrapper.getAllocator().allocateMultiple(singleAlloc, bc.getLength());
+ cacheWrapper.getAllocator().allocateMultiple(
+ singleAlloc, bc.getLength(), cacheWrapper.getDataBufferFactory());
MemoryBuffer buffer = singleAlloc[0];
cacheWrapper.reuseBuffer(buffer);
ByteBuffer dest = buffer.getByteBufferRaw();
@@ -1597,7 +1624,7 @@ class EncodedReaderImpl implements EncodedReader {
BufferChunk lastChunk, List<ProcCacheChunk> toDecompress, List<MemoryBuffer> cacheBuffers,
boolean doTrace) {
// Prepare future cache buffer.
- MemoryBuffer futureAlloc = cacheWrapper.getAllocator().createUnallocated();
+ MemoryBuffer futureAlloc = cacheWrapper.getDataBufferFactory().create();
// Add it to result in order we are processing.
cacheBuffers.add(futureAlloc);
// Add it to the list of work to decompress.
@@ -1706,4 +1733,297 @@ class EncodedReaderImpl implements EncodedReader {
});
}
}
+
+ // TODO: perhaps move to Orc InStream?
+ private static class IndexStream extends InputStream {
+ private List<MemoryBuffer> ranges;
+ private long currentOffset = 0, length;
+ private ByteBuffer range;
+ private int rangeIx = -1;
+
+ public IndexStream(List<MemoryBuffer> input, long length) {
+ this.ranges = input;
+ this.length = length;
+ }
+
+ @Override
+ public int read() {
+ if (!ensureRangeWithData()) {
+ return -1;
+ }
+ currentOffset += 1;
+ return 0xff & range.get();
+ }
+
+ private boolean ensureRangeWithData() {
+ if (range != null && range.remaining() > 0) return true;
+ ++rangeIx;
+ if (rangeIx == ranges.size()) return false;
+ range = ranges.get(rangeIx).getByteBufferDup();
+ return true;
+ }
+
+ @Override
+ public int read(byte[] data, int offset, int length) {
+ if (!ensureRangeWithData()) {
+ return -1;
+ }
+ int actualLength = Math.min(length, range.remaining());
+ range.get(data, offset, actualLength);
+ currentOffset += actualLength;
+ return actualLength;
+ }
+
+ @Override
+ public int available() {
+ if (range != null && range.remaining() > 0) {
+ return range.remaining();
+ }
+ return (int) (length - currentOffset);
+ }
+
+ @Override
+ public void close() {
+ rangeIx = ranges.size();
+ currentOffset = length;
+ ranges.clear();
+ }
+
+ @Override
+ public String toString() {
+ return "position: " + currentOffset + " length: " + length + " range: " + rangeIx +
+ " offset: " + (range == null ? 0 : range.position())
+ + " limit: " + (range == null ? 0 : range.limit());
+ }
+ }
+
+ @Override
+ public void readIndexStreams(OrcIndex index, StripeInformation stripe,
+ List<OrcProto.Stream> streams, boolean[] included, boolean[] sargColumns)
+ throws IOException {
+ long stripeOffset = stripe.getOffset();
+ DiskRangeList indexRanges = planIndexReading(
+ fileSchema, streams, true, included, sargColumns, version, index.getBloomFilterKinds());
+ if (indexRanges == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Nothing to read for stripe [" + stripe + "]");
+ }
+ return;
+ }
+ ReadContext[] colCtxs = new ReadContext[included.length];
+ int colRgIx = -1;
+ for (int i = 0; i < included.length; ++i) {
+ if (!included[i] && (sargColumns == null || !sargColumns[i])) continue;
+ colCtxs[i] = new ReadContext(i, ++colRgIx);
+ if (isTracingEnabled) {
+ LOG.trace("Creating context: " + colCtxs[i].toString());
+ }
+ }
+ long offset = 0;
+ for (OrcProto.Stream stream : streams) {
+ long length = stream.getLength();
+ int colIx = stream.getColumn();
+ OrcProto.Stream.Kind streamKind = stream.getKind();
+ // See planIndexReading - only read non-row-index streams if involved in SARGs.
+ if ((StreamName.getArea(streamKind) == StreamName.Area.INDEX)
+ && ((sargColumns != null && sargColumns[colIx])
+ || (included[colIx] && streamKind == Kind.ROW_INDEX))) {
+ colCtxs[colIx].addStream(offset, stream, -1);
+ if (isTracingEnabled) {
+ LOG.trace("Adding stream for column " + colIx + ": "
+ + streamKind + " at " + offset + ", " + length);
+ }
+ }
+ offset += length;
+ }
+
+ boolean hasFileId = this.fileKey != null;
+
+ // 2. Now, read all of the ranges from cache or disk.
+ IdentityHashMap<ByteBuffer, Boolean> toRelease = new IdentityHashMap<>();
+ MutateHelper toRead = getDataFromCacheAndDisk(indexRanges, stripeOffset, hasFileId, toRelease);
+
+ // 3. For uncompressed case, we need some special processing before read.
+ DiskRangeList iter = preReadUncompressedStreams(stripeOffset, colCtxs, toRead, toRelease);
+
+ // 4. Decompress the data.
+ for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
+ ReadContext 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];
+ try {
+ if (isTracingEnabled) {
+ LOG.trace("Getting index stream " + sctx.kind + " for column " + ctx.colIx
+ + " at " + sctx.offset + ", " + sctx.length);
+ }
+ ColumnStreamData csd = POOLS.csdPool.take();
+ long endCOffset = sctx.offset + sctx.length;
+ DiskRangeList lastCached = readEncodedStream(stripeOffset, iter, sctx.offset,
+ endCOffset, csd, endCOffset, sctx.offset, toRelease);
+ if (lastCached != null) {
+ iter = lastCached;
+ }
+ CodedInputStream cis = CodedInputStream.newInstance(
+ new IndexStream(csd.getCacheBuffers(), sctx.length));
+ cis.setSizeLimit(InStream.PROTOBUF_MESSAGE_MAX_LIMIT);
+ switch (sctx.kind) {
+ case ROW_INDEX:
+ index.getRowGroupIndex()[colIx] = OrcProto.RowIndex.parseFrom(cis);
+ break;
+ case BLOOM_FILTER:
+ case BLOOM_FILTER_UTF8:
+ index.getBloomFilterIndex()[colIx] = OrcProto.BloomFilterIndex.parseFrom(cis);
+ break;
+ default:
+ throw new AssertionError("Unexpected index stream type " + sctx.kind);
+ }
+ // We are done with the buffers; unlike data blocks, we are also the consumer. Release.
+ for (MemoryBuffer buf : csd.getCacheBuffers()) {
+ if (buf == null) continue;
+ cacheWrapper.releaseBuffer(buf);
+ }
+ } catch (Exception ex) {
+ DiskRangeList drl = toRead == null ? null : toRead.next;
+ LOG.error("Error getting stream " + sctx.kind + " for column " + ctx.colIx
+ + " at " + sctx.offset + ", " + sctx.length + "; toRead "
+ + RecordReaderUtils.stringifyDiskRanges(drl), ex);
+ throw (ex instanceof IOException) ? (IOException)ex : new IOException(ex);
+ }
+ }
+ }
+
+ if (isTracingEnabled) {
+ LOG.trace("Disk ranges after preparing all the data "
+ + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ }
+
+ // Release the unreleased buffers. See class comment about refcounts.
+ releaseInitialRefcounts(toRead.next);
+ releaseBuffers(toRelease.keySet(), true);
+ releaseCacheChunksIntoObjectPool(toRead.next);
+ }
+
+
+ private DiskRangeList preReadUncompressedStreams(long stripeOffset, ReadContext[] colCtxs,
+ MutateHelper toRead, IdentityHashMap<ByteBuffer, Boolean> toRelease) throws IOException {
+ if (codec != null) return toRead.next;
+ DiskRangeList iter = toRead.next; // Keep "toRead" list for future use, don't extract().
+ boolean hasError = true;
+ try {
+ for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
+ ReadContext 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(
+ stripeOffset, iter, sctx.offset, sctx.offset + sctx.length, sctx.kind);
+ if (newIter != null) {
+ iter = newIter;
+ }
+ }
+ }
+ // Release buffers as we are done with all the streams... also see toRelease comment.\
+ // With uncompressed streams, we know we are done earlier.
+ if (toRelease != null) {
+ releaseBuffers(toRelease.keySet(), true);
+ toRelease.clear();
+ }
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Disk ranges after pre-read (file " + fileKey + ", base offset "
+ + stripeOffset + "): " + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ }
+ iter = toRead.next; // Reset the iter to start.
+ hasError = false;
+ } finally {
+ // At this point, everything in the list is going to have a refcount of one. Unless it
+ // failed between the allocation and the incref for a single item, we should be ok.
+ if (hasError) {
+ releaseInitialRefcounts(toRead.next);
+ if (toRelease != null) {
+ releaseBuffers(toRelease.keySet(), true);
+ toRelease.clear();
+ }
+ }
+ }
+ return toRead.next; // Reset the iter to start.
+ }
+
+ // TODO: temporary, need to expose from ORC utils (note the difference in null checks)
+ static DiskRangeList planIndexReading(TypeDescription fileSchema,
+ List<OrcProto.Stream> streams,
+ boolean ignoreNonUtf8BloomFilter,
+ boolean[] fileIncluded,
+ boolean[] sargColumns,
+ WriterVersion version,
+ OrcProto.Stream.Kind[] bloomFilterKinds) {
+ DiskRangeList.CreateHelper result = new DiskRangeList.CreateHelper();
+ // figure out which kind of bloom filter we want for each column
+ // picks bloom_filter_utf8 if its available, otherwise bloom_filter
+ if (sargColumns != null) {
+ for (OrcProto.Stream stream : streams) {
+ if (stream.hasKind() && stream.hasColumn()) {
+ int column = stream.getColumn();
+ if (sargColumns[column]) {
+ switch (stream.getKind()) {
+ case BLOOM_FILTER:
+ if (bloomFilterKinds[column] == null &&
+ !(ignoreNonUtf8BloomFilter &&
+ hadBadBloomFilters(fileSchema.findSubtype(column)
+ .getCategory(), version))) {
+ bloomFilterKinds[column] = OrcProto.Stream.Kind.BLOOM_FILTER;
+ }
+ break;
+ case BLOOM_FILTER_UTF8:
+ bloomFilterKinds[column] = OrcProto.Stream.Kind.BLOOM_FILTER_UTF8;
+ break;
+ default:
+ break;
+ }
+ }
+ }
+ }
+ }
+ long offset = 0;
+ for (OrcProto.Stream stream: streams) {
+ if (stream.hasKind() && stream.hasColumn()) {
+ int column = stream.getColumn();
+ if (fileIncluded == null || fileIncluded[column]) {
+ boolean needStream = false;
+ switch (stream.getKind()) {
+ case ROW_INDEX:
+ needStream = true;
+ break;
+ case BLOOM_FILTER:
+ case BLOOM_FILTER_UTF8:
+ needStream = (sargColumns != null) && (bloomFilterKinds[column] == stream.getKind());
+ break;
+ default:
+ // PASS
+ break;
+ }
+ if (needStream) {
+ result.addOrMerge(offset, offset + stream.getLength(), true, false);
+ }
+ }
+ }
+ offset += stream.getLength();
+ }
+ return result.get();
+ }
+
+ // TODO: see planIndexReading; this is not needed here.
+ private static boolean hadBadBloomFilters(TypeDescription.Category category,
+ WriterVersion version) {
+ switch(category) {
+ case STRING:
+ case CHAR:
+ case VARCHAR:
+ return !version.includes(WriterVersion.HIVE_12055);
+ case DECIMAL:
+ return true;
+ default:
+ return false;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/Reader.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/Reader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/Reader.java
index cdd58df..d2bb641 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/Reader.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/Reader.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hive.common.io.DataCache;
import org.apache.hadoop.hive.common.io.encoded.EncodedColumnBatch;
import org.apache.hadoop.hive.common.io.encoded.EncodedColumnBatch.ColumnStreamData;
import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.orc.CompressionCodec;
import org.apache.orc.DataReader;
import org.apache.orc.OrcProto;
@@ -47,6 +48,9 @@ public interface Reader extends org.apache.hadoop.hive.ql.io.orc.Reader {
EncodedReader encodedReader(Object fileKey, DataCache dataCache, DataReader dataReader,
PoolFactory pf, IoTrace trace) throws IOException;
+ /** Gets the compression codec for the underlying ORC file. */
+ CompressionCodec getCodec();
+
/** The factory that can create (or return) the pools used by encoded reader. */
public interface PoolFactory {
<T> Pool<T> createPool(int size, PoolObjectHelper<T> helper);
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/ReaderImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/ReaderImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/ReaderImpl.java
index d47ba6b..a916d58 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/ReaderImpl.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/ReaderImpl.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.io.DataCache;
+import org.apache.orc.CompressionCodec;
import org.apache.orc.DataReader;
import org.apache.hadoop.hive.ql.io.orc.OrcFile.ReaderOptions;
@@ -35,7 +36,13 @@ class ReaderImpl extends org.apache.hadoop.hive.ql.io.orc.ReaderImpl implements
@Override
public EncodedReader encodedReader(Object fileKey, DataCache dataCache, DataReader dataReader,
PoolFactory pf, IoTrace trace) throws IOException {
- return new EncodedReaderImpl(fileKey, types,
- codec, bufferSize, rowIndexStride, dataCache, dataReader, pf, trace);
+ return new EncodedReaderImpl(fileKey, types, getSchema(), codec, getWriterVersion(),
+ bufferSize, rowIndexStride, dataCache, dataReader, pf, trace);
}
+
+ @Override
+ public CompressionCodec getCodec() {
+ return codec;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/ql/src/test/results/clientpositive/llap/orc_llap_counters.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientpositive/llap/orc_llap_counters.q.out b/ql/src/test/results/clientpositive/llap/orc_llap_counters.q.out
index 8af84dc..9e9f8cf 100644
--- a/ql/src/test/results/clientpositive/llap/orc_llap_counters.q.out
+++ b/ql/src/test/results/clientpositive/llap/orc_llap_counters.q.out
@@ -253,7 +253,7 @@ PREHOOK: Input: default@orc_ppd
Stage-1 FILE SYSTEM COUNTERS:
HDFS_BYTES_READ: 16673
HDFS_BYTES_WRITTEN: 104
- HDFS_READ_OPS: 5
+ HDFS_READ_OPS: 6
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -263,6 +263,9 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2100
Stage-1 LLAP IO COUNTERS:
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 26
+ CACHE_MISS_BYTES: 24
METADATA_CACHE_MISS: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 3
@@ -288,7 +291,7 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 1344
+ HDFS_BYTES_READ: 1055
HDFS_BYTES_WRITTEN: 101
HDFS_READ_OPS: 4
HDFS_LARGE_READ_OPS: 0
@@ -300,11 +303,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 524288
- ALLOCATED_USED_BYTES: 269
- CACHE_MISS_BYTES: 249
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 1048576
+ ALLOCATED_USED_BYTES: 2732
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 1055
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 1
ROWS_EMITTED: 1000
@@ -327,7 +330,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 22
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -352,7 +355,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 16
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -377,7 +380,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -402,7 +405,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -427,7 +430,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 32
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -452,7 +455,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 32
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -477,7 +480,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1697
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -502,7 +505,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 12
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -527,7 +530,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1713
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -552,7 +555,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -577,7 +580,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 50
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -602,7 +605,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 318
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -655,6 +658,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 830
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -675,6 +680,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 830
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -697,7 +704,7 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 6786
+ HDFS_BYTES_READ: 5691
HDFS_BYTES_WRITTEN: 101
HDFS_READ_OPS: 4
HDFS_LARGE_READ_OPS: 0
@@ -709,12 +716,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 786432
- ALLOCATED_USED_BYTES: 11299
- CACHE_HIT_BYTES: 0
- CACHE_MISS_BYTES: 3980
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 1310720
+ ALLOCATED_USED_BYTES: 13796
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 5691
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 1
ROWS_EMITTED: 1000
@@ -737,7 +743,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -762,7 +768,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -801,7 +807,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2100
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -826,6 +832,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -846,7 +854,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -871,7 +879,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -896,7 +904,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -921,7 +929,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 81
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -946,7 +954,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 74
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -971,7 +979,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 12
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -996,7 +1004,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 13
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -1021,7 +1029,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -1046,7 +1054,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 7
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -1071,6 +1079,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -1091,6 +1101,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -1111,6 +1123,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -1131,7 +1145,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -1156,7 +1170,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -1181,7 +1195,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4229
+ CACHE_HIT_BYTES: 6770
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -1206,7 +1220,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4229
+ CACHE_HIT_BYTES: 6770
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/ql/src/test/results/clientpositive/llap/orc_llap_counters1.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientpositive/llap/orc_llap_counters1.q.out b/ql/src/test/results/clientpositive/llap/orc_llap_counters1.q.out
index 4536cbb..7b52493 100644
--- a/ql/src/test/results/clientpositive/llap/orc_llap_counters1.q.out
+++ b/ql/src/test/results/clientpositive/llap/orc_llap_counters1.q.out
@@ -253,7 +253,7 @@ PREHOOK: Input: default@orc_ppd
Stage-1 FILE SYSTEM COUNTERS:
HDFS_BYTES_READ: 17728
HDFS_BYTES_WRITTEN: 104
- HDFS_READ_OPS: 6
+ HDFS_READ_OPS: 7
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -263,9 +263,10 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2094
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 524288
- ALLOCATED_USED_BYTES: 269
- CACHE_MISS_BYTES: 249
+ ALLOCATED_BYTES: 1310720
+ ALLOCATED_USED_BYTES: 2758
+ CACHE_HIT_BYTES: 0
+ CACHE_MISS_BYTES: 1079
METADATA_CACHE_MISS: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
@@ -289,7 +290,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2094
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/ql/src/test/results/clientpositive/llap/orc_ppd_basic.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientpositive/llap/orc_ppd_basic.q.out b/ql/src/test/results/clientpositive/llap/orc_ppd_basic.q.out
index 77b7f5a..7e169f2 100644
--- a/ql/src/test/results/clientpositive/llap/orc_ppd_basic.q.out
+++ b/ql/src/test/results/clientpositive/llap/orc_ppd_basic.q.out
@@ -205,7 +205,7 @@ PREHOOK: Input: default@orc_ppd
Stage-1 FILE SYSTEM COUNTERS:
HDFS_BYTES_READ: 16673
HDFS_BYTES_WRITTEN: 104
- HDFS_READ_OPS: 5
+ HDFS_READ_OPS: 6
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -215,6 +215,9 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2100
Stage-1 LLAP IO COUNTERS:
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 26
+ CACHE_MISS_BYTES: 24
METADATA_CACHE_MISS: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 3
@@ -240,7 +243,7 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 1344
+ HDFS_BYTES_READ: 1055
HDFS_BYTES_WRITTEN: 101
HDFS_READ_OPS: 4
HDFS_LARGE_READ_OPS: 0
@@ -252,11 +255,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 8
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 524288
- ALLOCATED_USED_BYTES: 269
- CACHE_MISS_BYTES: 249
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 1048576
+ ALLOCATED_USED_BYTES: 2732
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 1055
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 1
ROWS_EMITTED: 1000
@@ -279,7 +282,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 22
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -304,7 +307,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 16
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -329,7 +332,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 18
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -354,7 +357,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -379,7 +382,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 32
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -404,7 +407,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 32
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -429,7 +432,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1697
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -454,7 +457,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 12
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -479,7 +482,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1713
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -504,7 +507,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -529,7 +532,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 50
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -554,7 +557,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 318
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 249
+ CACHE_HIT_BYTES: 1079
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -607,6 +610,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 830
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -627,6 +632,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 830
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -649,7 +656,7 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 6786
+ HDFS_BYTES_READ: 5691
HDFS_BYTES_WRITTEN: 101
HDFS_READ_OPS: 4
HDFS_LARGE_READ_OPS: 0
@@ -661,12 +668,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 786432
- ALLOCATED_USED_BYTES: 11299
- CACHE_HIT_BYTES: 0
- CACHE_MISS_BYTES: 3980
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 1310720
+ ALLOCATED_USED_BYTES: 13796
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 5691
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
NUM_VECTOR_BATCHES: 1
ROWS_EMITTED: 1000
@@ -689,7 +695,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -714,7 +720,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -753,7 +759,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2100
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -778,6 +784,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -798,7 +806,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -823,7 +831,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -848,7 +856,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -873,7 +881,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 81
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -898,7 +906,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 74
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -923,7 +931,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 12
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -948,7 +956,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 13
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -973,7 +981,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 1
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -998,7 +1006,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 7
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -1023,6 +1031,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -1043,6 +1053,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -1063,6 +1075,8 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 0
Stage-1 LLAP IO COUNTERS:
+ CACHE_HIT_BYTES: 1735
+ CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
SELECTED_ROWGROUPS: 0
0
@@ -1083,7 +1097,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -1108,7 +1122,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 6
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 3980
+ CACHE_HIT_BYTES: 5715
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 2
@@ -1133,7 +1147,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4229
+ CACHE_HIT_BYTES: 6770
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -1158,7 +1172,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4229
+ CACHE_HIT_BYTES: 6770
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 1
@@ -1171,9 +1185,9 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 7718
+ HDFS_BYTES_READ: 4912
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 4
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
@@ -1183,12 +1197,11 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- ALLOCATED_BYTES: 262144
- ALLOCATED_USED_BYTES: 8400
- CACHE_HIT_BYTES: 0
- CACHE_MISS_BYTES: 4809
- METADATA_CACHE_HIT: 1
- METADATA_CACHE_MISS: 1
+ ALLOCATED_BYTES: 524288
+ ALLOCATED_USED_BYTES: 8527
+ CACHE_HIT_BYTES: 24
+ CACHE_MISS_BYTES: 4912
+ METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
NUM_VECTOR_BATCHES: 3
ROWS_EMITTED: 2100
@@ -1199,25 +1212,27 @@ PREHOOK: type: QUERY
PREHOOK: Input: default@orc_ppd
#### A masked pattern was here ####
Stage-1 FILE SYSTEM COUNTERS:
- HDFS_BYTES_READ: 0
+ HDFS_BYTES_READ: 1751
HDFS_BYTES_WRITTEN: 101
- HDFS_READ_OPS: 2
+ HDFS_READ_OPS: 3
HDFS_LARGE_READ_OPS: 0
HDFS_WRITE_OPS: 2
Stage-1 HIVE COUNTERS:
CREATED_FILES: 1
DESERIALIZE_ERRORS: 0
- RECORDS_IN_Map_1: 2100
+ RECORDS_IN_Map_1: 100
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4809
- CACHE_MISS_BYTES: 0
+ ALLOCATED_BYTES: 262144
+ ALLOCATED_USED_BYTES: 2376
+ CACHE_HIT_BYTES: 4936
+ CACHE_MISS_BYTES: 1751
METADATA_CACHE_HIT: 2
- NUM_DECODED_BATCHES: 3
- NUM_VECTOR_BATCHES: 3
- ROWS_EMITTED: 2100
- SELECTED_ROWGROUPS: 3
+ NUM_DECODED_BATCHES: 1
+ NUM_VECTOR_BATCHES: 1
+ ROWS_EMITTED: 100
+ SELECTED_ROWGROUPS: 1
2
PREHOOK: query: select count(*) from orc_ppd where f=74.72
PREHOOK: type: QUERY
@@ -1236,7 +1251,7 @@ Stage-1 HIVE COUNTERS:
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4809
+ CACHE_HIT_BYTES: 4936
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
NUM_DECODED_BATCHES: 3
@@ -1257,17 +1272,17 @@ Stage-1 FILE SYSTEM COUNTERS:
Stage-1 HIVE COUNTERS:
CREATED_FILES: 1
DESERIALIZE_ERRORS: 0
- RECORDS_IN_Map_1: 2100
+ RECORDS_IN_Map_1: 100
RECORDS_OUT_0: 1
RECORDS_OUT_INTERMEDIATE_Map_1: 2
Stage-1 LLAP IO COUNTERS:
- CACHE_HIT_BYTES: 4809
+ CACHE_HIT_BYTES: 6687
CACHE_MISS_BYTES: 0
METADATA_CACHE_HIT: 2
- NUM_DECODED_BATCHES: 3
- NUM_VECTOR_BATCHES: 3
- ROWS_EMITTED: 2100
- SELECTED_ROWGROUPS: 3
+ NUM_DECODED_BATCHES: 1
+ NUM_VECTOR_BATCHES: 1
+ ROWS_EMITTED: 100
+ SELECTED_ROWGROUPS: 1
2
PREHOOK: query: create temporary table tmp_orcppd
stored as orc
[3/3] hive git commit: HIVE-15665 : LLAP: OrcFileMetadata objects in
cache can impact heap usage (Sergey Shelukhin,
reviewed by Prasanth Jayachandran)
Posted by se...@apache.org.
HIVE-15665 : LLAP: OrcFileMetadata objects in cache can impact heap usage (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/50fb6f3c
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/50fb6f3c
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/50fb6f3c
Branch: refs/heads/master
Commit: 50fb6f3cb4651be5a1e0c0bae1f59f193f2c7e09
Parents: 4a4ae12
Author: sergey <se...@apache.org>
Authored: Fri Sep 15 12:09:11 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 15 12:09:11 2017 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 6 -
.../hive/llap/cache/EvictionDispatcher.java | 27 +-
.../hive/llap/io/api/impl/LlapIoImpl.java | 42 +-
.../llap/io/decode/OrcColumnVectorProducer.java | 6 +-
.../llap/io/encoded/OrcEncodedDataReader.java | 345 +++++++-----
.../hive/llap/io/metadata/MetadataCache.java | 537 +++++++++++++++++++
.../llap/io/metadata/OrcFileEstimateErrors.java | 5 +-
.../hive/llap/io/metadata/OrcFileMetadata.java | 114 +---
.../hive/llap/io/metadata/OrcMetadataCache.java | 163 ------
.../llap/io/metadata/OrcStripeMetadata.java | 123 +----
.../io/metadata/ParquetMetadataCacheImpl.java | 353 ------------
.../TestIncrementalObjectSizeEstimator.java | 4 +-
.../hive/llap/cache/TestOrcMetadataCache.java | 159 ++++--
.../hadoop/hive/ql/io/orc/OrcInputFormat.java | 11 +
.../hive/ql/io/orc/encoded/EncodedReader.java | 14 +
.../ql/io/orc/encoded/EncodedReaderImpl.java | 534 ++++++++++++++----
.../hadoop/hive/ql/io/orc/encoded/Reader.java | 4 +
.../hive/ql/io/orc/encoded/ReaderImpl.java | 11 +-
.../clientpositive/llap/orc_llap_counters.q.out | 98 ++--
.../llap/orc_llap_counters1.q.out | 11 +-
.../clientpositive/llap/orc_ppd_basic.q.out | 147 ++---
.../llap/orc_ppd_schema_evol_3a.q.out | 156 +++---
.../hive/common/io/FileMetadataCache.java | 8 +-
23 files changed, 1616 insertions(+), 1262 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 b22a834..8a906ce 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -3010,12 +3010,6 @@ public class HiveConf extends Configuration {
LLAP_ALLOCATOR_MAX_ALLOC("hive.llap.io.allocator.alloc.max", "16Mb", new SizeValidator(),
"Maximum allocation possible from LLAP buddy allocator. For ORC, should be as large as\n" +
"the largest expected ORC compression buffer size. Must be a power of 2."),
- @Deprecated
- LLAP_IO_METADATA_FRACTION("hive.llap.io.metadata.fraction", 0.1f,
- "Temporary setting for on-heap metadata cache fraction of xmx, set to avoid potential\n" +
- "heap problems on very large datasets when on-heap metadata cache takes over\n" +
- "everything. -1 managed metadata and data together (which is more flexible). This\n" +
- "setting will be removed (in effect become -1) once ORC metadata cache is moved off-heap."),
LLAP_ALLOCATOR_ARENA_COUNT("hive.llap.io.allocator.arena.count", 8,
"Arena count for LLAP low-level cache; cache will be allocated in the steps of\n" +
"(size/arena_count) bytes. This size must be <= 1Gb and >= max allocation; if it is\n" +
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 c5248ce..b226906 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
@@ -19,11 +19,8 @@ package org.apache.hadoop.hive.llap.cache;
import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl.LlapSerDeDataBuffer;
import org.apache.hadoop.hive.llap.io.metadata.OrcFileEstimateErrors;
-import org.apache.hadoop.hive.llap.io.metadata.OrcFileMetadata;
-import org.apache.hadoop.hive.llap.io.metadata.OrcMetadataCache;
-import org.apache.hadoop.hive.llap.io.metadata.OrcStripeMetadata;
-import org.apache.hadoop.hive.llap.io.metadata.ParquetMetadataCacheImpl;
-import org.apache.hadoop.hive.llap.io.metadata.ParquetMetadataCacheImpl.LlapFileMetadataBuffer;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache.LlapMetadataBuffer;
/**
* Eviction dispatcher - uses double dispatch to route eviction notifications to correct caches.
@@ -31,17 +28,13 @@ import org.apache.hadoop.hive.llap.io.metadata.ParquetMetadataCacheImpl.LlapFile
public final class EvictionDispatcher implements EvictionListener, LlapOomDebugDump {
private final LowLevelCache dataCache;
private final SerDeLowLevelCacheImpl serdeCache;
- private final OrcMetadataCache metadataCache;
+ private final MetadataCache metadataCache;
private final EvictionAwareAllocator allocator;
- // TODO# temporary, will be merged with OrcMetadataCache after HIVE-15665.
- private final ParquetMetadataCacheImpl parquetMetadataCache;
public EvictionDispatcher(LowLevelCache dataCache, SerDeLowLevelCacheImpl serdeCache,
- OrcMetadataCache metadataCache, EvictionAwareAllocator allocator,
- ParquetMetadataCacheImpl parquetMetadataCache) {
+ MetadataCache metadataCache, EvictionAwareAllocator allocator) {
this.dataCache = dataCache;
this.metadataCache = metadataCache;
- this.parquetMetadataCache = parquetMetadataCache;
this.serdeCache = serdeCache;
this.allocator = allocator;
}
@@ -51,10 +44,6 @@ public final class EvictionDispatcher implements EvictionListener, LlapOomDebugD
buffer.notifyEvicted(this); // This will call one of the specific notifyEvicted overloads.
}
- public void notifyEvicted(LlapFileMetadataBuffer buffer) {
- this.parquetMetadataCache.notifyEvicted(buffer);
- }
-
public void notifyEvicted(LlapSerDeDataBuffer buffer) {
serdeCache.notifyEvicted(buffer);
allocator.deallocateEvicted(buffer);
@@ -65,12 +54,10 @@ public final class EvictionDispatcher implements EvictionListener, LlapOomDebugD
allocator.deallocateEvicted(buffer);
}
- public void notifyEvicted(OrcFileMetadata buffer) {
- metadataCache.notifyEvicted(buffer);
- }
-
- public void notifyEvicted(OrcStripeMetadata buffer) {
+ public void notifyEvicted(LlapMetadataBuffer<?> buffer) {
metadataCache.notifyEvicted(buffer);
+ // Note: the metadata cache may deallocate additional buffers, but not this one.
+ allocator.deallocateEvicted(buffer);
}
public void notifyEvicted(OrcFileEstimateErrors buffer) {
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 f42622b..77c8ade 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
@@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit;
import javax.management.ObjectName;
-import org.apache.hadoop.hive.llap.daemon.impl.LlapDaemon;
import org.apache.hadoop.hive.llap.daemon.impl.StatsRecordingThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,8 +60,7 @@ import org.apache.hadoop.hive.llap.io.api.LlapIo;
import org.apache.hadoop.hive.llap.io.decode.ColumnVectorProducer;
import org.apache.hadoop.hive.llap.io.decode.GenericColumnVectorProducer;
import org.apache.hadoop.hive.llap.io.decode.OrcColumnVectorProducer;
-import org.apache.hadoop.hive.llap.io.metadata.OrcMetadataCache;
-import org.apache.hadoop.hive.llap.io.metadata.ParquetMetadataCacheImpl;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonIOMetrics;
import org.apache.hadoop.hive.llap.metrics.MetricsUtils;
@@ -72,8 +70,6 @@ import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace;
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.metrics2.util.MBeans;
import org.apache.hive.common.util.FixedSizedObjectPool;
@@ -132,7 +128,7 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
LOG.info("Started llap daemon metrics with displayName: {} sessionId: {}", displayName,
sessionId);
- OrcMetadataCache metadataCache = null;
+ MetadataCache metadataCache = null;
SerDeLowLevelCacheImpl serdeCache = null; // TODO: extract interface when needed
BufferUsageManager bufferManagerOrc = null, bufferManagerGeneric = null;
boolean isEncodeEnabled = HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENCODE_ENABLED);
@@ -141,30 +137,12 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
boolean useLrfu = HiveConf.getBoolVar(conf, HiveConf.ConfVars.LLAP_USE_LRFU);
long totalMemorySize = HiveConf.getSizeVar(conf, ConfVars.LLAP_IO_MEMORY_MAX_SIZE);
int minAllocSize = (int)HiveConf.getSizeVar(conf, ConfVars.LLAP_ALLOCATOR_MIN_ALLOC);
- float metadataFraction = HiveConf.getFloatVar(conf, ConfVars.LLAP_IO_METADATA_FRACTION);
- long metaMem = 0;
- // TODO: this split a workaround until HIVE-15665.
- // Technically we don't have to do it for on-heap data cache but we'd do for testing.
- boolean isSplitCache = metadataFraction > 0f;
- if (isSplitCache) {
- metaMem = (long)(LlapDaemon.getTotalHeapSize() * metadataFraction);
- }
LowLevelCachePolicy cachePolicy = useLrfu ? new LowLevelLrfuCachePolicy(
minAllocSize, totalMemorySize, conf) : new LowLevelFifoCachePolicy();
// Allocator uses memory manager to request memory, so create the manager next.
LowLevelCacheMemoryManager memManager = new LowLevelCacheMemoryManager(
totalMemorySize, cachePolicy, cacheMetrics);
- LowLevelCachePolicy metaCachePolicy = null;
- LowLevelCacheMemoryManager metaMemManager = null;
- if (isSplitCache) {
- metaCachePolicy = useLrfu ? new LowLevelLrfuCachePolicy(
- minAllocSize, metaMem, conf) : new LowLevelFifoCachePolicy();
- metaMemManager = new LowLevelCacheMemoryManager(metaMem, metaCachePolicy, cacheMetrics);
- } else {
- metaCachePolicy = cachePolicy;
- metaMemManager = memManager;
- }
- cacheMetrics.setCacheCapacityTotal(totalMemorySize + metaMem);
+ cacheMetrics.setCacheCapacityTotal(totalMemorySize);
// Cache uses allocator to allocate and deallocate, create allocator and then caches.
BuddyAllocator allocator = new BuddyAllocator(conf, memManager, cacheMetrics);
this.allocator = allocator;
@@ -179,18 +157,12 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
}
boolean useGapCache = HiveConf.getBoolVar(conf, ConfVars.LLAP_CACHE_ENABLE_ORC_GAP_CACHE);
- metadataCache = new OrcMetadataCache(metaMemManager, metaCachePolicy, useGapCache);
- // TODO# temporary, see comments there
- ParquetMetadataCacheImpl parquetMc = new ParquetMetadataCacheImpl(
- allocator, memManager, cachePolicy, cacheMetrics);
- fileMetadataCache = parquetMc;
+ metadataCache = new MetadataCache(
+ allocator, memManager, cachePolicy, useGapCache, cacheMetrics);
+ fileMetadataCache = metadataCache;
// And finally cache policy uses cache to notify it of eviction. The cycle is complete!
EvictionDispatcher e = new EvictionDispatcher(
- dataCache, serdeCache, metadataCache, allocator, parquetMc);
- if (isSplitCache) {
- metaCachePolicy.setEvictionListener(e);
- metaCachePolicy.setParentDebugDumper(e);
- }
+ dataCache, serdeCache, metadataCache, allocator);
cachePolicy.setEvictionListener(e);
cachePolicy.setParentDebugDumper(e);
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 6edd84b..373af76 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
@@ -33,7 +33,7 @@ import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch;
import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
import org.apache.hadoop.hive.llap.io.encoded.OrcEncodedDataReader;
-import org.apache.hadoop.hive.llap.io.metadata.OrcMetadataCache;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
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;
@@ -52,7 +52,7 @@ import org.apache.orc.OrcConf;
public class OrcColumnVectorProducer implements ColumnVectorProducer {
- private final OrcMetadataCache metadataCache;
+ private final MetadataCache metadataCache;
private final LowLevelCache lowLevelCache;
private final BufferUsageManager bufferManager;
private final Configuration conf;
@@ -63,7 +63,7 @@ public class OrcColumnVectorProducer implements ColumnVectorProducer {
// TODO: should this rather use a threadlocal for NUMA affinity?
private final FixedSizedObjectPool<IoTrace> tracePool;
- public OrcColumnVectorProducer(OrcMetadataCache metadataCache,
+ public OrcColumnVectorProducer(MetadataCache metadataCache,
LowLevelCache lowLevelCache, BufferUsageManager bufferManager,
Configuration conf, LlapDaemonCacheMetrics cacheMetrics, LlapDaemonIOMetrics ioMetrics,
FixedSizedObjectPool<IoTrace> tracePool) {
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 b5db302..2e47a56 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
@@ -26,11 +26,21 @@ import java.util.Collections;
import java.util.List;
import org.apache.hadoop.hive.llap.counters.LlapIOCounters;
+import org.apache.orc.CompressionCodec;
+import org.apache.orc.OrcProto.BloomFilterIndex;
+import org.apache.orc.OrcProto.FileTail;
+import org.apache.orc.OrcProto.RowIndex;
+import org.apache.orc.OrcProto.Stream;
+import org.apache.orc.OrcProto.StripeStatistics;
import org.apache.orc.TypeDescription;
+import org.apache.orc.impl.BufferChunk;
import org.apache.orc.impl.DataReaderProperties;
+import org.apache.orc.impl.InStream;
import org.apache.orc.impl.OrcIndex;
import org.apache.orc.impl.OrcTail;
+import org.apache.orc.impl.ReaderImpl;
import org.apache.orc.impl.SchemaEvolution;
+import org.apache.orc.impl.WriterImpl;
import org.apache.tez.common.counters.TezCounters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,7 +68,8 @@ import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
import org.apache.hadoop.hive.llap.io.decode.OrcEncodedDataConsumer;
import org.apache.hadoop.hive.llap.io.metadata.OrcFileMetadata;
-import org.apache.hadoop.hive.llap.io.metadata.OrcMetadataCache;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
+import org.apache.hadoop.hive.llap.io.metadata.MetadataCache.LlapBufferOrBuffers;
import org.apache.hadoop.hive.llap.io.metadata.OrcStripeMetadata;
import org.apache.hadoop.hive.ql.exec.DDLTask;
import org.apache.hadoop.hive.ql.io.AcidUtils;
@@ -87,6 +98,8 @@ import org.apache.hive.common.util.FixedSizedObjectPool;
import org.apache.orc.OrcProto;
import org.apache.tez.common.CallableWithNdc;
+import com.google.common.collect.Lists;
+
/**
* This produces EncodedColumnBatch via ORC EncodedDataImpl.
* It serves as Consumer for EncodedColumnBatch too, for the high-level cache scenario where
@@ -135,7 +148,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
};
- private final OrcMetadataCache metadataCache;
+ private final MetadataCache metadataCache;
private final LowLevelCache lowLevelCache;
private final BufferUsageManager bufferManager;
private final Configuration daemonConf, jobConf;
@@ -153,8 +166,10 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
private OrcFileMetadata fileMetadata;
private Path path;
private Reader orcReader;
- private DataReader metadataReader;
+ private DataReader rawDataReader;
+ private boolean isRawDataReaderOpen = false;
private EncodedReader stripeReader;
+ private CompressionCodec codec;
private Object fileKey;
private FileSystem fs;
/**
@@ -166,12 +181,12 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
@SuppressWarnings("unused")
private volatile boolean isPaused = false;
- boolean[] globalIncludes = null;
+ boolean[] globalIncludes = null, sargColumns = null;
private final IoTrace trace;
private Pool<IoTrace> tracePool;
public OrcEncodedDataReader(LowLevelCache lowLevelCache, BufferUsageManager bufferManager,
- OrcMetadataCache metadataCache, Configuration daemonConf, Configuration jobConf,
+ MetadataCache metadataCache, Configuration daemonConf, Configuration jobConf,
FileSplit split, List<Integer> columnIds, SearchArgument sarg, String[] columnNames,
OrcEncodedDataConsumer consumer, QueryFragmentCounters counters,
TypeDescription readerSchema, Pool<IoTrace> tracePool)
@@ -206,7 +221,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
fileKey = determineFileId(fs, split,
HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_CACHE_ALLOW_SYNTHETIC_FILEID),
HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_CACHE_DEFAULT_FS_FILE_ID));
- fileMetadata = getOrReadFileMetadata();
+ fileMetadata = getFileFooterFromCacheOrDisk();
if (readerSchema == null) {
readerSchema = fileMetadata.getSchema();
}
@@ -288,7 +303,6 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
// 3. Apply SARG if needed, and otherwise determine what RGs to read.
int stride = fileMetadata.getRowIndexStride();
ArrayList<OrcStripeMetadata> stripeMetadatas = null;
- boolean[] sargColumns = null;
try {
if (sarg != null && stride != 0) {
// TODO: move this to a common method
@@ -307,7 +321,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
stripeMetadatas = readStripesMetadata(globalIncludes, sargColumns);
}
- // Now, apply SARG if any; w/o sarg, this will just initialize readState.
+ // Now, apply SARG if any; w/o sarg, this will just initialize stripeRgs.
boolean hasData = determineRgsToRead(globalIncludes, stride, stripeMetadatas);
if (!hasData) {
consumer.setDone();
@@ -327,11 +341,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
// 4. Create encoded data reader.
try {
- ensureOrcReader();
- // Reader creating updates HDFS counters, don't do it here.
- DataWrapperForOrc dw = new DataWrapperForOrc();
- stripeReader = orcReader.encodedReader(fileKey, dw, dw, POOL_FACTORY, trace);
- stripeReader.setTracing(LlapIoImpl.ORC_LOGGER.isTraceEnabled());
+ ensureDataReader();
} catch (Throwable t) {
handleReaderError(startTime, t);
return null;
@@ -349,16 +359,15 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
int stripeIx = stripeIxFrom + stripeIxMod;
boolean[] rgs = null;
OrcStripeMetadata stripeMetadata = null;
- StripeInformation stripe;
+ StripeInformation si;
try {
- stripe = fileMetadata.getStripes().get(stripeIx);
-
- LlapIoImpl.ORC_LOGGER.trace("Reading stripe {}: {}, {}", stripeIx, stripe.getOffset(),
- stripe.getLength());
- trace.logReadingStripe(stripeIx, stripe.getOffset(), stripe.getLength());
+ si = fileMetadata.getStripes().get(stripeIx);
+ LlapIoImpl.ORC_LOGGER.trace("Reading stripe {}: {}, {}", stripeIx, si.getOffset(),
+ si.getLength());
+ trace.logReadingStripe(stripeIx, si.getOffset(), si.getLength());
rgs = stripeRgs[stripeIxMod];
if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
- LlapIoImpl.ORC_LOGGER.trace("readState[{}]: {}", stripeIxMod, Arrays.toString(rgs));
+ LlapIoImpl.ORC_LOGGER.trace("stripeRgs[{}]: {}", stripeIxMod, Arrays.toString(rgs));
}
// 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.
@@ -367,47 +376,18 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
if (rgs == RecordReaderImpl.SargApplier.READ_NO_RGS) continue;
// 6.2. Ensure we have stripe metadata. We might have read it before for RG filtering.
- boolean isFoundInCache = false;
if (stripeMetadatas != null) {
stripeMetadata = stripeMetadatas.get(stripeIxMod);
} else {
- if (hasFileId && metadataCache != null) {
- stripeKey.stripeIx = stripeIx;
- stripeMetadata = metadataCache.getStripeMetadata(stripeKey);
- }
- isFoundInCache = (stripeMetadata != null);
- if (!isFoundInCache) {
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
- ensureMetadataReader();
- long startTimeHdfs = counters.startTimeCounter();
- stripeMetadata = new OrcStripeMetadata(new OrcBatchKey(fileKey, stripeIx, 0),
- metadataReader, stripe, globalIncludes, sargColumns,
- orcReader.getSchema(), orcReader.getWriterVersion());
- counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTimeHdfs);
- if (hasFileId && metadataCache != null) {
- OrcStripeMetadata newMetadata = metadataCache.putStripeMetadata(stripeMetadata);
- isFoundInCache = newMetadata != stripeMetadata; // May be cached concurrently.
- stripeMetadata = newMetadata;
- if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
- LlapIoImpl.ORC_LOGGER.trace("Caching stripe {} metadata with includes: {}",
- stripeKey.stripeIx, DebugUtils.toString(globalIncludes));
- }
- }
- }
+ stripeKey.stripeIx = stripeIx;
+ OrcProto.StripeFooter footer = getStripeFooterFromCacheOrDisk(si, stripeKey);
+ stripeMetadata = createOrcStripeMetadataObject(
+ stripeIx, si, footer, globalIncludes, sargColumns);
+ ensureDataReader();
+ stripeReader.readIndexStreams(stripeMetadata.getIndex(),
+ si, footer.getStreamsList(), globalIncludes, sargColumns);
consumer.setStripeMetadata(stripeMetadata);
}
- if (!stripeMetadata.hasAllIndexes(globalIncludes)) {
- if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
- LlapIoImpl.ORC_LOGGER.trace("Updating indexes in stripe {} metadata for includes: {}",
- stripeKey.stripeIx, DebugUtils.toString(globalIncludes));
- }
- assert isFoundInCache; // If it's not fresh from the cache, indexes should be present.
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
- ensureMetadataReader();
- updateLoadedIndexes(stripeMetadata, stripe, globalIncludes, sargColumns);
- } else if (isFoundInCache) {
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_HIT);
- }
} catch (Throwable t) {
handleReaderError(startTime, t);
return null;
@@ -424,7 +404,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
// consumer. It is potentially holding locked buffers, and must perform its own cleanup.
// 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(),
+ stripeReader.readEncodedColumns(stripeIx, si, stripeMetadata.getRowIndexes(),
stripeMetadata.getEncodings(), stripeMetadata.getStreams(), globalIncludes,
rgs, consumer);
} catch (Throwable t) {
@@ -452,6 +432,14 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
tracePool.offer(trace);
}
+ private void ensureDataReader() throws IOException {
+ ensureOrcReader();
+ // Reader creation updates HDFS counters, don't do it here.
+ DataWrapperForOrc dw = new DataWrapperForOrc();
+ stripeReader = orcReader.encodedReader(fileKey, dw, dw, POOL_FACTORY, trace);
+ stripeReader.setTracing(LlapIoImpl.ORC_LOGGER.isTraceEnabled());
+ }
+
private void recordReaderTime(long startTime) {
counters.incrTimeCounter(LlapIOCounters.TOTAL_IO_TIME_NS, startTime);
}
@@ -544,22 +532,6 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
/**
- * In case if stripe metadata in cache does not have all indexes for current query, load
- * the missing one. This is a temporary cludge until real metadata cache becomes available.
- */
- private void updateLoadedIndexes(OrcStripeMetadata stripeMetadata,
- StripeInformation stripe, boolean[] stripeIncludes, boolean[] sargColumns) throws IOException {
- // We only synchronize on write for now - design of metadata cache is very temporary;
- // we pre-allocate the array and never remove entries; so readers should be safe.
- synchronized (stripeMetadata) {
- if (stripeMetadata.hasAllIndexes(stripeIncludes)) return;
- long startTime = counters.startTimeCounter();
- stripeMetadata.loadMissingIndexes(metadataReader, stripe, stripeIncludes, sargColumns);
- counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTime);
- }
- }
-
- /**
* Closes the stripe readers (on error).
*/
private void cleanupReaders() {
@@ -570,9 +542,9 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
// Ignore.
}
}
- if (metadataReader != null) {
+ if (rawDataReader != null && isRawDataReaderOpen) {
try {
- metadataReader.close();
+ rawDataReader.close();
} catch (IOException ex) {
// Ignore.
}
@@ -604,90 +576,194 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
/**
+ * Ensure codec is created for the split, to decode values from cache. Can only be called
+ * after initializing fileMetadata.
+ */
+ private void ensureCodecFromFileMetadata() {
+ if (codec != null) return;
+ codec = WriterImpl.createCodec(fileMetadata.getCompressionKind());
+ }
+
+ /**
* Gets file metadata for the split from cache, or reads it from the file.
*/
- private OrcFileMetadata getOrReadFileMetadata() throws IOException {
- OrcFileMetadata metadata = null;
- if (fileKey != null && metadataCache != null) {
- metadata = metadataCache.getFileMetadata(fileKey);
- if (metadata != null) {
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_HIT);
- return metadata;
- } else {
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
+ private OrcFileMetadata getFileFooterFromCacheOrDisk() throws IOException {
+ LlapBufferOrBuffers tailBuffers = null;
+ List<StripeStatistics> stats = null;
+ List<StripeInformation> stripes = null;
+ boolean hasCache = fileKey != null && metadataCache != null;
+ if (hasCache) {
+ tailBuffers = metadataCache.getFileMetadata(fileKey);
+ if (tailBuffers != null) {
+ try {
+ MemoryBuffer tailBuffer = tailBuffers.getSingleBuffer();
+ ByteBuffer bb = null;
+ if (tailBuffer != null) {
+ bb = tailBuffer.getByteBufferDup();
+ // TODO: remove the copy after ORC-158 and ORC-197
+ // if (bb.isDirect()) {
+ ByteBuffer dupBb = tailBuffer.getByteBufferDup(); // Don't mess with the cached object.
+ bb = ByteBuffer.allocate(dupBb.remaining());
+ bb.put(dupBb);
+ bb.flip();
+ // }
+ } else {
+ // TODO: add the ability to extractFileTail to read from multiple buffers?
+ MemoryBuffer[] tailBufferArray = tailBuffers.getMultipleBuffers();
+ int totalSize = 0;
+ for (MemoryBuffer buf : tailBufferArray) {
+ totalSize += buf.getByteBufferRaw().remaining();
+ }
+ bb = ByteBuffer.allocate(totalSize);
+ for (MemoryBuffer buf : tailBufferArray) {
+ bb.put(buf.getByteBufferDup());
+ }
+ bb.flip();
+ }
+ OrcTail orcTail = ReaderImpl.extractFileTail(bb);
+ FileTail tail = orcTail.getFileTail();
+ stats = orcTail.getStripeStatisticsProto();
+ stripes = new ArrayList<>(tail.getFooter().getStripesCount());
+ for (OrcProto.StripeInformation stripeProto : tail.getFooter().getStripesList()) {
+ stripes.add(new ReaderImpl.StripeInformationImpl(stripeProto));
+ }
+ return new OrcFileMetadata(
+ fileKey, tail.getFooter(), tail.getPostscript(), stats, stripes);
+ } finally {
+ // We don't need the buffer anymore.
+ metadataCache.decRefBuffer(tailBuffers);
+ counters.incrCounter(LlapIOCounters.METADATA_CACHE_HIT);
+ }
}
+ counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
}
ensureOrcReader();
- // We assume this call doesn't touch HDFS because everything is already read; don't add time.
- metadata = new OrcFileMetadata(fileKey, orcReader);
- if (fileKey == null || metadataCache == null) return metadata;
- return metadataCache.putFileMetadata(metadata);
+ ByteBuffer tailBufferBb = orcReader.getSerializedFileFooter();
+ if (hasCache) {
+ tailBuffers = metadataCache.putFileMetadata(fileKey, tailBufferBb);
+ metadataCache.decRefBuffer(tailBuffers); // We don't use the cache's copy of the buffer.
+ }
+ FileTail ft = orcReader.getFileTail();
+ return new OrcFileMetadata(fileKey, ft.getFooter(), ft.getPostscript(),
+ orcReader.getOrcProtoStripeStatistics(), orcReader.getStripes());
+ }
+
+ private OrcProto.StripeFooter buildStripeFooter(
+ List<DiskRange> bcs, int len, CompressionCodec codec, int bufferSize) throws IOException {
+ return OrcProto.StripeFooter.parseFrom(InStream.createCodedInputStream(
+ "footer", bcs, len, codec, bufferSize));
}
/**
* Reads the metadata for all stripes in the file.
*/
private ArrayList<OrcStripeMetadata> readStripesMetadata(
- boolean[] globalInc, boolean[] sargColumns) throws IOException {
+ boolean[] includes, boolean[] sargColumns) throws IOException {
ArrayList<OrcStripeMetadata> result = new ArrayList<OrcStripeMetadata>(stripeRgs.length);
boolean hasFileId = this.fileKey != null;
OrcBatchKey stripeKey = hasFileId ? new OrcBatchKey(fileKey, 0, 0) : null;
for (int stripeIxMod = 0; stripeIxMod < stripeRgs.length; ++stripeIxMod) {
- OrcStripeMetadata value = null;
int stripeIx = stripeIxMod + stripeIxFrom;
- if (hasFileId && metadataCache != null) {
- stripeKey.stripeIx = stripeIx;
- value = metadataCache.getStripeMetadata(stripeKey);
- }
- if (value == null || !value.hasAllIndexes(globalInc)) {
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
- ensureMetadataReader();
- StripeInformation si = fileMetadata.getStripes().get(stripeIx);
- if (value == null) {
- long startTime = counters.startTimeCounter();
- value = new OrcStripeMetadata(new OrcBatchKey(fileKey, stripeIx, 0),
- metadataReader, si, globalInc, sargColumns, orcReader.getSchema(),
- orcReader.getWriterVersion());
- counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTime);
- if (hasFileId && metadataCache != null) {
- value = metadataCache.putStripeMetadata(value);
- if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
- LlapIoImpl.ORC_LOGGER.trace("Caching stripe {} metadata with includes: {}",
- stripeKey.stripeIx, DebugUtils.toString(globalInc));
+ stripeKey.stripeIx = stripeIx;
+ StripeInformation si = fileMetadata.getStripes().get(stripeIx);
+ OrcProto.StripeFooter footer = getStripeFooterFromCacheOrDisk(si, stripeKey);
+ OrcStripeMetadata osm = createOrcStripeMetadataObject(
+ stripeIx, si, footer, includes, sargColumns);
+
+ ensureDataReader();
+ OrcIndex index = osm.getIndex();
+ stripeReader.readIndexStreams(index, si, footer.getStreamsList(), includes, sargColumns);
+ result.add(osm);
+ consumer.setStripeMetadata(osm);
+ }
+ return result;
+ }
+
+ private OrcStripeMetadata createOrcStripeMetadataObject(int stripeIx, StripeInformation si,
+ OrcProto.StripeFooter footer, boolean[] includes, boolean[] sargColumns) throws IOException {
+ Stream.Kind[] bks = sargColumns == null ? null : new Stream.Kind[includes.length];
+ BloomFilterIndex[] bis = sargColumns == null ? null : new BloomFilterIndex[includes.length];
+ return new OrcStripeMetadata(new OrcBatchKey(fileKey, stripeIx, 0), footer,
+ new OrcIndex(new RowIndex[includes.length], bks, bis), si);
+ }
+
+ private OrcProto.StripeFooter getStripeFooterFromCacheOrDisk(
+ StripeInformation si, OrcBatchKey stripeKey) throws IOException {
+ boolean hasCache = fileKey != null && metadataCache != null;
+ if (hasCache) {
+ LlapBufferOrBuffers footerBuffers = metadataCache.getStripeTail(stripeKey);
+ if (footerBuffers != null) {
+ try {
+ counters.incrCounter(LlapIOCounters.METADATA_CACHE_HIT);
+ ensureCodecFromFileMetadata();
+ MemoryBuffer footerBuffer = footerBuffers.getSingleBuffer();
+ if (footerBuffer != null) {
+ ByteBuffer bb = footerBuffer.getByteBufferDup();
+ return buildStripeFooter(Lists.<DiskRange>newArrayList(new BufferChunk(bb, 0)),
+ bb.remaining(), codec, fileMetadata.getCompressionBufferSize());
+ } else {
+ MemoryBuffer[] footerBufferArray = footerBuffers.getMultipleBuffers();
+ int pos = 0;
+ List<DiskRange> bcs = new ArrayList<>(footerBufferArray.length);
+ for (MemoryBuffer buf : footerBufferArray) {
+ ByteBuffer bb = buf.getByteBufferDup();
+ bcs.add(new BufferChunk(bb, pos));
+ pos += bb.remaining();
}
+ return buildStripeFooter(bcs, pos, codec, fileMetadata.getCompressionBufferSize());
}
+ } finally {
+ metadataCache.decRefBuffer(footerBuffers);
}
- // We might have got an old value from cache; recheck it has indexes.
- if (!value.hasAllIndexes(globalInc)) {
- if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) {
- LlapIoImpl.ORC_LOGGER.trace("Updating indexes in stripe {} metadata for includes: {}",
- stripeKey.stripeIx, DebugUtils.toString(globalInc));
- }
- updateLoadedIndexes(value, si, globalInc, sargColumns);
- }
- } else {
- counters.incrCounter(LlapIOCounters.METADATA_CACHE_HIT);
}
- result.add(value);
- consumer.setStripeMetadata(value);
+ counters.incrCounter(LlapIOCounters.METADATA_CACHE_MISS);
}
- return result;
+ long offset = si.getOffset() + si.getIndexLength() + si.getDataLength();
+ long startTime = counters.startTimeCounter();
+ ensureRawDataReader(true);
+ // TODO: add this to metadatareader in ORC - SI => metadata buffer, not just metadata.
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Reading [" + offset + ", "
+ + (offset + si.getFooterLength()) + ") based on " + si);
+ }
+ DiskRangeList footerRange = rawDataReader.readFileData(
+ new DiskRangeList(offset, offset + si.getFooterLength()), 0, false);
+ // LOG.error("Got " + RecordReaderUtils.stringifyDiskRanges(footerRange));
+ counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTime);
+ assert footerRange.next == null; // Can only happens w/zcr for a single input buffer.
+ if (hasCache) {
+ LlapBufferOrBuffers cacheBuf = metadataCache.putStripeTail(stripeKey, footerRange.getData());
+ metadataCache.decRefBuffer(cacheBuf); // We don't use this one.
+ }
+ ByteBuffer bb = footerRange.getData();
+ return buildStripeFooter(Lists.<DiskRange>newArrayList(new BufferChunk(bb, 0)),
+ bb.remaining(), orcReader.getCodec(), orcReader.getCompressionSize());
}
- private void ensureMetadataReader() throws IOException {
+ private void ensureRawDataReader(boolean isOpen) throws IOException {
ensureOrcReader();
- if (metadataReader != null) return;
+ if (rawDataReader != null) {
+ if (!isRawDataReaderOpen && isOpen) {
+ long startTime = counters.startTimeCounter();
+ rawDataReader.open();
+ counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTime);
+ }
+ return;
+ }
long startTime = counters.startTimeCounter();
boolean useZeroCopy = (daemonConf != null) && OrcConf.USE_ZEROCOPY.getBoolean(daemonConf);
- metadataReader = RecordReaderUtils.createDefaultDataReader(
- DataReaderProperties.builder()
- .withBufferSize(orcReader.getCompressionSize())
+ rawDataReader = RecordReaderUtils.createDefaultDataReader(
+ DataReaderProperties.builder().withBufferSize(orcReader.getCompressionSize())
.withCompression(orcReader.getCompressionKind())
- .withFileSystem(fs)
- .withPath(path)
+ .withFileSystem(fs).withPath(path)
.withTypeCount(orcReader.getSchema().getMaximumId() + 1)
.withZeroCopy(useZeroCopy)
.build());
+
+ if (isOpen) {
+ rawDataReader.open();
+ isRawDataReaderOpen = true;
+ }
counters.incrTimeCounter(LlapIOCounters.HDFS_TIME_NS, startTime);
}
@@ -781,7 +857,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
/**
- * Determine which stripes to read for a split. Populates stripeIxFrom and readState.
+ * Determine which stripes to read for a split. Populates stripeIxFrom and stripeRgs.
*/
public void determineStripesToRead() {
// The unit of caching for ORC is (rg x column) (see OrcBatchKey).
@@ -837,8 +913,8 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
public DataWrapperForOrc() throws IOException {
- ensureMetadataReader();
- this.orcDataReader = metadataReader.clone();
+ ensureRawDataReader(false);
+ this.orcDataReader = rawDataReader.clone();
}
@Override
@@ -853,7 +929,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
}
if (gotAllData.value) return result;
return (metadataCache == null) ? result
- : metadataCache.getIncompleteCbs(fileKey, result, baseOffset, factory, gotAllData);
+ : metadataCache.getIncompleteCbs(fileKey, result, baseOffset, gotAllData);
}
@Override
@@ -887,9 +963,6 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
@Override
public void close() throws IOException {
orcDataReader.close();
- if (metadataReader != null) {
- metadataReader.close();
- }
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
new file mode 100644
index 0000000..9d7951e
--- /dev/null
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
@@ -0,0 +1,537 @@
+/**
+ * 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.metadata;
+
+import org.apache.hadoop.hive.common.FileUtils;
+
+import org.apache.hadoop.hive.common.io.FileMetadataCache;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.hadoop.hive.common.io.DiskRange;
+import org.apache.hadoop.hive.common.io.DiskRangeList;
+import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
+import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
+import org.apache.hadoop.hive.llap.cache.EvictionAwareAllocator;
+import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
+import org.apache.hadoop.hive.llap.cache.LlapAllocatorBuffer;
+import org.apache.hadoop.hive.llap.cache.LlapOomDebugDump;
+import org.apache.hadoop.hive.llap.cache.LowLevelCachePolicy;
+import org.apache.hadoop.hive.llap.cache.MemoryManager;
+import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
+import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
+import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
+import org.apache.hadoop.hive.ql.io.orc.encoded.OrcBatchKey;
+
+public class MetadataCache implements LlapOomDebugDump, FileMetadataCache {
+ private final ConcurrentHashMap<Object, LlapBufferOrBuffers> metadata =
+ new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<Object, OrcFileEstimateErrors> estimateErrors;
+ private final MemoryManager memoryManager;
+ private final LowLevelCachePolicy policy;
+ private final EvictionAwareAllocator allocator;
+ private final LlapDaemonCacheMetrics metrics;
+
+ public MetadataCache(EvictionAwareAllocator allocator, MemoryManager memoryManager,
+ LowLevelCachePolicy policy, boolean useEstimateCache, LlapDaemonCacheMetrics metrics) {
+ this.memoryManager = memoryManager;
+ this.allocator = allocator;
+ this.policy = policy;
+ this.metrics = metrics;
+ this.estimateErrors = useEstimateCache
+ ? new ConcurrentHashMap<Object, OrcFileEstimateErrors>() : null;
+ }
+
+ public void putIncompleteCbs(Object fileKey, DiskRange[] ranges, long baseOffset) {
+ if (estimateErrors == null) return;
+ OrcFileEstimateErrors errorData = estimateErrors.get(fileKey);
+ boolean isNew = false;
+ // We should technically update memory usage if updating the old object, but we don't do it
+ // for now; there is no mechanism to properly notify the cache policy/etc. wrt parallel evicts.
+ if (errorData == null) {
+ errorData = new OrcFileEstimateErrors(fileKey);
+ for (DiskRange range : ranges) {
+ errorData.addError(range.getOffset(), range.getLength(), baseOffset);
+ }
+ long memUsage = errorData.estimateMemoryUsage();
+ memoryManager.reserveMemory(memUsage);
+ OrcFileEstimateErrors old = estimateErrors.putIfAbsent(fileKey, errorData);
+ if (old != null) {
+ errorData = old;
+ memoryManager.releaseMemory(memUsage);
+ policy.notifyLock(errorData);
+ } else {
+ isNew = true;
+ policy.cache(errorData, Priority.NORMAL);
+ }
+ } else {
+ policy.notifyLock(errorData);
+ }
+ if (!isNew) {
+ for (DiskRange range : ranges) {
+ errorData.addError(range.getOffset(), range.getLength(), baseOffset);
+ }
+ }
+ policy.notifyUnlock(errorData);
+ }
+
+ public DiskRangeList getIncompleteCbs(
+ Object fileKey, DiskRangeList ranges, long baseOffset, BooleanRef gotAllData) {
+ if (estimateErrors == null) return ranges;
+ OrcFileEstimateErrors errors = estimateErrors.get(fileKey);
+ if (errors == null) return ranges;
+ policy.notifyLock(errors);
+ policy.notifyUnlock(errors); // Never locked for eviction; Java object.
+ return errors.getIncompleteCbs(ranges, baseOffset, gotAllData);
+ }
+
+ public void notifyEvicted(LlapMetadataBuffer<?> buffer) {
+ LlapBufferOrBuffers removed = metadata.remove(buffer.getKey());
+ if (removed == null) return;
+ if (removed.getSingleBuffer() != null) {
+ assert removed.getSingleBuffer() == buffer;
+ return;
+ }
+ discardMultiBuffer(removed);
+ }
+
+ public void notifyEvicted(OrcFileEstimateErrors buffer) {
+ estimateErrors.remove(buffer.getFileKey());
+ }
+
+ @Override
+ public String debugDumpForOom() {
+ StringBuilder sb = new StringBuilder();
+ debugDumpShort(sb);
+ return sb.toString();
+ }
+
+ @Override
+ public void debugDumpShort(StringBuilder sb) {
+ // TODO: perhaps add counters for separate things and multiple buffer cases.
+ sb.append("\nMetadata cache state: ").append(metadata.size()).append(
+ " files and stripes, ").append(estimateErrors.size()).append(" files w/ORC estimate");
+ }
+
+ @Override
+ public LlapBufferOrBuffers getFileMetadata(Object fileKey) {
+ return getInternal(fileKey);
+ }
+
+ public LlapBufferOrBuffers getStripeTail(OrcBatchKey stripeKey) {
+ return getInternal(new StripeKey(stripeKey.fileKey, stripeKey.stripeIx));
+ }
+
+ private LlapBufferOrBuffers getInternal(Object key) {
+ LlapBufferOrBuffers result = metadata.get(key);
+ if (result == null) return null;
+ if (!lockBuffer(result, true)) {
+ // No need to discard the buffer we cannot lock - eviction takes care of that.
+ metadata.remove(key, result);
+ return null;
+ }
+ return result;
+ }
+
+ @Override
+ public LlapBufferOrBuffers putFileMetadata(Object fileKey, ByteBuffer tailBuffer) {
+ return putInternal(fileKey, tailBuffer);
+ }
+
+ public LlapBufferOrBuffers putStripeTail(OrcBatchKey stripeKey, ByteBuffer tailBuffer) {
+ return putInternal(new StripeKey(stripeKey.fileKey, stripeKey.stripeIx), tailBuffer);
+ }
+
+
+ @Override
+ public LlapBufferOrBuffers putFileMetadata(
+ Object fileKey, int length, InputStream is) throws IOException {
+ LlapBufferOrBuffers result = null;
+ while (true) { // Overwhelmingly executes once, or maybe twice (replacing stale value).
+ LlapBufferOrBuffers oldVal = metadata.get(fileKey);
+ if (oldVal == null) {
+ result = wrapBbForFile(result, fileKey, length, is);
+ if (!lockBuffer(result, false)) {
+ throw new AssertionError("Cannot lock a newly created value " + result);
+ }
+ oldVal = metadata.putIfAbsent(fileKey, result);
+ if (oldVal == null) {
+ cacheInPolicy(result); // Cached successfully, add to policy.
+ return result;
+ }
+ }
+ if (lockOldVal(fileKey, result, oldVal)) {
+ return oldVal;
+ }
+ // We found some old value but couldn't incRef it; remove it.
+ metadata.remove(fileKey, oldVal);
+ }
+ }
+
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private LlapBufferOrBuffers wrapBbForFile(LlapBufferOrBuffers result,
+ Object fileKey, int length, InputStream stream) throws IOException {
+ if (result != null) return result;
+ int maxAlloc = allocator.getMaxAllocation();
+ LlapMetadataBuffer<Object>[] largeBuffers = null;
+ if (maxAlloc < length) {
+ largeBuffers = new LlapMetadataBuffer[length / maxAlloc];
+ for (int i = 0; i < largeBuffers.length; ++i) {
+ largeBuffers[i] = new LlapMetadataBuffer<Object>(fileKey);
+ }
+ allocator.allocateMultiple(largeBuffers, maxAlloc, null);
+ for (int i = 0; i < largeBuffers.length; ++i) {
+ readIntoCacheBuffer(stream, maxAlloc, largeBuffers[i]);
+ }
+ }
+ int smallSize = length % maxAlloc;
+ if (smallSize == 0) {
+ return new LlapMetadataBuffers(largeBuffers);
+ } else {
+ LlapMetadataBuffer<Object>[] smallBuffer = new LlapMetadataBuffer[1];
+ smallBuffer[0] = new LlapMetadataBuffer(fileKey);
+ allocator.allocateMultiple(smallBuffer, length, null);
+ readIntoCacheBuffer(stream, smallSize, smallBuffer[0]);
+ if (largeBuffers == null) {
+ return smallBuffer[0]; // This is the overwhelmingly common case.
+ } else {
+ LlapMetadataBuffer<Object>[] cacheData = new LlapMetadataBuffer[largeBuffers.length + 1];
+ System.arraycopy(largeBuffers, 0, cacheData, 0, largeBuffers.length);
+ cacheData[largeBuffers.length] = smallBuffer[0];
+ return new LlapMetadataBuffers<Object>(cacheData);
+ }
+ }
+ }
+
+ private static void readIntoCacheBuffer(
+ InputStream stream, int length, MemoryBuffer dest) throws IOException {
+ ByteBuffer bb = dest.getByteBufferRaw();
+ int pos = bb.position();
+ bb.limit(pos + length);
+ // TODO: SeekableInputStream.readFully eventually calls a Hadoop method that used to be
+ // buggy in 2.7 and also anyway just does a copy for a direct buffer. Do a copy here.
+ // ((SeekableInputStream)stream).readFully(bb);
+ FileUtils.readFully(stream, length, bb);
+ bb.position(pos);
+ }
+
+ private <T> LlapBufferOrBuffers putInternal(T key, ByteBuffer tailBuffer) {
+ LlapBufferOrBuffers result = null;
+ while (true) { // Overwhelmingly executes once, or maybe twice (replacing stale value).
+ LlapBufferOrBuffers oldVal = metadata.get(key);
+ if (oldVal == null) {
+ result = wrapBb(result, key, tailBuffer);
+ oldVal = metadata.putIfAbsent(key, result);
+ if (oldVal == null) {
+ cacheInPolicy(result); // Cached successfully, add to policy.
+ return result;
+ }
+ }
+ if (lockOldVal(key, result, oldVal)) {
+ return oldVal;
+ }
+ // We found some old value but couldn't incRef it; remove it.
+ metadata.remove(key, oldVal);
+ }
+ }
+
+ private void cacheInPolicy(LlapBufferOrBuffers buffers) {
+ LlapAllocatorBuffer singleBuffer = buffers.getSingleLlapBuffer();
+ if (singleBuffer != null) {
+ policy.cache(singleBuffer, Priority.HIGH);
+ return;
+ }
+ for (LlapAllocatorBuffer buffer : buffers.getMultipleLlapBuffers()) {
+ policy.cache(buffer, Priority.HIGH);
+ }
+ }
+
+ private <T extends LlapBufferOrBuffers> boolean lockOldVal(Object key, T newVal, T oldVal) {
+ if (LlapIoImpl.CACHE_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.CACHE_LOGGER.trace("Trying to cache when metadata is already cached for" +
+ " {}; old {}, new {}", key, oldVal, newVal);
+ }
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Locking {} due to cache collision", oldVal);
+ }
+ if (lockBuffer(oldVal, true)) {
+ // We found an old, valid block for this key in the cache.
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Unlocking {} due to cache collision with {}",
+ newVal, oldVal);
+ }
+
+ if (newVal != null) {
+ unlockBuffer(newVal, false);
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public void decRefBuffer(MemoryBufferOrBuffers buffer) {
+ if (!(buffer instanceof LlapBufferOrBuffers)) {
+ throw new AssertionError(buffer.getClass());
+ }
+ unlockBuffer((LlapBufferOrBuffers)buffer, true);
+ }
+
+ private <T> LlapBufferOrBuffers wrapBb(
+ LlapBufferOrBuffers result, T key, ByteBuffer tailBuffer) {
+ if (result != null) return result;
+ if (tailBuffer.remaining() <= allocator.getMaxAllocation()) {
+ // The common case by far.
+ return wrapSmallBb(new LlapMetadataBuffer<T>(key), tailBuffer);
+ } else {
+ int allocCount = determineAllocCount(tailBuffer);
+ @SuppressWarnings("unchecked")
+ LlapMetadataBuffer<T>[] results = new LlapMetadataBuffer[allocCount];
+ for (int i = 0; i < allocCount; ++i) {
+ results[i] = new LlapMetadataBuffer<T>(key);
+ }
+ wrapLargeBb(results, tailBuffer);
+ return new LlapMetadataBuffers<T>(results);
+ }
+ }
+
+ private <T extends LlapAllocatorBuffer> T wrapSmallBb(T result, ByteBuffer tailBuffer) {
+ // Note: we pass in null factory because we allocate objects here. We could also pass a
+ // per-call factory that would set fileKey; or set it after put.
+ allocator.allocateMultiple(new MemoryBuffer[] { result }, tailBuffer.remaining(), null);
+ return putBufferToDest(tailBuffer.duplicate(), result);
+ }
+
+ private <T extends LlapAllocatorBuffer> void wrapLargeBb(T[] results, ByteBuffer tailBuffer) {
+ // Note: we pass in null factory because we allocate objects here. We could also pass a
+ // per-call factory that would set fileKey; or set it after put.
+ allocator.allocateMultiple(results, allocator.getMaxAllocation(), null);
+ ByteBuffer src = tailBuffer.duplicate();
+ int pos = src.position(), remaining = src.remaining();
+ for (int i = 0; i < results.length; ++i) {
+ T result = results[i];
+ int toPut = Math.min(remaining, result.getByteBufferRaw().remaining());
+ assert toPut > 0;
+ src.position(pos);
+ src.limit(pos + toPut);
+ pos += toPut;
+ remaining -= toPut;
+ putBufferToDest(src, result);
+ }
+ }
+
+ private <T extends LlapAllocatorBuffer> T putBufferToDest(ByteBuffer src, T result) {
+ ByteBuffer dest = result.getByteBufferRaw();
+ int startPos = dest.position();
+ dest.put(src);
+ int newPos = dest.position();
+ dest.position(startPos);
+ dest.limit(newPos);
+ boolean canLock = lockOneBuffer(result, false);
+ assert canLock;
+ return result;
+ }
+
+ public int determineAllocCount(ByteBuffer tailBuffer) {
+ int total = tailBuffer.remaining(), maxAlloc = allocator.getMaxAllocation();
+ return total / maxAlloc + ((total % maxAlloc) > 0 ? 1 : 0);
+ }
+
+ private boolean lockBuffer(LlapBufferOrBuffers buffers, boolean doNotifyPolicy) {
+ LlapAllocatorBuffer buffer = buffers.getSingleLlapBuffer();
+ if (buffer != null) {
+ return lockOneBuffer(buffer, doNotifyPolicy);
+ }
+ LlapAllocatorBuffer[] bufferArray = buffers.getMultipleLlapBuffers();
+ for (int i = 0; i < bufferArray.length; ++i) {
+ if (lockOneBuffer(bufferArray[i], doNotifyPolicy)) continue;
+ for (int j = 0; j < i; ++j) {
+ unlockSingleBuffer(buffer, true);
+ }
+ discardMultiBuffer(buffers);
+ return false;
+ }
+ return true;
+ }
+
+ private void discardMultiBuffer(LlapBufferOrBuffers removed) {
+ long memoryFreed = 0;
+ for (LlapAllocatorBuffer buf : removed.getMultipleLlapBuffers()) {
+ long memUsage = buf.getMemoryUsage();
+ // We cannot just deallocate the buffer, as it can hypothetically have users.
+ int result = buf.invalidate();
+ switch (result) {
+ case LlapAllocatorBuffer.INVALIDATE_ALREADY_INVALID: continue; // Nothing to do.
+ case LlapAllocatorBuffer.INVALIDATE_FAILED: {
+ // Someone is using this buffer; eventually, it will be evicted.
+ continue;
+ }
+ case LlapAllocatorBuffer.INVALIDATE_OK: {
+ memoryFreed += memUsage;
+ allocator.deallocateEvicted(buf);
+ break;
+ }
+ default: throw new AssertionError(result);
+ }
+ }
+ memoryManager.releaseMemory(memoryFreed);
+ }
+
+ public boolean lockOneBuffer(LlapAllocatorBuffer buffer, boolean doNotifyPolicy) {
+ int rc = buffer.incRef();
+ if (rc > 0) {
+ metrics.incrCacheNumLockedBuffers();
+ }
+ if (doNotifyPolicy && rc == 1) {
+ // We have just locked a buffer that wasn't previously locked.
+ policy.notifyLock(buffer);
+ }
+ return rc > 0;
+ }
+
+ private void unlockBuffer(LlapBufferOrBuffers buffers, boolean isCached) {
+ LlapAllocatorBuffer singleBuffer = buffers.getSingleLlapBuffer();
+ if (singleBuffer != null) {
+ unlockSingleBuffer(singleBuffer, isCached);
+ return;
+ }
+ for (LlapAllocatorBuffer buffer : buffers.getMultipleLlapBuffers()) {
+ unlockSingleBuffer(buffer, isCached);
+ }
+ }
+
+ private void unlockSingleBuffer(LlapAllocatorBuffer buffer, boolean isCached) {
+ boolean isLastDecref = (buffer.decRef() == 0);
+ if (isLastDecref) {
+ if (isCached) {
+ policy.notifyUnlock(buffer);
+ } else {
+ allocator.deallocate(buffer);
+ }
+ }
+ metrics.decrCacheNumLockedBuffers();
+ }
+
+ private final static class StripeKey {
+ private final Object fileKey;
+ private final int stripeIx;
+
+ public StripeKey(Object fileKey, int stripeIx) {
+ this.fileKey = fileKey;
+ this.stripeIx = stripeIx;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ return (prime + ((fileKey == null) ? 0 : fileKey.hashCode())) * prime + stripeIx;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (!(obj instanceof StripeKey)) return false;
+ StripeKey other = (StripeKey)obj;
+ return ((fileKey == null) == (other.fileKey == null))
+ && (fileKey == null || fileKey.equals(other.fileKey)) && (stripeIx == other.stripeIx);
+ }
+ }
+
+ public static interface LlapBufferOrBuffers extends MemoryBufferOrBuffers {
+ LlapAllocatorBuffer getSingleLlapBuffer();
+ LlapAllocatorBuffer[] getMultipleLlapBuffers();
+ }
+
+
+ public final static class LlapMetadataBuffer<T>
+ extends LlapAllocatorBuffer implements LlapBufferOrBuffers {
+ private final T key;
+
+ public LlapMetadataBuffer(T key) {
+ this.key = key;
+ }
+
+ @Override
+ public void notifyEvicted(EvictionDispatcher evictionDispatcher) {
+ evictionDispatcher.notifyEvicted(this);
+ }
+
+ public T getKey() {
+ return key;
+ }
+
+ @Override
+ public LlapAllocatorBuffer getSingleBuffer() {
+ return this;
+ }
+
+ @Override
+ public LlapAllocatorBuffer[] getMultipleBuffers() {
+ return null;
+ }
+
+
+ @Override
+ public LlapAllocatorBuffer getSingleLlapBuffer() {
+ return this;
+ }
+
+ @Override
+ public LlapAllocatorBuffer[] getMultipleLlapBuffers() {
+ return null;
+ }
+ }
+
+ public final static class LlapMetadataBuffers<T> implements LlapBufferOrBuffers {
+ private final LlapMetadataBuffer<T>[] buffers;
+
+ public LlapMetadataBuffers(LlapMetadataBuffer<T>[] buffers) {
+ this.buffers = buffers;
+ }
+
+ @Override
+ public LlapAllocatorBuffer getSingleBuffer() {
+ return null;
+ }
+
+ @Override
+ public LlapAllocatorBuffer[] getMultipleBuffers() {
+ return buffers;
+ }
+
+ @Override
+ public LlapAllocatorBuffer getSingleLlapBuffer() {
+ return null;
+ }
+
+ @Override
+ public LlapAllocatorBuffer[] getMultipleLlapBuffers() {
+ return buffers;
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
index dc053ee..6cf9563 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
@@ -23,7 +23,6 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
-import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
import org.apache.hadoop.hive.common.io.DiskRangeList.MutateHelper;
import org.apache.hadoop.hive.llap.IncrementalObjectSizeEstimator;
import org.apache.hadoop.hive.llap.IncrementalObjectSizeEstimator.ObjectEstimator;
@@ -61,8 +60,8 @@ public class OrcFileEstimateErrors extends LlapCacheableBuffer {
}
}
- public DiskRangeList getIncompleteCbs(DiskRangeList ranges, long baseOffset,
- DiskRangeListFactory factory, BooleanRef gotAllData) {
+ public DiskRangeList getIncompleteCbs(
+ DiskRangeList ranges, long baseOffset, BooleanRef gotAllData) {
DiskRangeList prev = ranges.prev;
if (prev == null) {
prev = new MutateHelper(ranges);
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/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 b9d7a77..8af161f 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
@@ -18,31 +18,22 @@
package org.apache.hadoop.hive.llap.io.metadata;
-import com.google.common.annotations.VisibleForTesting;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import org.apache.hadoop.hive.llap.IncrementalObjectSizeEstimator;
-import org.apache.hadoop.hive.llap.IncrementalObjectSizeEstimator.ObjectEstimator;
-import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
-import org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer;
-import org.apache.hadoop.hive.ql.io.SyntheticFileId;
+
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
-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.OrcProto.StripeStatistics;
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
* of ORC use different info. Ideally we would get rid of protobuf structs in code beyond reading,
* or instead use protobuf structs everywhere instead of the mix of things like now.
*/
-public final class OrcFileMetadata extends LlapCacheableBuffer
- implements FileMetadata, ConsumerFileMetadata {
+public final class OrcFileMetadata implements FileMetadata, ConsumerFileMetadata {
private final List<StripeInformation> stripes;
private final List<Integer> versionList;
private final List<OrcProto.StripeStatistics> stripeStats;
@@ -58,91 +49,22 @@ public final class OrcFileMetadata extends LlapCacheableBuffer
private final long numberOfRows;
private final boolean isOriginalFormat;
- private final int estimatedMemUsage;
-
- private final static HashMap<Class<?>, ObjectEstimator> SIZE_ESTIMATORS;
- private final static ObjectEstimator SIZE_ESTIMATOR;
- static {
- OrcFileMetadata ofm = createDummy(new SyntheticFileId());
- SIZE_ESTIMATORS = IncrementalObjectSizeEstimator.createEstimators(ofm);
- IncrementalObjectSizeEstimator.addEstimator(
- "com.google.protobuf.LiteralByteString", SIZE_ESTIMATORS);
- // Add long for the regular file ID estimation.
- IncrementalObjectSizeEstimator.createEstimators(Long.class, SIZE_ESTIMATORS);
- SIZE_ESTIMATOR = SIZE_ESTIMATORS.get(OrcFileMetadata.class);
- }
-
- @VisibleForTesting
- public static OrcFileMetadata createDummy(Object fileKey) {
- OrcFileMetadata ofm = new OrcFileMetadata(fileKey);
- ofm.stripes.add(new ReaderImpl.StripeInformationImpl(
- OrcProto.StripeInformation.getDefaultInstance()));
- ofm.fileStats.add(OrcProto.ColumnStatistics.getDefaultInstance());
- ofm.stripeStats.add(OrcProto.StripeStatistics.newBuilder().addColStats(createStatsDummy()).build());
- ofm.types.add(OrcProto.Type.newBuilder().addFieldNames("a").addSubtypes(0).build());
- ofm.versionList.add(0);
- return ofm;
- }
-
- static OrcProto.ColumnStatistics.Builder createStatsDummy() {
- return OrcProto.ColumnStatistics.newBuilder().setBucketStatistics(
- OrcProto.BucketStatistics.newBuilder().addCount(0)).setStringStatistics(
- OrcProto.StringStatistics.newBuilder().setMaximum("zzz"));
- }
-
- // Ctor for memory estimation and tests
- private OrcFileMetadata(Object fileKey) {
- this.fileKey = fileKey;
- stripes = new ArrayList<StripeInformation>();
- versionList = new ArrayList<Integer>();
- fileStats = new ArrayList<>();
- stripeStats = new ArrayList<>();
- types = new ArrayList<>();
- writerVersionNum = metadataSize = compressionBufferSize = rowIndexStride = 0;
- contentLength = numberOfRows = 0;
- estimatedMemUsage = 0;
- isOriginalFormat = false;
- compressionKind = CompressionKind.NONE;
- }
-
- public OrcFileMetadata(Object fileKey, Reader reader) {
+ public OrcFileMetadata(Object fileKey, OrcProto.Footer footer, OrcProto.PostScript ps,
+ List<StripeStatistics> stats, List<StripeInformation> stripes) {
+ this.stripeStats = stats;
+ this.compressionKind = CompressionKind.valueOf(ps.getCompression().name());
+ this.compressionBufferSize = (int)ps.getCompressionBlockSize();
+ this.stripes = stripes;
+ this.isOriginalFormat = OrcInputFormat.isOriginal(footer);
+ this.writerVersionNum = ps.getWriterVersion();
+ this.versionList = ps.getVersionList();
+ this.metadataSize = (int) ps.getMetadataLength();
+ this.types = footer.getTypesList();
+ this.rowIndexStride = footer.getRowIndexStride();
+ this.contentLength = footer.getContentLength();
+ this.numberOfRows = footer.getNumberOfRows();
+ this.fileStats = footer.getStatisticsList();
this.fileKey = fileKey;
- this.stripeStats = reader.getOrcProtoStripeStatistics();
- this.compressionKind = reader.getCompressionKind();
- this.compressionBufferSize = reader.getCompressionSize();
- this.stripes = reader.getStripes();
- this.isOriginalFormat = OrcInputFormat.isOriginal(reader);
- this.writerVersionNum = reader.getWriterVersion().getId();
- this.versionList = reader.getVersionList();
- this.metadataSize = reader.getMetadataSize();
- this.types = reader.getTypes();
- this.rowIndexStride = reader.getRowIndexStride();
- this.contentLength = reader.getContentLength();
- this.numberOfRows = reader.getNumberOfRows();
- this.fileStats = reader.getOrcProtoFileStatistics();
-
- this.estimatedMemUsage = SIZE_ESTIMATOR.estimate(this, SIZE_ESTIMATORS);
- }
-
- // LlapCacheableBuffer
- @Override
- public void notifyEvicted(EvictionDispatcher evictionDispatcher) {
- evictionDispatcher.notifyEvicted(this);
- }
-
- @Override
- protected int invalidate() {
- return INVALIDATE_OK; // relies on GC, so it can always be evicted now.
- }
-
- @Override
- public long getMemoryUsage() {
- return estimatedMemUsage;
- }
-
- @Override
- protected boolean isLocked() {
- return false;
}
// FileMetadata
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcMetadataCache.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcMetadataCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcMetadataCache.java
deleted file mode 100644
index 601b622..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcMetadataCache.java
+++ /dev/null
@@ -1,163 +0,0 @@
-/**
- * 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.metadata;
-
-import org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer;
-
-import java.io.IOException;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.hadoop.hive.common.io.DiskRange;
-import org.apache.hadoop.hive.common.io.DiskRangeList;
-import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
-import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
-import org.apache.hadoop.hive.llap.cache.LlapOomDebugDump;
-import org.apache.hadoop.hive.llap.cache.LowLevelCachePolicy;
-import org.apache.hadoop.hive.llap.cache.MemoryManager;
-import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
-import org.apache.hadoop.hive.ql.io.orc.encoded.OrcBatchKey;
-
-public class OrcMetadataCache implements LlapOomDebugDump {
- private final ConcurrentHashMap<Object, OrcFileMetadata> metadata = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<OrcBatchKey, OrcStripeMetadata> stripeMetadata =
- new ConcurrentHashMap<>();
- private final ConcurrentHashMap<Object, OrcFileEstimateErrors> estimateErrors;
- private final MemoryManager memoryManager;
- private final LowLevelCachePolicy policy;
-
- public OrcMetadataCache(MemoryManager memoryManager, LowLevelCachePolicy policy,
- boolean useEstimateCache) {
- this.memoryManager = memoryManager;
- this.policy = policy;
- this.estimateErrors = useEstimateCache
- ? new ConcurrentHashMap<Object, OrcFileEstimateErrors>() : null;
- }
-
- public OrcFileMetadata putFileMetadata(OrcFileMetadata metaData) {
- long memUsage = metaData.getMemoryUsage();
- memoryManager.reserveMemory(memUsage);
- OrcFileMetadata val = metadata.putIfAbsent(metaData.getFileKey(), metaData);
- // See OrcFileMetadata; it is always unlocked, so we just "touch" it here to simulate use.
- return touchOnPut(metaData, val, memUsage);
- }
-
- public OrcStripeMetadata putStripeMetadata(OrcStripeMetadata metaData) {
- long memUsage = metaData.getMemoryUsage();
- memoryManager.reserveMemory(memUsage);
- OrcStripeMetadata val = stripeMetadata.putIfAbsent(metaData.getKey(), metaData);
- // See OrcStripeMetadata; it is always unlocked, so we just "touch" it here to simulate use.
- return touchOnPut(metaData, val, memUsage);
- }
-
- private <T extends LlapCacheableBuffer> T touchOnPut(T newVal, T oldVal, long memUsage) {
- if (oldVal == null) {
- oldVal = newVal;
- policy.cache(oldVal, Priority.HIGH);
- } else {
- memoryManager.releaseMemory(memUsage);
- policy.notifyLock(oldVal);
- }
- policy.notifyUnlock(oldVal);
- return oldVal;
- }
-
-
- public void putIncompleteCbs(Object fileKey, DiskRange[] ranges, long baseOffset) {
- if (estimateErrors == null) return;
- OrcFileEstimateErrors errorData = estimateErrors.get(fileKey);
- boolean isNew = false;
- // We should technically update memory usage if updating the old object, but we don't do it
- // for now; there is no mechanism to properly notify the cache policy/etc. wrt parallel evicts.
- if (errorData == null) {
- errorData = new OrcFileEstimateErrors(fileKey);
- for (DiskRange range : ranges) {
- errorData.addError(range.getOffset(), range.getLength(), baseOffset);
- }
- long memUsage = errorData.estimateMemoryUsage();
- memoryManager.reserveMemory(memUsage);
- OrcFileEstimateErrors old = estimateErrors.putIfAbsent(fileKey, errorData);
- if (old != null) {
- errorData = old;
- memoryManager.releaseMemory(memUsage);
- policy.notifyLock(errorData);
- } else {
- isNew = true;
- policy.cache(errorData, Priority.NORMAL);
- }
- }
- if (!isNew) {
- for (DiskRange range : ranges) {
- errorData.addError(range.getOffset(), range.getLength(), baseOffset);
- }
- }
- policy.notifyUnlock(errorData);
- }
-
- public OrcStripeMetadata getStripeMetadata(OrcBatchKey stripeKey) throws IOException {
- return touchOnGet(stripeMetadata.get(stripeKey));
- }
-
- public OrcFileMetadata getFileMetadata(Object fileKey) throws IOException {
- return touchOnGet(metadata.get(fileKey));
- }
-
- private <T extends LlapCacheableBuffer> T touchOnGet(T result) {
- if (result != null) {
- policy.notifyLock(result);
- policy.notifyUnlock(result); // Never locked for eviction; Java object.
- }
- return result;
- }
-
- public DiskRangeList getIncompleteCbs(Object fileKey, DiskRangeList ranges, long baseOffset,
- DiskRangeListFactory factory, BooleanRef gotAllData) {
- if (estimateErrors == null) return ranges;
- OrcFileEstimateErrors errors = estimateErrors.get(fileKey);
- if (errors == null) return ranges;
- return errors.getIncompleteCbs(ranges, baseOffset, factory, gotAllData);
- }
-
- public void notifyEvicted(OrcFileMetadata buffer) {
- metadata.remove(buffer.getFileKey());
- // See OrcFileMetadata - we don't clear the object, it will be GCed when released by users.
- }
-
- public void notifyEvicted(OrcStripeMetadata buffer) {
- stripeMetadata.remove(buffer.getKey());
- // See OrcStripeMetadata - we don't clear the object, it will be GCed when released by users.
- }
-
- public void notifyEvicted(OrcFileEstimateErrors buffer) {
- estimateErrors.remove(buffer.getFileKey());
- }
-
- @Override
- public String debugDumpForOom() {
- StringBuilder sb = new StringBuilder();
- debugDumpShort(sb);
- return sb.toString();
- }
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- sb.append("\nORC metadata cache state: ").append(metadata.size()).append(" files, ")
- .append(stripeMetadata.size()).append(" stripes, ").append(estimateErrors.size())
- .append(" files w/ORC estimate");
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/50fb6f3c/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
index 4565d11..92b9c8f 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
@@ -19,115 +19,30 @@ package org.apache.hadoop.hive.llap.io.metadata;
import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import org.apache.hadoop.hive.llap.IncrementalObjectSizeEstimator;
-import org.apache.hadoop.hive.llap.IncrementalObjectSizeEstimator.ObjectEstimator;
-import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
-import org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer;
-import org.apache.hadoop.hive.ql.io.SyntheticFileId;
-import org.apache.hadoop.hive.ql.io.orc.OrcFile;
import org.apache.hadoop.hive.ql.io.orc.encoded.OrcBatchKey;
-import org.apache.orc.DataReader;
import org.apache.orc.OrcProto;
import org.apache.orc.OrcProto.RowIndexEntry;
import org.apache.orc.StripeInformation;
-import org.apache.orc.TypeDescription;
import org.apache.orc.impl.OrcIndex;
-public class OrcStripeMetadata extends LlapCacheableBuffer implements ConsumerStripeMetadata {
- private final TypeDescription schema;
+public class OrcStripeMetadata implements ConsumerStripeMetadata {
private final OrcBatchKey stripeKey;
private final List<OrcProto.ColumnEncoding> encodings;
private final List<OrcProto.Stream> streams;
private final String writerTimezone;
private final long rowCount;
private OrcIndex rowIndex;
- private OrcFile.WriterVersion writerVersion;
-
- private final int estimatedMemUsage;
-
- private final static HashMap<Class<?>, ObjectEstimator> SIZE_ESTIMATORS;
- private final static ObjectEstimator SIZE_ESTIMATOR;
- static {
- OrcStripeMetadata osm = createDummy(new SyntheticFileId());
- SIZE_ESTIMATORS = IncrementalObjectSizeEstimator.createEstimators(osm);
- IncrementalObjectSizeEstimator.addEstimator(
- "com.google.protobuf.LiteralByteString", SIZE_ESTIMATORS);
- // Add long for the regular file ID estimation.
- IncrementalObjectSizeEstimator.createEstimators(Long.class, SIZE_ESTIMATORS);
- SIZE_ESTIMATOR = SIZE_ESTIMATORS.get(OrcStripeMetadata.class);
- }
- public OrcStripeMetadata(OrcBatchKey stripeKey, DataReader mr, StripeInformation stripe,
- boolean[] includes, boolean[] sargColumns, TypeDescription schema,
- OrcFile.WriterVersion writerVersion) throws IOException {
- this.schema = schema;
+ public OrcStripeMetadata(OrcBatchKey stripeKey, OrcProto.StripeFooter footer,
+ OrcIndex orcIndex, StripeInformation stripe) throws IOException {
this.stripeKey = stripeKey;
- OrcProto.StripeFooter footer = mr.readStripeFooter(stripe);
streams = footer.getStreamsList();
encodings = footer.getColumnsList();
writerTimezone = footer.getWriterTimezone();
rowCount = stripe.getNumberOfRows();
- rowIndex = mr.readRowIndex(stripe, schema, footer, true, includes, null,
- sargColumns, writerVersion, null, null);
-
- estimatedMemUsage = SIZE_ESTIMATOR.estimate(this, SIZE_ESTIMATORS);
- this.writerVersion = writerVersion;
- }
-
- private OrcStripeMetadata(Object id) {
- stripeKey = new OrcBatchKey(id, 0, 0);
- encodings = new ArrayList<>();
- streams = new ArrayList<>();
- writerTimezone = "";
- schema = TypeDescription.fromString("struct<x:int>");
- rowCount = estimatedMemUsage = 0;
- }
-
- @VisibleForTesting
- public static OrcStripeMetadata createDummy(Object id) {
- OrcStripeMetadata dummy = new OrcStripeMetadata(id);
- dummy.encodings.add(OrcProto.ColumnEncoding.getDefaultInstance());
- dummy.streams.add(OrcProto.Stream.getDefaultInstance());
- OrcProto.RowIndex ri = OrcProto.RowIndex.newBuilder().addEntry(
- OrcProto.RowIndexEntry.newBuilder().addPositions(1).setStatistics(
- OrcFileMetadata.createStatsDummy())).build();
- OrcProto.BloomFilterIndex bfi = OrcProto.BloomFilterIndex.newBuilder().addBloomFilter(
- OrcProto.BloomFilter.newBuilder().addBitset(0)).build();
- dummy.rowIndex = new OrcIndex(
- new OrcProto.RowIndex[] { ri },
- new OrcProto.Stream.Kind[] { OrcProto.Stream.Kind.BLOOM_FILTER_UTF8 },
- new OrcProto.BloomFilterIndex[] { bfi });
- return dummy;
- }
-
- public boolean hasAllIndexes(boolean[] includes) {
- for (int i = 0; i < includes.length; ++i) {
- if (includes[i] && rowIndex.getRowGroupIndex()[i] == null) return false;
- }
- return true;
- }
-
- public void loadMissingIndexes(DataReader mr, StripeInformation stripe, boolean[] includes,
- boolean[] sargColumns) throws IOException {
- // Do not lose the old indexes. Create a super set includes
- OrcProto.RowIndex[] existing = getRowIndexes();
- boolean superset[] = new boolean[Math.max(existing.length, includes.length)];
- for (int i = 0; i < includes.length; i++) {
- superset[i] = includes[i];
- }
- for (int i = 0; i < existing.length; i++) {
- superset[i] = superset[i] || (existing[i] != null);
- }
- // TODO: should we save footer to avoid a read here?
- rowIndex = mr.readRowIndex(stripe, schema, null, true, superset,
- rowIndex.getRowGroupIndex(),
- sargColumns, writerVersion, rowIndex.getBloomFilterKinds(),
- rowIndex.getBloomFilterIndex());
- // TODO: theoretically, we should re-estimate memory usage here and update memory manager
+ rowIndex = orcIndex;
}
public int getStripeIx() {
@@ -157,25 +72,6 @@ public class OrcStripeMetadata extends LlapCacheableBuffer implements ConsumerSt
public String getWriterTimezone() {
return writerTimezone;
}
- @Override
- public long getMemoryUsage() {
- return estimatedMemUsage;
- }
-
- @Override
- public void notifyEvicted(EvictionDispatcher evictionDispatcher) {
- evictionDispatcher.notifyEvicted(this);
- }
-
- @Override
- protected int invalidate() {
- return INVALIDATE_OK;
- }
-
- @Override
- protected boolean isLocked() {
- return false;
- }
public OrcBatchKey getKey() {
return stripeKey;
@@ -199,4 +95,15 @@ public class OrcStripeMetadata extends LlapCacheableBuffer implements ConsumerSt
public boolean supportsRowIndexes() {
return true;
}
+
+ public OrcIndex getIndex() {
+ return rowIndex;
+ }
+
+ @Override
+ public String toString() {
+ return "OrcStripeMetadata [stripeKey=" + stripeKey + ", rowCount="
+ + rowCount + ", writerTimezone=" + writerTimezone + ", encodings="
+ + encodings + ", streams=" + streams + ", rowIndex=" + rowIndex + "]";
+ }
}