HBASE-2453 Revisit compaction policies after HBASE-2248 commit

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

CHANGES.txt

@@ -272,6 +272,7 @@ Release 0.21.0 - Unreleased
    HBASE-2439  HBase can get stuck if updates to META are blocked
                (Kannan Muthukkaruppan via Stack)
    HBASE-2451  .META. by-passes cache; BLOCKCACHE=>'false'
+   HBASE-2453  Revisit compaction policies after HBASE-2248 commit
 
   IMPROVEMENTS
    HBASE-1760  Cleanup TODOs in HTable

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

@@ -35,13 +35,11 @@ import java.util.List;
 public class MinorCompactingStoreScanner implements KeyValueScanner, InternalScanner {
 
   private KeyValueHeap heap;
-  private ScanDeleteTracker deleteTracker;
   private KeyValue.KVComparator comparator;
 
   MinorCompactingStoreScanner(Store store,
                               KeyValueScanner [] scanners) {
     comparator = store.comparator;
-    deleteTracker = new ScanDeleteTracker();
     KeyValue firstKv = KeyValue.createFirstOnRow(HConstants.EMPTY_START_ROW);
     for (KeyValueScanner scanner : scanners ) {
       scanner.seek(firstKv);
@@ -53,7 +51,6 @@ public class MinorCompactingStoreScanner implements KeyValueScanner, InternalSca
   MinorCompactingStoreScanner(String cfName, KeyValue.KVComparator comparator,
                               KeyValueScanner [] scanners) {
     this.comparator = comparator;
-    deleteTracker = new ScanDeleteTracker();
 
     KeyValue firstKv = KeyValue.createFirstOnRow(HConstants.EMPTY_START_ROW);
     for (KeyValueScanner scanner : scanners ) {
@@ -89,9 +86,6 @@ public class MinorCompactingStoreScanner implements KeyValueScanner, InternalSca
       close();
       return false;
     }
-    // between rows.
-    deleteTracker.reset();
-
     KeyValue kv;
     while ((kv = heap.peek()) != null) {
       // check to see if this is a different row
@@ -99,27 +93,6 @@ public class MinorCompactingStoreScanner implements KeyValueScanner, InternalSca
         // reached next row
         return true;
       }
-
-      // if delete type, output no matter what:
-      if (kv.getType() != KeyValue.Type.Put.getCode()) {
-        deleteTracker.add(kv.getBuffer(),
-            kv.getQualifierOffset(),
-            kv.getQualifierLength(),
-            kv.getTimestamp(),
-            kv.getType());
-
-        writer.append(heap.next());
-        continue;
-      }
-
-      if (deleteTracker.isDeleted(kv.getBuffer(),
-          kv.getQualifierOffset(),
-          kv.getQualifierLength(),
-          kv.getTimestamp())) {
-        heap.next();
-        continue;
-      }
-
       writer.append(heap.next());
     }
     close();
@@ -133,9 +106,6 @@ public class MinorCompactingStoreScanner implements KeyValueScanner, InternalSca
       close();
       return false;
     }
-    // between rows.
-    deleteTracker.reset();
-
     KeyValue kv;
     while ((kv = heap.peek()) != null) {
       // check to see if this is a different row
@@ -143,27 +113,6 @@ public class MinorCompactingStoreScanner implements KeyValueScanner, InternalSca
         // reached next row
         return true;
       }
-
-      // if delete type, output no matter what:
-      if (kv.getType() != KeyValue.Type.Put.getCode()) {
-        deleteTracker.add(kv.getBuffer(),
-            kv.getQualifierOffset(),
-            kv.getQualifierLength(),
-            kv.getTimestamp(),
-            kv.getType());
-
-        results.add(heap.next());
-        continue;
-      }
-
-      if (deleteTracker.isDeleted(kv.getBuffer(),
-          kv.getQualifierOffset(),
-          kv.getQualifierLength(),
-          kv.getTimestamp())) {
-        heap.next();
-        continue;
-      }
-
       results.add(heap.next());
     }
     close();

core/src/test/java/org/apache/hadoop/hbase/regionserver/TestMinorCompactingStoreScanner.java

@@ -52,10 +52,18 @@ public class TestMinorCompactingStoreScanner extends TestCase {
         new MinorCompactingStoreScanner("cf", KeyValue.COMPARATOR, scanners);
     List<KeyValue> results = new ArrayList<KeyValue>();
     assertTrue(scan.next(results));
-    assertEquals(3, results.size());
+    assertEquals(11, results.size());
     assertEquals(kvs[0], results.get(0));
-    assertEquals(kvs[5], results.get(1));
+    assertEquals(kvs[1], results.get(1));
-    assertEquals(kvs[8], results.get(2));
+    assertEquals(kvs[2], results.get(2));
+    assertEquals(kvs[3], results.get(3));
+    assertEquals(kvs[5], results.get(4));
+    assertEquals(kvs[4], results.get(5));
+    assertEquals(kvs[6], results.get(6));
+    assertEquals(kvs[8], results.get(7));
+    assertEquals(kvs[7], results.get(8));
+    assertEquals(kvs[9], results.get(9));
+    assertEquals(kvs[10], results.get(10));
 
     results.clear();
     assertFalse(scan.next(results));
@@ -76,8 +84,9 @@ public class TestMinorCompactingStoreScanner extends TestCase {
         new MinorCompactingStoreScanner("cf", KeyValue.COMPARATOR, scanners);
     List<KeyValue> results = new ArrayList<KeyValue>();
     assertFalse(scan.next(results));
-    assertEquals(2, results.size());
+    assertEquals(3, results.size());
     assertEquals(kvs[0], results.get(0));
     assertEquals(kvs[1], results.get(1));
+    assertEquals(kvs[2], results.get(2));
   }
 }