set hbaseOpType to requests and encode to server

JackShi148 · JackShi148 · commit 493906539b3f · 2025-10-24T17:12:37.000+08:00
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java
@@ -24,6 +24,7 @@
 import com.alipay.oceanbase.rpc.get.Get;
 import com.alipay.oceanbase.rpc.mutation.result.BatchOperationResult;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
+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.ObTableOperationType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.mutate.ObTableQueryAndMutate;
@@ -53,6 +54,7 @@ public class BatchOperation {
     ObTableOperationType        lastType         = ObTableOperationType.INVALID;
     boolean                     isSameType       = true;
     protected ObTableEntityType entityType       = ObTableEntityType.KV;
+    protected OHOperationType   hbaseOpType      = OHOperationType.INVALID;
 
     /*
      * default constructor
@@ -90,6 +92,10 @@ public BatchOperation setTable(String tableName) {
         return this;
     }
 
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
+
     /*
      * add queries
      */
@@ -325,6 +331,7 @@ private BatchOperationResult executeWithLSBatchOp() throws Exception {
             batchOps.setEntityType(entityType);
             batchOps.setServerCanRetry(serverCanRetry);
             batchOps.setNeedTabletId(needTabletId);
+            batchOps.setHbaseOpType(hbaseOpType);
             for (Object operation : operations) {
                 if (operation instanceof CheckAndInsUp) {
                     checkAndInsUpCnt++;
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
@@ -3,24 +3,25 @@
 import java.util.*;
 
 public enum OHOperationType {
-    PUT(0),
-    PUT_LIST(1),
-    DELETE(2),
-    DELETE_LIST(3),
-    GET(4),
-    GET_LIST(5),
-    EXISTS(6),
-    EXISTS_LIST(7),
-    BATCH(8),
-    BATCH_CALLBACK(9),
-    SCAN(10),
-    CHECK_AND_PUT(11),
-    CHECK_AND_DELETE(12),
-    CHECK_AND_MUTATE(13),
-    APPEND(14),
-    INCREMENT(15),
-    INCREMENT_COLUMN_VALUE(16),
-    MUTATE_ROW(17);
+    INVALID(0),
+    PUT(1),
+    PUT_LIST(2),
+    DELETE(3),
+    DELETE_LIST(4),
+    GET(5),
+    GET_LIST(6),
+    EXISTS(7),
+    EXISTS_LIST(8),
+    BATCH(9),
+    BATCH_CALLBACK(10),
+    SCAN(11),
+    CHECK_AND_PUT(12),
+    CHECK_AND_DELETE(13),
+    CHECK_AND_MUTATE(14),
+    APPEND(15),
+    INCREMENT(16),
+    INCREMENT_COLUMN_VALUE(17),
+    MUTATE_ROW(18);
 
     private final int                                  value;
     private static final Map<Integer, OHOperationType> map = new HashMap<Integer, OHOperationType>();
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
@@ -53,6 +53,7 @@ public class ObHbaseRequest extends AbstractPayload implements Credentialable {
     protected ObTableOperationType    opType;
     protected List<ObObj>             keys       = new ArrayList<>();
     protected List<ObHbaseCfRows>     cfRows;
+    protected OHOperationType         hbaseOpType;
 
     public ObHbaseRequest() {
         this.credential = new ObBytesString();
@@ -184,6 +185,10 @@ public void setServerCanRetry(boolean canRetry) {
         optionFlag.setFlagServerCanRetry(canRetry);
     }
 
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
+
     public boolean getServerCanRetry() {
         return optionFlag.getFlagServerCanRetry();
     }
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
@@ -39,6 +39,7 @@ public abstract class ObTableAbstractOperationRequest extends AbstractPayload im
     protected ObTableOptionFlag       option_flag             = ObTableOptionFlag.DEFAULT;
     protected boolean                 returningAffectedEntity = false;
     protected boolean                 returningAffectedRows   = false;
+    protected OHOperationType         hbaseOpType             = OHOperationType.INVALID; // for table operations, this will be INVALID(0)
 
     /*
      * Get payload content size.
@@ -220,6 +221,10 @@ public void setNeedTabletId(boolean needTabletId) {
         option_flag.setNeedTabletId(needTabletId);
     }
 
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
+
     public boolean getNeedTabletId() {
         return option_flag.isNeedTabletId();
     }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java
@@ -33,13 +33,15 @@
                     credential_,
                     entity_type_,
                     consistency_level_,
-                    ls_op_);
+                    ls_op_,
+                    hbase_op_type_);
  */
 public class ObTableLSOpRequest extends AbstractPayload implements Credentialable {
     protected ObBytesString           credential;
     protected ObTableEntityType       entityType       = ObTableEntityType.KV;
     protected ObTableConsistencyLevel consistencyLevel = ObTableConsistencyLevel.STRONG;
     private ObTableLSOperation        lsOperation      = null;
+    protected OHOperationType         hbaseOpType      = OHOperationType.INVALID;
 
     /*
      * Get pcode.
@@ -70,6 +72,9 @@ public byte[] encode() {
 
         // 4. encode lsOperation
         lsOperation.encode(buf);
+
+        // 5. encode hbase op type, for table operations, this will be INVALID(0)
+        Serialization.encodeI8(buf, hbaseOpType.getByteValue());
         if (buf.pos != buf.bytes.length) {
             throw new IllegalArgumentException("error in encode lsOperationRequest (" +
                     "pos:" + buf.pos + ", buf.capacity:" + buf.bytes.length + ")");
@@ -99,7 +104,7 @@ public Object decode(ByteBuf buf) {
     public long getPayloadContentSize() {
         if (payLoadContentSize == INVALID_PAYLOAD_CONTENT_SIZE) {
             payLoadContentSize = lsOperation.getPayloadSize() + Serialization.getNeedBytes(credential) + 1 // entityType
-                    + 1; // consistencyLevel
+                    + 1 /* consistencyLevel */ + 1 /* hbaseOpType */;
         }
         return payLoadContentSize;
     }
@@ -161,6 +166,10 @@ public void setTableId(long tableId) {
         this.lsOperation.setTableId(tableId);
     }
 
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
+
     /**
      * Reset the cached payload content size and propagate to child objects
      */
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/mutate/ObTableQueryAndMutateRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/mutate/ObTableQueryAndMutateRequest.java
@@ -35,7 +35,8 @@
      entity_type_,
      query_and_mutate_,
      binlog_row_image_type_,
-     option_flag_);
+     option_flag_,
+     hbase_op_type_);
  *
  */
 public class ObTableQueryAndMutateRequest extends ObTableAbstractOperationRequest {
@@ -78,6 +79,9 @@ public byte[] encode() {
         idx += len;
         System.arraycopy(Serialization.encodeI8(option_flag.getByteValue()), 0, bytes, idx, 1);
 
+        idx += 1;
+        System.arraycopy(Serialization.encodeI8(hbaseOpType.getByteValue()), 0, bytes, idx, 1);
+
         return bytes;
     }
 
@@ -111,7 +115,7 @@ public long getPayloadContentSize() {
         if (ObGlobal.obVsnMajor() >= 4)
             return Serialization.getNeedBytes(credential) + Serialization.getNeedBytes(tableName)
                    + Serialization.getNeedBytes(tableId) + 8 + 1
-                   + tableQueryAndMutate.getPayloadSize() + Serialization.getNeedBytes(type.getValue()) + 1;
+                   + tableQueryAndMutate.getPayloadSize() + Serialization.getNeedBytes(type.getValue()) + 1 + 1;
         else
             return Serialization.getNeedBytes(credential) + Serialization.getNeedBytes(tableName)
                    + Serialization.getNeedBytes(tableId) + Serialization.getNeedBytes(partitionId)
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryRequest.java
@@ -34,7 +34,9 @@
      partition_id_,
      entity_type_,
      consistency_level_,
-     query_
+     query_,
+     option_flag_,
+     hbase_op_type_
      );
  *
  */
@@ -76,6 +78,9 @@ public byte[] encode() {
         idx += len;
         System.arraycopy(Serialization.encodeVi64(option_flag.getValue()), 0, bytes, idx, 1);
 
+        idx += 1;
+        System.arraycopy(Serialization.encodeI8(hbaseOpType.getByteValue()), 0, bytes, idx, 1);
+
         return bytes;
     }
 
@@ -109,7 +114,7 @@ public Object decode(ByteBuf buf) {
     public long getPayloadContentSize() {
         if (ObGlobal.obVsnMajor() >= 4)
             return Serialization.getNeedBytes(credential) + Serialization.getNeedBytes(tableName)
-                   + Serialization.getNeedBytes(tableId) + 8 + 2 + tableQuery.getPayloadSize() + 1;
+                   + Serialization.getNeedBytes(tableId) + 8 + 2 + tableQuery.getPayloadSize() + 1 + 1;
         else
             return Serialization.getNeedBytes(credential) + Serialization.getNeedBytes(tableName)
                    + Serialization.getNeedBytes(tableId) + Serialization.getNeedBytes(partitionId)
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java
@@ -69,7 +69,8 @@ public class ObTableClientLSBatchOpsImpl extends AbstractTableBatchOps {
     private boolean               returningAffectedEntity = false;
     private boolean               needAllProp             = false;
     private boolean               serverCanRetry          = false;
-    private boolean               needTabletId  = false;
+    private boolean               needTabletId            = false;
+    protected OHOperationType     hbaseOpType             = OHOperationType.INVALID;
     private List<ObTableSingleOp> batchOperation;
 
     /*
@@ -94,6 +95,10 @@ public List<ObTableSingleOp> getSingleOperations() {
         return batchOperation;
     }
 
+    public void setHbaseOpType(OHOperationType hbaseOpType) {
+        this.hbaseOpType = hbaseOpType;
+    }
+
     /*
      * Get.
      */
@@ -591,6 +596,7 @@ public void partitionExecute(ObTableSingleOpResult[] results,
        tableLsOpRequest.setTableId(tableId);
        tableLsOpRequest.setEntityType(entityType);
        tableLsOpRequest.setTimeout(operationTimeout);
+       tableLsOpRequest.setHbaseOpType(hbaseOpType);
 
         ObTableLSOpResult subLSOpResult;
         boolean needRefreshPartitionLocation = false;
