set hbase op type to query

JackShi148 · JackShi148 · commit 7305dd82d756 · 2025-11-12T14:29:46.000+08:00
diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java
@@ -2271,13 +2271,15 @@ public ObPayload execute(final ObTableAbstractOperationRequest request) throws E
             ObTableClientQueryImpl tableQuery = new ObTableClientQueryImpl(tableName,
                     ((ObTableQueryRequest) request).getTableQuery(), this);
             tableQuery.setEntityType(request.getEntityType());
+            tableQuery.setHbaseOpType(request.getHbaseOpType());
             return new ObClusterTableQuery(tableQuery).executeInternal();
         } else if (request instanceof ObTableQueryAsyncRequest) {
             // TableGroup -> TableName
             String tableName = request.getTableName();
             ObTableClientQueryImpl tableQuery = new ObTableClientQueryImpl(tableName,
                     ((ObTableQueryAsyncRequest) request).getObTableQueryRequest().getTableQuery(), this);
             tableQuery.setEntityType(request.getEntityType());
+            tableQuery.setHbaseOpType(request.getHbaseOpType());
             ObClusterTableQuery clusterTableQuery = new ObClusterTableQuery(tableQuery);
             clusterTableQuery.setAllowDistributeScan(((ObTableQueryAsyncRequest) request).isAllowDistributeScan());
             return clusterTableQuery.asyncExecuteInternal();
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/OHOperationType.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/OHOperationType.java
@@ -1,3 +1,20 @@
+/*-
+ * #%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.protocol.payload.impl.execute;
 
 import java.util.*;
@@ -47,4 +64,37 @@ public int getValue() {
     public byte getByteValue() {
         return (byte) value;
     }
+
+    /*
+     * CHECK_AND_PUT -> checkAndPut
+     * PUT -> put
+     */
+    public String toCamelCase() {
+        String name = this.name();
+        if (name == null || name.isEmpty()) {
+            return name;
+        }
+
+        String[] parts = name.split("_");
+        StringBuilder sb = new StringBuilder();
+
+        for (int i = 0; i < parts.length; i++) {
+            String part = parts[i];
+            if (part == null || part.isEmpty()) {
+                continue;
+            }
+
+            if (i == 0) {
+                sb.append(part.toLowerCase());
+            } else {
+                if (!part.isEmpty()) {
+                    sb.append(Character.toUpperCase(part.charAt(0)));
+                    if (part.length() > 1) {
+                        sb.append(part.substring(1).toLowerCase());
+                    }
+                }
+            }
+        }
+        return sb.toString();
+    }
 }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbaseRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbaseRequest.java
@@ -38,7 +38,8 @@
                     option_flag_,
                     op_type_,
                     keys_,
