From 0e32567cdd424482286d0585dc9bc9b726e501bb Mon Sep 17 00:00:00 2001
From: Zhihong Yu <tedyu@apache.org>
Date: Wed, 2 Nov 2011 16:28:09 +0000
Subject: [PATCH] HBASE-4716  Improve locking for single column family bulk
 load

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1196674 13f79535-47bb-0310-9956-ffa450edef68
---
 CHANGES.txt                                   |  1 +
 .../hadoop/hbase/regionserver/HRegion.java    | 43 ++++++++++++++++---
 2 files changed, 39 insertions(+), 5 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index debe64bbb8d..bc8c694a9f9 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -442,6 +442,7 @@ Release 0.92.0 - Unreleased
    HBASE-3515  [replication] ReplicationSource can miss a log after RS comes out of GC
    HBASE-4713  Raise debug level to warn on ExecutionException in
                HConnectionManager$HConnectionImplementation (Lucian George Iordache)
+   HBASE-4716  Improve locking for single column family bulk load
 
   TESTS
    HBASE-4450  test for number of blocks read: to serve as baseline for expected
diff --git a/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index 2695c10d790..95698fbb03f 100644
--- a/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -2781,6 +2781,34 @@ public class HRegion implements HeapSize { // , Writable{
     }
     return lid;
   }
+  
+  enum COLUMN_FAMILY_TYPES {
+    NONE,       // there is no column family
+    SINGLE,
+    MULTIPLE     // there are more than one column family
+  }
+  
+  /**
+   * Determines the type of column family based on list of family/hfilePath pairs
+   *
+   * @param familyPaths List of Pair<byte[] column family, String hfilePath>
+   */
+  public static COLUMN_FAMILY_TYPES getColumnFamilyType(
+      List<Pair<byte[], String>> familyPaths) {
+    if (familyPaths == null) return COLUMN_FAMILY_TYPES.NONE;
+    COLUMN_FAMILY_TYPES familyType = COLUMN_FAMILY_TYPES.SINGLE;
+    byte[] family = null;
+    for (Pair<byte[], String> pair : familyPaths) {
+      byte[] fam = pair.getFirst();
+      if (family == null) {
+        family = fam;
+      } else if (!Bytes.equals(family, fam)) {
+        familyType = COLUMN_FAMILY_TYPES.MULTIPLE;
+        break;
+      }
+    }
+    return familyType;
+  }
 
   /**
    * Attempts to atomically load a group of hfiles.  This is critical for loading
@@ -2790,7 +2818,8 @@ public class HRegion implements HeapSize { // , Writable{
    */
   public void bulkLoadHFiles(List<Pair<byte[], String>> familyPaths)
   throws IOException {
-    startBulkRegionOperation();
+    // we need writeLock for multi-family bulk load
+    startBulkRegionOperation(getColumnFamilyType(familyPaths) == COLUMN_FAMILY_TYPES.MULTIPLE);
     this.writeRequestsCount.increment();
     List<IOException> ioes = new ArrayList<IOException>();
     List<Pair<byte[], String>> failures = new ArrayList<Pair<byte[], String>>();
@@ -4453,14 +4482,17 @@ public class HRegion implements HeapSize { // , Writable{
    * Acquires a writelock and checks if the region is closing or closed.
    * @throws NotServingRegionException when the region is closing or closed
    */
-  private void startBulkRegionOperation() throws NotServingRegionException {
+  private void startBulkRegionOperation(boolean writeLockNeeded)
+  throws NotServingRegionException {
     if (this.closing.get()) {
       throw new NotServingRegionException(regionInfo.getRegionNameAsString() +
           " is closing");
     }
-    lock.writeLock().lock();
+    if (writeLockNeeded) lock.writeLock().lock();
+    else lock.readLock().lock();
     if (this.closed.get()) {
-      lock.writeLock().unlock();
+      if (writeLockNeeded) lock.writeLock().unlock();
+      else lock.readLock().unlock();
       throw new NotServingRegionException(regionInfo.getRegionNameAsString() +
           " is closed");
     }
@@ -4471,7 +4503,8 @@ public class HRegion implements HeapSize { // , Writable{
    * to the try block of #startRegionOperation
    */
   private void closeBulkRegionOperation(){
-    lock.writeLock().unlock();
+    if (lock.writeLock().isHeldByCurrentThread()) lock.writeLock().unlock();
+    else lock.readLock().unlock();
   }
 
   /**