Index: src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionServerCoprocessorExceptionWithAbort.java =================================================================== --- src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionServerCoprocessorExceptionWithAbort.java (revision 1177371) +++ src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionServerCoprocessorExceptionWithAbort.java (working copy) @@ -32,6 +32,7 @@ import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.JVMClusterUtil; +import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.zookeeper.ZooKeeperNodeTracker; import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; import org.junit.AfterClass; @@ -108,7 +109,7 @@ public static class BuggyRegionObserver extends SimpleRegionObserver { @Override public void prePut(final ObserverContext c, - final Map> familyMap, + final Put put, final WALEdit edit, final boolean writeToWAL) { String tableName = c.getEnvironment().getRegion().getRegionInfo().getTableNameAsString(); Index: src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java =================================================================== --- src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java (revision 1177371) +++ src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java (working copy) @@ -33,6 +33,8 @@ import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Increment; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; @@ -41,6 +43,7 @@ import org.apache.hadoop.hbase.regionserver.RegionScanner; import org.apache.hadoop.hbase.regionserver.Store; import org.apache.hadoop.hbase.regionserver.StoreFile; +import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.util.Bytes; /** @@ -255,8 +258,10 @@ } @Override - public void prePut(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void prePut(final ObserverContext c, + final Put put, final WALEdit edit, + final boolean writeToWAL) throws IOException { + Map> familyMap = put.getFamilyMap(); RegionCoprocessorEnvironment e = c.getEnvironment(); assertNotNull(e); assertNotNull(e.getRegion()); @@ -283,8 +288,10 @@ } @Override - public void postPut(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void postPut(final ObserverContext c, + final Put put, final WALEdit edit, + final boolean writeToWAL) throws IOException { + Map> familyMap = put.getFamilyMap(); RegionCoprocessorEnvironment e = c.getEnvironment(); assertNotNull(e); assertNotNull(e.getRegion()); @@ -311,8 +318,10 @@ } @Override - public void preDelete(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void preDelete(final ObserverContext c, + final Delete delete, final WALEdit edit, + final boolean writeToWAL) throws IOException { + Map> familyMap = delete.getFamilyMap(); RegionCoprocessorEnvironment e = c.getEnvironment(); assertNotNull(e); assertNotNull(e.getRegion()); @@ -323,8 +332,10 @@ } @Override - public void postDelete(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void postDelete(final ObserverContext c, + final Delete delete, final WALEdit edit, + final boolean writeToWAL) throws IOException { + Map> familyMap = delete.getFamilyMap(); RegionCoprocessorEnvironment e = c.getEnvironment(); assertNotNull(e); assertNotNull(e.getRegion()); Index: src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverBypass.java =================================================================== --- src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverBypass.java (revision 1177371) +++ src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverBypass.java (working copy) @@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.coprocessor.CoprocessorHost; import org.apache.hadoop.hbase.coprocessor.ObserverContext; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; +import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.util.Bytes; import org.junit.AfterClass; import org.junit.BeforeClass; @@ -181,8 +182,9 @@ public static class TestCoprocessor extends BaseRegionObserver { @Override public void prePut(final ObserverContext e, - final Map> familyMap, final boolean writeToWAL) + final Put put, final WALEdit edit, final boolean writeToWAL) throws IOException { + Map> familyMap = put.getFamilyMap(); if (familyMap.containsKey(test)) { e.bypass(); } Index: src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverStacking.java =================================================================== --- src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverStacking.java (revision 1177371) +++ src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverStacking.java (working copy) @@ -34,6 +34,7 @@ import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost; +import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.regionserver.HRegion; import org.apache.hadoop.hbase.util.Bytes; @@ -46,7 +47,8 @@ long id; @Override public void postPut(final ObserverContext c, - final Map> familyMap, final boolean writeToWAL) + final Put put, final WALEdit edit, + final boolean writeToWAL) throws IOException { id = System.currentTimeMillis(); try { @@ -60,7 +62,8 @@ long id; @Override public void postPut(final ObserverContext c, - final Map> familyMap, final boolean writeToWAL) + final Put put, final WALEdit edit, + final boolean writeToWAL) throws IOException { id = System.currentTimeMillis(); try { @@ -75,7 +78,8 @@ @Override public void postPut(final ObserverContext c, - final Map> familyMap, final boolean writeToWAL) + final Put put, final WALEdit edit, + final boolean writeToWAL) throws IOException { id = System.currentTimeMillis(); try { Index: src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionServerCoprocessorExceptionWithRemove.java =================================================================== --- src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionServerCoprocessorExceptionWithRemove.java (revision 1177371) +++ src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionServerCoprocessorExceptionWithRemove.java (working copy) @@ -33,6 +33,7 @@ import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.JVMClusterUtil; +import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.zookeeper.ZooKeeperNodeTracker; import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; import org.junit.AfterClass; @@ -52,7 +53,7 @@ public static class BuggyRegionObserver extends SimpleRegionObserver { @Override public void prePut(final ObserverContext c, - final Map> familyMap, + final Put put, final WALEdit edit, final boolean writeToWAL) { String tableName = c.getEnvironment().getRegion().getRegionInfo().getTableNameAsString(); Index: src/test/java/org/apache/hadoop/hbase/replication/TestMasterReplication.java =================================================================== --- src/test/java/org/apache/hadoop/hbase/replication/TestMasterReplication.java (revision 1177371) +++ src/test/java/org/apache/hadoop/hbase/replication/TestMasterReplication.java (working copy) @@ -49,6 +49,7 @@ import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.JVMClusterUtil; +import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; import org.junit.After; @@ -304,13 +305,15 @@ @Override public void prePut(final ObserverContext e, - final Map> familyMap, final boolean writeToWAL) + final Put put, final WALEdit edit, + final boolean writeToWAL) throws IOException { nCount++; } @Override public void postDelete(final ObserverContext c, - final Map> familyMap, final boolean writeToWAL) + final Delete delete, final WALEdit edit, + final boolean writeToWAL) throws IOException { nDelete++; } Index: src/main/java/org/apache/hadoop/hbase/client/Mutation.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/client/Mutation.java (revision 1177371) +++ src/main/java/org/apache/hadoop/hbase/client/Mutation.java (working copy) @@ -130,6 +130,14 @@ } /** + * Method for setting the put's familyMap + * @return familyMap + */ + public void setFamilyMap(Map> map) { + this.familyMap = map; + } + + /** * Method to check if the familyMap is empty * @return true if empty, false otherwise */ Index: src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java (revision 1177371) +++ src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java (working copy) @@ -640,12 +640,13 @@ } /** - * @param familyMap map of family to edits for the given family. + * @param put The Put object + * @param edit The WALEdit object. * @param writeToWAL true if the change should be written to the WAL * @return true if default processing should be bypassed * @exception IOException Exception */ - public boolean prePut(final Map> familyMap, + public boolean prePut(Put put, WALEdit edit, final boolean writeToWAL) throws IOException { boolean bypass = false; ObserverContext ctx = null; @@ -653,7 +654,7 @@ if (env.getInstance() instanceof RegionObserver) { ctx = ObserverContext.createAndPrepare(env, ctx); try { - ((RegionObserver)env.getInstance()).prePut(ctx, familyMap, writeToWAL); + ((RegionObserver)env.getInstance()).prePut(ctx, put, edit, writeToWAL); } catch (Throwable e) { handleCoprocessorThrowable(env, e); } @@ -667,18 +668,19 @@ } /** - * @param familyMap map of family to edits for the given family. + * @param put The Put object + * @param edit The WALEdit object. * @param writeToWAL true if the change should be written to the WAL * @exception IOException Exception */ - public void postPut(final Map> familyMap, + public void postPut(Put put, WALEdit edit, final boolean writeToWAL) throws IOException { ObserverContext ctx = null; for (RegionEnvironment env: coprocessors) { if (env.getInstance() instanceof RegionObserver) { ctx = ObserverContext.createAndPrepare(env, ctx); try { - ((RegionObserver)env.getInstance()).postPut(ctx, familyMap, writeToWAL); + ((RegionObserver)env.getInstance()).postPut(ctx, put, edit, writeToWAL); } catch (Throwable e) { handleCoprocessorThrowable(env, e); } @@ -690,12 +692,13 @@ } /** - * @param familyMap map of family to edits for the given family. + * @param delete The Delete object + * @param edit The WALEdit object. * @param writeToWAL true if the change should be written to the WAL * @return true if default processing should be bypassed * @exception IOException Exception */ - public boolean preDelete(final Map> familyMap, + public boolean preDelete(Delete delete, WALEdit edit, final boolean writeToWAL) throws IOException { boolean bypass = false; ObserverContext ctx = null; @@ -703,7 +706,7 @@ if (env.getInstance() instanceof RegionObserver) { ctx = ObserverContext.createAndPrepare(env, ctx); try { - ((RegionObserver)env.getInstance()).preDelete(ctx, familyMap, writeToWAL); + ((RegionObserver)env.getInstance()).preDelete(ctx, delete, edit, writeToWAL); } catch (Throwable e) { handleCoprocessorThrowable(env, e); } @@ -717,18 +720,19 @@ } /** - * @param familyMap map of family to edits for the given family. + * @param delete The Delete object + * @param edit The WALEdit object. * @param writeToWAL true if the change should be written to the WAL * @exception IOException Exception */ - public void postDelete(final Map> familyMap, + public void postDelete(Delete delete, WALEdit edit, final boolean writeToWAL) throws IOException { ObserverContext ctx = null; for (RegionEnvironment env: coprocessors) { if (env.getInstance() instanceof RegionObserver) { ctx = ObserverContext.createAndPrepare(env, ctx); try { - ((RegionObserver)env.getInstance()).postDelete(ctx, familyMap, writeToWAL); + ((RegionObserver)env.getInstance()).postDelete(ctx, delete, edit, writeToWAL); } catch (Throwable e) { handleCoprocessorThrowable(env, e); } Index: src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (revision 1177371) +++ src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (working copy) @@ -1402,7 +1402,7 @@ try { // All edits for the given row (across all column families) must happen atomically. prepareDelete(delete); - delete(delete.getFamilyMap(), delete.getClusterId(), writeToWAL); + internalDelete(delete, delete.getClusterId(), writeToWAL); } finally { if(lockid == null) releaseRowLock(lid); } @@ -1418,9 +1418,25 @@ */ public void delete(Map> familyMap, UUID clusterId, boolean writeToWAL) throws IOException { + Delete delete = new Delete(); + delete.setFamilyMap(familyMap); + delete.setClusterId(clusterId); + delete.setWriteToWAL(writeToWAL); + internalDelete(delete, clusterId, writeToWAL); + } + + /** + * @param familyMap map of family to edits for the given family. + * @param writeToWAL + * @throws IOException + */ + private void internalDelete(Delete delete, UUID clusterId, + boolean writeToWAL) throws IOException { + Map> familyMap = delete.getFamilyMap(); + WALEdit walEdit = new WALEdit(); /* Run coprocessor pre hook outside of locks to avoid deadlock */ if (coprocessorHost != null) { - if (coprocessorHost.preDelete(familyMap, writeToWAL)) { + if (coprocessorHost.preDelete(delete, walEdit, writeToWAL)) { return; } } @@ -1484,7 +1500,6 @@ // // bunch up all edits across all column families into a // single WALEdit. - WALEdit walEdit = new WALEdit(); addFamilyMapToWALEdit(familyMap, walEdit); this.log.append(regionInfo, this.htableDescriptor.getName(), walEdit, clusterId, now, this.htableDescriptor); @@ -1495,7 +1510,7 @@ flush = isFlushSize(this.addAndGetGlobalMemstoreSize(addedSize)); if (coprocessorHost != null) { - coprocessorHost.postDelete(familyMap, writeToWAL); + coprocessorHost.postDelete(delete, walEdit, writeToWAL); } } finally { this.updatesLock.readLock().unlock(); @@ -1561,8 +1576,7 @@ try { // All edits for the given row (across all column families) must happen atomically. - // Coprocessor interception happens in put(Map,boolean) - put(put.getFamilyMap(), put.getClusterId(), writeToWAL); + internalPut(put, put.getClusterId(), writeToWAL); } finally { if(lockid == null) releaseRowLock(lid); } @@ -1644,13 +1658,14 @@ @SuppressWarnings("unchecked") private long doMiniBatchPut( BatchOperationInProgress> batchOp) throws IOException { + + WALEdit walEdit = new WALEdit(); /* Run coprocessor pre hook outside of locks to avoid deadlock */ if (coprocessorHost != null) { for (int i = 0; i < batchOp.operations.length; i++) { Pair nextPair = batchOp.operations[i]; Put put = nextPair.getFirst(); - Map> familyMap = put.getFamilyMap(); - if (coprocessorHost.prePut(familyMap, put.getWriteToWAL())) { + if (coprocessorHost.prePut(put, walEdit, put.getWriteToWAL())) { // pre hook says skip this Put // mark as success and skip below batchOp.retCodeDetails[i] = new OperationStatus( @@ -1744,7 +1759,6 @@ // ------------------------------------ // STEP 3. Write to WAL // ---------------------------------- - WALEdit walEdit = new WALEdit(); for (int i = firstIndex; i < lastIndexExclusive; i++) { // Skip puts that were determined to be invalid during preprocessing if (batchOp.retCodeDetails[i].getOperationStatusCode() @@ -1787,7 +1801,7 @@ continue; } Put p = batchOp.operations[i].getFirst(); - coprocessorHost.postPut(familyMaps[i], p.getWriteToWAL()); + coprocessorHost.postPut(p, walEdit, p.getWriteToWAL()); } } @@ -1897,11 +1911,11 @@ // originating cluster. A slave cluster receives the result as a Put // or Delete if (isPut) { - put(((Put)w).getFamilyMap(), HConstants.DEFAULT_CLUSTER_ID, writeToWAL); + internalPut(((Put)w), HConstants.DEFAULT_CLUSTER_ID, writeToWAL); } else { Delete d = (Delete)w; prepareDelete(d); - delete(d.getFamilyMap(), HConstants.DEFAULT_CLUSTER_ID, writeToWAL); + internalDelete(d, HConstants.DEFAULT_CLUSTER_ID, writeToWAL); } return true; } @@ -1992,21 +2006,27 @@ familyMap = new HashMap>(); familyMap.put(family, edits); - this.put(familyMap, HConstants.DEFAULT_CLUSTER_ID, true); + Put p = new Put(); + p.setFamilyMap(familyMap); + p.setClusterId(HConstants.DEFAULT_CLUSTER_ID); + p.setWriteToWAL(true); + this.internalPut(p, HConstants.DEFAULT_CLUSTER_ID, true); } /** * Add updates first to the hlog (if writeToWal) and then add values to memstore. * Warning: Assumption is caller has lock on passed in row. - * @param familyMap map of family to edits for the given family. + * @param put The Put command * @param writeToWAL if true, then we should write to the log * @throws IOException */ - private void put(Map> familyMap, UUID clusterId, + private void internalPut(Put put, UUID clusterId, boolean writeToWAL) throws IOException { + Map> familyMap = put.getFamilyMap(); + WALEdit walEdit = new WALEdit(); /* run pre put hook outside of lock to avoid deadlock */ if (coprocessorHost != null) { - if (coprocessorHost.prePut(familyMap, writeToWAL)) { + if (coprocessorHost.prePut(put, walEdit, writeToWAL)) { return; } } @@ -2025,7 +2045,6 @@ // for some reason fail to write/sync to commit log, the memstore // will contain uncommitted transactions. if (writeToWAL) { - WALEdit walEdit = new WALEdit(); addFamilyMapToWALEdit(familyMap, walEdit); this.log.append(regionInfo, this.htableDescriptor.getName(), walEdit, clusterId, now, this.htableDescriptor); @@ -2038,7 +2057,7 @@ } if (coprocessorHost != null) { - coprocessorHost.postPut(familyMap, writeToWAL); + coprocessorHost.postPut(put, walEdit, writeToWAL); } if (flush) { Index: src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java (revision 1177371) +++ src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java (working copy) @@ -133,24 +133,23 @@ } @Override - public void prePut(final ObserverContext e, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void prePut(final ObserverContext e, + final Put put, final WALEdit edit, final boolean writeToWAL) throws IOException { } @Override - public void postPut(final ObserverContext e, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void postPut(final ObserverContext e, + final Put put, final WALEdit edit, final boolean writeToWAL) throws IOException { } @Override - public void preDelete(final ObserverContext e, final Map> familyMap, final boolean writeToWAL) throws IOException { + public void preDelete(final ObserverContext e, + final Delete delete, final WALEdit edit, final boolean writeToWAL) throws IOException { } @Override public void postDelete(final ObserverContext e, - final Map> familyMap, final boolean writeToWAL) - throws IOException { + final Delete delete, final WALEdit edit, final boolean writeToWAL) throws IOException { } @Override Index: src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java (revision 1177371) +++ src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java (working copy) @@ -267,12 +267,13 @@ * Call CoprocessorEnvironment#complete to skip any subsequent chained * coprocessors * @param c the environment provided by the region server - * @param familyMap map of family to edits for the given family + * @param put The Put object + * @param edit The WALEdit object that will be written to the wal * @param writeToWAL true if the change should be written to the WAL * @throws IOException if an error occurred on the coprocessor */ - void prePut(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) + void prePut(final ObserverContext c, + final Put put, final WALEdit edit, final boolean writeToWAL) throws IOException; /** @@ -281,12 +282,13 @@ * Call CoprocessorEnvironment#complete to skip any subsequent chained * coprocessors * @param c the environment provided by the region server - * @param familyMap map of family to edits for the given family + * @param put The Put object + * @param edit The WALEdit object for the wal * @param writeToWAL true if the change should be written to the WAL * @throws IOException if an error occurred on the coprocessor */ - void postPut(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) + void postPut(final ObserverContext c, + final Put put, final WALEdit edit, final boolean writeToWAL) throws IOException; /** @@ -297,12 +299,13 @@ * Call CoprocessorEnvironment#complete to skip any subsequent chained * coprocessors * @param c the environment provided by the region server - * @param familyMap map of family to edits for the given family + * @param delete The Delete object + * @param edit The WALEdit object for the wal * @param writeToWAL true if the change should be written to the WAL * @throws IOException if an error occurred on the coprocessor */ - void preDelete(final ObserverContext c, final Map> familyMap, final boolean writeToWAL) + void preDelete(final ObserverContext c, + final Delete delete, final WALEdit edit, final boolean writeToWAL) throws IOException; /** @@ -311,12 +314,13 @@ * Call CoprocessorEnvironment#complete to skip any subsequent chained * coprocessors * @param c the environment provided by the region server - * @param familyMap map of family to edits for the given family + * @param delete The Delete object + * @param edit The WALEdit object for the wal * @param writeToWAL true if the change should be written to the WAL * @throws IOException if an error occurred on the coprocessor */ void postDelete(final ObserverContext c, - final Map> familyMap, final boolean writeToWAL) + final Delete delete, final WALEdit edit, final boolean writeToWAL) throws IOException; /**