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 2012/11/07 21:16:31 UTC
svn commit: r1406789 - in /hbase/branches/0.89-fb/src:
main/java/org/apache/hadoop/hbase/
main/java/org/apache/hadoop/hbase/regionserver/
main/java/org/apache/hadoop/hbase/util/
test/java/org/apache/hadoop/hbase/client/ test/java/org/apache/hadoop/hbas...
Author: liyin
Date: Wed Nov 7 20:16:30 2012
New Revision: 1406789
URL: http://svn.apache.org/viewvc?rev=1406789&view=rev
Log:
[HBASE-2376] Enhance Flushes and Compactions to respect FLASHBACK_QUERY_LIMIT.
Author: pritam
Test Plan: 1) All unit tests.
Reviewers: kannan, kranganathan, aaiyer, mbautin, liyintang, adela
Reviewed By: liyintang
CC: hbase-eng@, erling
Differential Revision: https://phabricator.fb.com/D606550
Task ID: 1733764
Added:
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestFlashBackQueryCompaction.java
Modified:
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/HColumnDescriptor.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanWildcardColumnTracker.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/StoreScanner.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/HColumnDescriptor.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/HColumnDescriptor.java?rev=1406789&r1=1406788&r2=1406789&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/HColumnDescriptor.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/HColumnDescriptor.java Wed Nov 7 20:16:30 2012
@@ -656,6 +656,10 @@ public class HColumnDescriptor implement
* @return this (for chained invocation)
*/
public HColumnDescriptor setFlashBackQueryLimit(int flashBackQueryLimit) {
+ if (flashBackQueryLimit < 0) {
+ throw new IllegalArgumentException(
+ "FlashBackQueryLimit cannot be negative");
+ }
return setValue(FLASHBACK_QUERY_LIMIT,
Integer.toString(flashBackQueryLimit));
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java?rev=1406789&r1=1406788&r2=1406789&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java Wed Nov 7 20:16:30 2012
@@ -75,6 +75,8 @@ public class ScanQueryMatcher {
* */
private boolean hasNullColumn = true;
+ private final long oldestFlashBackTS;
+
/**
* Constructs a ScanQueryMatcher for a Scan.
* @param scan
@@ -87,6 +89,23 @@ public class ScanQueryMatcher {
int maxVersions, long readPointToUse,
long retainDeletesInOutputUntil,
long oldestUnexpiredTS) {
+ this(scan, family, columnSet, rowComparator, maxVersions, readPointToUse,
+ retainDeletesInOutputUntil, oldestUnexpiredTS,
+ HConstants.LATEST_TIMESTAMP);
+ }
+
+ /**
+ * Constructs a ScanQueryMatcher for a Scan.
+ *
+ * @param scan
+ * @param family
+ * @param columnSet
+ * @param rowComparator
+ */
+ public ScanQueryMatcher(Scan scan, byte[] family,
+ NavigableSet<byte[]> columnSet, KeyValue.KeyComparator rowComparator,
+ int maxVersions, long readPointToUse, long retainDeletesInOutputUntil,
+ long oldestUnexpiredTS, long oldestFlashBackTS) {
this.tr = scan.getTimeRange();
this.oldestStamp = oldestUnexpiredTS;
this.rowComparator = rowComparator;
@@ -97,6 +116,7 @@ public class ScanQueryMatcher {
this.filter = scan.getFilter();
this.retainDeletesInOutputUntil = retainDeletesInOutputUntil;
this.maxReadPointToTrackVersions = readPointToUse;
+ this.oldestFlashBackTS = oldestFlashBackTS;
// Single branch to deal with two types of reads (columns vs all in family)
if (columnSet == null || columnSet.size() == 0) {
@@ -104,14 +124,17 @@ public class ScanQueryMatcher {
hasNullColumn = true;
// use a specialized scan for wildcard column tracker.
- this.columns = new ScanWildcardColumnTracker(maxVersions);
+ this.columns = new ScanWildcardColumnTracker(maxVersions,
+ this.oldestFlashBackTS);
} else {
// whether there is null column in the explicit column query
hasNullColumn = (columnSet.first().length == 0);
// We can share the ExplicitColumnTracker, diff is we reset
// between rows, not between storefiles.
- this.columns = new ExplicitColumnTracker(columnSet,maxVersions);
+ // Note that we do not use oldestFlashBackTS here since
+ // ExplicitColumnTracker is never used in Compactions/Flushes.
+ this.columns = new ExplicitColumnTracker(columnSet, maxVersions);
}
}
@@ -199,7 +222,11 @@ public class ScanQueryMatcher {
}
byte type = kv.getType();
- if (isDelete(type)) {
+ if (isDelete(type) && kv.getTimestamp() > this.oldestFlashBackTS) {
+ // During Flushes and Compactions we don't want to process deletes unless
+ // they are older than the FLASHBACK_QUERY_LIMIT.
+ return MatchCode.INCLUDE;
+ } else if (isDelete(type)) {
if (tr.withinOrAfterTimeRange(timestamp)) {
this.deletes.add(bytes, offset, qualLength, timestamp, type);
// Can't early out now, because DelFam come before any other keys
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanWildcardColumnTracker.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanWildcardColumnTracker.java?rev=1406789&r1=1406788&r2=1406789&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanWildcardColumnTracker.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/ScanWildcardColumnTracker.java Wed Nov 7 20:16:30 2012
@@ -27,6 +27,7 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.regionserver.ScanQueryMatcher.MatchCode;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
/**
* Keeps track of the columns for a scan if they are not explicitly specified
@@ -44,11 +45,22 @@ public class ScanWildcardColumnTracker i
private long latestTSOfCurrentColumn;
/**
+ * This is the oldest timestamp upto which FlashBack Queries are supported.
+ */
+ private final long oldestFlashBackTS;
+
+ public ScanWildcardColumnTracker(int maxVersion) {
+ this(maxVersion, HConstants.LATEST_TIMESTAMP);
+ this.maxVersions = maxVersion;
+ }
+
+ /**
* Return maxVersions of every row.
* @param maxVersion
*/
- public ScanWildcardColumnTracker(int maxVersion) {
+ public ScanWildcardColumnTracker(int maxVersion, long oldestFlashBackTS) {
this.maxVersions = maxVersion;
+ this.oldestFlashBackTS = oldestFlashBackTS;
}
/**
@@ -77,7 +89,7 @@ public class ScanWildcardColumnTracker i
if (ignoreCount) return ScanQueryMatcher.MatchCode.INCLUDE;
- if (++currentCount > maxVersions) {
+ if (timestamp <= this.oldestFlashBackTS && ++currentCount > maxVersions) {
return ScanQueryMatcher.MatchCode.SEEK_NEXT_COL;
}
setTS(timestamp);
@@ -92,7 +104,7 @@ public class ScanWildcardColumnTracker i
if (sameAsPreviousTS(timestamp)) {
return ScanQueryMatcher.MatchCode.SKIP;
}
- if (++currentCount > maxVersions) {
+ if (timestamp <= this.oldestFlashBackTS && ++currentCount > maxVersions) {
return ScanQueryMatcher.MatchCode.SEEK_NEXT_COL; // skip to next col
}
setTS(timestamp);
@@ -111,7 +123,7 @@ public class ScanWildcardColumnTracker i
if (ignoreCount) return ScanQueryMatcher.MatchCode.INCLUDE;
- if (++currentCount > maxVersions)
+ if (timestamp <= this.oldestFlashBackTS && ++currentCount > maxVersions)
return ScanQueryMatcher.MatchCode.SEEK_NEXT_COL;
setTS(timestamp);
return ScanQueryMatcher.MatchCode.INCLUDE;
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=1406789&r1=1406788&r2=1406789&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 Wed Nov 7 20:16:30 2012
@@ -69,6 +69,8 @@ import org.apache.hadoop.hbase.regionser
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.InjectionEvent;
+import org.apache.hadoop.hbase.util.InjectionHandler;
import org.apache.hadoop.util.StringUtils;
import com.google.common.base.Preconditions;
@@ -112,6 +114,10 @@ public class Store extends SchemaConfigu
final CacheConfig cacheConf;
// ttl in milliseconds.
protected long ttl;
+ // The amount of time in seconds in the past upto which we support FlashBack
+ // queries. Ex. 60 * 60 * 24 indicates we support FlashBack queries upto 1 day
+ // ago.
+ public final long flashBackQueryLimit;
private long timeToPurgeDeletes;
private long lastCompactSize = 0;
volatile boolean forceMajor = false;
@@ -191,6 +197,7 @@ public class Store extends SchemaConfigu
family.getDataBlockEncoding());
this.comparator = info.getComparator();
+ this.flashBackQueryLimit = family.getFlashBackQueryLimit() * 1000;
// getTimeToLive returns ttl in seconds. Convert to milliseconds.
this.ttl = family.getTimeToLive();
if (ttl == HConstants.FOREVER) {
@@ -667,7 +674,8 @@ public class Store extends SchemaConfigu
InternalScanner scanner = new StoreScanner(this, scan,
MemStore.getSnapshotScanners(snapshot, this.comparator),
this.region.getSmallestReadPoint(),
- Long.MIN_VALUE, getAggregator()); // include all deletes
+ Long.MIN_VALUE, getAggregator(),
+ flashBackQueryLimit); // include all deletes
String fileName;
try {
@@ -966,6 +974,10 @@ public class Store extends SchemaConfigu
/**
* Compact the most recent N files.
+ *
+ * @param N
+ * the number of store files to compact, pass -1 to compact
+ * everything
*/
public void compactRecentForTesting(int N) throws IOException {
List<StoreFile> filesToCompact;
@@ -985,11 +997,13 @@ public class Store extends SchemaConfigu
filesToCompact.subList(0, idx + 1).clear();
}
int count = filesToCompact.size();
- if (N > count) {
+ if (N > count && N != -1) {
throw new RuntimeException("Not enough files");
}
- filesToCompact = filesToCompact.subList(count - N, count);
+ if (N != -1) {
+ filesToCompact = filesToCompact.subList(count - N, count);
+ }
maxId = StoreFile.getMaxSequenceIdInList(filesToCompact, true);
isMajor = (filesToCompact.size() == storefiles.size());
filesCompacting.addAll(filesToCompact);
@@ -1181,7 +1195,7 @@ public class Store extends SchemaConfigu
(System.currentTimeMillis() - this.timeToPurgeDeletes))
: Long.MIN_VALUE;
scanner = new StoreScanner(this, scan, scanners, smallestReadPoint,
- retainDeletesUntil, getAggregator());
+ retainDeletesUntil, getAggregator(), flashBackQueryLimit);
int bytesWritten = 0;
// since scanner.next() can return 'false' but still be delivering data,
// we have to use a do/while loop.
@@ -1209,6 +1223,8 @@ public class Store extends SchemaConfigu
}
} else {
writer.append(kv);
+ InjectionHandler.processEvent(
+ InjectionEvent.STORE_AFTER_APPEND_KV, (Object) kv);
}
// check periodically to see if a system stop is requested
if (Store.closeCheckInterval > 0) {
@@ -1854,8 +1870,8 @@ public class Store extends SchemaConfigu
public static final long FIXED_OVERHEAD =
ClassSize.align(SchemaConfigured.SCHEMA_CONFIGURED_UNALIGNED_HEAP_SIZE +
- + (19 * ClassSize.REFERENCE) + (5 * Bytes.SIZEOF_LONG)
- + (4 * Bytes.SIZEOF_INT) + 2 * Bytes.SIZEOF_BOOLEAN) ;
+ + (19 * ClassSize.REFERENCE) + (6 * Bytes.SIZEOF_LONG)
+ + (4 * Bytes.SIZEOF_INT) + 2 * Bytes.SIZEOF_BOOLEAN);
public static final long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD +
ClassSize.OBJECT + ClassSize.REENTRANT_LOCK +
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=1406789&r1=1406788&r2=1406789&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 Wed Nov 7 20:16:30 2012
@@ -29,6 +29,7 @@ import java.util.NavigableSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.ipc.HBaseServer.Call;
@@ -76,9 +77,15 @@ public class StoreScanner extends NonLaz
// if heap == null and lastTop != null, you need to reseek given the key below
private KeyValue lastTop = null;
- /** An internal constructor. */
private StoreScanner(Store store, boolean cacheBlocks, Scan scan,
final NavigableSet<byte[]> columns, long ttl, KeyValueAggregator keyValueAggregator) {
+ this(store, cacheBlocks, scan, columns, ttl, keyValueAggregator, 0);
+ }
+
+ /** An internal constructor. */
+ private StoreScanner(Store store, boolean cacheBlocks, Scan scan,
+ final NavigableSet<byte[]> columns, long ttl,
+ KeyValueAggregator keyValueAggregator, long flashBackQueryLimit) {
this.store = store;
initializeMetricNames();
this.cacheBlocks = cacheBlocks;
@@ -88,7 +95,8 @@ public class StoreScanner extends NonLaz
this.scan = scan;
this.keyValueAggregator = keyValueAggregator;
this.columns = columns;
- oldestUnexpiredTS = EnvironmentEdgeManager.currentTimeMillis() - ttl;
+ oldestUnexpiredTS = EnvironmentEdgeManager.currentTimeMillis() - ttl
+ - flashBackQueryLimit;
// We look up row-column Bloom filters for multi-column queries as part of
// the seek operation. However, we also look the row-column Bloom filter
@@ -145,6 +153,14 @@ public class StoreScanner extends NonLaz
this.store.addChangedReaderObserver(this);
}
+ StoreScanner(Store store, Scan scan,
+ List<? extends KeyValueScanner> scanners, long smallestReadPoint,
+ long retainDeletesInOutputUntil, KeyValueAggregator keyValueAggregator)
+ throws IOException {
+ this(store, scan, scanners, smallestReadPoint, retainDeletesInOutputUntil,
+ keyValueAggregator, 0);
+ }
+
/**
* Opens a scanner across specified StoreFiles. Can be used in compactions.
* @param store who we scan
@@ -156,14 +172,20 @@ public class StoreScanner extends NonLaz
*/
StoreScanner(Store store, Scan scan,
List<? extends KeyValueScanner> scanners, long smallestReadPoint,
- long retainDeletesInOutputUntil, KeyValueAggregator keyValueAggregator) throws IOException {
- this(store, false, scan, null, store.ttl, keyValueAggregator);
+ long retainDeletesInOutputUntil, KeyValueAggregator keyValueAggregator,
+ long flashBackQueryLimit) throws IOException {
+ this(store, false, scan, null, store.ttl, keyValueAggregator, flashBackQueryLimit);
+
+ // This is the oldest TS upto which flashback queries are supported.
+ long oldestFlashBackTS = (flashBackQueryLimit != 0) ? EnvironmentEdgeManager
+ .currentTimeMillis() - flashBackQueryLimit
+ : HConstants.LATEST_TIMESTAMP;
matcher =
new ScanQueryMatcher(scan, store.getFamily().getName(), null,
store.comparator.getRawComparator(),
store.versionsToReturn(scan.getMaxVersions()), smallestReadPoint,
- retainDeletesInOutputUntil, oldestUnexpiredTS);
+ retainDeletesInOutputUntil, oldestUnexpiredTS, oldestFlashBackTS);
// Filter the list of scanners using Bloom filters, time range, TTL, etc.
scanners = selectScannersFrom(scanners);
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java?rev=1406789&r1=1406788&r2=1406789&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java Wed Nov 7 20:16:30 2012
@@ -33,5 +33,6 @@ public enum InjectionEvent {
HMASTER_ENABLE_TABLE,
HMASTER_DISABLE_TABLE,
ZKUNASSIGNEDWATCHER_REGION_OPENED,
+ STORE_AFTER_APPEND_KV,
SPLITLOGWORKER_SPLIT_LOG_START
}
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=1406789&r1=1406788&r2=1406789&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 Wed Nov 7 20:16:30 2012
@@ -136,7 +136,7 @@ public class TestFromClientSide {
for (int i = 0; i < expected.length; i++) {
byte[] FAMILY = Bytes.toBytes("foo" + i);
HColumnDescriptor desc = new HColumnDescriptor(FAMILY);
- int fbt = random.nextInt();
+ int fbt = random.nextInt(Integer.MAX_VALUE);
desc.setFlashBackQueryLimit(fbt);
expected[i] = desc;
}
Added: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestFlashBackQueryCompaction.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestFlashBackQueryCompaction.java?rev=1406789&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestFlashBackQueryCompaction.java (added)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestFlashBackQueryCompaction.java Wed Nov 7 20:16:30 2012
@@ -0,0 +1,436 @@
+/**
+ * Copyright 2010 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.
+ */
+package org.apache.hadoop.hbase.regionserver;
+
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.HashMap;
+import java.util.PriorityQueue;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.Type;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
+import org.apache.hadoop.hbase.util.InjectionEvent;
+import org.apache.hadoop.hbase.util.InjectionHandler;
+import org.apache.hadoop.hbase.util.ManualEnvironmentEdge;
+
+import org.junit.AfterClass;
+import static org.junit.Assert.*;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestFlashBackQueryCompaction {
+
+ private static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+ private static Random random = new Random();
+ private static final int CURRENT_TIME = 60000;
+ private static final int MAX_MAXVERSIONS = 100;
+ private static final HashSet<KeyValue> goodKvs = new HashSet<KeyValue>();
+ private static final HashMap<String, HashMap<String, PriorityQueue<KeyValue>>> heapKvs = new HashMap<String, HashMap<String, PriorityQueue<KeyValue>>>();
+ private static final HashMap<String, HashMap<String, PriorityQueue<KeyValue>>> heapKvsNormalScan = new HashMap<String, HashMap<String, PriorityQueue<KeyValue>>>();
+ private static final Log LOG = LogFactory
+ .getLog(TestFlashBackQueryCompaction.class);
+ private static final byte[][] families = new byte[][] { "a".getBytes(),
+ "b".getBytes(), "c".getBytes(), "d".getBytes(), "e".getBytes(),
+ "f".getBytes(), "g".getBytes(), "h".getBytes(), "i".getBytes(),
+ "j".getBytes() };
+
+ private static class KVComparator implements Comparator<KeyValue> {
+
+ @Override
+ public int compare(KeyValue kv1, KeyValue kv2) {
+ return (kv1.getTimestamp() < kv2.getTimestamp() ? -1 : (kv1
+ .getTimestamp() > kv2.getTimestamp() ? 1 : 0));
+ }
+ }
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+
+ ManualEnvironmentEdge mock = new ManualEnvironmentEdge();
+ mock.setValue(CURRENT_TIME);
+ EnvironmentEdgeManagerTestHelper.injectEdge(mock);
+ TEST_UTIL.startMiniCluster(1);
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ TEST_UTIL.shutdownMiniCluster();
+ EnvironmentEdgeManagerTestHelper.reset();
+ }
+
+ private void processHeapKvs(
+ HashMap<String, HashMap<String, PriorityQueue<KeyValue>>> heapKvs,
+ byte[] row, HColumnDescriptor hcd, KeyValue kv) {
+ HashMap<String, PriorityQueue<KeyValue>> rowMap = heapKvs.get(new String(
+ row));
+ if (rowMap == null) {
+ rowMap = new HashMap<String, PriorityQueue<KeyValue>>();
+ heapKvs.put(new String(row), rowMap);
+ }
+ PriorityQueue<KeyValue> q = rowMap.get(new String(hcd.getName()));
+ if (q == null) {
+ q = new PriorityQueue<KeyValue>(1, new KVComparator());
+ rowMap.put(new String(hcd.getName()), q);
+ }
+ q.add(kv);
+ if (q.size() > hcd.getMaxVersions()) {
+ q.poll();
+ }
+ }
+
+ private KeyValue getRandomKv(byte[] row, HColumnDescriptor hcd, int start,
+ int size) {
+ byte[] value = new byte[10];
+ random.nextBytes(value);
+ long ts = start + random.nextInt(size);
+ return new KeyValue(row, hcd.getName(), null, ts, value);
+ }
+
+ private KeyValue processKV(byte[] row, HColumnDescriptor hcd, int start,
+ int size) {
+ KeyValue kv = getRandomKv(row, hcd, start, size);
+ if (kv.getTimestamp() >= CURRENT_TIME - hcd.getTimeToLive() * 1000) {
+ processHeapKvs(heapKvsNormalScan, row, hcd, kv);
+ }
+ if (kv.getTimestamp() >= CURRENT_TIME - hcd.getFlashBackQueryLimit() * 1000
+ - hcd.getTimeToLive() * 1000) {
+ if (kv.getTimestamp() > CURRENT_TIME - hcd.getFlashBackQueryLimit()
+ * 1000) {
+ goodKvs.add(kv);
+ } else {
+ processHeapKvs(heapKvs, row, hcd, kv);
+ }
+ }
+ return kv;
+ }
+
+ private HTable setupTable(int ttl, int flashBackQueryLimit, int maxVersions)
+ throws Exception {
+ LOG.info("maxVersions : " + maxVersions);
+ LOG.info("TTL : " + ttl);
+ LOG.info("FBQ: " + flashBackQueryLimit);
+ byte[] tableName = ("loadTable" + random.nextInt()).getBytes();
+ HColumnDescriptor[] hcds = new HColumnDescriptor[random
+ .nextInt(families.length - 1) + 1];
+ for (int i = 0; i < hcds.length; i++) {
+ hcds[i] = new HColumnDescriptor(families[i]);
+ hcds[i].setTimeToLive(ttl);
+ hcds[i].setFlashBackQueryLimit(flashBackQueryLimit);
+ hcds[i].setMaxVersions(maxVersions);
+ }
+ return TEST_UTIL.createTable(tableName, hcds);
+ }
+
+ private void flushAllRegions() throws Exception {
+ HRegionServer server = TEST_UTIL.getHBaseCluster().getRegionServer(0);
+ for (HRegion region : server.getOnlineRegions()) {
+ server.flushRegion(region.getRegionName());
+ }
+ }
+
+ private void loadTable(int ttl, int flashBackQueryLimit, int maxVersions)
+ throws Exception {
+ HTable table = setupTable(ttl, flashBackQueryLimit, maxVersions);
+ HColumnDescriptor[] hcds = table.getTableDescriptor().getColumnFamilies();
+ byte[] tableName = table.getTableName();
+ byte[] row = new byte[10];
+ for (int i = 0; i < 10; i++) {
+ random.nextBytes(row);
+ Put put = new Put(row);
+ int size = CURRENT_TIME / MAX_MAXVERSIONS;
+ for (HColumnDescriptor hcd : hcds) {
+ for (int versions = 0, start = 0; versions < MAX_MAXVERSIONS; versions++, start += size) {
+ put.add(this.processKV(row, hcd, start, size));
+ }
+ }
+ table.put(put);
+ }
+
+ // flush all regions synchronously.
+ flushAllRegions();
+
+ // major compact everything.
+ majorCompact(tableName, hcds);
+
+ setStoreProps(tableName, hcds, false);
+
+ // verify all kvs in all store files.
+ Scan scan = new Scan();
+ scan.setMaxVersions(Integer.MAX_VALUE);
+ HashSet<KeyValue> tableSet = new HashSet<KeyValue>();
+ for (Result result : table.getScanner(scan)) {
+ for (KeyValue kv : result.list()) {
+ assertTrue("KV : " + kv + " should not exist", goodKvs.contains(kv)
+ || inHeapKvs(kv, heapKvs));
+ tableSet.add(kv);
+ }
+ }
+
+ for (KeyValue kv : goodKvs) {
+ assertTrue("KV in goodKvs: " + kv + " does not exist in table",
+ tableSet.contains(kv));
+ }
+
+ verifyHeapKvs(heapKvs, tableSet);
+
+ setStoreProps(tableName, hcds, true);
+ LOG.info("Scanning normal");
+ scan = new Scan();
+ scan.setMaxVersions(Integer.MAX_VALUE);
+ tableSet = new HashSet<KeyValue>();
+ for (Result result : table.getScanner(scan)) {
+ for (KeyValue kv : result.list()) {
+ LOG.info("Last SCAN KV : " + kv);
+ assertTrue("KV : " + kv + " should not exist",
+ inHeapKvs(kv, heapKvsNormalScan));
+ tableSet.add(kv);
+ }
+ }
+
+ verifyHeapKvs(heapKvsNormalScan, tableSet);
+ }
+
+ private void majorCompact(byte[] tableName, HColumnDescriptor[] hcds)
+ throws Exception {
+ HRegionServer server = TEST_UTIL.getHBaseCluster().getRegionServer(0);
+ for (HRegion region : server.getOnlineRegions()) {
+ if (!new String(region.getTableDesc().getName()).equals(new String(
+ tableName))) {
+ LOG.info("Skipping since name is : "
+ + new String(region.getTableDesc().getName()));
+ continue;
+ }
+ for (HColumnDescriptor hcd : hcds) {
+ Store store = region.getStore(hcd.getName());
+ store.compactRecentForTesting(-1);
+ }
+ }
+ }
+
+ private void setStoreProps(byte[] tableName, HColumnDescriptor[] hcds,
+ boolean def) {
+ HRegionServer server = TEST_UTIL.getHBaseCluster().getRegionServer(0);
+ for (HRegion region : server.getOnlineRegions()) {
+ if (!new String(region.getTableDesc().getName()).equals(new String(
+ tableName))) {
+ LOG.info("Skipping since name is : "
+ + new String(region.getTableDesc().getName()));
+ continue;
+ }
+ for (HColumnDescriptor hcd : hcds) {
+ Store store = region.getStore(hcd.getName());
+ // Reset back to original values.
+ if (def) {
+ store.ttl = hcd.getTimeToLive() * 1000;
+ store.getFamily().setMaxVersions(hcd.getMaxVersions());
+ } else {
+ store.ttl = HConstants.FOREVER;
+ store.getFamily().setMaxVersions(Integer.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ private void verifyHeapKvs(
+ HashMap<String, HashMap<String, PriorityQueue<KeyValue>>> heapKvs,
+ HashSet<KeyValue> tableSet) {
+ for (HashMap<String, PriorityQueue<KeyValue>> rowMap : heapKvs.values()) {
+ for (PriorityQueue<KeyValue> q : rowMap.values()) {
+ for (KeyValue kv : q) {
+ assertTrue("KV in heapKvs: " + kv + " does not exist in table",
+ tableSet.contains(kv));
+ }
+ }
+ }
+ }
+
+ private boolean inHeapKvs(KeyValue kv,
+ HashMap<String, HashMap<String, PriorityQueue<KeyValue>>> heapKvs) {
+ HashMap<String, PriorityQueue<KeyValue>> rowMap = heapKvs.get(new String(kv
+ .getRow()));
+ if (rowMap == null)
+ return false;
+ return rowMap.get(new String(kv.getFamily())).contains(kv);
+ }
+
+ @Test
+ public void testRandom() throws Exception {
+ loadTable(random.nextInt(CURRENT_TIME) / 1000,
+ random.nextInt(CURRENT_TIME) / 1000, random.nextInt(MAX_MAXVERSIONS));
+ }
+
+ private class CompactionHookHandler extends InjectionHandler {
+ protected void _processEvent(InjectionEvent event, Object... args) {
+ if (InjectionEvent.STORE_AFTER_APPEND_KV == event) {
+ KeyValue kv = (KeyValue) args[0];
+ if (Type.codeToType(kv.getType()) == Type.Put) {
+ compactionSet.add((KeyValue) args[0]);
+ }
+ }
+ }
+ }
+
+ private static HashSet<KeyValue> compactionSet = new HashSet<KeyValue>();
+
+ private KeyValue loadTableAndDelete(boolean family) throws Exception {
+ HTable table = setupTable(30, 20, 100);
+ byte[] row = new byte[10];
+ Put put = new Put(row);
+ KeyValue kv1 = new KeyValue(row, families[0], null, 55000, row);
+ KeyValue kv2 = new KeyValue(row, families[0], null, 35000, row);
+ put.add(kv1);
+ put.add(kv2);
+ table.put(put);
+
+ Delete delete = new Delete(row);
+ if (!family) {
+ delete.deleteRow(57000);
+ } else {
+ delete.deleteFamily(families[0], 57000);
+ }
+ table.delete(delete);
+
+ flushAllRegions();
+
+ compactionSet.clear();
+ majorCompact(table.getTableName(), table.getTableDescriptor()
+ .getColumnFamilies());
+
+ assertEquals(2, compactionSet.size());
+
+ delete = new Delete(row);
+ if (!family) {
+ delete.deleteRow(37000);
+ } else {
+ delete.deleteFamily(families[0], 37000);
+ }
+ table.delete(delete);
+
+ flushAllRegions();
+
+ compactionSet.clear();
+ majorCompact(table.getTableName(), table.getTableDescriptor()
+ .getColumnFamilies());
+
+ return kv1;
+ }
+
+ @Test
+ public void testDeleteRow() throws Exception {
+ InjectionHandler.set(new CompactionHookHandler());
+ try {
+ KeyValue kv1 = loadTableAndDelete(false);
+ assertEquals(1, compactionSet.size());
+ assertTrue(compactionSet.contains(kv1));
+ } finally {
+ InjectionHandler.clear();
+ }
+ }
+
+ @Test
+ public void testDeleteFamily() throws Exception {
+ InjectionHandler.set(new CompactionHookHandler());
+ try {
+ KeyValue kv1 = loadTableAndDelete(true);
+ assertEquals(1, compactionSet.size());
+ assertTrue(compactionSet.contains(kv1));
+ } finally {
+ InjectionHandler.clear();
+ }
+ }
+
+ @Test
+ public void testCornerCaseTTL() throws Exception {
+ HTable table = setupTable(30, 20, 100);
+ byte[] row = new byte[10];
+ Put put = new Put(row);
+ KeyValue kv1 = new KeyValue(row, families[0], null, 10000, row);
+ KeyValue kv2 = new KeyValue(row, families[0], null, 9999, row);
+ put.add(kv1);
+ put.add(kv2);
+ table.put(put);
+
+ flushAllRegions();
+
+ majorCompact(table.getTableName(), table.getTableDescriptor()
+ .getColumnFamilies());
+
+ // Get all kvs.
+ setStoreProps(table.getTableName(), table.getTableDescriptor()
+ .getColumnFamilies(), false);
+
+ Scan scan = new Scan(row);
+ scan.setMaxVersions();
+ HashSet<KeyValue> kvSet = new HashSet<KeyValue>();
+ for (Result result : table.getScanner(scan)) {
+ kvSet.addAll(result.list());
+ }
+ assertTrue(kvSet.contains(kv1));
+ assertFalse(kvSet.contains(kv2));
+ }
+
+ @Test
+ public void testCornerCaseMaxVersions() throws Exception {
+ HTable table = setupTable(30, 20, 2);
+ byte[] row = new byte[10];
+ Put put = new Put(row);
+ KeyValue kv1 = new KeyValue(row, families[0], null, 50000, row);
+ KeyValue kv2 = new KeyValue(row, families[0], null, 40000, row);
+ KeyValue kv3 = new KeyValue(row, families[0], null, 15000, row);
+ KeyValue kv4 = new KeyValue(row, families[0], null, 10000, row);
+ put.add(kv1);
+ put.add(kv2);
+ put.add(kv3);
+ put.add(kv4);
+ table.put(put);
+
+ flushAllRegions();
+
+ majorCompact(table.getTableName(), table.getTableDescriptor()
+ .getColumnFamilies());
+
+ // Get all kvs.
+ setStoreProps(table.getTableName(), table.getTableDescriptor()
+ .getColumnFamilies(), false);
+
+ Scan scan = new Scan(row);
+ scan.setMaxVersions();
+ HashSet<KeyValue> kvSet = new HashSet<KeyValue>();
+ for (Result result : table.getScanner(scan)) {
+ kvSet.addAll(result.list());
+ }
+ assertTrue(kvSet.contains(kv1));
+ assertTrue(kvSet.contains(kv2));
+ assertTrue(kvSet.contains(kv3));
+ assertFalse(kvSet.contains(kv4));
+ }
+}