From 4e446e768912ca08b30d74b3c570ddb336b837a6 Mon Sep 17 00:00:00 2001
From: Michael Stack <stack@apache.org>
Date: Fri, 25 Sep 2009 23:15:10 +0000
Subject: [PATCH] HBASE-1869 IndexedTable delete fails when used in conjunction
 with RowLock()

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@819060 13f79535-47bb-0310-9956-ffa450edef68
---
 .../tableindexed/IndexedRegion.java           |  4 +--
 .../client/tableindexed/TestIndexedTable.java | 30 +++++++++++++++++++
 2 files changed, 32 insertions(+), 2 deletions(-)

diff --git a/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java b/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java
index 4d9563bc567..fcd12873ccc 100644
--- a/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java
+++ b/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java
@@ -244,7 +244,7 @@ class IndexedRegion extends TransactionalRegion {
         }
       }
       
-      Result oldRow = super.get(get, null);
+      Result oldRow = super.get(get, lockid);
       SortedMap<byte[], byte[]> oldColumnValues = convertToValueMap(oldRow);
       
       
@@ -255,7 +255,7 @@ class IndexedRegion extends TransactionalRegion {
       // Handle if there is still a version visible.
       if (delete.getTimeStamp() != HConstants.LATEST_TIMESTAMP) {
         get.setTimeRange(1, delete.getTimeStamp());
-        oldRow = super.get(get, null);
+        oldRow = super.get(get, lockid);
         SortedMap<byte[], byte[]> currentColumnValues = convertToValueMap(oldRow);
         
         for (IndexSpecification indexSpec : getIndexes()) {
diff --git a/src/contrib/transactional/src/test/org/apache/hadoop/hbase/client/tableindexed/TestIndexedTable.java b/src/contrib/transactional/src/test/org/apache/hadoop/hbase/client/tableindexed/TestIndexedTable.java
index 0cd440acd14..d7c1b15c020 100644
--- a/src/contrib/transactional/src/test/org/apache/hadoop/hbase/client/tableindexed/TestIndexedTable.java
+++ b/src/contrib/transactional/src/test/org/apache/hadoop/hbase/client/tableindexed/TestIndexedTable.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
 import org.apache.hadoop.hbase.client.RowLock; 
+import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.regionserver.tableindexed.IndexedRegionServer;
 import org.apache.hadoop.hbase.util.Bytes;
 
@@ -144,6 +145,24 @@ public class TestIndexedTable extends HBaseClusterTestCase {
                                     Bytes.toString(persistedRowValue));
   }
 
+  private void assertRowDeleted(int numRowsExpected)
+      throws IndexNotFoundException, IOException {
+    // Check the size of the primary table
+    ResultScanner scanner = table.getScanner(new Scan());
+    int numRows = 0;
+    for (Result rowResult : scanner) {
+      byte[] colA = rowResult.getValue(FAMILY, QUAL_A);
+      LOG.info("primary scan : row [" + Bytes.toString(rowResult.getRow())
+          + "] value [" + Bytes.toString(colA) + "]");
+      numRows++;
+    }
+    scanner.close();
+    Assert.assertEquals(numRowsExpected, numRows);
+
+    // Check the size of the index tables
+    assertRowsInOrder(numRowsExpected); 
+  }
+
   private void updateRow(int row, int newValue) throws IOException {
       Put update = new Put(PerformanceEvaluation.format(row));
       byte[] valueA = PerformanceEvaluation.format(newValue);
@@ -220,4 +239,15 @@ public class TestIndexedTable extends HBaseClusterTestCase {
     updateLockedRowNoAutoFlush(row, value);
     assertRowUpdated(row, value);
   } 
+
+  public void testLockedRowDelete() throws IOException {
+    writeInitalRows();
+    // Delete the first row;
+    byte[] row = PerformanceEvaluation.format(0);
+    RowLock lock = table.lockRow(row);
+    table.delete(new Delete(row, HConstants.LATEST_TIMESTAMP, lock));
+    table.unlockRow(lock);    
+
+    assertRowDeleted(NUM_ROWS - 1);  
+  }
 }