HBASE-1620 Need to use special StoreScanner constructor for major compactions (passed sf, no caching, etc)

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

CHANGES.txt

@@ -443,6 +443,8 @@ Release 0.20.0 - Unreleased
    HBASE-1218  Implement in-memory column (Jon Gray via Stack)
    HBASE-1606  Remove zoo.cfg, put config options into hbase-site.xml
    HBASE-1575  HMaster does not handle ZK session expiration
+   HBASE-1620  Need to use special StoreScanner constructor for major compactions
+               (passed sf, no caching, etc) (Jon Gray via Stack)
 
   OPTIMIZATIONS
    HBASE-1412  Change values for delete column and column family in KeyValue

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

@@ -859,8 +859,7 @@ public class Store implements HConstants, HeapSize {
       try {
         Scan scan = new Scan();
         scan.setMaxVersions(family.getMaxVersions());
-        // TODO pass in the scanners/store files.
-        scanner = new StoreScanner(this, scan, null);
+        scanner = new StoreScanner(this, scan, scanners);
 
         // since scanner.next() can return 'false' but still be delivering data,
         // we have to use a do/while loop.

src/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java

@@ -70,6 +70,26 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
     this.store.addChangedReaderObserver(this);
   }
 
+  /**
+   * Used for major compactions.<p>
+   * 
+   * Opens a scanner across specified StoreFiles.
+   */
+  StoreScanner(Store store, Scan scan, KeyValueScanner [] scanners) {
+    this.store = store;
+    matcher = new ScanQueryMatcher(scan, store.getFamily().getName(),
+        null, store.ttl, store.comparator.getRawComparator(),
+        store.versionsToReturn(scan.getMaxVersions()));
+
+    // Seek all scanners to the initial key
+    for(KeyValueScanner scanner : scanners) {
+      scanner.seek(matcher.getStartKey());
+    }
+
+    // Combine all seeked scanners with a heap
+    heap = new KeyValueHeap(scanners, store.comparator);
+  }
+
   // Constructor for testing.
   StoreScanner(final Scan scan, final byte [] colFamily, final long ttl,
       final KeyValue.KVComparator comparator,