HBASE-15323 Hbase Rest CheckAndDeleteAPi should be able to delete more cells (Ajith)

hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RowResource.java

@@ -553,9 +553,12 @@ public class RowResource extends ResourceBase {
           .build();
       }
 
+      List<CellModel> cellModels = rowModel.getCells();
+      int cellModelCount = cellModels.size();
+
       delete = new Delete(key);
       boolean retValue;
-      CellModel valueToDeleteCell = rowModel.getCells().get(0);
+      CellModel valueToDeleteCell = rowModel.getCells().get(cellModelCount -1);
       byte[] valueToDeleteColumn = valueToDeleteCell.getColumn();
       if (valueToDeleteColumn == null) {
         try {
@@ -567,25 +570,62 @@ public class RowResource extends ResourceBase {
             .build();
         }
       }
-      byte[][] parts = KeyValue.parseColumn(valueToDeleteColumn);
+
+      byte[][] parts ;
+      // Copy all the cells to the Delete request if extra cells are sent
+      if(cellModelCount > 1) {
+        for (int i = 0, n = cellModelCount - 1; i < n; i++) {
+          CellModel cell = cellModels.get(i);
+          byte[] col = cell.getColumn();
+
+          if (col == null) {
+            servlet.getMetrics().incrementFailedPutRequests(1);
+            return Response.status(Response.Status.BAD_REQUEST)
+                    .type(MIMETYPE_TEXT).entity("Bad request: Column found to be null." + CRLF)
+                    .build();
+          }
+
+          parts = KeyValue.parseColumn(col);
+
+          if (parts.length == 1) {
+            // Only Column Family is specified
+            delete.addFamily(parts[0], cell.getTimestamp());
+          } else if (parts.length == 2) {
+            delete.addColumn(parts[0], parts[1], cell.getTimestamp());
+          } else {
+            servlet.getMetrics().incrementFailedDeleteRequests(1);
+            return Response.status(Response.Status.BAD_REQUEST)
+                    .type(MIMETYPE_TEXT)
+                    .entity("Bad request: Column to delete incorrectly specified." + CRLF)
+                    .build();
+          }
+        }
+      }
+
+      parts = KeyValue.parseColumn(valueToDeleteColumn);
       if (parts.length == 2) {
         if (parts[1].length != 0) {
-          delete.addColumns(parts[0], parts[1]);
+          // To support backcompat of deleting a cell
+          // if that is the only cell passed to the rest api
+          if(cellModelCount == 1) {
+            delete.addColumns(parts[0], parts[1]);
+          }
           retValue = table.checkAndDelete(key, parts[0], parts[1],
             valueToDeleteCell.getValue(), delete);
         } else {
           // The case of empty qualifier.
-          delete.addColumns(parts[0], Bytes.toBytes(StringUtils.EMPTY));
+          if(cellModelCount == 1) {
+            delete.addColumns(parts[0], Bytes.toBytes(StringUtils.EMPTY));
+          }
           retValue = table.checkAndDelete(key, parts[0], Bytes.toBytes(StringUtils.EMPTY),
             valueToDeleteCell.getValue(), delete);
         }
       } else {
         servlet.getMetrics().incrementFailedDeleteRequests(1);
         return Response.status(Response.Status.BAD_REQUEST)
-          .type(MIMETYPE_TEXT).entity("Bad request: Column incorrectly specified." + CRLF)
+          .type(MIMETYPE_TEXT).entity("Bad request: Column to check incorrectly specified." + CRLF)
           .build();
       }
-      delete.addColumns(parts[0], parts[1]);
 
       if (LOG.isDebugEnabled()) {
         LOG.debug("CHECK-AND-DELETE " + delete.toString() + ", returns "

hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/client/RemoteHTable.java

@@ -722,6 +722,7 @@ public class RemoteHTable implements Table {
   public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier,
       byte[] value, Delete delete) throws IOException {
     Put put = new Put(row);
+    put.setFamilyCellMap(delete.getFamilyCellMap());
     // column to check-the-value
     put.add(new KeyValue(row, family, qualifier, value));
     CellSetModel model = buildModelFromPut(put);

hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/RowResourceBase.java

@@ -320,9 +320,16 @@ public class RowResourceBase {
   }
 
   protected static Response checkAndDeleteXML(String url, String table,
-      String row, String column, String valueToCheck)
+      String row, String column, String valueToCheck, HashMap<String,String> cellsToDelete)
         throws IOException, JAXBException {
     RowModel rowModel = new RowModel(row);
+
+    if(cellsToDelete != null) {
+      for (Map.Entry<String,String> entry :cellsToDelete.entrySet()) {
+        rowModel.addCell(new CellModel(Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue())));
+      }
+    }
+    // Add this at the end
     rowModel.addCell(new CellModel(Bytes.toBytes(column),
       Bytes.toBytes(valueToCheck)));
     CellSetModel cellSetModel = new CellSetModel();
@@ -337,30 +344,46 @@ public class RowResourceBase {
 
   protected static Response checkAndDeleteXML(String table, String row,
       String column, String valueToCheck) throws IOException, JAXBException {
+    return checkAndDeleteXML(table, row, column, valueToCheck, null);
+  }
+  protected static Response checkAndDeleteXML(String table, String row,
+      String column, String valueToCheck, HashMap<String,String> cellsToDelete) throws IOException, JAXBException {
     StringBuilder path = new StringBuilder();
     path.append('/');
     path.append(table);
     path.append('/');
     path.append(row);
     path.append("?check=delete");
-    return checkAndDeleteXML(path.toString(), table, row, column, valueToCheck);
+    return checkAndDeleteXML(path.toString(), table, row, column, valueToCheck, cellsToDelete);
   }
 
   protected static Response checkAndDeleteJson(String table, String row,
       String column, String valueToCheck) throws IOException, JAXBException {
+    return checkAndDeleteJson(table, row, column, valueToCheck, null);
+  }
+
+  protected static Response checkAndDeleteJson(String table, String row,
+      String column, String valueToCheck, HashMap<String,String> cellsToDelete) throws IOException, JAXBException {
     StringBuilder path = new StringBuilder();
     path.append('/');
     path.append(table);
     path.append('/');
     path.append(row);
     path.append("?check=delete");
-    return checkAndDeleteJson(path.toString(), table, row, column, valueToCheck);
+    return checkAndDeleteJson(path.toString(), table, row, column, valueToCheck, cellsToDelete);
   }
 
   protected static Response checkAndDeleteJson(String url, String table,
-      String row, String column, String valueToCheck)
+      String row, String column, String valueToCheck, HashMap<String,String> cellsToDelete)
         throws IOException, JAXBException {
     RowModel rowModel = new RowModel(row);
+
+    if(cellsToDelete != null) {
+      for (Map.Entry<String,String> entry :cellsToDelete.entrySet()) {
+        rowModel.addCell(new CellModel(Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue())));
+      }
+    }
+    // Add this at the end
     rowModel.addCell(new CellModel(Bytes.toBytes(column),
       Bytes.toBytes(valueToCheck)));
     CellSetModel cellSetModel = new CellSetModel();
@@ -374,19 +397,31 @@ public class RowResourceBase {
 
   protected static Response checkAndDeletePB(String table, String row,
       String column, String value) throws IOException {
+
+    return checkAndDeletePB(table, row, column, value, null);
+  }
+
+  protected static Response checkAndDeletePB(String table, String row,
+      String column, String value, HashMap<String,String> cellsToDelete) throws IOException {
     StringBuilder path = new StringBuilder();
     path.append('/');
     path.append(table);
     path.append('/');
     path.append(row);
     path.append("?check=delete");
-    return checkAndDeleteValuePB(path.toString(), table, row, column, value);
+    return checkAndDeleteValuePB(path.toString(), table, row, column, value, cellsToDelete);
   }
-
   protected static Response checkAndDeleteValuePB(String url, String table,
-      String row, String column, String valueToCheck)
+      String row, String column, String valueToCheck, HashMap<String,String> cellsToDelete)
       throws IOException {
     RowModel rowModel = new RowModel(row);
+
+    if(cellsToDelete != null) {
+      for (Map.Entry<String,String> entry :cellsToDelete.entrySet()) {
+        rowModel.addCell(new CellModel(Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue())));
+      }
+    }
+    // Add this at the end
     rowModel.addCell(new CellModel(Bytes.toBytes(column), Bytes
         .toBytes(valueToCheck)));
     CellSetModel cellSetModel = new CellSetModel();

hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/TestGetAndPutResource.java

@@ -197,6 +197,59 @@ public class TestGetAndPutResource extends RowResourceBase {
     assertEquals(response.getCode(), 200);
   }
 
+  @Test
+  public void testMultipleCellCheckDeletePB() throws IOException, JAXBException {
+    Response response = getValuePB(TABLE, ROW_1, COLUMN_1);
+    assertEquals(response.getCode(), 404);
+
+    // Add 3 Columns to setup the test
+    response = putValuePB(TABLE, ROW_1, COLUMN_1, VALUE_1);
+    assertEquals(response.getCode(), 200);
+    checkValuePB(TABLE, ROW_1, COLUMN_1, VALUE_1);
+
+    response = putValuePB(TABLE, ROW_1, COLUMN_2, VALUE_2);
+    assertEquals(response.getCode(), 200);
+    checkValuePB(TABLE, ROW_1, COLUMN_2, VALUE_2);
+
+    response = putValuePB(TABLE, ROW_1, COLUMN_3, VALUE_3);
+    assertEquals(response.getCode(), 200);
+    checkValuePB(TABLE, ROW_1, COLUMN_3, VALUE_3);
+
+    // Deletes the following columns based on Column1 check
+    HashMap<String,String> cellsToDelete = new HashMap<String, String>();
+    cellsToDelete.put(COLUMN_2,VALUE_2); // Value does not matter
+    cellsToDelete.put(COLUMN_3,VALUE_3); // Value does not matter
+
+    // On Success update both the cells
+    response = checkAndDeletePB(TABLE, ROW_1, COLUMN_1, VALUE_1, cellsToDelete);
+    assertEquals(response.getCode(), 200);
+
+    checkValuePB(TABLE, ROW_1, COLUMN_1, VALUE_1);
+
+    response = getValuePB(TABLE, ROW_1, COLUMN_2);
+    assertEquals(response.getCode(), 404);
+
+    response = getValuePB(TABLE, ROW_1, COLUMN_3);
+    assertEquals(response.getCode(), 404);
+
+    response = putValuePB(TABLE, ROW_1, COLUMN_2, VALUE_2);
+    assertEquals(response.getCode(), 200);
+    checkValuePB(TABLE, ROW_1, COLUMN_2, VALUE_2);
+
+    response = putValuePB(TABLE, ROW_1, COLUMN_3, VALUE_3);
+    assertEquals(response.getCode(), 200);
+    checkValuePB(TABLE, ROW_1, COLUMN_3, VALUE_3);
+
+    // On Failure, we dont update any cells
+    response = checkAndDeletePB(TABLE, ROW_1, COLUMN_1, VALUE_3, cellsToDelete);
+    assertEquals(response.getCode(), 304);
+    checkValuePB(TABLE, ROW_1, COLUMN_1, VALUE_1);
+    checkValuePB(TABLE, ROW_1, COLUMN_2, VALUE_2);
+    checkValuePB(TABLE, ROW_1, COLUMN_3, VALUE_3);
+
+    response = deleteRow(TABLE, ROW_1);
+    assertEquals(response.getCode(), 200);
+  }
   @Test
   public void testSingleCellGetPutBinary() throws IOException {
     final String path = "/" + TABLE + "/" + ROW_3 + "/" + COLUMN_1;