CheckAndMutateBuilder and CheckAndMutate interface

src/main/java/com/alipay/oceanbase/hbase/OHTable.java

@@ -63,6 +63,7 @@
 
 import static com.alipay.oceanbase.hbase.constants.OHConstants.*;
 import static com.alipay.oceanbase.hbase.util.Preconditions.checkArgument;
+import static com.alipay.oceanbase.hbase.util.Preconditions.checkNotNull;
 import static com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory.LCD;
 import static com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory.TABLE_HBASE_LOGGER_SPACE;
 import static com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperation.getInstance;
@@ -104,6 +105,16 @@ public class OHTable implements Table {
      */
     private int                  rpcTimeout;
 
+    /**
+     * timeout for each read rpc request
+     */
+    private int                  readRpcTimeout;
+
+    /**
+     * timeout for each write rpc request
+     */
+    private int                  writeRpcTimeout;
+
     /**
      * if the <code>Get</code> executing pool is specified by user cleanupPoolOnClose will be false ,
      * which means that user is responsible for the pool
@@ -292,7 +303,7 @@ public OHTable(final byte[] tableName, final ObTableClient obTableClient,
     public OHTable(TableName tableName, Connection connection,
                    OHConnectionConfiguration connectionConfig, ExecutorService executePool)
                                                                                            throws IOException {
-        checkArgument(connection.getConfiguration() != null, "configuration is null.");
+        checkArgument(connection != null, "connection is null.");
         checkArgument(tableName != null, "tableName is null.");
         checkArgument(connection.getConfiguration() != null, "configuration is null.");
         checkArgument(tableName.getName() != null, "tableNameString is null.");
@@ -311,6 +322,8 @@ public OHTable(TableName tableName, Connection connection,
             this.cleanupPoolOnClose = false;
         }
         this.rpcTimeout = connectionConfig.getRpcTimeout();
+        this.readRpcTimeout = connectionConfig.getReadRpcTimeout();
+        this.writeRpcTimeout = connectionConfig.getWriteRpcTimeout();
         this.operationTimeout = connectionConfig.getOperationTimeout();
         this.operationExecuteInPool = this.configuration.getBoolean(
             HBASE_CLIENT_OPERATION_EXECUTE_IN_POOL,
@@ -331,6 +344,50 @@ public OHTable(TableName tableName, Connection connection,
         finishSetUp();
     }
 
+    public OHTable(Connection connection, ObTableBuilderBase builder,
+                   OHConnectionConfiguration connectionConfig, ExecutorService executePool)
+                                                                                           throws IOException {
+        checkArgument(connection != null, "connection is null.");
+        checkArgument(connection.getConfiguration() != null, "configuration is null.");
+        checkArgument(builder != null, "builder is null");
+        checkArgument(connectionConfig != null, "connectionConfig is null.");
+        TableName builderTableName = builder.getTableName();
+        this.tableName = builderTableName.getName();
+        this.tableNameString = builderTableName.getNameAsString();
+        this.configuration = connection.getConfiguration();
+        this.executePool = executePool;
+        if (executePool == null) {
+            int maxThreads = configuration.getInt(HBASE_HTABLE_PRIVATE_THREADS_MAX,
+                DEFAULT_HBASE_HTABLE_PRIVATE_THREADS_MAX);
+            long keepAliveTime = configuration.getLong(HBASE_HTABLE_THREAD_KEEP_ALIVE_TIME,
+                DEFAULT_HBASE_HTABLE_THREAD_KEEP_ALIVE_TIME);
+            this.executePool = createDefaultThreadPoolExecutor(1, maxThreads, keepAliveTime);
+            this.cleanupPoolOnClose = true;
+        } else {
+            this.cleanupPoolOnClose = false;
+        }
+        this.rpcTimeout = builder.getRpcTimeout();
+        this.readRpcTimeout = builder.getReadRpcTimeout();
+        this.writeRpcTimeout = builder.getWriteRpcTimeout();
+        this.operationTimeout = builder.getOperationTimeout();
+        this.operationExecuteInPool = this.configuration.getBoolean(
+            HBASE_CLIENT_OPERATION_EXECUTE_IN_POOL,
+            (this.operationTimeout != HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT));
+        this.maxKeyValueSize = connectionConfig.getMaxKeyValueSize();
+        this.putWriteBufferCheck = this.configuration.getInt(HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK,
+            DEFAULT_HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK);
+        this.writeBufferSize = connectionConfig.getWriteBufferSize();
+        int numRetries = configuration.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+            HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
+        this.obTableClient = ObTableClientManager.getOrCreateObTableClient(setUserDefinedNamespace(
+            this.tableNameString, connectionConfig));
+        this.obTableClient.setRpcExecuteTimeout(rpcTimeout);
+        this.obTableClient.setRuntimeRetryTimes(numRetries);
+        setOperationTimeout(operationTimeout);
+
+        finishSetUp();
+    }
+
     /**
      * 创建默认的线程池
      * Using the "direct handoff" approach, new threads will only be created
@@ -370,11 +427,15 @@ private void finishSetUp() {
             HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
             HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
             HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
-        this.rpcTimeout = configuration.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
-            HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
-        this.operationTimeout = this.configuration.getInt(
+        this.rpcTimeout = this.rpcTimeout <= 0 ? configuration.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
+            HConstants.DEFAULT_HBASE_RPC_TIMEOUT) : this.rpcTimeout;
+        this.readRpcTimeout = this.readRpcTimeout <= 0 ? configuration.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY,
+                HConstants.DEFAULT_HBASE_RPC_TIMEOUT) : this.readRpcTimeout;
+        this.writeRpcTimeout = this.writeRpcTimeout <= 0 ? configuration.getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
+                HConstants.DEFAULT_HBASE_RPC_TIMEOUT) : this.writeRpcTimeout;
+        this.operationTimeout = this.operationTimeout <= 0 ? this.configuration.getInt(
             HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
-            HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
+            HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT) : this.operationTimeout;
         this.operationExecuteInPool = this.configuration.getBoolean(
             HBASE_CLIENT_OPERATION_EXECUTE_IN_POOL,
             (this.operationTimeout != HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT));
@@ -388,7 +449,7 @@ private void finishSetUp() {
 
     private OHConnectionConfiguration setUserDefinedNamespace(String tableNameString,
                                                               OHConnectionConfiguration ohConnectionConf)
-                                                                                                         throws IOException {
+                                                                                                         throws IllegalArgumentException {
         if (tableNameString.indexOf(':') != -1) {
             String[] params = tableNameString.split(":");
             if (params.length != 2) {
@@ -903,7 +964,7 @@ public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier,
         RowMutations rowMutations = new RowMutations(row);
         rowMutations.add(put);
         try {
-            return checkAndMutation(row, family, qualifier, compareOp, value, rowMutations);
+            return checkAndMutation(row, family, qualifier, compareOp, value, null, rowMutations);
         } catch (Exception e) {
             logger.error(LCD.convert("01-00005"), put, tableNameString, e);
             throw new IOException("checkAndPut type table:" + tableNameString + " e.msg:"
@@ -959,7 +1020,7 @@ public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier,
         RowMutations rowMutations = new RowMutations(row);
         rowMutations.add(delete);
         try {
-            return checkAndMutation(row, family, qualifier, compareOp, value, rowMutations);
+            return checkAndMutation(row, family, qualifier, compareOp, value, null, rowMutations);
         } catch (Exception e) {
             logger.error(LCD.convert("01-00005"), delete, tableNameString, e);
             throw new IOException("checkAndDelete type table:" + tableNameString + " e.msg:"
@@ -972,26 +1033,29 @@ public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier,
                                   CompareFilter.CompareOp compareOp, byte[] value,
                                   RowMutations rowMutations) throws IOException {
         try {
-            return checkAndMutation(row, family, qualifier, compareOp, value, rowMutations);
+            return checkAndMutation(row, family, qualifier, compareOp, value, null, rowMutations);
         } catch (Exception e) {
             logger.error(LCD.convert("01-00005"), rowMutations, tableNameString, e);
             throw new IOException("checkAndMutate type table:" + tableNameString + " e.msg:"
                                   + e.getMessage() + " error.", e);
         }
     }
 
+    @Override
+    public CheckAndMutateBuilder checkAndMutate(byte[]row , byte[] family) {
+        return new ObCheckAndMutateBuilderImpl(row, family);
+    }
+
     private boolean checkAndMutation(byte[] row, byte[] family, byte[] qualifier, CompareFilter.CompareOp compareOp, byte[] value,
-                                     RowMutations rowMutations) throws Exception {
+                                     TimeRange timeRange, RowMutations rowMutations) throws Exception {
             checkArgument(row != null, "row is null");
             checkArgument(isNotBlank(Bytes.toString(family)), "family is blank");
             checkArgument(Bytes.equals(row, rowMutations.getRow()),
-                "mutation row is not equal check row");
-
+                    "mutation row is not equal check row");
             checkArgument(!rowMutations.getMutations().isEmpty(), "mutation is empty");
-
             byte[] filterString = buildCheckAndMutateFilterString(family, qualifier, compareOp, value);
 
-            ObHTableFilter filter = buildObHTableFilter(filterString, null, 1, qualifier);
+            ObHTableFilter filter = buildObHTableFilter(filterString, timeRange, 1, qualifier);
             List<Mutation> mutations = rowMutations.getMutations();
             List<Cell> keyValueList = new LinkedList<>();
             // only one family operation is allowed
@@ -1320,6 +1384,16 @@ public int getRpcTimeout() {
         return this.rpcTimeout;
     }
 
+    @Override
+    public int getReadRpcTimeout() {
+        return this.readRpcTimeout;
+    }
+
+    @Override
+    public int getWriteRpcTimeout() {
+        return this.writeRpcTimeout;
+    }
+
     public void setRuntimeBatchExecutor(ExecutorService runtimeBatchExecutor) {
         this.obTableClient.setRuntimeBatchExecutor(runtimeBatchExecutor);
     }
@@ -1841,4 +1915,109 @@ public static OHOpType getDeleteType(Cell.Type type) {
                 throw new IllegalArgumentException("illegal mutation type " + type);
         }
     }
+
+    private class ObCheckAndMutateBuilderImpl implements CheckAndMutateBuilder {
+        private final byte[] row;
+        private final byte[] family;
+        private byte[] qualifier;
+        private byte[] value;
+        private TimeRange timeRange;
+        private CompareOperator cmpOp;
+
+        ObCheckAndMutateBuilderImpl(byte[] row, byte[] family) {
+            this.row = checkNotNull(row, "The provided row is null.");
+            this.family = checkNotNull(family, "The provided family is null.");
+        }
+
+        @Override
+        public CheckAndMutateBuilder qualifier(byte[] qualifier) {
+            this.qualifier = checkNotNull(qualifier, "The provided qualifier is null. You could" +
+                    " use an empty byte array, or do not call this method if you want a null qualifier.");
+            return this;
+        }
+
+        @Override
+        public CheckAndMutateBuilder timeRange(TimeRange timeRange) {
+            this.timeRange = timeRange;
+            return this;
+        }
+
+        @Override
+        public CheckAndMutateBuilder ifNotExists() {
+            this.cmpOp = CompareOperator.EQUAL;
+            this.value = null;
+            return this;
+        }
+
+        @Override
+        public CheckAndMutateBuilder ifMatches(CompareOperator cmpOp, byte[] value) {
+            this.cmpOp = checkNotNull(cmpOp, "The provided cmpOp is null.");
+            this.value = checkNotNull(value , "The provided value is null.");
+            return this;
+        }
+
+        @Override
+        public boolean thenPut(Put put) throws IOException {
+            checkCmpOp();
+            RowMutations rowMutations = new RowMutations(row);
+            rowMutations.add(put);
+            try {
+                return checkAndMutation(row, family, qualifier, getCompareOp(cmpOp), value, timeRange, rowMutations);
+            } catch(Exception e) {
+                logger.error(LCD.convert("01-00005"), rowMutations, tableNameString, e);
+                throw new IOException("checkAndMutate type table: " + tableNameString + " e.msg: "
+                        + e.getMessage() + " error.", e);
+            }
+        }
+
+        @Override
+        public boolean thenDelete(Delete delete) throws IOException {
+            checkCmpOp();
+            RowMutations rowMutations = new RowMutations(row);
+            rowMutations.add(delete);
+            try {
+                return checkAndMutation(row, family, qualifier, getCompareOp(cmpOp), value, timeRange, rowMutations);
+            } catch(Exception e) {
+                logger.error(LCD.convert("01-00005"), rowMutations, tableNameString, e);
+                throw new IOException("checkAndMutate type table: " + tableNameString + " e.msg: "
+                        + e.getMessage() + " error.", e);
+            }
+        }
+
+        @Override
+        public boolean thenMutate(RowMutations mutation) throws IOException {
+            checkCmpOp();
+            try {
+                return checkAndMutation(row, family, qualifier, getCompareOp(cmpOp), value, timeRange, mutation);
+            } catch(Exception e) {
+                logger.error(LCD.convert("01-00005"), mutation, tableNameString, e);
+                throw new IOException("checkAndMutate type table: " + tableNameString + " e.msg: "
+                        + e.getMessage() + " error.", e);
+            }
+        }
+
+        private void checkCmpOp() {
+            checkNotNull(this.cmpOp, "The compare condition is null. Please use"
+                    + " ifNotExists/ifEquals/ifMatches before executing the request");
+        }
+
+        private CompareFilter.CompareOp getCompareOp(CompareOperator cmpOp) {
+            switch (cmpOp) {
+                case LESS:
+                    return CompareFilter.CompareOp.LESS;
+                case LESS_OR_EQUAL:
+                    return CompareFilter.CompareOp.LESS_OR_EQUAL;
+                case EQUAL:
+                    return CompareFilter.CompareOp.EQUAL;
+                case NOT_EQUAL:
+                    return CompareFilter.CompareOp.NOT_EQUAL;
+                case GREATER_OR_EQUAL:
+                    return CompareFilter.CompareOp.GREATER_OR_EQUAL;
+                case GREATER:
+                    return CompareFilter.CompareOp.GREATER;
+                default:
+                    return CompareFilter.CompareOp.NO_OP;
+            }
+        }
+    }
 }

src/main/java/com/alipay/oceanbase/hbase/OHTableClient.java

@@ -173,6 +173,12 @@ public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier,
         return ohTable.checkAndMutate(row, family, qualifier, compareOp, value, mutations);
     }
 
+    @Override
+    public CheckAndMutateBuilder checkAndMutate(byte[] row, byte[] family) {
+        checkStatus();
+        return ohTable.checkAndMutate(row, family);
+    }
+
     @Override
     public void setOperationTimeout(int i) {
         checkStatus();

src/main/java/com/alipay/oceanbase/hbase/OHTablePool.java

@@ -826,6 +826,11 @@ public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier,
             return table.checkAndDelete(row, family, qualifier, compareOp, value, delete);
         }
 
+        @Override
+        public CheckAndMutateBuilder checkAndMutate(byte[] row, byte[] family) {
+            return table.checkAndMutate(row, family);
+        }
+
         @Override
         public Result increment(Increment increment) throws IOException {
             return table.increment(increment);

src/main/java/com/alipay/oceanbase/hbase/util/OHConnectionConfiguration.java

@@ -47,10 +47,13 @@ public class OHConnectionConfiguration {
     private final boolean    odpMode;
     private final long       writeBufferSize;
     private final int        operationTimeout;
+    private final int        metaOperationTimeout;
     private final int        scannerCaching;
     private final long       scannerMaxResultSize;
     private final int        maxKeyValueSize;
     private final int        rpcTimeout;
+    private final int        readRpcTimeout;
+    private final int        writeRpcTimeout;
     private final int        rpcConnectTimeout;
     private final long       writeBufferPeriodicFlushTimeoutMs;
     private final long       writeBufferPeriodicFlushTimerTickMs;
@@ -70,10 +73,16 @@ public OHConnectionConfiguration(Configuration conf) {
         }
         this.database = database;
         this.writeBufferSize = conf.getLong(WRITE_BUFFER_SIZE_KEY, WRITE_BUFFER_SIZE_DEFAULT);
+        this.metaOperationTimeout = conf.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT,
+            HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
         this.operationTimeout = conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
             HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
         this.rpcTimeout = conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
             HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
+        this.readRpcTimeout = conf.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY,
+            HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
+        this.writeRpcTimeout = conf.getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
+            HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
         this.writeBufferPeriodicFlushTimeoutMs = conf.getLong(
             WRITE_BUFFER_PERIODIC_FLUSH_TIMEOUT_MS, WRITE_BUFFER_PERIODIC_FLUSH_TIMEOUT_MS_DEFAULT);
         this.writeBufferPeriodicFlushTimerTickMs = conf.getLong(
@@ -117,6 +126,10 @@ public long getWriteBufferSize() {
         return this.writeBufferSize;
     }
 
+    public int getMetaOperationTimeout() {
+        return this.metaOperationTimeout;
+    }
+
     public int getOperationTimeout() {
         return this.operationTimeout;
     }
@@ -133,6 +146,14 @@ public int getRpcTimeout() {
         return this.rpcTimeout;
     }
 
+    public int getReadRpcTimeout() {
+        return this.readRpcTimeout;
+    }
+
+    public int getWriteRpcTimeout() {
+        return this.writeRpcTimeout;
+    }
+
     public int getRpcConnectTimeout() {
         return this.rpcConnectTimeout;
     }