From 020f520d173e2bc8c3bf23a4f5d472c3bb248ca6 Mon Sep 17 00:00:00 2001
From: Ramkrishna <ramkrishna.s.vasudevan@intel.com>
Date: Fri, 16 Jun 2017 11:03:04 +0530
Subject: [PATCH] HBASE-18220 Compaction scanners need not reopen storefile
 scanners while trying to switch over from pread to stream (Ram)

---
 .../hadoop/hbase/regionserver/StoreScanner.java       | 19 ++++++++++++++-----
 .../hadoop/hbase/regionserver/TestStoreScanner.java   | 16 ++++++++++++++++
 2 files changed, 30 insertions(+), 5 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index 4593a4dddb..985af4b47c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -168,7 +168,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
 
   /** An internal constructor. */
   protected StoreScanner(Store store, Scan scan, final ScanInfo scanInfo,
-      final NavigableSet<byte[]> columns, long readPt, boolean cacheBlocks) {
+      final NavigableSet<byte[]> columns, long readPt, boolean cacheBlocks, ScanType scanType) {
     this.readPt = readPt;
     this.store = store;
     this.cacheBlocks = cacheBlocks;
@@ -198,7 +198,12 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
       }
       // Always start with pread unless user specific stream. Will change to stream later if
       // readType is default if the scan keeps running for a long time.
-      this.scanUsePread = this.readType != Scan.ReadType.STREAM;
+      if (scanType != ScanType.COMPACT_DROP_DELETES
+          && scanType != ScanType.COMPACT_RETAIN_DELETES) {
+        // For compaction scanners never use Pread as already we have stream based scanners on the
+        // store files to be compacted
+        this.scanUsePread = this.readType != Scan.ReadType.STREAM;
+      }
     }
     this.preadMaxBytes = scanInfo.getPreadMaxBytes();
     this.cellsPerHeartbeatCheck = scanInfo.getCellsPerTimeoutCheck();
@@ -228,7 +233,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
   public StoreScanner(Store store, ScanInfo scanInfo, Scan scan, final NavigableSet<byte[]> columns,
       long readPt)
   throws IOException {
-    this(store, scan, scanInfo, columns, readPt, scan.getCacheBlocks());
+    this(store, scan, scanInfo, columns, readPt, scan.getCacheBlocks(), ScanType.USER_SCAN);
     if (columns != null && scan.isRaw()) {
       throw new DoNotRetryIOException("Cannot specify any column for a raw scan");
     }
@@ -302,7 +307,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
       List<? extends KeyValueScanner> scanners, ScanType scanType, long smallestReadPoint,
       long earliestPutTs, byte[] dropDeletesFromRow, byte[] dropDeletesToRow) throws IOException {
     this(store, scan, scanInfo, null,
-        ((HStore) store).getHRegion().getReadPoint(IsolationLevel.READ_COMMITTED), false);
+        ((HStore) store).getHRegion().getReadPoint(IsolationLevel.READ_COMMITTED), false, scanType);
     if (scan.hasFilter() || (scan.getStartRow() != null && scan.getStartRow().length > 0)
         || (scan.getStopRow() != null && scan.getStopRow().length > 0)
         || !scan.getTimeRange().isAllTime()) {
@@ -351,7 +356,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
       final NavigableSet<byte[]> columns, final List<? extends KeyValueScanner> scanners, long earliestPutTs,
       long readPt) throws IOException {
     this(null, scan, scanInfo, columns, readPt,
-        scanType == ScanType.USER_SCAN ? scan.getCacheBlocks() : false);
+        scanType == ScanType.USER_SCAN ? scan.getCacheBlocks() : false, scanType);
     if (scanType == ScanType.USER_SCAN) {
       this.matcher = UserScanQueryMatcher.create(scan, scanInfo, columns, oldestUnexpiredTS, now,
         null);
@@ -385,6 +390,10 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
         scan.includeStartRow(), scan.getStopRow(), scan.includeStopRow(), this.readPt));
   }
 
+  @VisibleForTesting
+  boolean isScanUsePread() {
+    return this.scanUsePread;
+  }
   /**
    * Seek the specified scanners with the given key
    * @param scanners
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
index 040a53e9a4..10f00a6928 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
@@ -992,4 +992,20 @@ public class TestStoreScanner {
       EnvironmentEdgeManagerTestHelper.reset();
     }
   }
+
+  @Test
+  public void testPreadNotEnabledForCompactionStoreScanners() throws Exception {
+    ScanType scanType = ScanType.COMPACT_RETAIN_DELETES;
+    long now = System.currentTimeMillis();
+    KeyValue[] kvs = new KeyValue[] {
+        new KeyValue(Bytes.toBytes("R1"), Bytes.toBytes("cf"), null, now - 1000,
+            KeyValue.Type.DeleteFamily),
+        KeyValueTestUtil.create("R1", "cf", "a", now - 10, KeyValue.Type.Put, "dont-care"), };
+    List<KeyValueScanner> scanners = scanFixture(kvs);
+    Scan scan = new Scan();
+    ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, 1, 500, KeepDeletedCells.FALSE,
+        HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.COMPARATOR);
+    StoreScanner storeScanner = new StoreScanner(scan, scanInfo, scanType, null, scanners);
+    assertFalse(storeScanner.isScanUsePread());
+  }
 }
\ No newline at end of file
-- 
2.11.4.GIT