hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anoopsamj...@apache.org
Subject [1/2] hbase git commit: HBASE-16608 Introducing the ability to merge ImmutableSegments without copy-compaction or SQM usage. (Anastasia)
Date Mon, 24 Oct 2016 17:40:11 GMT
Repository: hbase
Updated Branches:
  refs/heads/master 1b12a6039 -> 988d1f9bc


http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
index 3405b49..277eb48 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
@@ -36,11 +36,13 @@ import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Threads;
 import org.apache.hadoop.hbase.wal.WAL;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
@@ -121,7 +123,7 @@ public class TestWalAndCompactingMemStoreFlush {
   }
 
   @Test(timeout = 180000)
-  public void testSelectiveFlushWhenEnabled() throws IOException {
+  public void testSelectiveFlushWithDataCompaction() throws IOException {
 
     // Set up the configuration
     Configuration conf = HBaseConfiguration.create();
@@ -130,9 +132,11 @@ public class TestWalAndCompactingMemStoreFlush {
         FlushNonSloppyStoresFirstPolicy.class.getName());
     conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
200 * 1024);
     conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.25);
+    // set memstore to do data compaction
+    conf.set("hbase.hregion.compacting.memstore.type", "data-compaction");
 
     // Intialize the region
-    Region region = initHRegion("testSelectiveFlushWhenEnabled", conf);
+    Region region = initHRegion("testSelectiveFlushWithDataCompaction", conf);
 
     // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
     for (int i = 1; i <= 1200; i++) {
@@ -313,7 +317,7 @@ public class TestWalAndCompactingMemStoreFlush {
     assertEquals(smallestSeqCF3PhaseII, smallestSeqCF3PhaseIV);
 
     // CF3 should be bottleneck for WAL
-     assertEquals(s, smallestSeqInRegionCurrentMemstorePhaseIV, smallestSeqCF3PhaseIV);
+    assertEquals(s, smallestSeqInRegionCurrentMemstorePhaseIV, smallestSeqCF3PhaseIV);
 
     // Flush!!!!!!!!!!!!!!!!!!!!!!
     // Trying to clean the existing memstores, CF2 all flushed to disk. The single
@@ -330,7 +334,7 @@ public class TestWalAndCompactingMemStoreFlush {
         .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
 
     assertTrue(
-        CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD < cf1MemstoreSizePhaseV);
+        CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD <= cf1MemstoreSizePhaseV);
     assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
         cf2MemstoreSizePhaseV);
     assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
@@ -371,9 +375,13 @@ public class TestWalAndCompactingMemStoreFlush {
     HBaseTestingUtility.closeRegionAndWAL(region);
   }
 
+  /*------------------------------------------------------------------------------*/
+  /* Check the same as above but for index-compaction type of compacting memstore */
   @Test(timeout = 180000)
-  public void testSelectiveFlushWhenEnabledNoFlattening() throws IOException {
+  public void testSelectiveFlushWithIndexCompaction() throws IOException {
 
+    /*------------------------------------------------------------------------------*/
+    /* SETUP */
     // Set up the configuration
     Configuration conf = HBaseConfiguration.create();
     conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 600 * 1024);
@@ -381,18 +389,17 @@ public class TestWalAndCompactingMemStoreFlush {
         FlushNonSloppyStoresFirstPolicy.class.getName());
     conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
200 * 1024);
     conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.5);
+    // set memstore to index-compaction
+    conf.set("hbase.hregion.compacting.memstore.type", "index-compaction");
 
-    // set memstore segment flattening to false and compact to skip-list
-    conf.setBoolean("hbase.hregion.compacting.memstore.flatten", false);
-    conf.setInt("hbase.hregion.compacting.memstore.type",1);
-
-    // Intialize the region
-    Region region = initHRegion("testSelectiveFlushWhenEnabled", conf);
+    // Initialize the region
+    Region region = initHRegion("testSelectiveFlushWithIndexCompaction", conf);
 
