HBASE-2338 log recovery: deleted items may be resurrected

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@928342 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -456,6 +456,8 @@ Release 0.21.0 - Unreleased
    HBASE-2331  [shell] count command needs a way to specify scan caching
                (Alexey Kovyrin via Stack)
    HBASE-2364  Ignore Deprecations during build (Paul Smith via Stack)
+   HBASE-2338  log recovery: deleted items may be resurrected
+               (Aravind Menon via Stack)
 
   NEW FEATURES
    HBASE-1961  HBase EC2 scripts

core/src/main/java/org/apache/hadoop/hbase/io/hfile/HFile.java

@@ -1754,7 +1754,7 @@ public class HFile {
           KeyValue kv = scanner.getKeyValue();
           // dump key value
           if (printKeyValue) {
-            System.out.println("K: " + Bytes.toStringBinary(kv.getKey()) +
+            System.out.println("K: " + kv +
               " V: " + Bytes.toStringBinary(kv.getValue()));
           }
           // check if rows are in order

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

@@ -1167,6 +1167,41 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
     this.updatesLock.readLock().lock();
 
     try {
+
+      for (Map.Entry<byte[], List<KeyValue>> e : familyMap.entrySet()) {
+
+        byte[] family = e.getKey(); 
+        List<KeyValue> kvs = e.getValue();
+        
+        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()) {
+            List<KeyValue> result = new ArrayList<KeyValue>(1);
+            Get g = new Get(kv.getRow());
+            NavigableSet<byte []> qualifiers =
+              new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
+              byte [] q = kv.getQualifier();
+              if(q == null) q = HConstants.EMPTY_BYTE_ARRAY;
+              qualifiers.add(q);
+              get(store, g, qualifiers, result);
+              if (result.isEmpty()) {
+                // Nothing to delete
+                continue;
+              }
+              if (result.size() > 1) {
+                throw new RuntimeException("Unexpected size: " + result.size());
+              }
+              KeyValue getkv = result.get(0);
+              Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(),
+                  getkv.getBuffer(), getkv.getTimestampOffset(), Bytes.SIZEOF_LONG);
+          } else {
+            kv.updateLatestStamp(byteNow);
+          }
+        }
+      }
+
       if (writeToWAL) {
         //
         // write/sync to WAL should happen before we touch memstore.
@@ -1204,30 +1239,6 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
         
         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()) {
-            List<KeyValue> result = new ArrayList<KeyValue>(1);
-            Get g = new Get(kv.getRow());
-            NavigableSet<byte []> qualifiers =
-              new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
-              byte [] q = kv.getQualifier();
-              if(q == null) q = HConstants.EMPTY_BYTE_ARRAY;
-              qualifiers.add(q);
-              get(store, g, qualifiers, result);
-              if (result.isEmpty()) {
-                // Nothing to delete
-                continue;
-              }
-              if (result.size() > 1) {
-                throw new RuntimeException("Unexpected size: " + result.size());
-              }
-              KeyValue getkv = result.get(0);
-              Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(),
-                  getkv.getBuffer(), getkv.getTimestampOffset(), Bytes.SIZEOF_LONG);
-          } else {
-            kv.updateLatestStamp(byteNow);
-          }
           size = this.memstoreSize.addAndGet(store.delete(kv));
         }
       }