Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-2338 log recovery: deleted items may be resurrected; 2nd attempt... fixes broke build 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@929962 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Michael Stack 2010-04-01 13:27:11 +00:00
parent 3e6935613c
commit 4cbfa4b594
1 changed files with 26 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
core/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
View File

@ -1172,28 +1172,39 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
byte[] family = e.getKey();
List<KeyValue> kvs = e.getValue();
Map<byte[], Integer> kvCount = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);
Store store = getStore(family);
for (KeyValue kv: kvs) {
// Check if time is LATEST, change to time of most recent addition if so
// This is expensive.
if (kv.isLatestTimestamp() && kv.isDeleteType()) {
byte[] qual = kv.getQualifier();
if (qual == null) qual = HConstants.EMPTY_BYTE_ARRAY;
Integer count = kvCount.get(qual);
if (count == null) {
kvCount.put(qual, new Integer(1));
} else {
kvCount.put(qual, new Integer(count+1));
}
count = kvCount.get(qual);
List<KeyValue> result = new ArrayList<KeyValue>(1);
Get g = new Get(kv.getRow());
g.setMaxVersions(count);
NavigableSet<byte []> qualifiers =
new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
byte [] q = kv.getQualifier();
if(q == null) q = HConstants.EMPTY_BYTE_ARRAY;
qualifiers.add(q);
qualifiers.add(qual);
get(store, g, qualifiers, result);
if (result.isEmpty()) {
if (result.size() < count) {
// Nothing to delete
kv.updateLatestStamp(byteNow);
continue;
}
if (result.size() > 1) {
if (result.size() > count) {
throw new RuntimeException("Unexpected size: " + result.size());
}
KeyValue getkv = result.get(0);
KeyValue getkv = result.get(count - 1);
Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(),
getkv.getBuffer(), getkv.getTimestampOffset(), Bytes.SIZEOF_LONG);
} else {