joshelser commented on a change in pull request #1648: URL: https://github.com/apache/hbase/pull/1648#discussion_r432189368
########## File path: hbase-client/src/main/java/org/apache/hadoop/hbase/client/CheckAndMutate.java ########## @@ -0,0 +1,384 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.client; + +import java.util.Collections; +import java.util.List; +import java.util.NavigableMap; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellBuilder; +import org.apache.hadoop.hbase.CellBuilderType; +import org.apache.hadoop.hbase.CompareOperator; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.filter.Filter; +import org.apache.hadoop.hbase.io.TimeRange; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.yetus.audience.InterfaceStability; +import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; + +/** + * Used to perform CheckAndMutate operations. Currently {@link Put}, {@link Delete} + * and {@link RowMutations} are supported. + * <p> + * Use the builder class to instantiate a CheckAndMutate object. + * This builder class is fluent style APIs, the code are like: + * <pre> + * <code> + * // A CheckAndMutate operation where do the specified action if the column (specified by the + * // family and the qualifier) of the row equals to the specified value + * CheckAndMutate checkAndMutate = CheckAndMutate.builder(row) + * .ifEquals(family, qualifier, value) + * .build(put); + * + * // A CheckAndMutate operation where do the specified action if the column (specified by the + * // family and the qualifier) of the row doesn't exist + * CheckAndMutate checkAndMutate = CheckAndMutate.builder(row) + * .ifNotExists(family, qualifier) + * .build(put); + * + * // A CheckAndMutate operation where do the specified action if the row matches the filter + * CheckAndMutate checkAndMutate = CheckAndMutate.builder(row) + * .ifMatches(filter) + * .build(delete); + * </code> + * </pre> + */ [email protected] [email protected] +public final class CheckAndMutate extends Mutation { + + /** + * A builder class for building a CheckAndMutate object. + */ + public static final class Builder { + private final byte[] row; + private byte[] family; + private byte[] qualifier; + private CompareOperator op; + private byte[] value; + private Filter filter; + private TimeRange timeRange; + + private Builder(byte[] row) { + this.row = Preconditions.checkNotNull(row, "row is null"); + } + + /** + * Check for lack of column + * + * @param family family to check + * @param qualifier qualifier to check + * @return the CheckAndMutate object + */ + public Builder ifNotExists(byte[] family, byte[] qualifier) { + return ifEquals(family, qualifier, null); + } + + /** + * Check for equality + * + * @param family family to check + * @param qualifier qualifier to check + * @param value the expected value + * @return the CheckAndMutate object + */ + public Builder ifEquals(byte[] family, byte[] qualifier, byte[] value) { + return ifMatches(family, qualifier, CompareOperator.EQUAL, value); + } + + /** + * @param family family to check + * @param qualifier qualifier to check + * @param compareOp comparison operator to use + * @param value the expected value + * @return the CheckAndMutate object + */ + public Builder ifMatches(byte[] family, byte[] qualifier, CompareOperator compareOp, + byte[] value) { + this.family = Preconditions.checkNotNull(family, "family is null"); + this.qualifier = qualifier; + this.op = Preconditions.checkNotNull(compareOp, "compareOp is null"); + this.value = value; + return this; + } + + /** + * @param filter filter to check + * @return the CheckAndMutate object + */ + public Builder ifMatches(Filter filter) { + this.filter = Preconditions.checkNotNull(filter, "filter is null"); + return this; + } + + /** + * @param timeRange time range to check + * @return the CheckAndMutate object + */ + public Builder timeRange(TimeRange timeRange) { + this.timeRange = timeRange; + return this; + } + + private void preCheck(Row action) { + Preconditions.checkNotNull(action, "action (Put/Delete/RowMutations) is null"); + if (!Bytes.equals(row, action.getRow())) { + throw new IllegalArgumentException("The row of the action (Put/Delete/RowMutations) <" + + Bytes.toStringBinary(action.getRow()) + "> doesn't match the original one <" + + Bytes.toStringBinary(this.row) + ">"); + } + Preconditions.checkState(op != null || filter != null, "condition is null. You need to" + + " specify the condition by calling ifNotExists/ifEquals/ifMatches before building a" + + " CheckAndMutate object"); + } + + /** + * @param put data to put if check succeeds + * @return a CheckAndMutate object + */ + public CheckAndMutate build(Put put) { + preCheck(put); + if (filter != null) { + return new CheckAndMutate(row, filter, timeRange, put); + } else { + return new CheckAndMutate(row, family, qualifier, op, value, timeRange, put); + } + } + + /** + * @param delete data to delete if check succeeds + * @return a CheckAndMutate object + */ + public CheckAndMutate build(Delete delete) { + preCheck(delete); + if (filter != null) { + return new CheckAndMutate(row, filter, timeRange, delete); + } else { + return new CheckAndMutate(row, family, qualifier, op, value, timeRange, delete); + } + } + + /** + * @param mutation mutations to perform if check succeeds + * @return a CheckAndMutate object + */ + public CheckAndMutate build(RowMutations mutation) { + preCheck(mutation); + if (filter != null) { + return new CheckAndMutate(row, filter, timeRange, mutation); + } else { + return new CheckAndMutate(row, family, qualifier, op, value, timeRange, mutation); + } + } + } + + /** + * returns a builder object to build a CheckAndMutate object + * + * @param row row + * @return a builder object + */ + public static Builder builder(byte[] row) { + return new Builder(row); + } + + private final byte[] family; + private final byte[] qualifier; + private final CompareOperator op; + private final byte[] value; + private final Filter filter; + private final TimeRange timeRange; + private final Row action; + + private CheckAndMutate(byte[] row, byte[] family, byte[] qualifier,final CompareOperator op, + byte[] value, TimeRange timeRange, Row action) { + super(row, HConstants.LATEST_TIMESTAMP, Collections.emptyNavigableMap()); + this.family = family; + this.qualifier = qualifier; + this.op = op; + this.value = value; + this.filter = null; + this.timeRange = timeRange; + this.action = action; + } + + private CheckAndMutate(byte[] row, Filter filter, TimeRange timeRange, Row action) { + super(row, HConstants.LATEST_TIMESTAMP, Collections.emptyNavigableMap()); + this.family = null; + this.qualifier = null; + this.op = null; + this.value = null; + this.filter = filter; + this.timeRange = timeRange; + this.action = action; + } + + /** + * @return whether it has an action or not + */ + public boolean hasAction() { + return action != null; + } + + /** + * @return whether it has a condition or not + */ + public boolean hasCondition() { + return op != null || filter != null; + } + + /** + * @return the family to check + */ + public byte[] getFamily() { + return family; + } + + /** + * @return the qualifier to check + */ + public byte[] getQualifier() { + return qualifier; + } + + /** + * @return the comparison operator + */ + public CompareOperator getCompareOp() { + return op; + } + + /** + * @return the expected value + */ + public byte[] getValue() { + return value; + } + + /** + * @return the filter to check + */ + public Filter getFilter() { + return filter; + } + + /** + * @return the time range to check + */ + public TimeRange getTimeRange() { + return timeRange; + } + + /** + * @return the action done if check succeeds + */ + public Row getAction() { + return action; + } + + @Override + public NavigableMap<byte[], List<Cell>> getFamilyCellMap() { + if (action instanceof Mutation) { + return ((Mutation) action).getFamilyCellMap(); + } else { Review comment: nit: can drop the `else` and just make this `if (..) { ...} throw new ..` ########## File path: hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java ########## @@ -2827,37 +2905,113 @@ public MultiResponse multi(final RpcController rpcc, final MultiRequest request) quota.close(); continue; } - // How does this call happen? It may need some work to play well w/ the surroundings. - // Need to return an item per Action along w/ Action index. TODO. + try { - if (request.hasCondition()) { - Condition condition = request.getCondition(); - byte[] row = condition.getRow().toByteArray(); - byte[] family = condition.hasFamily() ? condition.getFamily().toByteArray() : null; - byte[] qualifier = condition.hasQualifier() ? - condition.getQualifier().toByteArray() : null; - CompareOperator op = condition.hasCompareType() ? - CompareOperator.valueOf(condition.getCompareType().name()) : null; - ByteArrayComparable comparator = condition.hasComparator() ? - ProtobufUtil.toComparator(condition.getComparator()) : null; - Filter filter = condition.hasFilter() ? - ProtobufUtil.toFilter(condition.getFilter()) : null; - TimeRange timeRange = condition.hasTimeRange() ? - ProtobufUtil.toTimeRange(condition.getTimeRange()) : - TimeRange.allTime(); + Condition condition = regionAction.getCondition(); + byte[] row = condition.getRow().toByteArray(); + byte[] family = condition.hasFamily() ? condition.getFamily().toByteArray() : null; + byte[] qualifier = condition.hasQualifier() ? + condition.getQualifier().toByteArray() : null; + CompareOperator op = condition.hasCompareType() ? + CompareOperator.valueOf(condition.getCompareType().name()) : null; + ByteArrayComparable comparator = condition.hasComparator() ? + ProtobufUtil.toComparator(condition.getComparator()) : null; + Filter filter = condition.hasFilter() ? + ProtobufUtil.toFilter(condition.getFilter()) : null; + TimeRange timeRange = condition.hasTimeRange() ? + ProtobufUtil.toTimeRange(condition.getTimeRange()) : + TimeRange.allTime(); + + boolean processed; + if (regionAction.hasAtomic() && regionAction.getAtomic()) { + // RowMutations processed = checkAndRowMutate(region, regionAction.getActionList(), cellScanner, row, family, qualifier, op, comparator, filter, timeRange, regionActionResultBuilder, spaceQuotaEnforcement); } else { - doAtomicBatchOp(regionActionResultBuilder, region, quota, regionAction.getActionList(), - cellScanner, spaceQuotaEnforcement); - processed = Boolean.TRUE; + if (regionAction.getActionList().isEmpty()) { + // If the region action list is empty, do nothing. + regionActionResultBuilder.setProcessed(true); + quota.close(); Review comment: should be covered in the finally? ########## File path: hbase-client/src/main/java/org/apache/hadoop/hbase/client/CheckAndMutate.java ########## @@ -0,0 +1,384 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.client; + +import java.util.Collections; +import java.util.List; +import java.util.NavigableMap; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellBuilder; +import org.apache.hadoop.hbase.CellBuilderType; +import org.apache.hadoop.hbase.CompareOperator; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.filter.Filter; +import org.apache.hadoop.hbase.io.TimeRange; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.yetus.audience.InterfaceStability; +import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; + +/** + * Used to perform CheckAndMutate operations. Currently {@link Put}, {@link Delete} + * and {@link RowMutations} are supported. + * <p> + * Use the builder class to instantiate a CheckAndMutate object. + * This builder class is fluent style APIs, the code are like: + * <pre> + * <code> + * // A CheckAndMutate operation where do the specified action if the column (specified by the + * // family and the qualifier) of the row equals to the specified value + * CheckAndMutate checkAndMutate = CheckAndMutate.builder(row) + * .ifEquals(family, qualifier, value) + * .build(put); + * + * // A CheckAndMutate operation where do the specified action if the column (specified by the + * // family and the qualifier) of the row doesn't exist + * CheckAndMutate checkAndMutate = CheckAndMutate.builder(row) + * .ifNotExists(family, qualifier) + * .build(put); + * + * // A CheckAndMutate operation where do the specified action if the row matches the filter + * CheckAndMutate checkAndMutate = CheckAndMutate.builder(row) + * .ifMatches(filter) + * .build(delete); + * </code> + * </pre> + */ [email protected] [email protected] +public final class CheckAndMutate extends Mutation { + + /** + * A builder class for building a CheckAndMutate object. + */ + public static final class Builder { + private final byte[] row; + private byte[] family; + private byte[] qualifier; + private CompareOperator op; + private byte[] value; + private Filter filter; + private TimeRange timeRange; + + private Builder(byte[] row) { + this.row = Preconditions.checkNotNull(row, "row is null"); + } + + /** + * Check for lack of column + * + * @param family family to check + * @param qualifier qualifier to check + * @return the CheckAndMutate object + */ + public Builder ifNotExists(byte[] family, byte[] qualifier) { + return ifEquals(family, qualifier, null); + } + + /** + * Check for equality + * + * @param family family to check + * @param qualifier qualifier to check + * @param value the expected value + * @return the CheckAndMutate object + */ + public Builder ifEquals(byte[] family, byte[] qualifier, byte[] value) { + return ifMatches(family, qualifier, CompareOperator.EQUAL, value); + } + + /** + * @param family family to check + * @param qualifier qualifier to check + * @param compareOp comparison operator to use + * @param value the expected value + * @return the CheckAndMutate object + */ + public Builder ifMatches(byte[] family, byte[] qualifier, CompareOperator compareOp, + byte[] value) { + this.family = Preconditions.checkNotNull(family, "family is null"); + this.qualifier = qualifier; + this.op = Preconditions.checkNotNull(compareOp, "compareOp is null"); + this.value = value; + return this; + } + + /** + * @param filter filter to check + * @return the CheckAndMutate object + */ + public Builder ifMatches(Filter filter) { + this.filter = Preconditions.checkNotNull(filter, "filter is null"); + return this; + } + + /** + * @param timeRange time range to check + * @return the CheckAndMutate object + */ + public Builder timeRange(TimeRange timeRange) { + this.timeRange = timeRange; + return this; + } + + private void preCheck(Row action) { + Preconditions.checkNotNull(action, "action (Put/Delete/RowMutations) is null"); + if (!Bytes.equals(row, action.getRow())) { + throw new IllegalArgumentException("The row of the action (Put/Delete/RowMutations) <" + + Bytes.toStringBinary(action.getRow()) + "> doesn't match the original one <" + + Bytes.toStringBinary(this.row) + ">"); + } + Preconditions.checkState(op != null || filter != null, "condition is null. You need to" + + " specify the condition by calling ifNotExists/ifEquals/ifMatches before building a" + + " CheckAndMutate object"); + } + + /** + * @param put data to put if check succeeds + * @return a CheckAndMutate object + */ + public CheckAndMutate build(Put put) { + preCheck(put); + if (filter != null) { + return new CheckAndMutate(row, filter, timeRange, put); + } else { + return new CheckAndMutate(row, family, qualifier, op, value, timeRange, put); + } + } + + /** + * @param delete data to delete if check succeeds + * @return a CheckAndMutate object + */ + public CheckAndMutate build(Delete delete) { + preCheck(delete); + if (filter != null) { + return new CheckAndMutate(row, filter, timeRange, delete); + } else { + return new CheckAndMutate(row, family, qualifier, op, value, timeRange, delete); + } + } + + /** + * @param mutation mutations to perform if check succeeds + * @return a CheckAndMutate object + */ + public CheckAndMutate build(RowMutations mutation) { + preCheck(mutation); + if (filter != null) { + return new CheckAndMutate(row, filter, timeRange, mutation); + } else { + return new CheckAndMutate(row, family, qualifier, op, value, timeRange, mutation); + } + } + } + + /** + * returns a builder object to build a CheckAndMutate object + * + * @param row row + * @return a builder object + */ + public static Builder builder(byte[] row) { Review comment: nit: `newBuilder(byte[] row)`? It would match the protobuf-java API, but i'm not sure if we have a convention in HBase itself? ########## File path: hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTable.java ########## @@ -599,11 +612,748 @@ public void testCheckAndMutateWithFilterAndTimeRange() throws Throwable { } @Test(expected = NullPointerException.class) - public void testCheckAndMutateWithNotSpecifyingCondition() throws Throwable { + @Deprecated + public void testCheckAndMutateWithoutConditionForOldApi() { getTable.get().checkAndMutate(row, FAMILY) .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d"))); } + // Tests for new checkAndMutate API + + @SuppressWarnings("FutureReturnValueIgnored") + @Test + public void testCheckAndPut() throws InterruptedException, ExecutionException { + AsyncTable<?> table = getTable.get(); + AtomicInteger successCount = new AtomicInteger(0); + AtomicInteger successIndex = new AtomicInteger(-1); + int count = 10; + CountDownLatch latch = new CountDownLatch(count); + + IntStream.range(0, count) + .forEach(i -> table.checkAndMutate(new CheckAndMutate(row) + .ifNotExists(FAMILY, QUALIFIER) + .action(new Put(row).addColumn(FAMILY, QUALIFIER, concat(VALUE, i)))) + .thenAccept(x -> { + if (x) { + successCount.incrementAndGet(); + successIndex.set(i); + } + latch.countDown(); + })); + latch.await(); + assertEquals(1, successCount.get()); + String actual = Bytes.toString(table.get(new Get(row)).get().getValue(FAMILY, QUALIFIER)); + assertTrue(actual.endsWith(Integer.toString(successIndex.get()))); + } + + @SuppressWarnings("FutureReturnValueIgnored") + @Test + public void testCheckAndDelete() throws InterruptedException, ExecutionException { + AsyncTable<?> table = getTable.get(); + int count = 10; + CountDownLatch putLatch = new CountDownLatch(count + 1); + table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown()); + IntStream.range(0, count) + .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE)) + .thenRun(() -> putLatch.countDown())); + putLatch.await(); + + AtomicInteger successCount = new AtomicInteger(0); + AtomicInteger successIndex = new AtomicInteger(-1); + CountDownLatch deleteLatch = new CountDownLatch(count); + + IntStream.range(0, count) + .forEach(i -> table.checkAndMutate(new CheckAndMutate(row) + .ifEquals(FAMILY, QUALIFIER, VALUE) + .action( + new Delete(row).addColumn(FAMILY, QUALIFIER).addColumn(FAMILY, concat(QUALIFIER, i)))) + .thenAccept(x -> { + if (x) { + successCount.incrementAndGet(); + successIndex.set(i); + } + deleteLatch.countDown(); + })); + deleteLatch.await(); + assertEquals(1, successCount.get()); + Result result = table.get(new Get(row)).get(); + IntStream.range(0, count).forEach(i -> { + if (i == successIndex.get()) { + assertFalse(result.containsColumn(FAMILY, concat(QUALIFIER, i))); + } else { + assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i))); + } + }); + } + + @SuppressWarnings("FutureReturnValueIgnored") + @Test + public void testCheckAndMutate() throws InterruptedException, ExecutionException { + AsyncTable<?> table = getTable.get(); + int count = 10; + CountDownLatch putLatch = new CountDownLatch(count + 1); + table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown()); + IntStream.range(0, count) + .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE)) + .thenRun(() -> putLatch.countDown())); + putLatch.await(); + + AtomicInteger successCount = new AtomicInteger(0); + AtomicInteger successIndex = new AtomicInteger(-1); + CountDownLatch mutateLatch = new CountDownLatch(count); + IntStream.range(0, count).forEach(i -> { + RowMutations mutation = new RowMutations(row); + try { + mutation.add((Mutation) new Delete(row).addColumn(FAMILY, QUALIFIER)); Review comment: ah, bummer. Thanks for the explanation. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
