HADOOP-1550 No means of deleting a'row'

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

src/contrib/hbase/CHANGES.txt

@@ -12,6 +12,7 @@ Trunk (unreleased changes)
     HADOOP-2068 RESTful interface (Bryan Duxbury via Stack)
     HADOOP-2316 Run REST servlet outside of master
                 (Bryan Duxbury & Stack)
+    HADOOP-1550 No means of deleting a'row' (Bryan Duxbuery via Stack)
 
   OPTIMIZATIONS
 

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java

@@ -1060,7 +1060,7 @@ public class HRegion implements HConstants {
    */
   private List<HStoreKey> getKeys(final HStoreKey origin, final int versions)
     throws IOException {
-    
+
     List<HStoreKey> keys = null;
     Text colFamily = HStoreKey.extractFamily(origin.getColumn());
     HStore targetStore = stores.get(colFamily);
@@ -1071,7 +1071,7 @@ public class HRegion implements HConstants {
     }
     return keys;
   }
-
+  
   /**
    * Return an iterator that scans over the HRegion, returning the indicated 
    * columns for only the rows that match the data filter.  This Iterator must
@@ -1248,6 +1248,33 @@ public class HRegion implements HConstants {
       releaseRowLock(row);
     }
   }
+
+  /**
+   * Delete all cells of the same age as the passed timestamp or older.
+   * @param row
+   * @param ts Delete all entries that have this timestamp or older
+   * @throws IOException
+   */
+  public void deleteAll(final Text row, final long ts)
+    throws IOException {
+    
+    obtainRowLock(row);    
+    
+    try {
+      for(Map.Entry<Text, HStore> store : stores.entrySet()){
+        List<HStoreKey> keys = store.getValue().getKeys(new HStoreKey(row, ts), ALL_VERSIONS);
+
+        TreeMap<HStoreKey, byte []> edits = new TreeMap<HStoreKey, byte []>();
+        for (HStoreKey key: keys) {
+          edits.put(key, HLogEdit.deleteBytes.get());
+        }
+        update(edits);
+      }
+    } finally {
+      releaseRowLock(row);
+    }
+  }
+
   
   /**
    * Delete one or many cells.

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInterface.java

@@ -134,6 +134,18 @@ public interface HRegionInterface extends VersionedProtocol {
    */
   public void deleteAll(Text regionName, Text row, Text column, long timestamp)
   throws IOException;
+
+  /**
+   * Delete all cells that match the passed row and whose
+   * timestamp is equal-to or older than the passed timestamp.
+   *
+   * @param regionName region name
+   * @param row row key
+   * @param timestamp Delete all entries that have this timestamp or older
+   * @throws IOException
+   */
+  public void deleteAll(Text regionName, Text row, long timestamp)
+  throws IOException;
   
   //
   // remote scanner interface

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java

@@ -1516,6 +1516,14 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
     region.deleteAll(row, column, timestamp);
   }
 
+  /** {@inheritDoc} */
+  public void deleteAll(final Text regionName, final Text row,
+      final long timestamp) 
+  throws IOException {
+    HRegion region = getRegion(regionName);
+    region.deleteAll(row, timestamp);
+  }
+
   /**
    * @return Info on this server.
    */

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java

@@ -268,7 +268,7 @@ class HStore implements HConstants {
                 (versions - results.size())));
         }
         return results;
-
+        
       } finally {
         this.lock.readLock().unlock();
       }
