Skip to content

Commit

Permalink
HBASE-25574 Revisit put/delete/increment/append related RegionObserve…
Browse files Browse the repository at this point in the history 
…r methods

Signed-off-by: Duo Zhang <zhangduo@apache.org>
  • Loading branch information
@brfrn169
brfrn169 committed Feb 22, 2021
1 parent c1a9e89 commit b78366c
Show file tree
Hide file tree
Showing 3 changed files with 208 additions and 53 deletions.
187 changes: 172 additions & 15 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -374,10 +374,30 @@ default boolean postExists(ObserverContext<RegionCoprocessorEnvironment> c, Get
* @param put The Put object
* @param edit The WALEdit object that will be written to the wal
* @param durability Persistence guarantee for this Put
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #prePut(ObserverContext, Put, WALEdit)} instead.
*/
@Deprecated
default void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit,
Durability durability) throws IOException {}

/**
* Called before the client stores a value.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions.
* If 'bypass' is set, we skip out on calling any subsequent chained coprocessors.
* <p>
* Note: Do not retain references to any Cells in 'put' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param put The Put object
* @param edit The WALEdit object that will be written to the wal
*/
default void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit)
throws IOException {
prePut(c, put, edit, put.getDurability());
}

/**
* Called after the client stores a value.
* <p>
Expand All @@ -387,10 +407,27 @@ default void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WA
* @param put The Put object
* @param edit The WALEdit object for the wal
* @param durability Persistence guarantee for this Put
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #postPut(ObserverContext, Put, WALEdit)} instead.
*/
@Deprecated
default void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit,
Durability durability) throws IOException {}

/**
* Called after the client stores a value.
* <p>
* Note: Do not retain references to any Cells in 'put' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param put The Put object
* @param edit The WALEdit object for the wal
*/
default void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit)
throws IOException {
postPut(c, put, edit, put.getDurability());
}

/**
* Called before the client deletes a value.
* <p>
Expand All @@ -403,10 +440,30 @@ default void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, W
* @param delete The Delete object
* @param edit The WALEdit object for the wal
* @param durability Persistence guarantee for this Delete
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #preDelete(ObserverContext, Delete, WALEdit)} instead.
*/
@Deprecated
default void preDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete,
WALEdit edit, Durability durability) throws IOException {}

/**
* Called before the client deletes a value.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions.
* If 'bypass' is set, we skip out on calling any subsequent chained coprocessors.
* <p>
* Note: Do not retain references to any Cells in 'delete' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param delete The Delete object
* @param edit The WALEdit object for the wal
*/
default void preDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete,
WALEdit edit) throws IOException {
preDelete(c, delete, edit, delete.getDurability());
}

/**
* Called before the server updates the timestamp for version delete with latest timestamp.
* <p>
Expand Down Expand Up @@ -434,10 +491,27 @@ default void prePrepareTimeStampForDeleteVersion(ObserverContext<RegionCoprocess
* @param delete The Delete object
* @param edit The WALEdit object for the wal
* @param durability Persistence guarantee for this Delete
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #postDelete(ObserverContext, Delete, WALEdit)} instead.
*/
@Deprecated
default void postDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete,
WALEdit edit, Durability durability) throws IOException {}

/**
* Called after the client deletes a value.
* <p>
* Note: Do not retain references to any Cells in 'delete' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param delete The Delete object
* @param edit The WALEdit object for the wal
*/
default void postDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete,
WALEdit edit) throws IOException {
postDelete(c, delete, edit, delete.getDurability());
}

