add query_and_mutate check_and_mutate (#305)

* add query_and_mutate check_and_mutate

* fix review

Author: maochongxin

src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java

@@ -26,6 +26,9 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperationType;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.mutate.ObTableQueryAndMutate;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQuery;
+import com.alipay.oceanbase.rpc.queryandmutate.QueryAndMutate;
 import com.alipay.oceanbase.rpc.table.ObTableClientLSBatchOpsImpl;
 import com.alipay.oceanbase.rpc.table.api.Table;
 import com.alipay.oceanbase.rpc.table.api.TableBatchOps;
@@ -110,6 +113,12 @@ public BatchOperation addOperation(TableQuery... queries) {
         return this;
     }
 
+    public BatchOperation addOperation(QueryAndMutate qm) {
+        lastType = ObTableOperationType.QUERY_AND_MUTATE;
+        this.operations.add(qm);
+        return this;
+    }
+
     /*
      * add get
      */
@@ -336,6 +345,9 @@ private BatchOperationResult executeWithLSBatchOp() throws Exception {
                 } else if (operation instanceof TableQuery) {
                     TableQuery query = (TableQuery) operation;
                     batchOps.addOperation(query);
+                } else if (operation instanceof QueryAndMutate) {
+                    QueryAndMutate qm = (QueryAndMutate) operation;
+                    batchOps.addOperation(qm);
                 } else {
                     throw new IllegalArgumentException(
                         "The operations in batch must be all checkAndInsUp or all non-checkAndInsUp");

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableOperationType.java

@@ -30,7 +30,8 @@ public enum ObTableOperationType {
     TTL(9), // observer internal type, not used by client
     CHECK_AND_INSERT_UP(10), PUT(11), // override row
     TRIGGER(12), // internal op type
-    INVALID(15);
+    REDIS(13), // not used
+    QUERY_AND_MUTATE(14), CHECK_AND_MUTATE(15), INVALID(16);
 
     private int                                       value;
     private static Map<Integer, ObTableOperationType> map             = new HashMap<Integer, ObTableOperationType>();
@@ -46,6 +47,10 @@ public enum ObTableOperationType {
                                                                           false, // TTL
                                                                           true, // CHECK_AND_INSERT_UP
                                                                           false, // PUT
+                                                                          false, // TRIGGER
+                                                                          false, // REDIS
+                                                                          true, // QUERY_AND_MUTATE,
+                                                                          true, // CHECK_AND_MUTATE,
                                                                           false // INVALID
                                                                           );
 

src/main/java/com/alipay/oceanbase/rpc/queryandmutate/QueryAndMutate.java

@@ -0,0 +1,39 @@
+/*-
+ * #%L
+ * com.oceanbase:obkv-table-client
+ * %%
+ * Copyright (C) 2021 - 2025 OceanBase
+ * %%
+ * OBKV Table Client Framework is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *          http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * #L%
+ */
+
+package com.alipay.oceanbase.rpc.queryandmutate;
+
+import com.alipay.oceanbase.rpc.mutation.Mutation;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQuery;
+
+public class QueryAndMutate {
+    private ObTableQuery query;
+    private Mutation     mutation;
+
+    public QueryAndMutate(ObTableQuery query, Mutation mutation) {
+        this.query = query;
+        this.mutation = mutation;
+    }
+
+    public ObTableQuery getQuery() {
+        return query;
+    }
+
+    public Mutation getMutation() {
+        return mutation;
+    }
+}

src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java

@@ -36,6 +36,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.*;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObNewRange;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQuery;
+import com.alipay.oceanbase.rpc.queryandmutate.QueryAndMutate;
 import com.alipay.oceanbase.rpc.threadlocal.ThreadLocalMap;
 import com.alipay.oceanbase.rpc.util.MonitorUtil;
 import com.alipay.oceanbase.rpc.util.TableClientLoggerFactory;
@@ -225,6 +226,31 @@ public void addOperation(TableQuery query) throws Exception {
         addOperation(singleOp);
     }
 
+    public void addOperation(QueryAndMutate queryAndMutate) {
+
+        ObTableSingleOp singleOp = new ObTableSingleOp();
+        ObTableQuery obTableQuery = queryAndMutate.getQuery();
+        if (queryAndMutate.getMutation() instanceof Delete) {
+            Delete delete = (Delete) queryAndMutate.getMutation();
+            ObTableSingleOpQuery singleOpQuery = ObTableSingleOpQuery.getInstance(obTableQuery.getIndexName(),
+                    obTableQuery.getKeyRanges(), obTableQuery.getSelectColumns(),
+                    obTableQuery.getScanOrder(), obTableQuery.isHbaseQuery(),
+                    obTableQuery.gethTableFilter(), obTableQuery.getObKVParams(),
+                    obTableQuery.getFilterString());
+            singleOp.setQuery(singleOpQuery);
+            singleOp.setQuery(singleOpQuery);
+            singleOp.setSingleOpType(ObTableOperationType.QUERY_AND_MUTATE);
+            String[] rowKeyNames = delete.getRowKey().getColumns();
+            Object[] rowKeyValues = delete.getRowKey().getValues();
+            ObTableSingleOpEntity entity = ObTableSingleOpEntity.getInstance(rowKeyNames, rowKeyValues,
+                    null, null);
+            singleOp.addEntity(entity);
+            addOperation(singleOp);
+        } else {
+            throw new ObTableException("invalid operation type " + queryAndMutate.getMutation().getOperationType());
+        }
+    }
+
     public void addOperation(Mutation mutation) throws Exception {
         // entity
         String[] rowKeyNames = null;