@@ -289,9 +289,28 @@ class HStore implements HConstants {
       SortedMap<HStoreKey, byte []> tailMap = map.tailMap(origin);
       for (Map.Entry<HStoreKey, byte []> es: tailMap.entrySet()) {
         HStoreKey key = es.getKey();
-        if (!key.matchesRowCol(origin)) {
-          break;
+    
+        // if there's no column name, then compare rows and timestamps
+        if (origin.getColumn().toString().equals("")) {
+          // if the current and origin row don't match, then we can jump
+          // out of the loop entirely.
+          if (!key.getRow().equals(origin.getRow())) {
+            break;
+          }
+          // if the rows match but the timestamp is newer, skip it so we can
+          // get to the ones we actually want.
+          if (key.getTimestamp() > origin.getTimestamp()) {
+            continue;
+          }
+        }
+        else{ // compare rows and columns
+          // if the key doesn't match the row and column, then we're done, since 
+          // all the cells are ordered.
+          if (!key.matchesRowCol(origin)) {
+            break;
+          }
         }
+
         if (!HLogEdit.isDeleted(es.getValue())) {
           result.add(key);
           if (versions != HConstants.ALL_VERSIONS && result.size() >= versions) {
@@ -1429,6 +1448,7 @@ class HStore implements HConstants {
    */
   void getFull(HStoreKey key, TreeMap<Text, byte []> results)
     throws IOException {
+    Map<Text, List<Long>> deletes = new HashMap<Text, List<Long>>();
     
     this.lock.readLock().lock();
     memcache.getFull(key, results);
@@ -1447,7 +1467,7 @@ class HStore implements HConstants {
             Text readcol = readkey.getColumn();
             if (results.get(readcol) == null
                 && key.matchesWithoutColumn(readkey)) {
-              if(HLogEdit.isDeleted(readval.get())) {
+              if(isDeleted(readkey, readval.get(), true, deletes)) {
                 break;
               }
               results.put(new Text(readcol), readval.get());
@@ -1598,6 +1618,8 @@ class HStore implements HConstants {
         MapFile.Reader map = maparray[i];
         synchronized(map) {
           map.reset();
+          
+          // do the priming read
           ImmutableBytesWritable readval = new ImmutableBytesWritable();
           HStoreKey readkey = (HStoreKey)map.getClosest(origin, readval);
           if (readkey == null) {
@@ -1607,33 +1629,82 @@ class HStore implements HConstants {
             // BEFORE.
             continue;
           }
-          if (!readkey.matchesRowCol(origin)) {
-            continue;
-          }
-          if (!isDeleted(readkey, readval.get(), false, null) &&
-              !keys.contains(readkey)) {
-            keys.add(new HStoreKey(readkey));
-          }
-          for (readval = new ImmutableBytesWritable();
-              map.next(readkey, readval) &&
-              readkey.matchesRowCol(origin);
-              readval = new ImmutableBytesWritable()) {
-            if (!isDeleted(readkey, readval.get(), false, null) &&
-                !keys.contains(readkey)) {
-              keys.add(new HStoreKey(readkey));
-              if (versions != ALL_VERSIONS && keys.size() >= versions) {
-                break;
+          
+          do{
+            // if the row matches, we might want this one.
+            if(rowMatches(origin, readkey)){
+              // if the cell matches, then we definitely want this key.
+              if (cellMatches(origin, readkey)) {
+                // store the key if it isn't deleted or superceeded by what's
+                // in the memcache
+                if (!isDeleted(readkey, readval.get(), false, null) &&
+                    !keys.contains(readkey)) {
+                  keys.add(new HStoreKey(readkey));
+
+                  // if we've collected enough versions, then exit the loop.
+                  if (versions != ALL_VERSIONS && keys.size() >= versions) {
+                    break;
+                  }
+                }
+              } else {
+                // the cell doesn't match, but there might be more with different
+                // timestamps, so move to the next key
+                continue;
               }
+            } else{
+              // the row doesn't match, so we've gone too far.
+              break;
             }
-          }
+          }while(map.next(readkey, readval)); // advance to the next key
         }
       }
+      
       return keys;
     } finally {
       this.lock.readLock().unlock();
     }
   }
   
+  /**
+   * Test that the <i>target</i> matches the <i>origin</i>. If the 
+   * <i>origin</i> has an empty column, then it's assumed to mean any column 
+   * matches and only match on row and timestamp. Otherwise, it compares the
+   * keys with HStoreKey.matchesRowCol().
+   * @param origin The key we're testing against
+   * @param target The key we're testing
+   */
+  private boolean cellMatches(HStoreKey origin, HStoreKey target){
+    // if the origin's column is empty, then we're matching any column
+    if (origin.getColumn().equals(new Text())){
+      // if the row matches, then...
+      if (target.getRow().equals(origin.getRow())) {
+        // check the timestamp
+        return target.getTimestamp() <= origin.getTimestamp();
+      } else {
+        return false;
+      }
+    } else { // otherwise, we want to match on row and column
+      return target.matchesRowCol(origin);
+    }
+  }
+  
+  /**
+   * Test that the <i>target</i> matches the <i>origin</i>. If the <i>origin</i>
+   * has an empty column, then it just tests row equivalence. Otherwise, it uses
+   * HStoreKey.matchesRowCol().
+   * @param origin Key we're testing against
+   * @param target Key we're testing
+   */
+  private boolean rowMatches(HStoreKey origin, HStoreKey target){
+    // if the origin's column is empty, then we're matching any column
+    if (origin.getColumn().equals(new Text())){
+      // if the row matches, then...
+      return target.getRow().equals(origin.getRow());
+    } else { // otherwise, we want to match on row and column
+      return target.matchesRowCol(origin);
+    }
+  }
+  
   /*
    * Data structure to hold result of a look at store file sizes.
    */

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HTable.java

@@ -742,6 +742,53 @@ public class HTable implements HConstants {
     }
   }
   
+  /**
+   * Completely delete the row's cells of the same timestamp or older.
+   *
+   * @param row Key of the row you want to completely delete.
+   * @param ts Timestamp of cells to delete
+   */
+  public void deleteAll(final Text row, long ts)
+  throws IOException {
+    checkClosed();
+    for(int tries = 0; tries < numRetries; tries++) {
+      HRegionLocation r = getRegionLocation(row);
+      HRegionInterface server =
+        connection.getHRegionConnection(r.getServerAddress());
+      try {
+        server.deleteAll(r.getRegionInfo().getRegionName(), row, ts);
+        break;
+
+      } catch (IOException e) {
+        if (e instanceof RemoteException) {
+          e = RemoteExceptionHandler.decodeRemoteException((RemoteException) e);
+        }
+        if (tries == numRetries - 1) {
+          throw e;
+        }
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("reloading table servers because: " + e.getMessage());
+        }
+        tableServers = connection.reloadTableServers(tableName);
+      }
+      try {
+        Thread.sleep(this.pause);
+      } catch (InterruptedException x) {
+        // continue
+      }
+    }
+  }
+      
+  /**
+   * Completely delete the row's cells.
+   *
+   * @param row Key of the row you want to completely delete.
+   */
+  public void deleteAll(final Text row)
+  throws IOException {
+    deleteAll(row, HConstants.LATEST_TIMESTAMP);
+  }
+  
   /** 
    * Abort a row mutation.
    *