HBASE-3961 Add Delete.setWriteToWAL functionality

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

CHANGES.txt

@@ -247,6 +247,7 @@ Release 0.91.0 - Unreleased
                (Alejandro Abdelnur)
    HBASE-3941  "hbase version" command line should print version info
                (Jolly Chen)
+   HBASE-3961  Add Delete.setWriteToWAL functionality (Bruno Dumon)
 
   TASKS
    HBASE-3559  Move report of split to master OFF the heartbeat channel

src/main/java/org/apache/hadoop/hbase/client/Delete.java

@@ -69,12 +69,13 @@ import java.util.TreeMap;
  * timestamp.  The constructor timestamp is not referenced.
  */
 public class Delete implements Writable, Row, Comparable<Row> {
-  private static final byte DELETE_VERSION = (byte)2;
+  private static final byte DELETE_VERSION = (byte)3;
 
   private byte [] row = null;
   // This ts is only used when doing a deleteRow.  Anything less,
   private long ts;
   private long lockId = -1L;
+  private boolean writeToWAL = true;
   private final Map<byte [], List<KeyValue>> familyMap =
     new TreeMap<byte [], List<KeyValue>>(Bytes.BYTES_COMPARATOR);
 
@@ -128,6 +129,7 @@ public class Delete implements Writable, Row, Comparable<Row> {
     this.ts = d.getTimeStamp();
     this.lockId = d.getLockId();
     this.familyMap.putAll(d.getFamilyMap());
+    this.writeToWAL = d.writeToWAL;
   }
 
   public int compareTo(final Row d) {
@@ -382,6 +384,9 @@ public class Delete implements Writable, Row, Comparable<Row> {
     this.row = Bytes.readByteArray(in);
     this.ts = in.readLong();
     this.lockId = in.readLong();
+    if (version > 2) {
+      this.writeToWAL = in.readBoolean();
+    }
     this.familyMap.clear();
     int numFamilies = in.readInt();
     for(int i=0;i<numFamilies;i++) {
@@ -413,6 +418,7 @@ public class Delete implements Writable, Row, Comparable<Row> {
     Bytes.writeByteArray(out, this.row);
     out.writeLong(this.ts);
     out.writeLong(this.lockId);
+    out.writeBoolean(this.writeToWAL);
     out.writeInt(familyMap.size());
     for(Map.Entry<byte [], List<KeyValue>> entry : familyMap.entrySet()) {
       Bytes.writeByteArray(out, entry.getKey());
@@ -460,4 +466,20 @@ public class Delete implements Writable, Row, Comparable<Row> {
     this.deleteColumn(parts[0], parts[1], HConstants.LATEST_TIMESTAMP);
     return this;
   }
+
+  /**
+   * @return true if edits should be applied to WAL, false if not
+   */
+  public boolean getWriteToWAL() {
+    return this.writeToWAL;
+  }
+
+  /**
+   * Set whether this Delete should be written to the WAL or not.
+   * Not writing the WAL means you may lose edits on server crash.
+   * @param write true if edits should be written to WAL, false if not
+   */
+  public void setWriteToWAL(boolean write) {
+    this.writeToWAL = write;
+  }
 }

src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -2068,7 +2068,7 @@ public class HRegionServer implements HRegionInterface, HBaseRPCErrorHandler,
       throws IOException {
     checkOpen();
     try {
-      boolean writeToWAL = true;
+      boolean writeToWAL = delete.getWriteToWAL();
       this.requestCount.incrementAndGet();
       HRegion region = getRegion(regionName);
       if (!region.getRegionInfo().isMetaTable()) {
@@ -2088,7 +2088,6 @@ public class HRegionServer implements HRegionInterface, HBaseRPCErrorHandler,
     checkOpen();
     HRegion region = null;
     try {
-      boolean writeToWAL = true;
       region = getRegion(regionName);
       if (!region.getRegionInfo().isMetaTable()) {
         this.cacheFlusher.reclaimMemStoreMemory();
@@ -2098,7 +2097,7 @@ public class HRegionServer implements HRegionInterface, HBaseRPCErrorHandler,
       for (Delete delete : deletes) {
         this.requestCount.incrementAndGet();
         locks[i] = getLockFromId(delete.getLockId());
-        region.delete(delete, locks[i], writeToWAL);
+        region.delete(delete, locks[i], delete.getWriteToWAL());
         i++;
       }
     } catch (WrongRegionException ex) {