support weak read (#406)

Author: WeiXinChan

src/main/java/com/alipay/oceanbase/rpc/ObClusterTableBatchOps.java

@@ -184,6 +184,12 @@ public void setReturnOneResult(boolean returnOneResult) {
         tableBatchOps.setReturnOneResult(returnOneResult);
     }
 
+    @Override
+    public void setIsWeakRead(boolean isWeakRead) {
+        super.setIsWeakRead(isWeakRead);
+        tableBatchOps.setIsWeakRead(isWeakRead);
+    }
+
     void preCheck() {
         List<ObTableOperation> operations = this.tableBatchOps.getObTableBatchOperation()
             .getTableOperations();

src/main/java/com/alipay/oceanbase/rpc/ObClusterTableQuery.java

@@ -271,6 +271,26 @@ public TableQuery setSearchText(String searchText) {
         return this;
     }
 
+    @Override
+    public TableQuery setReadConsistency(String readConsistency) {
+        // 同时设置父类和内部 tableClientQuery 的 readConsistency
+        super.setReadConsistency(readConsistency);
+        tableClientQuery.setReadConsistency(readConsistency);
+        return this;
+    }
+
+    @Override
+    public String getReadConsistency() {
+        // 返回内部 tableClientQuery 的 readConsistency
+        return tableClientQuery.getReadConsistency();
+    }
+
+    @Override
+    public TableQuery setScanRangeColumns(String... columns) {
+        tableClientQuery.setScanRangeColumns(columns);
+        return this;
+    }
+
     public void setAllowDistributeScan(boolean allowDistributeScan) {
         tableClientQuery.setAllowDistributeScan(allowDistributeScan);
     }

src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java

@@ -17,8 +17,6 @@
 
 package com.alipay.oceanbase.rpc.bolt.protocol;
 
-import com.alipay.oceanbase.rpc.exception.ObTableRoutingWrongException;
-import com.alipay.oceanbase.rpc.meta.ObTableMetaRequest;
 import com.alipay.oceanbase.rpc.meta.ObTableMetaResponse;
 import com.alipay.oceanbase.rpc.protocol.packet.ObRpcPacketHeader;
 import com.alipay.oceanbase.rpc.protocol.payload.ObPayload;
@@ -36,9 +34,6 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.login.ObTableLoginResult;
 import com.alipay.remoting.CommandCode;
 
-import static com.alipay.oceanbase.rpc.protocol.payload.Pcodes.OB_TABLE_API_HBASE_EXECUTE;
-import static com.alipay.oceanbase.rpc.protocol.payload.Pcodes.OB_TABLE_API_META_INFO_EXECUTE;
-
 public enum ObTablePacketCode implements CommandCode {
 
     OB_TABLE_API_LOGIN(Pcodes.OB_TABLE_API_LOGIN) {

src/main/java/com/alipay/oceanbase/rpc/bolt/protocol/ObTablePacketCode.java

@@ -22,7 +22,6 @@
 import com.alipay.remoting.InvokeContext;
 import com.alipay.remoting.InvokeFuture;
 import com.alipay.remoting.RemotingCommand;
-import com.alipay.oceanbase.rpc.exception.ObTableTimeoutExcetion;
 import io.netty.util.Timeout;
 
 import java.net.InetSocketAddress;

src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObClientFuture.java

@@ -119,7 +119,8 @@ private boolean connect() throws Exception {
 
         if (tries >= maxTryTimes) {
             if (!obTable.isOdpMode()) {
-                RouteTableRefresher.SuspectObServer suspectAddr = new RouteTableRefresher.SuspectObServer(obTable.getObServerAddr());
+                RouteTableRefresher.SuspectObServer suspectAddr = new RouteTableRefresher.SuspectObServer(
+                    obTable.getObServerAddr());
                 RouteTableRefresher.addIntoSuspectIPs(suspectAddr);
             }
             LOGGER.warn("connect failed after max " + maxTryTimes + " tries "

src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableConnection.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.exception;
 
 public class ObTableSessionNotExistException extends ObTableException {

src/main/java/com/alipay/oceanbase/rpc/exception/ObTableSessionNotExistException.java

@@ -20,22 +20,17 @@
 import com.alipay.oceanbase.rpc.ObTableClient;
 import com.alipay.oceanbase.rpc.mutation.ColumnValue;
 import com.alipay.oceanbase.rpc.mutation.Row;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableConsistencyLevel;
 import com.alipay.oceanbase.rpc.table.api.Table;
 
 import java.util.Map;
 
 public class Get {
-    private Table      client;
-    private String     tableName;
-    protected Row      rowKey;
-    protected String[] selectColumns;
-
-    public Get() {
-        tableName = null;
-        client = null;
-        rowKey = null;
-        selectColumns = null;
-    }
+    private Table    client          = null;
+    private String   tableName       = null;
+    private Row      rowKey          = null;
+    private String[] selectColumns   = null;
+    private String   readConsistency = "";
 
     public Get(Table client, String tableName) {
         this.client = client;
@@ -67,14 +62,27 @@ public Get select(String... columns) {
         return this;
     }
 
+    public Get setReadConsistency(String readConsistency) {
+        this.readConsistency = readConsistency;
+        return this;
+    }
+
+    public String getReadConsistency() {
+        return readConsistency;
+    }
+
     public String[] getSelectColumns() {
         return selectColumns;
     }
 
     public Map<String, Object> execute() throws Exception {
+        ObTableConsistencyLevel readConsistency = null;
+        if (this.readConsistency != null && !this.readConsistency.isEmpty()) {
+            readConsistency = ObTableConsistencyLevel.getByName(this.readConsistency);
+        }
         if (client == null) {
             throw new IllegalArgumentException("client is null");
         }
-        return ((ObTableClient) client).get(tableName, rowKey, selectColumns);
+        return ((ObTableClient) client).get(tableName, rowKey, selectColumns, readConsistency);
     }
 }