/**
* This will be called for every batch mutation operation happening at the server. This will be
* called after acquiring the locks on the mutating rows and after applying the proper timestamp
Expand All @@ -457,10 +531,10 @@ default void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
/**
* This will be called after applying a batch of Mutations on a region. The Mutations are added
* to memstore and WAL. The difference of this one with
* {@link #postPut(ObserverContext, Put, WALEdit, Durability)}
* and {@link #postDelete(ObserverContext, Delete, WALEdit, Durability)}
* and {@link #postIncrement(ObserverContext, Increment, Result)}
* and {@link #postAppend(ObserverContext, Append, Result)} is
* {@link #postPut(ObserverContext, Put, WALEdit)}
* and {@link #postDelete(ObserverContext, Delete, WALEdit)}
* and {@link #postIncrement(ObserverContext, Increment, Result, WALEdit)}
* and {@link #postAppend(ObserverContext, Append, Result, WALEdit)} is
* this hook will be executed before the mvcc transaction completion.
* <p>
* Note: Do not retain references to any Cells in Mutations beyond the life of this invocation.
Expand Down Expand Up @@ -968,12 +1042,33 @@ default CheckAndMutateResult postCheckAndMutate(ObserverContext<RegionCoprocesso
* @param c the environment provided by the region server
* @param append Append object
* @return result to return to the client if bypassing default processing
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #preAppend(ObserverContext, Append, WALEdit)} instead.
*/
@Deprecated
default Result preAppend(ObserverContext<RegionCoprocessorEnvironment> c, Append append)
throws IOException {
throws IOException {
return null;
}

/**
* Called before Append.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions.
* If 'bypass' is set, we skip out on calling any subsequent chained coprocessors.
* <p>
* Note: Do not retain references to any Cells in 'append' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param append Append object
* @param edit The WALEdit object that will be written to the wal
* @return result to return to the client if bypassing default processing
*/
default Result preAppend(ObserverContext<RegionCoprocessorEnvironment> c, Append append,
WALEdit edit) throws IOException {
return preAppend(c, append);
}

/**
* Called before Append but after acquiring rowlock.
* <p>
Expand All @@ -989,9 +1084,12 @@ default Result preAppend(ObserverContext<RegionCoprocessorEnvironment> c, Append
* @param c the environment provided by the region server
* @param append Append object
* @return result to return to the client if bypassing default processing
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #preBatchMutate(ObserverContext, MiniBatchOperationInProgress)} instead.
*/
@Deprecated
default Result preAppendAfterRowLock(ObserverContext<RegionCoprocessorEnvironment> c,
Append append) throws IOException {
Append append) throws IOException {
return null;
}

Expand All @@ -1004,12 +1102,31 @@ default Result preAppendAfterRowLock(ObserverContext<RegionCoprocessorEnvironmen
* @param append Append object
* @param result the result returned by increment
* @return the result to return to the client
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #postAppend(ObserverContext, Append, Result, WALEdit)} instead.
*/
@Deprecated
default Result postAppend(ObserverContext<RegionCoprocessorEnvironment> c, Append append,
Result result) throws IOException {
Result result) throws IOException {
return result;
}

/**
* Called after Append
* <p>
* Note: Do not retain references to any Cells in 'append' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param append Append object
* @param result the result returned by increment
* @param edit The WALEdit object for the wal
* @return the result to return to the client
*/
default Result postAppend(ObserverContext<RegionCoprocessorEnvironment> c, Append append,
Result result, WALEdit edit) throws IOException {
return postAppend(c, append, result);
}

/**
* Called before Increment.
* <p>
Expand All @@ -1021,12 +1138,33 @@ default Result postAppend(ObserverContext<RegionCoprocessorEnvironment> c, Appen
* @param c the environment provided by the region server
* @param increment increment object
* @return result to return to the client if bypassing default processing
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #preIncrement(ObserverContext, Increment, WALEdit)} instead.
*/
@Deprecated
default Result preIncrement(ObserverContext<RegionCoprocessorEnvironment> c, Increment increment)
throws IOException {
throws IOException {
return null;
}

/**
* Called before Increment.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions.
* If 'bypass' is set, we skip out on calling any subsequent chained coprocessors.
* <p>
* Note: Do not retain references to any Cells in 'increment' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param increment increment object
* @param edit The WALEdit object that will be written to the wal
* @return result to return to the client if bypassing default processing
*/
default Result preIncrement(ObserverContext<RegionCoprocessorEnvironment> c, Increment increment,
WALEdit edit) throws IOException {
return preIncrement(c, increment);
}

/**
* Called before Increment but after acquiring rowlock.
* <p>
Expand All @@ -1040,15 +1178,15 @@ default Result preIncrement(ObserverContext<RegionCoprocessorEnvironment> c, Inc
* Note: Do not retain references to any Cells in 'increment' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
*
* @param c
* the environment provided by the region server
* @param increment
* increment object
* @param c the environment provided by the region server
* @param increment increment object
* @return result to return to the client if bypassing default processing
* if an error occurred on the coprocessor
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #preBatchMutate(ObserverContext, MiniBatchOperationInProgress)} instead.
*/
@Deprecated
default Result preIncrementAfterRowLock(ObserverContext<RegionCoprocessorEnvironment> c,
Increment increment) throws IOException {
Increment increment) throws IOException {
return null;
}

Expand All @@ -1061,12 +1199,31 @@ default Result preIncrementAfterRowLock(ObserverContext<RegionCoprocessorEnviron
* @param increment increment object
* @param result the result returned by increment
* @return the result to return to the client
* @deprecated since 2.4.2 and will be removed in 4.0.0. Use
* {@link #postIncrement(ObserverContext, Increment, Result, WALEdit)} instead.
*/
@Deprecated
default Result postIncrement(ObserverContext<RegionCoprocessorEnvironment> c, Increment increment,
Result result) throws IOException {
Result result) throws IOException {
return result;
}

/**
* Called after increment
* <p>
* Note: Do not retain references to any Cells in 'increment' beyond the life of this invocation.
* If need a Cell reference for later use, copy the cell and use that.
* @param c the environment provided by the region server
* @param increment increment object
* @param result the result returned by increment
* @param edit The WALEdit object for the wal
* @return the result to return to the client
*/
default Result postIncrement(ObserverContext<RegionCoprocessorEnvironment> c, Increment increment,
Result result, WALEdit edit) throws IOException {
return postIncrement(c, increment, result);
}

/**
* Called before the client opens a new scanner.
* <p>
Expand Down
17 changes: 9 additions & 8 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4190,18 +4190,19 @@ public void doPostOpCleanupForMiniBatch(MiniBatchOperationInProgress<Mutation> m
if (retCodeDetails[i].getOperationStatusCode() == OperationStatusCode.SUCCESS) {
Mutation m = getMutation(i);
if (m instanceof Put) {
region.coprocessorHost.postPut((Put) m, walEdit, m.getDurability());
region.coprocessorHost.postPut((Put) m, walEdit);
} else if (m instanceof Delete) {
region.coprocessorHost.postDelete((Delete) m, walEdit, m.getDurability());
region.coprocessorHost.postDelete((Delete) m, walEdit);
} else if (m instanceof Increment) {
Result result = region.getCoprocessorHost().postIncrement((Increment) m,
results[i]);
results[i], walEdit);
if (result != results[i]) {
retCodeDetails[i] =
new OperationStatus(retCodeDetails[i].getOperationStatusCode(), result);
}
} else if (m instanceof Append) {
Result result = region.getCoprocessorHost().postAppend((Append) m, results[i]);
Result result = region.getCoprocessorHost().postAppend((Append) m, results[i],
walEdit);
if (result != results[i]) {
retCodeDetails[i] =
new OperationStatus(retCodeDetails[i].getOperationStatusCode(), result);
Expand Down Expand Up @@ -4263,7 +4264,7 @@ private void callPreMutateCPHook(int index, final WALEdit walEdit, final int[] m
throws IOException {
Mutation m = getMutation(index);
if (m instanceof Put) {
if (region.coprocessorHost.prePut((Put) m, walEdit, m.getDurability())) {
if (region.coprocessorHost.prePut((Put) m, walEdit)) {
// pre hook says skip this Put
// mark as success and skip in doMiniBatchMutation
metrics[0]++;
Expand All @@ -4277,15 +4278,15 @@ private void callPreMutateCPHook(int index, final WALEdit walEdit, final int[] m
// Can this be avoided?
region.prepareDelete(curDel);
}
if (region.coprocessorHost.preDelete(curDel, walEdit, m.getDurability())) {
if (region.coprocessorHost.preDelete(curDel, walEdit)) {
// pre hook says skip this Delete
// mark as success and skip in doMiniBatchMutation
metrics[1]++;
retCodeDetails[index] = OperationStatus.SUCCESS;
}
} else if (m instanceof Increment) {
Increment increment = (Increment) m;
Result result = region.coprocessorHost.preIncrement(increment);
Result result = region.coprocessorHost.preIncrement(increment, walEdit);
if (result != null) {
// pre hook says skip this Increment
// mark as success and skip in doMiniBatchMutation
Expand All @@ -4294,7 +4295,7 @@ private void callPreMutateCPHook(int index, final WALEdit walEdit, final int[] m
}
} else if (m instanceof Append) {
Append append = (Append) m;
Result result = region.coprocessorHost.preAppend(append);
Result result = region.coprocessorHost.preAppend(append, walEdit);
if (result != null) {
// pre hook says skip this Append
// mark as success and skip in doMiniBatchMutation
Expand Down
Loading

0 comments on commit b78366c

Please sign in to comment.