-                    cf_rows_);
+                    cf_rows_,
+                    hbase_op_type_);
  */
 /*
     [k1][k2][k3]...
@@ -53,7 +54,7 @@ public class ObHbaseRequest extends AbstractPayload implements Credentialable {
     protected ObTableOperationType    opType;
     protected List<ObObj>             keys       = new ArrayList<>();
     protected List<ObHbaseCfRows>     cfRows;
-    protected OHOperationType         hbaseOpType;
+    protected OHOperationType         hbaseOpType = OHOperationType.INVALID;
 
     public ObHbaseRequest() {
         this.credential = new ObBytesString();
@@ -113,6 +114,9 @@ public byte[] encode() {
             ObHbaseCfRows sameCfRows = cfRows.get(i);
             sameCfRows.encode(buf);
         }
+
+        // 7. encode hbase op type, to differentiate put and put list
+        Serialization.encodeI8(buf, hbaseOpType.getByteValue());
         
         if (buf.pos != buf.bytes.length) {
             throw new IllegalArgumentException("error in encode ObHbaseRequest (" +
@@ -152,6 +156,7 @@ public long getPayloadContentSize() {
             for (ObHbaseCfRows cfRows : cfRows) {
                 payLoadContentSize += cfRows.getPayloadSize();
             }
+            payLoadContentSize += 1; // hbase_op_type_
         }
         return payLoadContentSize;
     }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableAbstractOperationRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableAbstractOperationRequest.java
@@ -225,6 +225,10 @@ public void setHbaseOpType(OHOperationType hbaseOpType) {
         this.hbaseOpType = hbaseOpType;
     }
 
+    public OHOperationType getHbaseOpType() {
+        return hbaseOpType;
+    }
+
     public boolean getNeedTabletId() {
         return option_flag.isNeedTabletId();
     }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java
@@ -30,11 +30,8 @@
 import com.alipay.oceanbase.rpc.protocol.payload.Pcodes;
 import com.alipay.oceanbase.rpc.protocol.payload.ResultCodes;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
-import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableApiMove;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.*;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObRowKey;
-import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType;
-import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableStreamRequest;
-import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.QueryStreamResult;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.syncquery.ObTableQueryAsyncRequest;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.syncquery.ObTableQueryAsyncResult;
 import com.alipay.oceanbase.rpc.table.ObTable;
@@ -65,6 +62,7 @@ public abstract class AbstractQueryStreamResult extends AbstractPayload implemen
     // global index: key is index table name (be like: __idx_<data_table_id>_<index_name>)
     protected String                                                           indexTableName;
     protected ObTableEntityType                                                entityType;
+    protected OHOperationType                                                  hbaseOpType = OHOperationType.INVALID;
     protected Map<Long, ObPair<Long, ObTableParam>>                            expectant;
     protected List<String>                                                     cacheProperties     = new LinkedList<String>();
     protected LinkedList<List<ObObj>>                                          cacheRows           = new LinkedList<List<ObObj>>();
@@ -832,4 +830,12 @@ public ObTableClient getClient() {
     public void setClient(ObTableClient client) {
         this.client = client;
     }
+
+    public OHOperationType getHbaseOpType() {
+        return hbaseOpType;
+    }
+
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
 }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java
@@ -58,6 +58,7 @@ public void init() throws Exception {
         request.setTableQuery(tableQuery);
         request.setEntityType(entityType);
         request.setConsistencyLevel(getReadConsistency().toObTableConsistencyLevel());
+        request.setHbaseOpType(hbaseOpType);
 
         // construct async query request
         asyncRequest.setObTableQueryRequest(request);
diff --git a/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryStreamResult.java
@@ -50,6 +50,7 @@ protected ObTableQueryResult referToNewPartition(ObPair<Long, ObTableParam> part
         request.setPartitionId(partitionId);
         request.setTableId(partIdWithObTable.getRight().getTableId());
         request.setEntityType(entityType);
+        request.setHbaseOpType(hbaseOpType);
         if (operationTimeout > 0) {
             request.setTimeout(operationTimeout);
         } else {
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientQueryImpl.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientQueryImpl.java
@@ -28,6 +28,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.ObPayload;
 import com.alipay.oceanbase.rpc.protocol.payload.ResultCodes;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObRowKey;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.OHOperationType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.aggregation.ObTableAggregationType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.*;
@@ -53,6 +54,7 @@ public class ObTableClientQueryImpl extends AbstractTableQueryImpl {
     private Row                                   rowKey;                    // only used by BatchOperation
 
     private boolean                               allowDistributeScan = true;
+    private OHOperationType                       hbaseOpType = OHOperationType.INVALID;
 
     /*
      * Add aggregation.
@@ -162,6 +164,7 @@ private void setCommonParams2Result(AbstractQueryStreamResult result) throws Exc
         result.setExpectant(partitionObTables);
         result.setOperationTimeout(operationTimeout);
         result.setReadConsistency(obTableClient.getReadConsistency());
+        result.setHbaseOpType(hbaseOpType);
     }
 
     private abstract static class InitQueryResultCallback<T> {
@@ -445,4 +448,8 @@ public Long getPartId() {
     public void setAllowDistributeScan(boolean allowDistributeScan) {
         this.allowDistributeScan = allowDistributeScan;
     }
+
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -225,6 +225,10 @@ public void setHbaseOpType(OHOperationType hbaseOpType) {`
`225`	`225`	`this.hbaseOpType = hbaseOpType;`
`226`	`226`	`}`
`227`	`227`
	`228`	`+ public OHOperationType getHbaseOpType() {`
	`229`	`+ return hbaseOpType;`
	`230`	`+ }`
	`231`	`+`
`228`	`232`	`public boolean getNeedTabletId() {`
`229`	`233`	`return option_flag.isNeedTabletId();`
`230`	`234`	`}`