diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java index 9c2f27c0..c5b6218f 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java +++ b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java @@ -2271,6 +2271,7 @@ 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 @@ -2278,6 +2279,7 @@ public ObPayload execute(final ObTableAbstractOperationRequest request) throws E 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(); @@ -2387,6 +2389,15 @@ public ObPayload execute(final ObTableAbstractOperationRequest request) throws E } else { if (ex instanceof ObTableException && (((ObTableException) ex).isNeedRefreshTableEntry() || ((ObTableException) ex).isNeedRetryError())) { + if (ex instanceof ObTableNotExistException) { + String logMessage = String.format( + "exhaust retry while meet TableNotExist Exception, table name: %s, errorCode: %d", + request.getTableName(), + ((ObTableException) ex).getErrorCode() + ); + logger.warn(logMessage, ex); + throw ex; + } logger.warn( "tablename:{} partition id:{} batch ops refresh table while meet ObTableMasterChangeException, errorCode: {}", request.getTableName(), routeTabletId, ((ObTableException) ex).getErrorCode(), ex); diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java index e1ddc69e..ce5dde92 100644 --- 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 new file mode 100644 index 00000000..d667ad78 --- /dev/null +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/OHOperationType.java @@ -0,0 +1,100 @@ +/*- + * #%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.*; + +public enum OHOperationType { + 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 map = new HashMap(); + + static { + for (OHOperationType type : OHOperationType.values()) { + map.put(type.value, type); + } + } + + OHOperationType(int value) { + this.value = value; + } + + public static OHOperationType valueOf(int value) { + return map.get(value); + } + + public int getValue() { + return value; + } + + 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 index 6f423694..027c40f4 100644 --- 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,6 +54,7 @@ public class ObHbaseRequest extends AbstractPayload implements Credentialable { protected ObTableOperationType opType; protected List keys = new ArrayList<>(); protected List cfRows; + protected OHOperationType hbaseOpType = OHOperationType.INVALID; public ObHbaseRequest() { this.credential = new ObBytesString(); @@ -112,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 (" + @@ -151,6 +156,7 @@ public long getPayloadContentSize() { for (ObHbaseCfRows cfRows : cfRows) { payLoadContentSize += cfRows.getPayloadSize(); } + payLoadContentSize += 1; // hbase_op_type_ } return payLoadContentSize; } @@ -184,6 +190,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 index d037a1b4..309f7318 100644 --- 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,14 @@ public void setNeedTabletId(boolean needTabletId) { option_flag.setNeedTabletId(needTabletId); } + 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/ObTableLSOpRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java index fdb8b89e..a8589155 100644 --- 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 index f72e764e..85bd0a2c 100644 --- 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/AbstractQueryStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java index 213b1f67..ffe8697f 100644 --- 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__) protected String indexTableName; protected ObTableEntityType entityType; + protected OHOperationType hbaseOpType = OHOperationType.INVALID; protected Map> expectant; protected List cacheProperties = new LinkedList(); protected LinkedList> cacheRows = new LinkedList>(); @@ -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/protocol/payload/impl/execute/query/ObTableQueryRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryRequest.java index 1bdbc99b..3382b4f5 100644 --- 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/stream/ObTableClientQueryAsyncStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java index 9c2bc4ec..508adf76 100644 --- 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 index 3ae915f3..3dce44c2 100644 --- 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 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/ObTableClientLSBatchOpsImpl.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java index b5840cce..08b4c34d 100644 --- 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 batchOperation; /* @@ -94,6 +95,10 @@ public List 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; diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientQueryImpl.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientQueryImpl.java index 64123e5d..50baca59 100644 --- 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 { @@ -445,4 +448,8 @@ public Long getPartId() { public void setAllowDistributeScan(boolean allowDistributeScan) { this.allowDistributeScan = allowDistributeScan; } + + public void setHbaseOpType(OHOperationType hbaseOpType) { + this.hbaseOpType = hbaseOpType; + } }