HBASE-3290 fix #2. Fixes bugs found with major compaction logic for

threshold files & major compaction jitter generation encountered
during testing.

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

src/main/java/org/apache/hadoop/hbase/regionserver/Store.java

@@ -716,32 +716,55 @@ public class Store implements HeapSize {
   }
 
   /*
-   * Gets lowest timestamp from files in a dir
+   * Gets lowest timestamp from candidate StoreFiles
    *
    * @param fs
    * @param dir
    * @throws IOException
    */
-  private static long getLowestTimestamp(FileSystem fs, Path dir) throws IOException {
-    FileStatus[] stats = fs.listStatus(dir);
+  private static long getLowestTimestamp(FileSystem fs, 
+      final List<StoreFile> candidates) throws IOException {
+    long minTs = Long.MAX_VALUE;
+    if (candidates.isEmpty()) {
+      return minTs; 
+    }
+    Path[] p = new Path[candidates.size()];
+    for (int i = 0; i < candidates.size(); ++i) {
+      p[i] = candidates.get(i).getPath();
+    }
+    
+    FileStatus[] stats = fs.listStatus(p);
     if (stats == null || stats.length == 0) {
-      return 0l;
+      return minTs;
     }
-    long lowTimestamp = Long.MAX_VALUE;
-    for (int i = 0; i < stats.length; i++) {
-      long timestamp = stats[i].getModificationTime();
-      if (timestamp < lowTimestamp){
-        lowTimestamp = timestamp;
-      }
+    for (FileStatus s : stats) {
+      minTs = Math.min(minTs, s.getModificationTime());
     }
-    return lowTimestamp;
+    return minTs;
   }
 
   /*
    * @return True if we should run a major compaction.
    */
   boolean isMajorCompaction() throws IOException {
-    return isMajorCompaction(storefiles);
+    for (StoreFile sf : this.storefiles) {
+      if (sf.getReader() == null) {
+        LOG.debug("StoreFile " + sf + " has null Reader");
+        return false;
+      }
+    }
+    
+    List<StoreFile> candidates = new ArrayList<StoreFile>(this.storefiles);
+
+    // exclude files above the max compaction threshold
+    // except: save all references. we MUST compact them
+    int pos = 0;
+    while (pos < candidates.size() && 
+           candidates.get(pos).getReader().length() > this.maxCompactSize &&
+           !candidates.get(pos).isReference()) ++pos;
+    candidates.subList(0, pos).clear();
+
+    return isMajorCompaction(candidates);
   }
 
   /*
@@ -755,8 +778,7 @@ public class Store implements HeapSize {
       return result;
         }
     // TODO: Use better method for determining stamp of last major (HBASE-2990)
-    long lowTimestamp = getLowestTimestamp(fs,
-        filesToCompact.get(0).getPath().getParent());
+    long lowTimestamp = getLowestTimestamp(fs, filesToCompact);
     long now = System.currentTimeMillis();
     if (lowTimestamp > 0l && lowTimestamp < (now - this.majorCompactionTime)) {
       // Major compaction time has elapsed.
@@ -778,7 +800,6 @@ public class Store implements HeapSize {
               "; time since last major compaction " + (now - lowTimestamp) + "ms");
         }
         result = true;
-        this.majorCompactionTime = getNextMajorCompactTime();
       }
     }
     return result;
@@ -849,15 +870,18 @@ public class Store implements HeapSize {
              !filesToCompact.get(pos).isReference()) ++pos;
       filesToCompact.subList(0, pos).clear();
     }
-    
-    // major compact on user action or age (caveat: we have too many files)
-    boolean majorcompaction = (forcemajor || isMajorCompaction(filesToCompact))
-      && filesToCompact.size() < this.maxFilesToCompact;
 
     if (filesToCompact.isEmpty()) {
       LOG.debug(this.storeNameStr + ": no store files to compact");
       return filesToCompact;
     }
+    
+    // major compact on user action or age (caveat: we have too many files)
+    boolean majorcompaction = (forcemajor || isMajorCompaction(filesToCompact))
+      && filesToCompact.size() < this.maxFilesToCompact;
+    if (majorcompaction) {
+      this.majorCompactionTime = getNextMajorCompactTime();
+    }
 
     if (!majorcompaction && !hasReferences(filesToCompact)) {
       // we're doing a minor compaction, let's see what files are applicable

src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactSelection.java

@@ -51,6 +51,7 @@ public class TestCompactSelection extends TestCase {
   private Store store;
   private static final String DIR
     = HBaseTestingUtility.getTestDir() + "/TestCompactSelection/";
+  private static Path TEST_FILE;
 
   private static final int minFiles = 3;
   private static final int maxFiles = 5;
@@ -86,6 +87,8 @@ public class TestCompactSelection extends TestCase {
     HRegion region = new HRegion(basedir, hlog, fs, conf, info, null);
 
     store = new Store(basedir, region, hcd, fs, conf);
+    TEST_FILE = StoreFile.getRandomFilename(fs, store.getHomedir());
+    fs.create(TEST_FILE);
   }
 
   // used so our tests don't deal with actual StoreFiles
@@ -94,7 +97,7 @@ public class TestCompactSelection extends TestCase {
     boolean isRef = false;
 
     MockStoreFile(long length, boolean isRef) throws IOException {
-      super(TEST_UTIL.getTestFileSystem(), new Path("_"), false,
+      super(TEST_UTIL.getTestFileSystem(), TEST_FILE, false,
             TEST_UTIL.getConfiguration(), BloomType.NONE, false);
       this.length = length;
       this.isRef  = isRef;