HBASE-12583 Allow creating reference files even the split row not lies in the storefile range if required(Rajeshbabu)

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -6715,4 +6715,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
       configurationManager.get().deregisterObserver(s);
     }
   }
+
+  /**
+   * @return split policy for this region.
+   */
+  public RegionSplitPolicy getSplitPolicy() {
+    return this.splitPolicy;
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java

@@ -568,11 +568,13 @@ public class HRegionFileSystem {
    * @param splitRow Split Row
    * @param top True if we are referring to the top half of the hfile.
    * @return Path to created reference.
+   * @param splitPolicy
    * @throws IOException
    */
-  Path splitStoreFile(final HRegionInfo hri, final String familyName,
-      final StoreFile f, final byte[] splitRow, final boolean top) throws IOException {
+  Path splitStoreFile(final HRegionInfo hri, final String familyName, final StoreFile f,
+      final byte[] splitRow, final boolean top, RegionSplitPolicy splitPolicy) throws IOException {
 
+    if (splitPolicy == null || !splitPolicy.skipStoreFileRangeCheck()) {
       // Check whether the split row lies in the range of the store file
       // If it is outside the range, return directly.
       if (top) {
@@ -596,6 +598,7 @@ public class HRegionFileSystem {
           return null;
         }
       }
+    }
 
     f.getReader().close(true);
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionSplitPolicy.java

@@ -125,4 +125,17 @@ public abstract class RegionSplitPolicy extends Configured {
           e);
     }
   }
+
+  /**
+   * In {@link HRegionFileSystem#splitStoreFile(org.apache.hadoop.hbase.HRegionInfo,
+   * String, StoreFile, byte[], boolean)} we are not creating the split reference if split row
+   * not lies in the StoreFile range. But some use cases we may need to create the split reference
+   * even the split row not lies in the range.
+   * This method can be used to whether to skip the the StoreRile range check or not.
+   * @return whether to skip the StoreFile range check or or not
+   */
+  protected boolean skipStoreFileRangeCheck() {
+    return false;
+  }
+
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitTransaction.java

@@ -753,8 +753,12 @@ public class SplitTransaction {
     HRegionFileSystem fs = this.parent.getRegionFileSystem();
     String familyName = Bytes.toString(family);
 
-    Path path_a = fs.splitStoreFile(this.hri_a, familyName, sf, this.splitrow, false);
-    Path path_b = fs.splitStoreFile(this.hri_b, familyName, sf, this.splitrow, true);
+    Path path_a =
+        fs.splitStoreFile(this.hri_a, familyName, sf, this.splitrow, false,
+          this.parent.getSplitPolicy());
+    Path path_b =
+        fs.splitStoreFile(this.hri_b, familyName, sf, this.splitrow, true,
+          this.parent.getSplitPolicy());
     return new Pair<Path,Path>(path_a, path_b);
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java

@@ -28,6 +28,7 @@ import static org.junit.Assert.fail;
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -1130,6 +1131,37 @@ public class TestSplitTransactionOnCluster {
       TESTING_UTIL.deleteTable(tableName);
     }
   }
+
+  @Test
+  public void testStoreFileReferenceCreationWhenSplitPolicySaysToSkipRangeCheck()
+      throws Exception {
+    final TableName tableName =
+        TableName.valueOf("testStoreFileReferenceCreationWhenSplitPolicySaysToSkipRangeCheck");
+    try {
+      HTableDescriptor htd = new HTableDescriptor(tableName);
+      htd.addFamily(new HColumnDescriptor("f"));
+      htd.setRegionSplitPolicyClassName(CustomSplitPolicy.class.getName());
+      admin.createTable(htd);
+      List<HRegion> regions = awaitTableRegions(tableName);
+      HRegion region = regions.get(0);
+      for(int i = 3;i<9;i++) {
+        Put p = new Put(Bytes.toBytes("row"+i));
+        p.add(Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("value"+i));
+        region.put(p);
+      }
+      region.flushcache();
+      Store store = region.getStore(Bytes.toBytes("f"));
+      Collection<StoreFile> storefiles = store.getStorefiles();
+      assertEquals(storefiles.size(), 1);
+      assertFalse(region.hasReferences());
+      Path referencePath = region.getRegionFileSystem().splitStoreFile(region.getRegionInfo(), "f",
+        storefiles.iterator().next(), Bytes.toBytes("row1"), false, region.getSplitPolicy());
+      assertNotNull(referencePath);
+    } finally {
+      TESTING_UTIL.deleteTable(tableName);
+    }
+  }
+
     public static class MockedCoordinatedStateManager extends ZkCoordinatedStateManager {
 
         public void initialize(Server server, HRegion region) {
@@ -1445,5 +1477,18 @@ public class TestSplitTransactionOnCluster {
     }
 
   }
+
+  static class CustomSplitPolicy extends RegionSplitPolicy {
+
+    @Override
+    protected boolean shouldSplit() {
+      return true;
+    }
+
+    @Override
+    public boolean skipStoreFileRangeCheck() {
+      return true;
+    }
+  }
 }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFile.java

@@ -940,7 +940,7 @@ public class TestStoreFile extends HBaseTestCase {
       final String family, final StoreFile sf, final byte[] splitKey, boolean isTopRef)
       throws IOException {
     FileSystem fs = regionFs.getFileSystem();
-    Path path = regionFs.splitStoreFile(hri, family, sf, splitKey, isTopRef);
+    Path path = regionFs.splitStoreFile(hri, family, sf, splitKey, isTopRef, null);
     if (null == path) {
       return null;
     }