You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by li...@apache.org on 2013/03/26 19:19:09 UTC
svn commit: r1461255 [1/2] - in /hbase/branches/0.89-fb/src:
main/java/org/apache/hadoop/hbase/ main/java/org/apache/hadoop/hbase/io/
main/java/org/apache/hadoop/hbase/io/hfile/
main/java/org/apache/hadoop/hbase/regionserver/ main/java/org/apache/hadoo...
Author: liyin
Date: Tue Mar 26 18:19:08 2013
New Revision: 1461255
URL: http://svn.apache.org/r1461255
Log:
[HBASE-5263] Auto Cache hot blocks on compaction v1
Author: rshroff
Summary:
This change will try to cache blocks which are hot on compaction.
The design is same as mentioned in the JIRA.
The implementation is very simple as of now:
1. Adds a boolean in the scan part to fetch the information, whether the KV was
obtained from a block present in cache or not.
2. The same information is relayed down to the writer during compaction.
3. The writer keeps a track of % of KV's obtained from cache and then
when it dumps a block, it will use that and compare it with the caching
threshold for compaction and cache the block if it meets the criteria.
4. It also adds knobs to enable/disable this feature and change the
caching threshold used during compaction.
I was planning to further improve upon the feature in the following
areas:
1. The cached block from which the KV was read might be a candidate for
next round of eviction. Hence, need to improve on that part, where we
can consider the probability whether or not the block is hot enough to
be cached after compaction, depending upon the number of times it was
accessed.
2. If recaching on compaction is trashing the cache, then we can limit
the number of blocks we should cache during compaction.
Wanted to get this diff out for review to get an early response of the
overall design and the change.
Test Plan:
Added 4 unit test cases to test the change. I will am also planning to
test this on dev cluster. Will update it with more results.
MR Unit test link:
http://hbasectrl056.ash3.facebook.com:50030/jobdetails.jsp?jobid=job_201209211223_44686
Reviewers: liyintang, aaiyer, manukranthk
Reviewed By: manukranthk
CC: hbase-eng@, manukranthk
Differential Revision: https://phabricator.fb.com/D686137
Task ID: 2016630
Added:
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/KeyValueContext.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestCachingOnCompaction.java
Modified:
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/HalfStoreFileReader.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileReader.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileWriter.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/CacheConfig.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFile.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV1.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV2.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV1.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV2.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/InternalScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CompoundBloomFilter.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/RandomSeek.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompoundBloomFilter.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFile.java
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/KeyValueContext.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/KeyValueContext.java?rev=1461255&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/KeyValueContext.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/KeyValueContext.java Tue Mar 26 18:19:08 2013
@@ -0,0 +1,34 @@
+package org.apache.hadoop.hbase;
+
+/**
+* Copyright 2013 The Apache Software Foundation
+*
+* 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.
+*/
+
+public class KeyValueContext {
+ private boolean obtainedFromCache;
+
+ public boolean getObtainedFromCache() {
+ return obtainedFromCache;
+ }
+
+ public void setObtainedFromCache(boolean obtainedFromCache) {
+ this.obtainedFromCache = obtainedFromCache;
+ }
+
+}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/HalfStoreFileReader.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/HalfStoreFileReader.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/HalfStoreFileReader.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/HalfStoreFileReader.java Tue Mar 26 18:19:08 2013
@@ -245,6 +245,11 @@ public class HalfStoreFileReader extends
public boolean isSeeked() {
return this.delegate.isSeeked();
}
+
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return this.delegate.currKeyValueObtainedFromCache();
+ }
};
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileReader.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileReader.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileReader.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileReader.java Tue Mar 26 18:19:08 2013
@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.HFile.FileInfo;
import org.apache.hadoop.hbase.regionserver.metrics.SchemaConfigured;
@@ -274,6 +275,7 @@ public abstract class AbstractHFileReade
protected int currValueLen;
protected int currMemstoreTSLen;
protected long currMemstoreTS;
+ protected KeyValueContext kvContext = new KeyValueContext();
protected int blockFetches;
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileWriter.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileWriter.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileWriter.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/AbstractHFileWriter.java Tue Mar 26 18:19:08 2013
@@ -111,6 +111,16 @@ public abstract class AbstractHFileWrite
*/
protected final String name;
+ /**
+ * No of cached keys in the current block
+ */
+ protected int numCachedKeysInCurrentBlock;
+
+ /**
+ * No of keys in the current block
+ */
+ protected int numKeysInCurrentBlock;
+
public AbstractHFileWriter(Configuration conf, CacheConfig cacheConf,
FSDataOutputStream outputStream, Path path, int blockSize,
Compression.Algorithm compressAlgo,
@@ -130,6 +140,8 @@ public abstract class AbstractHFileWrite
closeOutputStream = path != null;
this.cacheConf = cacheConf;
+ this.numKeysInCurrentBlock = 0;
+ this.numCachedKeysInCurrentBlock = 0;
}
/**
@@ -303,5 +315,4 @@ public abstract class AbstractHFileWrite
fs.getDefaultReplication(), fs.getDefaultBlockSize(), bytesPerChecksum,
null);
}
-
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/CacheConfig.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/CacheConfig.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/CacheConfig.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/CacheConfig.java Tue Mar 26 18:19:08 2013
@@ -38,7 +38,7 @@ public class CacheConfig {
* Configuration key to cache data blocks on write. There are separate
* switches for bloom blocks and non-root index blocks.
*/
- public static final String CACHE_BLOCKS_ON_WRITE_KEY =
+ public static final String CACHE_BLOCKS_ON_FLUSH_KEY =
"hbase.rs.cacheblocksonwrite";
/**
@@ -67,16 +67,34 @@ public class CacheConfig {
*/
public static final String EVICT_BLOCKS_ON_CLOSE_KEY =
"hbase.rs.evictblocksonclose";
+ /**
+ * Configuration key to turn on cache hot blocks on compaction.
+ */
+ public static final String CACHE_DATA_BLOCKS_ON_COMPACTION =
+ "hfile.block.cacheoncompaction";
+
+ /**
+ * Configuration key to determine the threshold for hot blocks
+ * (No of KVs obtained from cache during compaction)/
+ * (Total number of KVs in the block)
+ * During compaction, the blocks in which the % of KVs cached exceed this
+ * threshold will be automatically cached on write. For scenarios, where cacheOnWrite
+ * is already set, this parameter will not have any effect.
+ */
+ public static final String CACHE_DATA_BLOCKS_ON_COMPACTION_THRESHOLD =
+ "hfile.block.cacheoncompaction.threshold";
// Defaults
public static final boolean DEFAULT_CACHE_DATA_ON_READ = true;
- public static final boolean DEFAULT_CACHE_DATA_ON_WRITE = false;
+ public static final boolean DEFAULT_CACHE_DATA_ON_FLUSH = false;
public static final boolean DEFAULT_IN_MEMORY = false;
public static final boolean DEFAULT_CACHE_INDEXES_ON_WRITE = false;
public static final boolean DEFAULT_CACHE_BLOOMS_ON_WRITE = false;
public static final boolean DEFAULT_EVICT_ON_CLOSE = false;
public static final boolean DEFAULT_COMPRESSED_CACHE = false;
+ public static final boolean DEFAULT_CACHE_DATA_BLOCKS_ON_COMPACTION = false;
+ public static final float DEFAULT_CACHE_DATA_BLOCKS_ON_COMPACTION_THRESHOLD = 0.75F;
/** Local reference to the block cache, null if completely disabled */
private final BlockCache blockCache;
@@ -90,8 +108,9 @@ public class CacheConfig {
/** Whether blocks should be flagged as in-memory when being cached */
private final boolean inMemory;
- /** Whether data blocks should be cached when new files are written */
- private boolean cacheDataOnWrite;
+ /** Whether data blocks should be cached when new files are written because
+ * of a memstore flush */
+ private boolean cacheDataOnFlush;
/** Whether index blocks should be cached when new files are written */
private final boolean cacheIndexesOnWrite;
@@ -105,6 +124,12 @@ public class CacheConfig {
/** Whether data blocks should be stored in compressed form in the cache */
private final boolean cacheCompressed;
+ /** Whether data blocks during compaction are cached */
+ private boolean cacheOnCompaction;
+
+ /** Threshold for caching hot data blocks during compaction */
+ private final float cacheOnCompactionThreshold;
+
/**
* Create a cache configuration using the specified configuration object and
* family descriptor.
@@ -114,13 +139,17 @@ public class CacheConfig {
public CacheConfig(Configuration conf, HColumnDescriptor family) {
this(CacheConfig.instantiateBlockCache(conf),
family.isBlockCacheEnabled(), family.isInMemory(),
- conf.getBoolean(CACHE_BLOCKS_ON_WRITE_KEY, DEFAULT_CACHE_DATA_ON_WRITE),
+ conf.getBoolean(CACHE_BLOCKS_ON_FLUSH_KEY, DEFAULT_CACHE_DATA_ON_FLUSH),
conf.getBoolean(CACHE_INDEX_BLOCKS_ON_WRITE_KEY,
DEFAULT_CACHE_INDEXES_ON_WRITE),
conf.getBoolean(CACHE_BLOOM_BLOCKS_ON_WRITE_KEY,
DEFAULT_CACHE_BLOOMS_ON_WRITE),
conf.getBoolean(EVICT_BLOCKS_ON_CLOSE_KEY, DEFAULT_EVICT_ON_CLOSE),
- conf.getBoolean(CACHE_DATA_BLOCKS_COMPRESSED_KEY, DEFAULT_COMPRESSED_CACHE)
+ conf.getBoolean(CACHE_DATA_BLOCKS_COMPRESSED_KEY, DEFAULT_COMPRESSED_CACHE),
+ conf.getBoolean(CACHE_DATA_BLOCKS_ON_COMPACTION,
+ DEFAULT_CACHE_DATA_BLOCKS_ON_COMPACTION),
+ conf.getFloat(CACHE_DATA_BLOCKS_ON_COMPACTION_THRESHOLD,
+ DEFAULT_CACHE_DATA_BLOCKS_ON_COMPACTION_THRESHOLD)
);
}
@@ -134,14 +163,18 @@ public class CacheConfig {
DEFAULT_CACHE_DATA_ON_READ,
DEFAULT_IN_MEMORY, // This is a family-level setting so can't be set
// strictly from conf
- conf.getBoolean(CACHE_BLOCKS_ON_WRITE_KEY, DEFAULT_CACHE_DATA_ON_WRITE),
+ conf.getBoolean(CACHE_BLOCKS_ON_FLUSH_KEY, DEFAULT_CACHE_DATA_ON_FLUSH),
conf.getBoolean(CACHE_INDEX_BLOCKS_ON_WRITE_KEY,
DEFAULT_CACHE_INDEXES_ON_WRITE),
conf.getBoolean(CACHE_BLOOM_BLOCKS_ON_WRITE_KEY,
DEFAULT_CACHE_BLOOMS_ON_WRITE),
conf.getBoolean(EVICT_BLOCKS_ON_CLOSE_KEY, DEFAULT_EVICT_ON_CLOSE),
conf.getBoolean(CACHE_DATA_BLOCKS_COMPRESSED_KEY,
- DEFAULT_COMPRESSED_CACHE)
+ DEFAULT_COMPRESSED_CACHE),
+ conf.getBoolean(CACHE_DATA_BLOCKS_ON_COMPACTION,
+ DEFAULT_CACHE_DATA_BLOCKS_ON_COMPACTION),
+ conf.getFloat(CACHE_DATA_BLOCKS_ON_COMPACTION_THRESHOLD,
+ DEFAULT_CACHE_DATA_BLOCKS_ON_COMPACTION_THRESHOLD)
);
}
@@ -153,23 +186,28 @@ public class CacheConfig {
* @param inMemory whether blocks should be flagged as in-memory
* @param cacheDataOnWrite whether data blocks should be cached on write
* @param cacheIndexesOnWrite whether index blocks should be cached on write
- * @param cacheBloomsOnWrite whether blooms should be cached on write
+ * @param cacheBloomsOnFlush whether blooms should be cached on write
* @param evictOnClose whether blocks should be evicted when HFile is closed
* @param cacheCompressed whether to store blocks as compressed in the cache
+ * @param cacheOnCompaction whether to cache blocks during compaction
+ * @param cacheOnCompactionThreshold threshold used of caching of blocks during compaction
*/
CacheConfig(final BlockCache blockCache,
final boolean cacheDataOnRead, final boolean inMemory,
- final boolean cacheDataOnWrite, final boolean cacheIndexesOnWrite,
+ final boolean cacheDataOnFlush, final boolean cacheIndexesOnWrite,
final boolean cacheBloomsOnWrite, final boolean evictOnClose,
- final boolean cacheCompressed) {
+ final boolean cacheCompressed, final boolean cacheOnCompaction,
+ final float cacheOnCompactionThreshold) {
this.blockCache = blockCache;
this.cacheDataOnRead = cacheDataOnRead;
this.inMemory = inMemory;
- this.cacheDataOnWrite = cacheDataOnWrite;
+ this.cacheDataOnFlush = cacheDataOnFlush;
this.cacheIndexesOnWrite = cacheIndexesOnWrite;
this.cacheBloomsOnWrite = cacheBloomsOnWrite;
this.evictOnClose = evictOnClose;
this.cacheCompressed = cacheCompressed;
+ this.cacheOnCompaction = cacheOnCompaction;
+ this.cacheOnCompactionThreshold = cacheOnCompactionThreshold;
}
/**
@@ -178,9 +216,10 @@ public class CacheConfig {
*/
public CacheConfig(CacheConfig cacheConf) {
this(cacheConf.blockCache, cacheConf.cacheDataOnRead, cacheConf.inMemory,
- cacheConf.cacheDataOnWrite, cacheConf.cacheIndexesOnWrite,
+ cacheConf.cacheDataOnFlush, cacheConf.cacheIndexesOnWrite,
cacheConf.cacheBloomsOnWrite, cacheConf.evictOnClose,
- cacheConf.cacheCompressed);
+ cacheConf.cacheCompressed, cacheConf.cacheOnCompaction,
+ cacheConf.cacheOnCompactionThreshold);
}
/**
@@ -234,8 +273,8 @@ public class CacheConfig {
* @return true if data blocks should be written to the cache when an HFile is
* written, false if not
*/
- public boolean shouldCacheDataOnWrite() {
- return isBlockCacheEnabled() && this.cacheDataOnWrite;
+ public boolean shouldCacheDataOnFlush() {
+ return isBlockCacheEnabled() && this.cacheDataOnFlush;
}
/**
@@ -243,8 +282,8 @@ public class CacheConfig {
* @param cacheDataOnWrite whether data blocks should be written to the cache
* when an HFile is written
*/
- public void setCacheDataOnWrite(boolean cacheDataOnWrite) {
- this.cacheDataOnWrite = cacheDataOnWrite;
+ public void setCacheDataOnFlush(boolean cacheDataOnFlush) {
+ this.cacheDataOnFlush = cacheDataOnFlush;
}
/**
@@ -287,6 +326,22 @@ public class CacheConfig {
return isBlockCacheEnabled() && this.cacheCompressed;
}
+ public boolean shouldCacheOnCompaction() {
+ return isBlockCacheEnabled() && cacheOnCompaction;
+ }
+
+ public void setCacheDataOnRead(final boolean cacheDataOnRead) {
+ this.cacheDataOnRead = cacheDataOnRead;
+ }
+
+ public void setCacheOnCompaction(final boolean cacheOnCompaction) {
+ this.cacheOnCompaction = cacheOnCompaction;
+ }
+
+ public float getCacheOnCompactionThreshold() {
+ return cacheOnCompactionThreshold;
+ }
+
@Override
public String toString() {
if (!isBlockCacheEnabled()) {
@@ -294,11 +349,13 @@ public class CacheConfig {
}
return "CacheConfig:enabled " +
"[cacheDataOnRead=" + shouldCacheDataOnRead() + "] " +
- "[cacheDataOnWrite=" + shouldCacheDataOnWrite() + "] " +
+ "[cacheDataOnWrite=" + shouldCacheDataOnFlush() + "] " +
"[cacheIndexesOnWrite=" + shouldCacheIndexesOnWrite() + "] " +
"[cacheBloomsOnWrite=" + shouldCacheBloomsOnWrite() + "] " +
"[cacheEvictOnClose=" + shouldEvictOnClose() + "] " +
- "[cacheCompressed=" + shouldCacheCompressed() + "]";
+ "[cacheCompressed=" + shouldCacheCompressed() + "] " +
+ "[cacheOnCompaction=" + shouldCacheOnCompaction() + "] " +
+ "[cacheOnCompactionThreshold=" + getCacheOnCompactionThreshold() + "]";
}
// Static block cache reference and methods
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFile.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFile.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFile.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFile.java Tue Mar 26 18:19:08 2013
@@ -43,6 +43,7 @@ import org.apache.hadoop.hbase.HColumnDe
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue.KeyComparator;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.HbaseMapWritable;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics.SchemaAware;
@@ -225,6 +226,8 @@ public class HFile {
void append(KeyValue kv) throws IOException;
+ void append(KeyValue kv, KeyValueContext vc) throws IOException;
+
void append(byte[] key, byte[] value) throws IOException;
/** @return the path to this {@link HFile} */
@@ -392,7 +395,7 @@ public class HFile {
public interface CachingBlockReader {
HFileBlock readBlock(long offset, long onDiskBlockSize,
boolean cacheBlock, final boolean isCompaction,
- BlockType expectedBlockType)
+ BlockType expectedBlockType, KeyValueContext kvContext)
throws IOException;
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java Tue Mar 26 18:19:08 2013
@@ -38,6 +38,7 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.HFile.CachingBlockReader;
@@ -181,10 +182,10 @@ public class HFileBlockIndex {
*/
public HFileBlock seekToDataBlock(final byte[] key, int keyOffset,
int keyLength, HFileBlock currentBlock, boolean cacheBlocks,
- boolean isCompaction)
+ boolean isCompaction, KeyValueContext kvContext)
throws IOException {
BlockWithScanInfo blockWithScanInfo = loadDataBlockWithScanInfo(key, keyOffset, keyLength,
- currentBlock, cacheBlocks, isCompaction);
+ currentBlock, cacheBlocks, isCompaction, kvContext);
if (blockWithScanInfo == null) {
return null;
} else {
@@ -204,12 +205,14 @@ public class HFileBlockIndex {
* block
* @param cacheBlocks
* @param isCompaction
+ * @param kvContext
* @return the BlockWithScanInfo which contains the DataBlock with other scan info
* such as nextIndexedKey.
* @throws IOException
*/
public BlockWithScanInfo loadDataBlockWithScanInfo(final byte[] key, int keyOffset,
- int keyLength, HFileBlock currentBlock, boolean cacheBlocks, boolean isCompaction)
+ int keyLength, HFileBlock currentBlock, boolean cacheBlocks,
+ boolean isCompaction, KeyValueContext kvContext)
throws IOException {
int rootLevelIndex = rootBlockContainingKey(key, keyOffset, keyLength);
if (rootLevelIndex < 0 || rootLevelIndex >= blockOffsets.length) {
@@ -257,7 +260,7 @@ public class HFileBlockIndex {
}
block = cachingBlockReader.readBlock(currentOffset,
currentOnDiskSize, shouldCache, isCompaction,
- expectedBlockType);
+ expectedBlockType, kvContext);
}
if (block == null) {
@@ -333,7 +336,7 @@ public class HFileBlockIndex {
// Caching, using pread, assuming this is not a compaction.
HFileBlock midLeafBlock = cachingBlockReader.readBlock(
midLeafBlockOffset, midLeafBlockOnDiskSize, true, false,
- BlockType.LEAF_INDEX);
+ BlockType.LEAF_INDEX, null);
ByteBuffer b = midLeafBlock.getBufferWithoutHeader();
int numDataBlocks = b.getInt();
@@ -874,7 +877,7 @@ public class HFileBlockIndex {
if (rootChunk.getNumEntries() > 0)
throw new IOException("Root-level entries already added in " +
- "single-level mode");
+ "single-level mode");
rootChunk = curInlineChunk;
curInlineChunk = new BlockIndexChunk();
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV1.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV1.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV1.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV1.java Tue Mar 26 18:19:08 2013
@@ -30,6 +30,7 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.BlockType.BlockCategory;
import org.apache.hadoop.hbase.io.hfile.HFile.FileInfo;
@@ -268,7 +269,8 @@ public class HFileReaderV1 extends Abstr
* @throws IOException
*/
ByteBuffer readBlockBuffer(int block, boolean cacheBlock,
- final boolean isCompaction) throws IOException {
+ final boolean isCompaction, KeyValueContext kvContext)
+ throws IOException {
if (dataBlockIndexReader == null) {
throw new IOException("Block index not loaded");
}
@@ -293,6 +295,9 @@ public class HFileReaderV1 extends Abstr
cacheConf.shouldCacheDataOnRead());
if (cachedBlock != null) {
cacheHits.incrementAndGet();
+ if (kvContext != null) {
+ kvContext.setObtainedFromCache(true);
+ }
getSchemaMetrics().updateOnCacheHit(
cachedBlock.getBlockType().getCategory(), isCompaction);
return cachedBlock.getBufferWithoutHeader();
@@ -329,7 +334,9 @@ public class HFileReaderV1 extends Abstr
}
getSchemaMetrics().updateOnCacheMiss(BlockCategory.DATA, isCompaction,
delta);
-
+ if (kvContext != null) {
+ kvContext.setObtainedFromCache(false);
+ }
// Cache the block
if (cacheBlock && cacheConf.shouldCacheBlockOnRead(
hfileBlock.getBlockType().getCategory())) {
@@ -562,7 +569,8 @@ public class HFileReaderV1 extends Abstr
blockBuffer = null;
return false;
}
- blockBuffer = reader.readBlockBuffer(currBlock, cacheBlocks, isCompaction);
+ blockBuffer = reader.readBlockBuffer(currBlock, cacheBlocks, isCompaction,
+ kvContext);
currKeyLen = blockBuffer.getInt();
currValueLen = blockBuffer.getInt();
blockFetches++;
@@ -640,7 +648,8 @@ public class HFileReaderV1 extends Abstr
return true;
}
currBlock = 0;
- blockBuffer = reader.readBlockBuffer(currBlock, cacheBlocks, isCompaction);
+ blockBuffer = reader.readBlockBuffer(currBlock, cacheBlocks, isCompaction,
+ kvContext);
currKeyLen = blockBuffer.getInt();
currValueLen = blockBuffer.getInt();
blockFetches++;
@@ -650,12 +659,14 @@ public class HFileReaderV1 extends Abstr
@Override
protected void loadBlock(int bloc, boolean rewind) throws IOException {
if (blockBuffer == null) {
- blockBuffer = reader.readBlockBuffer(bloc, cacheBlocks, isCompaction);
+ blockBuffer = reader.readBlockBuffer(bloc, cacheBlocks, isCompaction,
+ kvContext);
currBlock = bloc;
blockFetches++;
} else {
if (bloc != currBlock) {
- blockBuffer = reader.readBlockBuffer(bloc, cacheBlocks, isCompaction);
+ blockBuffer = reader.readBlockBuffer(bloc, cacheBlocks, isCompaction,
+ kvContext);
currBlock = bloc;
blockFetches++;
} else {
@@ -671,12 +682,17 @@ public class HFileReaderV1 extends Abstr
}
}
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return this.kvContext.getObtainedFromCache();
+ }
+
}
@Override
public HFileBlock readBlock(long offset, long onDiskBlockSize,
boolean cacheBlock, boolean isCompaction,
- BlockType expectedBlockType) {
+ BlockType expectedBlockType, KeyValueContext kvContext) {
throw new UnsupportedOperationException();
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV2.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV2.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV2.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderV2.java Tue Mar 26 18:19:08 2013
@@ -32,6 +32,7 @@ import org.apache.hadoop.fs.FSDataInputS
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoder;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.BlockType.BlockCategory;
@@ -247,16 +248,18 @@ public class HFileReaderV2 extends Abstr
* read operation, or null to read whatever block type is available
* and avoid checking (that might reduce caching efficiency of
* encoded data blocks)
+ * @param obtainedFromCache
* @return Block wrapped in a ByteBuffer.
* @throws IOException
*/
public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize,
final boolean cacheBlock, final boolean isCompaction,
- BlockType expectedBlockType)
+ BlockType expectedBlockType, KeyValueContext kvContext)
throws IOException {
if (dataBlockIndexReader == null) {
throw new IOException("Block index not loaded");
}
+
if (dataBlockOffset < 0
|| dataBlockOffset >= trailer.getLoadOnOpenDataOffset()) {
throw new IOException("Requested block is out of range: "
@@ -278,6 +281,13 @@ public class HFileReaderV2 extends Abstr
HFileBlock cachedBlock = this.getCachedBlock(cacheKey, cacheBlock, isCompaction,
expectedBlockType, true);
if (cachedBlock != null) {
+ if (kvContext != null) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Read block from file " + name + ", offset "
+ + dataBlockOffset + " from cache.");
+ }
+ kvContext.setObtainedFromCache(true);
+ }
return cachedBlock;
}
@@ -287,9 +297,11 @@ public class HFileReaderV2 extends Abstr
cachedBlock = this.getCachedBlock(cacheKey, cacheBlock, isCompaction,
expectedBlockType, false);
if (cachedBlock != null) {
+ if (kvContext != null) {
+ kvContext.setObtainedFromCache(true);
+ }
return cachedBlock;
}
-
// Load block from filesystem.
long startTimeNs = System.nanoTime();
HFileBlock hfileBlock = fsBlockReader.readBlockData(dataBlockOffset,
@@ -309,10 +321,14 @@ public class HFileReaderV2 extends Abstr
HFile.preadTimeNano.addAndGet(deltaNs);
HFile.preadOps.incrementAndGet();
}
+ if (kvContext != null) {
+ kvContext.setObtainedFromCache(false);
+ }
getSchemaMetrics().updateOnCacheMiss(blockCategory, isCompaction,
TimeUnit.NANOSECONDS.toMillis(deltaNs));
// Cache the block if necessary
+
if (cacheBlock && cacheConf.shouldCacheBlockOnRead(
hfileBlock.getBlockType().getCategory())) {
cacheConf.getBlockCache().cacheBlock(cacheKey, hfileBlock,
@@ -490,7 +506,7 @@ public class HFileReaderV2 extends Abstr
reader.getDataBlockIndexReader();
BlockWithScanInfo blockWithScanInfo =
indexReader.loadDataBlockWithScanInfo(key, offset, length, block,
- cacheBlocks, isCompaction);
+ cacheBlocks, isCompaction, this.kvContext);
if (blockWithScanInfo == null || blockWithScanInfo.getHFileBlock() == null) {
// This happens if the key e.g. falls before the beginning of the file.
return -1;
@@ -546,7 +562,7 @@ public class HFileReaderV2 extends Abstr
throws IOException {
HFileBlock seekToBlock =
reader.getDataBlockIndexReader().seekToDataBlock(key, offset, length,
- block, cacheBlocks, isCompaction);
+ block, cacheBlocks, isCompaction, this.kvContext);
if (seekToBlock == null) {
return false;
}
@@ -567,7 +583,7 @@ public class HFileReaderV2 extends Abstr
// figure out the size.
seekToBlock = reader.readBlock(previousBlockOffset,
seekToBlock.getOffset() - previousBlockOffset, cacheBlocks,
- isCompaction, BlockType.DATA);
+ isCompaction, BlockType.DATA, this.kvContext);
// TODO shortcut: seek forward in this block to the last key of the
// block.
}
@@ -604,7 +620,7 @@ public class HFileReaderV2 extends Abstr
curBlock = reader.readBlock(curBlock.getOffset()
+ curBlock.getOnDiskSizeWithHeader(),
curBlock.getNextBlockOnDiskSizeWithHeader(), cacheBlocks,
- isCompaction, null);
+ isCompaction, null, this.kvContext);
} while (!(curBlock.getBlockType().equals(BlockType.DATA) ||
curBlock.getBlockType().equals(BlockType.ENCODED_DATA)));
@@ -695,7 +711,6 @@ public class HFileReaderV2 extends Abstr
setNonSeekedState();
return false;
}
-
// read the next block
HFileBlock nextBlock = readNextDataBlock();
if (nextBlock == null) {
@@ -739,7 +754,8 @@ public class HFileReaderV2 extends Abstr
}
block = reader.readBlock(firstDataBlockOffset, -1, cacheBlocks,
- isCompaction, BlockType.DATA);
+ isCompaction, BlockType.DATA, this.kvContext);
+
if (block.getOffset() < 0) {
throw new IOException("Invalid block offset: " + block.getOffset());
}
@@ -925,6 +941,11 @@ public class HFileReaderV2 extends Abstr
+ blockBuffer.position() + KEY_VALUE_LEN_SIZE + currKeyLen,
currValueLen);
}
+
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return this.kvContext.getObtainedFromCache();
+ }
}
/**
@@ -1004,7 +1025,7 @@ public class HFileReaderV2 extends Abstr
}
block = reader.readBlock(firstDataBlockOffset, -1, cacheBlocks,
- isCompaction, BlockType.DATA);
+ isCompaction, BlockType.DATA, this.kvContext);
if (block.getOffset() < 0) {
throw new IOException("Invalid block offset: " + block.getOffset());
}
@@ -1082,6 +1103,11 @@ public class HFileReaderV2 extends Abstr
this.nextIndexedKey = nextIndexedKey;
return seeker.seekToKeyInBlock(key, offset, length, seekBefore);
}
+
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return this.kvContext.getObtainedFromCache();
+ }
}
/**
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java Tue Mar 26 18:19:08 2013
@@ -143,4 +143,6 @@ public interface HFileScanner {
* Otherwise returns false.
*/
public boolean isSeeked();
-}
\ No newline at end of file
+
+ public boolean currKeyValueObtainedFromCache();
+}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV1.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV1.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV1.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV1.java Tue Mar 26 18:19:08 2013
@@ -32,6 +32,7 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.HFile.FileInfo;
import org.apache.hadoop.hbase.io.hfile.HFile.Writer;
@@ -128,7 +129,7 @@ public class HFileWriterV1 extends Abstr
HFile.writeTimeNano.addAndGet(System.nanoTime() - startTimeNs);
HFile.writeOps.incrementAndGet();
- if (cacheConf.shouldCacheDataOnWrite()) {
+ if (cacheConf.shouldCacheDataOnFlush() || cacheCurrentBlockDuringCompaction()) {
baosDos.flush();
// we do not do data block encoding on disk for HFile v1
byte[] bytes = baos.toByteArray();
@@ -146,6 +147,11 @@ public class HFileWriterV1 extends Abstr
blockNumber++;
}
+ protected boolean cacheCurrentBlockDuringCompaction() {
+ return false;
+ }
+
+
/**
* Ready a new block for writing.
*
@@ -157,7 +163,9 @@ public class HFileWriterV1 extends Abstr
this.out = getCompressingStream();
BlockType.DATA.write(out);
firstKeyInBlock = null;
- if (cacheConf.shouldCacheDataOnWrite()) {
+ this.numKeysInCurrentBlock = 0;
+ this.numCachedKeysInCurrentBlock = 0;
+ if (cacheConf.shouldCacheDataOnFlush()) {
this.baos = new ByteArrayOutputStream();
this.baosDos = new DataOutputStream(baos);
baosDos.write(HFileBlock.DUMMY_HEADER);
@@ -245,8 +253,13 @@ public class HFileWriterV1 extends Abstr
* @throws IOException
*/
public void append(final KeyValue kv) throws IOException {
+ append(kv, null);
+ }
+
+ @Override
+ public void append(KeyValue kv, final KeyValueContext cv) throws IOException {
append(kv.getBuffer(), kv.getKeyOffset(), kv.getKeyLength(),
- kv.getBuffer(), kv.getValueOffset(), kv.getValueLength());
+ kv.getBuffer(), kv.getValueOffset(), kv.getValueLength(), cv);
}
/**
@@ -260,7 +273,7 @@ public class HFileWriterV1 extends Abstr
* @throws IOException
*/
public void append(final byte[] key, final byte[] value) throws IOException {
- append(key, 0, key.length, value, 0, value.length);
+ append(key, 0, key.length, value, 0, value.length, null);
}
/**
@@ -276,7 +289,7 @@ public class HFileWriterV1 extends Abstr
* @throws IOException
*/
private void append(final byte[] key, final int koffset, final int klength,
- final byte[] value, final int voffset, final int vlength)
+ final byte[] value, final int voffset, final int vlength, final KeyValueContext cv)
throws IOException {
boolean dupKey = checkKey(key, koffset, klength);
checkValue(value, voffset, vlength);
@@ -300,8 +313,12 @@ public class HFileWriterV1 extends Abstr
this.lastKeyOffset = koffset;
this.lastKeyLength = klength;
this.entryCount++;
+ this.numKeysInCurrentBlock++;
+ if (cv != null && cv.getObtainedFromCache()) {
+ this.numCachedKeysInCurrentBlock++;
+ }
// If we are pre-caching blocks on write, fill byte array stream
- if (cacheConf.shouldCacheDataOnWrite()) {
+ if (cacheConf.shouldCacheDataOnFlush()) {
this.baosDos.writeInt(klength);
this.baosDos.writeInt(vlength);
this.baosDos.write(key, koffset, klength);
@@ -431,5 +448,4 @@ public class HFileWriterV1 extends Abstr
}
return pos;
}
-
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV2.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV2.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV2.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterV2.java Tue Mar 26 18:19:08 2013
@@ -30,6 +30,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.hfile.HFile.Writer;
import org.apache.hadoop.hbase.io.hfile.HFile.WriterFactory;
import org.apache.hadoop.hbase.io.hfile.HFileBlock.BlockWritable;
@@ -160,11 +161,33 @@ public class HFileWriterV2 extends Abstr
HFile.writeTimeNano.addAndGet(System.nanoTime() - startTimeNs);
HFile.writeOps.incrementAndGet();
- if (cacheConf.shouldCacheDataOnWrite()) {
+ boolean cacheOnCompaction = cacheCurrentBlockForCompaction();
+ if (cacheConf.shouldCacheDataOnFlush() || cacheOnCompaction) {
doCacheOnWrite(lastDataBlockOffset);
}
}
+ protected boolean cacheCurrentBlockForCompaction() {
+
+ if (this.numKeysInCurrentBlock == 0) {
+ return false;
+ }
+
+ float blockCachingPercentage =
+ ((float)this.numCachedKeysInCurrentBlock)/this.numKeysInCurrentBlock;
+
+ LOG.debug("Block Caching %: " + blockCachingPercentage +
+ " CachingOnCompaction: " + this.cacheConf.shouldCacheOnCompaction() +
+ " Threshold: " + this.cacheConf.getCacheOnCompactionThreshold());
+
+ // Get the threshold from the config
+ if (this.cacheConf.shouldCacheOnCompaction() &&
+ blockCachingPercentage >= this.cacheConf.getCacheOnCompactionThreshold()) {
+ return true;
+ }
+ return false;
+ }
+
/** Gives inline block writers an opportunity to contribute blocks. */
private void writeInlineBlocks(boolean closing) throws IOException {
for (InlineBlockWriter ibw : inlineBlockWriters) {
@@ -190,8 +213,6 @@ public class HFileWriterV2 extends Abstr
* the cache key.
*/
private void doCacheOnWrite(long offset) {
- // We don't cache-on-write data blocks on compaction, so assume this is not
- // a compaction.
final boolean isCompaction = false;
HFileBlock cacheFormatBlock = blockEncoder.diskToCacheFormat(
fsBlockWriter.getBlockForCaching(), isCompaction);
@@ -210,6 +231,8 @@ public class HFileWriterV2 extends Abstr
// This is where the next block begins.
fsBlockWriter.startWriting(BlockType.DATA);
firstKeyInBlock = null;
+ this.numKeysInCurrentBlock = 0;
+ this.numCachedKeysInCurrentBlock = 0;
}
/**
@@ -249,8 +272,22 @@ public class HFileWriterV2 extends Abstr
*/
@Override
public void append(final KeyValue kv) throws IOException {
+ append(kv, null);
+ this.maxMemstoreTS = Math.max(this.maxMemstoreTS, kv.getMemstoreTS());
+ }
+
+ /**
+ * Add key/value to file. Keys must be added in an order that agrees with the
+ * Comparator passed on construction.
+ *
+ * @param kv KeyValue to add. Cannot be empty nor null.
+ * @param appendForCompaction Whether the KV was read from cache or not
+ * @throws IOException
+ */
+ @Override
+ public void append(final KeyValue kv, KeyValueContext cv) throws IOException {
append(kv.getMemstoreTS(), kv.getBuffer(), kv.getKeyOffset(), kv.getKeyLength(),
- kv.getBuffer(), kv.getValueOffset(), kv.getValueLength());
+ kv.getBuffer(), kv.getValueOffset(), kv.getValueLength(), cv);
this.maxMemstoreTS = Math.max(this.maxMemstoreTS, kv.getMemstoreTS());
}
@@ -266,7 +303,7 @@ public class HFileWriterV2 extends Abstr
*/
@Override
public void append(final byte[] key, final byte[] value) throws IOException {
- append(0, key, 0, key.length, value, 0, value.length);
+ append(0, key, 0, key.length, value, 0, value.length, null);
}
/**
@@ -279,11 +316,13 @@ public class HFileWriterV2 extends Abstr
* @param value
* @param voffset
* @param vlength
+ * @param KeyValueContext
* @throws IOException
*/
private void append(final long memstoreTS, final byte[] key,
final int koffset, final int klength, final byte[] value,
- final int voffset, final int vlength) throws IOException {
+ final int voffset, final int vlength, final KeyValueContext cv)
+ throws IOException {
boolean dupKey = checkKey(key, koffset, klength);
checkValue(value, voffset, vlength);
if (!dupKey) {
@@ -299,6 +338,10 @@ public class HFileWriterV2 extends Abstr
voffset, vlength);
totalKeyLength += klength;
totalValueLength += vlength;
+ this.numKeysInCurrentBlock++;
+ if (cv != null && cv.getObtainedFromCache()) {
+ this.numCachedKeysInCurrentBlock++;
+ }
// Are we the first key in this block?
if (firstKeyInBlock == null) {
@@ -435,5 +478,7 @@ public class HFileWriterV2 extends Abstr
}
});
}
-
}
+
+
+
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Tue Mar 26 18:19:08 2013
@@ -46,9 +46,7 @@ import java.util.concurrent.ConcurrentSk
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
-import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -72,7 +70,6 @@ import org.apache.hadoop.hbase.HRegionIn
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.NotServingRegionException;
-import org.apache.hadoop.hbase.UnknownScannerException;
import org.apache.hadoop.hbase.client.RowMutations;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
@@ -81,14 +78,11 @@ import org.apache.hadoop.hbase.client.Pu
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.RowLock;
import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.filter.Filter;
-import org.apache.hadoop.hbase.filter.IncompatibleFilterException;
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.io.Reference.Range;
import org.apache.hadoop.hbase.io.hfile.BlockCache;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.ipc.HRegionInterface;
-import org.apache.hadoop.hbase.metrics.RequestMetrics;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.monitoring.TaskMonitor;
import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
@@ -97,11 +91,9 @@ import org.apache.hadoop.hbase.regionser
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
-import org.apache.hadoop.hbase.util.DaemonThreadFactory;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.util.Writables;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.Progressable;
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/InternalScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/InternalScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/InternalScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/InternalScanner.java Tue Mar 26 18:19:08 2013
@@ -20,6 +20,7 @@
package org.apache.hadoop.hbase.regionserver;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import java.io.Closeable;
import java.io.IOException;
@@ -79,6 +80,37 @@ public interface InternalScanner extends
public boolean next(List<KeyValue> result, int limit, String metric) throws IOException;
/**
+ * Grab the next row's worth of values with a limit on the number of values
+ * to return.
+ * @param result return output array
+ * @param limit limit on row count to get
+ * @param kvContext more information about the KeyValues, can be null
+ * @return true if more rows exist after this one, false if scanner is done
+ * @throws IOException e
+ */
+ public boolean next(List<KeyValue> result, int limit,
+ KeyValueContext kvContext) throws IOException;
+
+ /**
+ * Grab the next row's worth of values with a limit on the number of values
+ * to return.
+ * @param result return output array
+ * @param limit limit on row count to get
+ * @param metric the metric name
+ * @param kvContext whether or not the KeyValues were obtained from cache, can be null
+ * @return true if more rows exist after this one, false if scanner is done
+ * @throws IOException e
+ */
+ public boolean next(List<KeyValue> result, int limit, String metric,
+ KeyValueContext kvContext) throws IOException;
+
+ /**
+ * Tells whether the current KeyValue is obtained from cache or not
+ * @return boolean whether the KeyValue exists or not
+ */
+ public boolean currKeyValueObtainedFromCache();
+
+ /**
* Closes the scanner and releases any resources it has allocated
* @throws IOException
*/
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java Tue Mar 26 18:19:08 2013
@@ -22,6 +22,7 @@ package org.apache.hadoop.hbase.regionse
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.KVComparator;
+import org.apache.hadoop.hbase.KeyValueContext;
import java.io.IOException;
import java.util.ArrayList;
@@ -121,7 +122,19 @@ public class KeyValueHeap extends NonLaz
* @return true if there are more keys, false if all scanners are done
*/
public boolean next(List<KeyValue> result, int limit) throws IOException {
- return next(result, limit, null);
+ return next(result, limit, null, null);
+ }
+
+ @Override
+ public boolean next(List<KeyValue> kvs, int limit,
+ KeyValueContext kvContext) throws IOException {
+ return next(kvs, limit, null, kvContext);
+ }
+
+ @Override
+ public boolean next(List<KeyValue> result, int limit,
+ String metric) throws IOException {
+ return next(result, limit, metric, null);
}
/**
@@ -136,16 +149,18 @@ public class KeyValueHeap extends NonLaz
* @param metric the metric name
* @return true if there are more keys, false if all scanners are done
*/
- public boolean next(List<KeyValue> result, int limit, String metric) throws IOException {
+ public boolean next(List<KeyValue> result, int limit, String metric,
+ KeyValueContext kvContext) throws IOException {
if (this.current == null) {
return false;
}
InternalScanner currentAsInternal = (InternalScanner)this.current;
- boolean mayContainsMoreRows = currentAsInternal.next(result, limit, metric);
+ boolean mayContainsMoreRows =
+ currentAsInternal.next(result, limit, metric, kvContext);
+
KeyValue pee = this.current.peek();
/*
* By definition, any InternalScanner must return false only when it has no
- * further rows to be fetched. So, we can close a scanner if it returns
* false. All existing implementations seem to be fine with this. It is much
* more efficient to close scanners which are not needed than keep them in
* the heap. This is also required for certain optimizations.
@@ -155,7 +170,9 @@ public class KeyValueHeap extends NonLaz
} else {
this.heap.add(this.current);
}
+
this.current = pollRealKV();
+
return (this.current != null);
}
@@ -419,4 +436,11 @@ public class KeyValueHeap extends NonLaz
public boolean passesDeleteColumnCheck(KeyValue kv) {
return true;
}
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ if (this.current == null) {
+ return false;
+ }
+ return this.current.currKeyValueObtainedFromCache();
+ }
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java Tue Mar 26 18:19:08 2013
@@ -66,6 +66,8 @@ public interface KeyValueScanner {
*/
public long getSequenceID();
+ public boolean currKeyValueObtainedFromCache();
+
/**
* Close the KeyValue scanner.
*/
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java Tue Mar 26 18:19:08 2013
@@ -687,6 +687,11 @@ public class MemStore implements HeapSiz
return true;
return false;
}
+
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return true;
+ }
}
public final static long FIXED_OVERHEAD = ClassSize.align(
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java Tue Mar 26 18:19:08 2013
@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.UnknownScannerException;
import org.apache.hadoop.hbase.client.Result;
@@ -123,7 +124,7 @@ public class RegionScanner implements In
@Override
public boolean next(List<KeyValue> outResults, int limit)
throws IOException {
- return next(outResults, limit, null);
+ return next(outResults, limit, null, null);
}
private void preCondition() throws IOException{
@@ -196,7 +197,7 @@ public class RegionScanner implements In
getOriginalScan().setCurrentPartialResponseSize(0);
int maxResponseSize = getOriginalScan().getMaxResponseSize();
do {
- moreRows = nextInternal(tmpList, limit, metric);
+ moreRows = nextInternal(tmpList, limit, metric, null);
if (!tmpList.isEmpty()) {
currentNbRows++;
if (outResults != null) {
@@ -287,17 +288,23 @@ public class RegionScanner implements In
* used by Scans
*/
@Override
- public synchronized boolean next(List<KeyValue> outResults, int limit,
+ public boolean next(List<KeyValue> outResults, int limit,
String metric) throws IOException {
+ return next(outResults, limit, metric, null);
+ }
+
+ @Override
+ public synchronized boolean next(List<KeyValue> outResults, int limit, String metric,
+ KeyValueContext kvContext) throws IOException {
preCondition();
boolean returnResult;
if (outResults.isEmpty()) {
// Usually outResults is empty. This is true when next is called
// to handle scan or get operation.
- returnResult = nextInternal(outResults, limit, metric);
+ returnResult = nextInternal(outResults, limit, metric, kvContext);
} else {
List<KeyValue> tmpList = new ArrayList<KeyValue>();
- returnResult = nextInternal(tmpList, limit, metric);
+ returnResult = nextInternal(tmpList, limit, metric, kvContext);
outResults.addAll(tmpList);
}
rowReadCnt.incrementAndGet();
@@ -307,12 +314,12 @@ public class RegionScanner implements In
}
return returnResult;
}
-
+
@Override
public boolean next(List<KeyValue> outResults)
throws IOException {
// apply the batching limit by default
- return next(outResults, batch, null);
+ return next(outResults, batch, null, null);
}
@Override
@@ -332,7 +339,8 @@ public class RegionScanner implements In
/**
* @param results empty list in which results will be stored
*/
- private boolean nextInternal(List<KeyValue> results, int limit, String metric)
+ private boolean nextInternal(List<KeyValue> results, int limit, String metric,
+ KeyValueContext kvContext)
throws IOException {
if (!results.isEmpty()) {
@@ -358,7 +366,7 @@ public class RegionScanner implements In
} else {
byte [] nextRow;
do {
- this.storeHeap.next(results, limit - results.size(), metric);
+ this.storeHeap.next(results, limit - results.size(), metric, kvContext);
if (limit > 0 && results.size() == limit) {
if (this.filter != null && filter.hasFilterRow())
throw new IncompatibleFilterException(
@@ -446,4 +454,17 @@ public class RegionScanner implements In
public Scan getOriginalScan() {
return originalScan;
}
+
+ @Override
+ public boolean next(List<KeyValue> result, int limit,
+ KeyValueContext kvContext) throws IOException {
+ return next(result, limit, null, kvContext);
+ }
+
+
+
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return this.storeHeap.currKeyValueObtainedFromCache();
+ }
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Tue Mar 26 18:19:08 2013
@@ -48,6 +48,7 @@ import org.apache.hadoop.hbase.HColumnDe
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.RemoteExceptionHandler;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.HeapSize;
@@ -267,6 +268,7 @@ public class Store extends SchemaConfigu
throw new IllegalArgumentException(e);
}
}
+
try {
this.compactHook = (CompactionHook) Class.forName(compactHookString).newInstance();
} catch (InstantiationException e) {
@@ -781,7 +783,7 @@ public class Store extends SchemaConfigu
if (isCompaction) {
// Don't cache data on write on compactions.
writerCacheConf = new CacheConfig(cacheConf);
- writerCacheConf.setCacheDataOnWrite(false);
+ writerCacheConf.setCacheDataOnFlush(false);
} else {
writerCacheConf = cacheConf;
}
@@ -1209,8 +1211,10 @@ public class Store extends SchemaConfigu
if (maxCompactingSequcenceId == this.getMaxSequenceId(true)) {
writer = createWriterInTmp(maxKeyCount, compression, true);
}
+ KeyValueContext kvContext = new KeyValueContext();
+
do {
- hasMore = scanner.next(kvs, 1);
+ hasMore = scanner.next(kvs, 1, kvContext);
if (!kvs.isEmpty()) {
if (writer == null) {
writer = createWriterInTmp(maxKeyCount, compression, true);
@@ -1224,10 +1228,10 @@ public class Store extends SchemaConfigu
RestrictedKeyValue restrictedKv = new RestrictedKeyValue(kv);
RestrictedKeyValue modifiedKv = compactHook.transform(restrictedKv);
if (modifiedKv != null) {
- writer.append(modifiedKv.getKeyValue());
+ writer.append(modifiedKv.getKeyValue(), kvContext);
}
} else {
- writer.append(kv);
+ writer.append(kv, kvContext);
}
// check periodically to see if a system stop is requested
if (Store.closeCheckInterval > 0) {
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java Tue Mar 26 18:19:08 2013
@@ -44,6 +44,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue.KVComparator;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.HalfStoreFileReader;
import org.apache.hadoop.hbase.io.Reference;
@@ -1117,10 +1118,15 @@ public class StoreFile extends SchemaCon
}
public void append(final KeyValue kv) throws IOException {
+ append(kv, null);
+ }
+
+ public void append(final KeyValue kv, final KeyValueContext cv) throws IOException{
appendGeneralBloomfilter(kv);
appendDeleteFamilyBloomFilter(kv);
appendDeleteColumnBloomFilter(kv);
- writer.append(kv);
+
+ writer.append(kv, cv);
includeInTimeRangeTracker(kv);
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileScanner.java Tue Mar 26 18:19:08 2013
@@ -372,4 +372,8 @@ class StoreFileScanner implements KeyVal
public boolean passesDeleteColumnCheck(KeyValue kv) {
return this.reader.passesDeleteColumnBloomFilter(kv);
}
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return this.hfs.currKeyValueObtainedFromCache();
+ }
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java Tue Mar 26 18:19:08 2013
@@ -31,6 +31,7 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.ipc.HBaseServer.Call;
import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
@@ -329,19 +330,23 @@ public class StoreScanner extends NonLaz
*/
@Override
public synchronized boolean next(List<KeyValue> outResult, int limit) throws IOException {
- return next(outResult, limit, null);
+ return next(outResult, limit, null, null);
}
/**
* Get the next row of values from this Store.
* @param outResult
* @param limit
+ * @param metic
+ * @param kvContext
* @return true if there are more rows, false if scanner is done
*/
- @Override
public synchronized boolean next(List<KeyValue> outResult, int limit,
- String metric) throws IOException {
+ String metric, KeyValueContext kvContext) throws IOException {
+ if (kvContext != null) {
+ kvContext.setObtainedFromCache(false);
+ }
checkReseek();
// if the heap was left null, then the scanners had previously run out anyways, close and
@@ -368,7 +373,6 @@ public class StoreScanner extends NonLaz
Call call = HRegionServer.callContext.get();
long quotaRemaining = (call == null) ? Long.MAX_VALUE
: HRegionServer.getResponseSizeLimit() - call.getPartialResponseSize();
-
// Only do a sanity-check if store and comparator are available.
KeyValue.KVComparator comparator =
store != null ? store.getComparator() : null;
@@ -413,7 +417,7 @@ public class StoreScanner extends NonLaz
this.countPerRow > (storeLimit + storeOffset)) {
// do what SEEK_NEXT_ROW does.
if (!matcher.moreRowsMayExistAfter(kv)) {
- numNewKeyValues += processLastKeyValue(outResult);
+ numNewKeyValues += processLastKeyValue(outResult, kvContext);
return false;
}
reseek(matcher.getKeyForNextRow(kv));
@@ -430,13 +434,21 @@ public class StoreScanner extends NonLaz
+ HRegionServer.getResponseSizeLimit() + " bytes.");
throw new DoNotRetryIOException("Result too large");
}
+ if (kvContext != null) {
+
+ // If any of the KV is obtained from cache, then lets tell the caller
+ // that all the KVs are obtained from cache
+ kvContext.setObtainedFromCache(
+ kvContext.getObtainedFromCache() |
+ this.currKeyValueObtainedFromCache());
+ }
outResult.add(copyKv);
numNewKeyValues++;
}
if (qcode == ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_ROW) {
if (!matcher.moreRowsMayExistAfter(kv)) {
- numNewKeyValues += processLastKeyValue(outResult);
+ numNewKeyValues += processLastKeyValue(outResult, kvContext);
return false;
}
reseek(matcher.getKeyForNextRow(kv));
@@ -464,19 +476,19 @@ public class StoreScanner extends NonLaz
continue;
case DONE:
- numNewKeyValues += processLastKeyValue(outResult);
+ numNewKeyValues += processLastKeyValue(outResult, kvContext);
return true;
case DONE_SCAN:
close();
- numNewKeyValues += processLastKeyValue(outResult);
+ numNewKeyValues += processLastKeyValue(outResult, kvContext);
return false;
case SEEK_NEXT_ROW:
// This is just a relatively simple end of scan fix, to short-cut end
// us if there is an endKey in the scan.
if (!matcher.moreRowsMayExistAfter(kv)) {
- numNewKeyValues += processLastKeyValue(outResult);
+ numNewKeyValues += processLastKeyValue(outResult, kvContext);
return false;
}
@@ -550,7 +562,7 @@ public class StoreScanner extends NonLaz
}
}
- numNewKeyValues += processLastKeyValue(outResult);
+ numNewKeyValues += processLastKeyValue(outResult, kvContext);
if (numNewKeyValues > 0) {
return true;
}
@@ -560,10 +572,16 @@ public class StoreScanner extends NonLaz
return false;
}
- private byte processLastKeyValue(List<KeyValue> outResult){
+ private byte processLastKeyValue(List<KeyValue> outResult, KeyValueContext kvContext){
KeyValue lastKV = keyValueAggregator.finalizeKeyValues();
if (lastKV != null) {
outResult.add(lastKV);
+ if (kvContext != null) {
+ // If any of the KV is obtained from cache, then lets tell the caller
+ // that all the KVs are obtained from cache
+ kvContext.setObtainedFromCache(kvContext.getObtainedFromCache() |
+ this.currKeyValueObtainedFromCache());
+ }
return 1;
}
return 0;
@@ -571,7 +589,7 @@ public class StoreScanner extends NonLaz
@Override
public synchronized boolean next(List<KeyValue> outResult) throws IOException {
- return next(outResult, -1, null);
+ return next(outResult, -1, null, null);
}
@Override
@@ -678,4 +696,23 @@ public class StoreScanner extends NonLaz
public boolean passesDeleteColumnCheck(KeyValue kv) {
return true;
}
+ @Override
+ public boolean next(List<KeyValue> kvs, int limit,
+ KeyValueContext kvContext) throws IOException {
+ return next(kvs, limit, null, kvContext);
+ }
+
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ if (this.heap != null) {
+ return this.heap.currKeyValueObtainedFromCache();
+ }
+ return false;
+ }
+
+ @Override
+ public boolean next(List<KeyValue> result, int limit, String metric)
+ throws IOException {
+ return next(result, limit, metric, null);
+ }
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java Tue Mar 26 18:19:08 2013
@@ -131,4 +131,8 @@ public class CollectionBackedScanner ext
return true;
}
+ @Override
+ public boolean currKeyValueObtainedFromCache() {
+ return false;
+ }
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CompoundBloomFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CompoundBloomFilter.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CompoundBloomFilter.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/CompoundBloomFilter.java Tue Mar 26 18:19:08 2013
@@ -97,7 +97,7 @@ public class CompoundBloomFilter extends
// We cache the block and use a positional read.
bloomBlock = reader.readBlock(index.getRootBlockOffset(block),
index.getRootBlockDataSize(block), true, false,
- BlockType.BLOOM_CHUNK);
+ BlockType.BLOOM_CHUNK, null);
} catch (IOException ex) {
// The Bloom filter is broken, turn it off.
throw new IllegalArgumentException(
Modified: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java (original)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java Tue Mar 26 18:19:08 2013
@@ -4293,7 +4293,7 @@ public class TestFromClientSide {
tableName).getOnlineRegionByFullName(regionName);
Store store = region.getStores().values().iterator().next();
CacheConfig cacheConf = store.getCacheConfig();
- cacheConf.setCacheDataOnWrite(true);
+ cacheConf.setCacheDataOnFlush(true);
cacheConf.setEvictOnClose(true);
BlockCache cache = cacheConf.getBlockCache();
// Get rid of blocks marked to be evicted.
Modified: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/RandomSeek.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/RandomSeek.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/RandomSeek.java (original)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/RandomSeek.java Tue Mar 26 18:19:08 2013
@@ -69,7 +69,7 @@ public class RandomSeek {
SimpleBlockCache cache = new SimpleBlockCache();
//LruBlockCache cache = new LruBlockCache();
CacheConfig cacheConf = new CacheConfig(cache, true, false, false, false,
- false, false, false);
+ false, false, false, false, 1);
Reader reader = HFile.createReader(lfs, path, cacheConf);
reader.loadFileInfo();
System.out.println(reader.getTrailer());
Modified: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java (original)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java Tue Mar 26 18:19:08 2013
@@ -88,7 +88,7 @@ public class TestCacheOnWrite {
private static final BloomType BLOOM_TYPE = StoreFile.BloomType.ROWCOL;
private static enum CacheOnWriteType {
- DATA_BLOCKS(CacheConfig.CACHE_BLOCKS_ON_WRITE_KEY,
+ DATA_BLOCKS(CacheConfig.CACHE_BLOCKS_ON_FLUSH_KEY,
BlockType.DATA, BlockType.ENCODED_DATA),
BLOOM_BLOCKS(CacheConfig.CACHE_BLOOM_BLOCKS_ON_WRITE_KEY,
BlockType.BLOOM_CHUNK),
@@ -179,7 +179,7 @@ public class TestCacheOnWrite {
conf.setInt(HFileBlockIndex.MAX_CHUNK_SIZE_KEY, INDEX_BLOCK_SIZE);
conf.setInt(BloomFilterFactory.IO_STOREFILE_BLOOM_BLOCK_SIZE,
BLOOM_BLOCK_SIZE);
- conf.setBoolean(CacheConfig.CACHE_BLOCKS_ON_WRITE_KEY,
+ conf.setBoolean(CacheConfig.CACHE_BLOCKS_ON_FLUSH_KEY,
cowType.shouldBeCached(BlockType.DATA));
conf.setBoolean(CacheConfig.CACHE_INDEX_BLOCKS_ON_WRITE_KEY,
cowType.shouldBeCached(BlockType.LEAF_INDEX));
@@ -227,7 +227,7 @@ public class TestCacheOnWrite {
// Flags: don't cache the block, use pread, this is not a compaction.
// Also, pass null for expected block type to avoid checking it.
HFileBlock block = reader.readBlock(offset, onDiskSize, false,
- false, null);
+ false, null, null);
BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(),
offset, encodingInCache, block.getBlockType());
boolean isCached = blockCache.getBlock(blockCacheKey, true) != null;
Modified: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java?rev=1461255&r1=1461254&r2=1461255&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java (original)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java Tue Mar 26 18:19:08 2013
@@ -45,6 +45,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueContext;
import org.apache.hadoop.hbase.io.hfile.HFileBlockIndex.BlockIndexChunk;
import org.apache.hadoop.hbase.io.hfile.HFileBlockIndex.BlockIndexReader;
import org.apache.hadoop.hbase.util.Bytes;
@@ -140,7 +141,7 @@ public class TestHFileBlockIndex {
@Override
public HFileBlock readBlock(long offset, long onDiskSize,
boolean cacheBlock, boolean isCompaction,
- BlockType expectedBlockType)
+ BlockType expectedBlockType, KeyValueContext kvContext)
throws IOException {
if (offset == prevOffset && onDiskSize == prevOnDiskSize) {
hitCount += 1;
@@ -151,7 +152,9 @@ public class TestHFileBlockIndex {
prevBlock = realReader.readBlockData(offset, onDiskSize, -1);
prevOffset = offset;
prevOnDiskSize = onDiskSize;
-
+ if (kvContext != null) {
+ kvContext.setObtainedFromCache(false);
+ }
return prevBlock;
}
}
@@ -181,7 +184,7 @@ public class TestHFileBlockIndex {
assertTrue(key != null);
assertTrue(indexReader != null);
HFileBlock b = indexReader.seekToDataBlock(key, 0, key.length, null,
- true, false);
+ true, false, null);
if (Bytes.BYTES_RAWCOMPARATOR.compare(key, firstKeyInFile) < 0) {
assertTrue(b == null);
++i;
@@ -604,4 +607,4 @@ public class TestHFileBlockIndex {
valueRead);
}
-}
\ No newline at end of file
+}