+    /*------------------------------------------------------------------------------*/
+    /* PHASE I - insertions */
     // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
     for (int i = 1; i <= 1200; i++) {
       region.put(createPut(1, i));        // compacted memstore
-
       if (i <= 100) {
         region.put(createPut(2, i));
         if (i <= 50) {
@@ -400,41 +407,32 @@ public class TestWalAndCompactingMemStoreFlush {
         }
       }
     }
-
     // Now add more puts for CF2, so that we only flush CF2 to disk
     for (int i = 100; i < 2000; i++) {
       region.put(createPut(2, i));
     }
 
-    long totalMemstoreSize = region.getMemstoreSize();
-
+    /*------------------------------------------------------------------------------*/
+    /*------------------------------------------------------------------------------*/
+    /* PHASE I - collect sizes */
+    long totalMemstoreSizePhaseI = region.getMemstoreSize();
     // Find the smallest LSNs for edits wrt to each CF.
     long smallestSeqCF1PhaseI = region.getOldestSeqIdOfStore(FAMILY1);
     long smallestSeqCF2PhaseI = region.getOldestSeqIdOfStore(FAMILY2);
     long smallestSeqCF3PhaseI = region.getOldestSeqIdOfStore(FAMILY3);
-
     // Find the sizes of the memstores of each CF.
     long cf1MemstoreSizePhaseI = region.getStore(FAMILY1).getMemStoreSize();
     long cf2MemstoreSizePhaseI = region.getStore(FAMILY2).getMemStoreSize();
     long cf3MemstoreSizePhaseI = region.getStore(FAMILY3).getMemStoreSize();
-
     // Get the overall smallest LSN in the region's memstores.
     long smallestSeqInRegionCurrentMemstorePhaseI = getWAL(region)
         .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
 
-    String s = "\n\n----------------------------------\n"
-        + "Upon initial insert and before any flush, size of CF1 is:"
-        + cf1MemstoreSizePhaseI + ", is CF1 compacted memstore?:"
-        + region.getStore(FAMILY1).isSloppyMemstore() + ". Size of CF2 is:"
-        + cf2MemstoreSizePhaseI + ", is CF2 compacted memstore?:"
-        + region.getStore(FAMILY2).isSloppyMemstore() + ". Size of CF3 is:"
-        + cf3MemstoreSizePhaseI + ", is CF3 compacted memstore?:"
-        + region.getStore(FAMILY3).isSloppyMemstore() + "\n";
-
+    /*------------------------------------------------------------------------------*/
+    /* PHASE I - validation */
     // The overall smallest LSN in the region's memstores should be the same as
     // the LSN of the smallest edit in CF1
     assertEquals(smallestSeqCF1PhaseI, smallestSeqInRegionCurrentMemstorePhaseI);
-
     // Some other sanity checks.
     assertTrue(smallestSeqCF1PhaseI < smallestSeqCF2PhaseI);
     assertTrue(smallestSeqCF2PhaseI < smallestSeqCF3PhaseI);
@@ -444,149 +442,169 @@ public class TestWalAndCompactingMemStoreFlush {
 
     // The total memstore size should be the same as the sum of the sizes of
     // memstores of CF1, CF2 and CF3.
-    String msg = "totalMemstoreSize="+totalMemstoreSize +
-        " DefaultMemStore.DEEP_OVERHEAD="+DefaultMemStore.DEEP_OVERHEAD +
-        " cf1MemstoreSizePhaseI="+cf1MemstoreSizePhaseI +
-        " cf2MemstoreSizePhaseI="+cf2MemstoreSizePhaseI +
-        " cf3MemstoreSizePhaseI="+cf3MemstoreSizePhaseI ;
-    assertEquals(msg,
-        totalMemstoreSize + 2 * (CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)
-            + (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD),
+    assertEquals(
+        totalMemstoreSizePhaseI
+            + 1 * DefaultMemStore.DEEP_OVERHEAD
+            + 2 * CompactingMemStore.DEEP_OVERHEAD
+            + 3 * MutableSegment.DEEP_OVERHEAD,
         cf1MemstoreSizePhaseI + cf2MemstoreSizePhaseI + cf3MemstoreSizePhaseI);
 
-    // Flush!!!!!!!!!!!!!!!!!!!!!!
-    // We have big compacting memstore CF1 and two small memstores:
-    // CF2 (not compacted) and CF3 (compacting)
-    // All together they are above the flush size lower bound.
-    // Since CF1 and CF3 should be flushed to memory (not to disk),
-    // CF2 is going to be flushed to disk.
-    // CF1 - nothing to compact, CF3 - should be twice compacted
+    /*------------------------------------------------------------------------------*/
+    /* PHASE I - Flush */
+    // First Flush in Test!!!!!!!!!!!!!!!!!!!!!!
+    // CF1, CF2, CF3, all together they are above the flush size lower bound.
+    // Since CF1 and CF3 are compacting, CF2 is going to be flushed to disk.
+    // CF1 and CF3 - flushed to memory and flatten explicitly
+    region.flush(false);
     CompactingMemStore cms1 = (CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore;
     CompactingMemStore cms3 = (CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore;
     cms1.flushInMemory();
     cms3.flushInMemory();
-    region.flush(false);
 
+    // CF3/CF1 should be merged so wait here to be sure the compaction is done
+    while (((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+    while (((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+
+    /*------------------------------------------------------------------------------*/
+    /*------------------------------------------------------------------------------*/
+    /* PHASE II - collect sizes */
     // Recalculate everything
     long cf1MemstoreSizePhaseII = region.getStore(FAMILY1).getMemStoreSize();
     long cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
     long cf3MemstoreSizePhaseII = region.getStore(FAMILY3).getMemStoreSize();
-
     long smallestSeqInRegionCurrentMemstorePhaseII = getWAL(region)
         .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
     // Find the smallest LSNs for edits wrt to each CF.
-    long smallestSeqCF1PhaseII = region.getOldestSeqIdOfStore(FAMILY1);
-    long smallestSeqCF2PhaseII = region.getOldestSeqIdOfStore(FAMILY2);
     long smallestSeqCF3PhaseII = region.getOldestSeqIdOfStore(FAMILY3);
+    long totalMemstoreSizePhaseII = region.getMemstoreSize();
 
-    s = s + "DefaultMemStore DEEP_OVERHEAD is:" + DefaultMemStore.DEEP_OVERHEAD
-        + ", CompactingMemStore DEEP_OVERHEAD is:" + CompactingMemStore.DEEP_OVERHEAD
-        + "\n----After first flush! CF1 should be flushed to memory, but not compacted.---\n"
-        + "Size of CF1 is:" + cf1MemstoreSizePhaseII + ", size of CF2 is:" + cf2MemstoreSizePhaseII
-        + ", size of CF3 is:" + cf3MemstoreSizePhaseII + "\n";
-
-    // CF1 was flushed to memory, but there is nothing to compact, should
-    // remain the same size plus renewed empty skip-list
-    assertEquals(s, cf1MemstoreSizePhaseII, cf1MemstoreSizePhaseI
-        + ImmutableSegment.DEEP_OVERHEAD_CAM + CompactionPipeline.ENTRY_OVERHEAD);
-
+    /*------------------------------------------------------------------------------*/
+    /* PHASE II - validation */
+    // CF1 was flushed to memory, should be flattened and take less space
+    assertTrue(cf1MemstoreSizePhaseII < cf1MemstoreSizePhaseI);
     // CF2 should become empty
     assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
         cf2MemstoreSizePhaseII);
-
-    // verify that CF3 was flushed to memory and was compacted (this is approximation check)
-    assertTrue(cf3MemstoreSizePhaseI / 2 + CompactingMemStore.DEEP_OVERHEAD
-        + ImmutableSegment.DEEP_OVERHEAD_CAM
-        + CompactionPipeline.ENTRY_OVERHEAD > cf3MemstoreSizePhaseII);
-    assertTrue(cf3MemstoreSizePhaseI/2 < cf3MemstoreSizePhaseII);
-
+    // verify that CF3 was flushed to memory and was not compacted (this is an approximation
check)
+    // if compacted CF# should be at least twice less because its every key was duplicated
+    assertTrue(cf3MemstoreSizePhaseI / 2 < cf3MemstoreSizePhaseII);
 
     // Now the smallest LSN in the region should be the same as the smallest
     // LSN in the memstore of CF1.
     assertEquals(smallestSeqInRegionCurrentMemstorePhaseII, smallestSeqCF1PhaseI);
-
+    // The total memstore size should be the same as the sum of the sizes of
+    // memstores of CF1, CF2 and CF3. Counting the empty active segments in CF1/2/3 and pipeline
+    // items in CF1/2
+    assertEquals(
+        totalMemstoreSizePhaseII
+            + 1 * DefaultMemStore.DEEP_OVERHEAD
+            + 2 * CompactingMemStore.DEEP_OVERHEAD
+            + 3 * MutableSegment.DEEP_OVERHEAD
+            + 2 * CompactionPipeline.ENTRY_OVERHEAD
+            + 2 * ImmutableSegment.DEEP_OVERHEAD_CAM,
+        cf1MemstoreSizePhaseII + cf2MemstoreSizePhaseII + cf3MemstoreSizePhaseII);
+
+    /*------------------------------------------------------------------------------*/
+    /*------------------------------------------------------------------------------*/
+    /* PHASE III - insertions */
     // Now add more puts for CF1, so that we also flush CF1 to disk instead of
-    // memory in next flush
-    for (int i = 1200; i < 2000; i++) {
+    // memory in next flush. This is causing the CF! to be flushed to memory twice.
+    for (int i = 1200; i < 8000; i++) {
       region.put(createPut(1, i));
     }
 
-    s = s + "The smallest sequence in region WAL is: " + smallestSeqInRegionCurrentMemstorePhaseII
-        + ", the smallest sequence in CF1:" + smallestSeqCF1PhaseII + ", " +
-        "the smallest sequence in CF2:"
-        + smallestSeqCF2PhaseII +", the smallest sequence in CF3:" + smallestSeqCF3PhaseII
+ "\n";
+    // CF1 should be flatten and merged so wait here to be sure the compaction is done
+    while (((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
 
-    // How much does the CF1 memstore occupy? Will be used later.
+    /*------------------------------------------------------------------------------*/
+    /* PHASE III - collect sizes */
+    // How much does the CF1 memstore occupy now? Will be used later.
     long cf1MemstoreSizePhaseIII = region.getStore(FAMILY1).getMemStoreSize();
-    long smallestSeqCF1PhaseIII = region.getOldestSeqIdOfStore(FAMILY1);
-
-    s = s + "----After more puts into CF1 its size is:" + cf1MemstoreSizePhaseIII
-        + ", and its sequence is:" + smallestSeqCF1PhaseIII + " ----\n" ;
+    long totalMemstoreSizePhaseIII = region.getMemstoreSize();
 
-
-    // Flush!!!!!!!!!!!!!!!!!!!!!!
-    // Flush again, CF1 is flushed to disk
-    // CF2 is flushed to disk, because it is not in-memory compacted memstore
-    // CF3 is flushed empty to memory (actually nothing happens to CF3)
+    /*------------------------------------------------------------------------------*/
+    /* PHASE III - validation */
+    // The total memstore size should be the same as the sum of the sizes of
+    // memstores of CF1, CF2 and CF3. Counting the empty active segments in CF1/2/3 and pipeline
+    // items in CF1/2
+    assertEquals(
+        totalMemstoreSizePhaseIII
+            + 1 * DefaultMemStore.DEEP_OVERHEAD
+            + 2 * CompactingMemStore.DEEP_OVERHEAD
+            + 3 * MutableSegment.DEEP_OVERHEAD
+            + 2 * CompactionPipeline.ENTRY_OVERHEAD
+            + 2 * ImmutableSegment.DEEP_OVERHEAD_CAM,
+        cf1MemstoreSizePhaseIII + cf2MemstoreSizePhaseII + cf3MemstoreSizePhaseII);
+
+    /*------------------------------------------------------------------------------*/
+    /* PHASE III - Flush */
+    // Second Flush in Test!!!!!!!!!!!!!!!!!!!!!!
+    // CF1 is flushed to disk, but not entirely emptied.
+    // CF2 was and remained empty, same way nothing happens to CF3
     region.flush(false);
 
+    /*------------------------------------------------------------------------------*/
+    /*------------------------------------------------------------------------------*/
+    /* PHASE IV - collect sizes */
     // Recalculate everything
     long cf1MemstoreSizePhaseIV = region.getStore(FAMILY1).getMemStoreSize();
     long cf2MemstoreSizePhaseIV = region.getStore(FAMILY2).getMemStoreSize();
     long cf3MemstoreSizePhaseIV = region.getStore(FAMILY3).getMemStoreSize();
-
     long smallestSeqInRegionCurrentMemstorePhaseIV = getWAL(region)
         .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
-    long smallestSeqCF1PhaseIV = region.getOldestSeqIdOfStore(FAMILY1);
-    long smallestSeqCF2PhaseIV = region.getOldestSeqIdOfStore(FAMILY2);
     long smallestSeqCF3PhaseIV = region.getOldestSeqIdOfStore(FAMILY3);
 
-    s = s + "----After SECOND FLUSH, CF1 size is:" + cf1MemstoreSizePhaseIV + ", CF2 size
is:"
-        + cf2MemstoreSizePhaseIV + " and CF3 size is:" + cf3MemstoreSizePhaseIV
-        + "\n";
-
-    s = s + "The smallest sequence in region WAL is: " + smallestSeqInRegionCurrentMemstorePhaseIV
-        + ", the smallest sequence in CF1:" + smallestSeqCF1PhaseIV + ", " +
-        "the smallest sequence in CF2:"
-        + smallestSeqCF2PhaseIV +", the smallest sequence in CF3:" + smallestSeqCF3PhaseIV
-        + "\n";
-
-    // CF1's pipeline component (inserted before first flush) should be flushed to disk
-    // CF2 should be flushed to disk
-    assertEquals(cf1MemstoreSizePhaseIII - cf1MemstoreSizePhaseI + CompactingMemStore.DEEP_OVERHEAD
-        + MutableSegment.DEEP_OVERHEAD, cf1MemstoreSizePhaseIV);
+    /*------------------------------------------------------------------------------*/
+    /* PHASE IV - validation */
+    // CF1's biggest pipeline component (inserted before first flush) should be flushed to
disk
+    // CF2 should remain empty
+    assertTrue(cf1MemstoreSizePhaseIII > cf1MemstoreSizePhaseIV);
     assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
         cf2MemstoreSizePhaseIV);
-
     // CF3 shouldn't have been touched.
     assertEquals(cf3MemstoreSizePhaseIV, cf3MemstoreSizePhaseII);
-
     // the smallest LSN of CF3 shouldn't change
     assertEquals(smallestSeqCF3PhaseII, smallestSeqCF3PhaseIV);
-
     // CF3 should be bottleneck for WAL
-    assertEquals(s, smallestSeqInRegionCurrentMemstorePhaseIV, smallestSeqCF3PhaseIV);
+    assertEquals(smallestSeqInRegionCurrentMemstorePhaseIV, smallestSeqCF3PhaseIV);
 
-    // Flush!!!!!!!!!!!!!!!!!!!!!!
-    // Clearing the existing memstores, CF2 all flushed to disk. The single
-    // memstore segment in the compaction pipeline of CF1 and CF3 should be flushed to disk.
-    // Note that active sets of CF1 and CF3 are empty
+    /*------------------------------------------------------------------------------*/
+    /* PHASE IV - Flush */
+    // Third Flush in Test!!!!!!!!!!!!!!!!!!!!!!
+    // Force flush to disk on all memstores (flush parameter true).
+    // CF1/CF3 all flushed to disk. Note that active sets of CF1 and CF3 are empty
     region.flush(true);
 
+    /*------------------------------------------------------------------------------*/
+    /*------------------------------------------------------------------------------*/
+    /* PHASE V - collect sizes */
     // Recalculate everything
     long cf1MemstoreSizePhaseV = region.getStore(FAMILY1).getMemStoreSize();
     long cf2MemstoreSizePhaseV = region.getStore(FAMILY2).getMemStoreSize();
     long cf3MemstoreSizePhaseV = region.getStore(FAMILY3).getMemStoreSize();
     long smallestSeqInRegionCurrentMemstorePhaseV = getWAL(region)
         .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
+    long totalMemstoreSizePhaseV = region.getMemstoreSize();
 
+    /*------------------------------------------------------------------------------*/
+    /* PHASE V - validation */
     assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
         cf1MemstoreSizePhaseV);
     assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
         cf2MemstoreSizePhaseV);
     assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
         cf3MemstoreSizePhaseV);
-
+    // The total memstores size should be empty
+    assertEquals(totalMemstoreSizePhaseV, 0);
     // Because there is nothing in any memstore the WAL's LSN should be -1
     assertEquals(smallestSeqInRegionCurrentMemstorePhaseV, HConstants.NO_SEQNUM);
 
@@ -594,6 +612,9 @@ public class TestWalAndCompactingMemStoreFlush {
     // any Column Family above the threshold?
     // In that case, we should flush all the CFs.
 
+    /*------------------------------------------------------------------------------*/
+    /*------------------------------------------------------------------------------*/
+    /* PHASE VI - insertions */
     // The memstore limit is 200*1024 and the column family flush threshold is
     // around 50*1024. We try to just hit the memstore limit with each CF's
     // memstore being below the CF flush threshold.
@@ -605,24 +626,32 @@ public class TestWalAndCompactingMemStoreFlush {
       region.put(createPut(5, i));
     }
 
-    region.flush(false);
-
-    s = s + "----AFTER THIRD AND FORTH FLUSH, The smallest sequence in region WAL is: "
-        + smallestSeqInRegionCurrentMemstorePhaseV
-        + ". After additional inserts and last flush, the entire region size is:" + region
-        .getMemstoreSize()
-        + "\n----------------------------------\n";
+    long cf1ActiveSizePhaseVI = region.getStore(FAMILY1).getMemStoreSize();
+    long cf3ActiveSizePhaseVI = region.getStore(FAMILY3).getMemStoreSize();
+    long cf5ActiveSizePhaseVI = region.getStore(FAMILIES[4]).getMemStoreSize();
 
+    /*------------------------------------------------------------------------------*/
+    /* PHASE VI - Flush */
+    // Fourth Flush in Test!!!!!!!!!!!!!!!!!!!!!!
+    // None among compacting memstores was flushed to memory due to previous puts.
+    // But is going to be moved to pipeline and flatten due to the flush.
+    region.flush(false);
     // Since we won't find any CF above the threshold, and hence no specific
     // store to flush, we should flush all the memstores
-    // Also compacted memstores are flushed to disk.
-    assertEquals(0, region.getMemstoreSize());
-    System.out.println(s);
+    // Also compacted memstores are flushed to disk, but not entirely emptied
+    long cf1ActiveSizePhaseVII = region.getStore(FAMILY1).getMemStoreSize();
+    long cf3ActiveSizePhaseVII = region.getStore(FAMILY3).getMemStoreSize();
+    long cf5ActiveSizePhaseVII = region.getStore(FAMILIES[4]).getMemStoreSize();
+
+    assertTrue(cf1ActiveSizePhaseVII < cf1ActiveSizePhaseVI);
+    assertTrue(cf3ActiveSizePhaseVII < cf3ActiveSizePhaseVI);
+    assertTrue(cf5ActiveSizePhaseVII < cf5ActiveSizePhaseVI);
+
     HBaseTestingUtility.closeRegionAndWAL(region);
   }
 
   @Test(timeout = 180000)
-  public void testSelectiveFlushWhenEnabledAndWALinCompaction() throws IOException {
+  public void testSelectiveFlushAndWALinDataCompaction() throws IOException {
     // Set up the configuration
     Configuration conf = HBaseConfiguration.create();
     conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 600 * 1024);
@@ -631,9 +660,11 @@ public class TestWalAndCompactingMemStoreFlush {
     conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
200 *
         1024);
     conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.5);
+    // set memstore to do data compaction and not to use the speculative scan
+    conf.set("hbase.hregion.compacting.memstore.type", "data-compaction");
 
     // Intialize the HRegion
-    HRegion region = initHRegion("testSelectiveFlushWhenNotEnabled", conf);
+    HRegion region = initHRegion("testSelectiveFlushAndWALinDataCompaction", conf);
     // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
     for (int i = 1; i <= 1200; i++) {
       region.put(createPut(1, i));
@@ -752,6 +783,253 @@ public class TestWalAndCompactingMemStoreFlush {
     HBaseTestingUtility.closeRegionAndWAL(region);
   }
 
+  @Test(timeout = 180000)
+  public void testSelectiveFlushAndWALinIndexCompaction() throws IOException {
+    // Set up the configuration
+    Configuration conf = HBaseConfiguration.create();
+    conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 600 * 1024);
+    conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY,
+        FlushNonSloppyStoresFirstPolicy.class.getName());
+    conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
+        200 * 1024);
+    conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.5);
+    // set memstore to do data compaction and not to use the speculative scan
+    conf.set("hbase.hregion.compacting.memstore.type", "index-compaction");
+
+    // Intialize the HRegion
+    HRegion region = initHRegion("testSelectiveFlushAndWALinDataCompaction", conf);
+    // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
+    for (int i = 1; i <= 1200; i++) {
+      region.put(createPut(1, i));
+      if (i <= 100) {
+        region.put(createPut(2, i));
+        if (i <= 50) {
+          region.put(createPut(3, i));
+        }
+      }
+    }
+    // Now add more puts for CF2, so that we only flush CF2 to disk
+    for (int i = 100; i < 2000; i++) {
+      region.put(createPut(2, i));
+    }
+
+    long totalMemstoreSize = region.getMemstoreSize();
+
+    // Find the sizes of the memstores of each CF.
+    long cf1MemstoreSizePhaseI = region.getStore(FAMILY1).getMemStoreSize();
+    long cf2MemstoreSizePhaseI = region.getStore(FAMILY2).getMemStoreSize();
+    long cf3MemstoreSizePhaseI = region.getStore(FAMILY3).getMemStoreSize();
+
+    // Some other sanity checks.
+    assertTrue(cf1MemstoreSizePhaseI > 0);
+    assertTrue(cf2MemstoreSizePhaseI > 0);
+    assertTrue(cf3MemstoreSizePhaseI > 0);
+
+    // The total memstore size should be the same as the sum of the sizes of
+    // memstores of CF1, CF2 and CF3.
+    assertEquals(
+        totalMemstoreSize
+            + 1 * DefaultMemStore.DEEP_OVERHEAD
+            + 2 * CompactingMemStore.DEEP_OVERHEAD
+            + 3 * MutableSegment.DEEP_OVERHEAD,
+        cf1MemstoreSizePhaseI + cf2MemstoreSizePhaseI + cf3MemstoreSizePhaseI);
+
+    // Flush!
+    ((CompactingMemStore) ((HStore)region.getStore(FAMILY1)).memstore).flushInMemory();
+    ((CompactingMemStore) ((HStore)region.getStore(FAMILY3)).memstore).flushInMemory();
+    // CF1 and CF3 should be compacted so wait here to be sure the compaction is done
+    while (((CompactingMemStore) ((HStore)region.getStore(FAMILY1)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+    while (((CompactingMemStore) ((HStore)region.getStore(FAMILY3)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+    region.flush(false);
+
+    long cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
+
+    long smallestSeqInRegionCurrentMemstorePhaseII = region.getWAL()
+        .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
+    long smallestSeqCF1PhaseII = region.getOldestSeqIdOfStore(FAMILY1);
+    long smallestSeqCF2PhaseII = region.getOldestSeqIdOfStore(FAMILY2);
+    long smallestSeqCF3PhaseII = region.getOldestSeqIdOfStore(FAMILY3);
+
+    // CF2 should have been cleared
+    assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
+        cf2MemstoreSizePhaseII);
+
+    // Add same entries to compact them later
+    for (int i = 1; i <= 1200; i++) {
+      region.put(createPut(1, i));
+      if (i <= 100) {
+        region.put(createPut(2, i));
+        if (i <= 50) {
+          region.put(createPut(3, i));
+        }
+      }
+    }
+    // Now add more puts for CF2, so that we only flush CF2 to disk
+    for (int i = 100; i < 2000; i++) {
+      region.put(createPut(2, i));
+    }
+
+    long smallestSeqInRegionCurrentMemstorePhaseIII = region.getWAL()
+        .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
+    long smallestSeqCF1PhaseIII = region.getOldestSeqIdOfStore(FAMILY1);
+    long smallestSeqCF2PhaseIII = region.getOldestSeqIdOfStore(FAMILY2);
+    long smallestSeqCF3PhaseIII = region.getOldestSeqIdOfStore(FAMILY3);
+
+    // Flush!
+    ((CompactingMemStore) ((HStore)region.getStore(FAMILY1)).memstore).flushInMemory();
+    ((CompactingMemStore) ((HStore)region.getStore(FAMILY3)).memstore).flushInMemory();
+    // CF1 and CF3 should be compacted so wait here to be sure the compaction is done
+    while (((CompactingMemStore) ((HStore)region.getStore(FAMILY1)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+    while (((CompactingMemStore) ((HStore)region.getStore(FAMILY3)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+    region.flush(false);
+
+    long smallestSeqInRegionCurrentMemstorePhaseIV = region.getWAL()
+        .getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
+    long smallestSeqCF1PhaseIV = region.getOldestSeqIdOfStore(FAMILY1);
+    long smallestSeqCF2PhaseIV = region.getOldestSeqIdOfStore(FAMILY2);
+    long smallestSeqCF3PhaseIV = region.getOldestSeqIdOfStore(FAMILY3);
+
+    // now check that the LSN of the entire WAL, of CF1 and of CF3 has NOT progressed due
to merge
+    assertFalse(
+        smallestSeqInRegionCurrentMemstorePhaseIV > smallestSeqInRegionCurrentMemstorePhaseIII);
+    assertFalse(smallestSeqCF1PhaseIV > smallestSeqCF1PhaseIII);
+    assertFalse(smallestSeqCF3PhaseIV > smallestSeqCF3PhaseIII);
+
+    HBaseTestingUtility.closeRegionAndWAL(region);
+  }
+
+  // should end in 300 seconds (5 minutes)
+  @Test(timeout = 300000)
+  public void testStressFlushAndWALinIndexCompaction() throws IOException {
+    // Set up the configuration
+    Configuration conf = HBaseConfiguration.create();
+    conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 600 * 1024);
+    conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY,
+        FlushNonSloppyStoresFirstPolicy.class.getName());
+    conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
+        200 * 1024);
+    conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.5);
+    // set memstore to do data compaction and not to use the speculative scan
+    conf.set("hbase.hregion.compacting.memstore.type", "index-compaction");
+
+    // Successfully initialize the HRegion
+    HRegion region = initHRegion("testSelectiveFlushAndWALinDataCompaction", conf);
+
+    Thread[] threads = new Thread[25];
+    for (int i = 0; i < threads.length; i++) {
+      int id = i * 10000;
+      ConcurrentPutRunnable runnable = new ConcurrentPutRunnable(region, id);
+      threads[i] = new Thread(runnable);
+      threads[i].start();
+    }
+    Threads.sleep(10000); // let other threads start
+    region.flush(true); // enforce flush of everything TO DISK while there are still ongoing
puts
+    Threads.sleep(10000); // let other threads continue
+    region.flush(true); // enforce flush of everything TO DISK while there are still ongoing
puts
+
+    ((CompactingMemStore) ((HStore)region.getStore(FAMILY1)).memstore).flushInMemory();
+    ((CompactingMemStore) ((HStore)region.getStore(FAMILY3)).memstore).flushInMemory();
+    while (((CompactingMemStore) ((HStore)region.getStore(FAMILY1)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+    while (((CompactingMemStore) ((HStore)region.getStore(FAMILY3)).memstore)
+        .isMemStoreFlushingInMemory()) {
+      Threads.sleep(10);
+    }
+
+    for (int i = 0; i < threads.length; i++) {
+      try {
+        threads[i].join();
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  /**
+   * The in-memory-flusher thread performs the flush asynchronously. There is at most one
thread per
+   * memstore instance. It takes the updatesLock exclusively, pushes active into the pipeline,
+   * releases updatesLock and compacts the pipeline.
+   */
+  private class ConcurrentPutRunnable implements Runnable {
+    private final HRegion stressedRegion;
+    private final int startNumber;
+
+    ConcurrentPutRunnable(HRegion r, int i) {
+      this.stressedRegion = r;
+      this.startNumber = i;
+    }
+
+    @Override
+    public void run() {
+
+      try {
+        int dummy = startNumber / 10000;
+        System.out.print("Thread " + dummy + " with start number " + startNumber + " starts\n");
+        // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
+        for (int i = startNumber; i <= startNumber + 3000; i++) {
+          stressedRegion.put(createPut(1, i));
+          if (i <= startNumber + 2000) {
+            stressedRegion.put(createPut(2, i));
+            if (i <= startNumber + 1000) {
+              stressedRegion.put(createPut(3, i));
+            }
+          }
+        }
+        System.out.print("Thread with start number " + startNumber + " continues to more
puts\n");
+        // Now add more puts for CF2, so that we only flush CF2 to disk
+        for (int i = startNumber + 3000; i < startNumber + 5000; i++) {
+          stressedRegion.put(createPut(2, i));
+        }
+        // And add more puts for CF1
+        for (int i = startNumber + 5000; i < startNumber + 7000; i++) {
+          stressedRegion.put(createPut(1, i));
+        }
+        System.out.print("Thread with start number " + startNumber + " flushes\n");
+        // flush (IN MEMORY) one of the stores (each thread flushes different store)
+        // and wait till the flush and the following action are done
+        if (startNumber == 0) {
+          ((CompactingMemStore) ((HStore) stressedRegion.getStore(FAMILY1)).memstore)
+              .flushInMemory();
+          while (((CompactingMemStore) ((HStore) stressedRegion.getStore(FAMILY1)).memstore)
+              .isMemStoreFlushingInMemory()) {
+            Threads.sleep(10);
+          }
+        }
+        if (startNumber == 10000) {
+          ((CompactingMemStore) ((HStore) stressedRegion.getStore(FAMILY2)).memstore).flushInMemory();
+          while (((CompactingMemStore) ((HStore) stressedRegion.getStore(FAMILY2)).memstore)
+              .isMemStoreFlushingInMemory()) {
+            Threads.sleep(10);
+          }
+        }
+        if (startNumber == 20000) {
+          ((CompactingMemStore) ((HStore) stressedRegion.getStore(FAMILY3)).memstore).flushInMemory();
+          while (((CompactingMemStore) ((HStore) stressedRegion.getStore(FAMILY3)).memstore)
+              .isMemStoreFlushingInMemory()) {
+            Threads.sleep(10);
+          }
+        }
+        System.out.print("Thread with start number " + startNumber + " finishes\n");
+      } catch (IOException e) {
+        assert false;
+      }
+    }
+  }
+
   private WAL getWAL(Region region) {
     return ((HRegion)region).getWAL();
   }


Mime
View raw message