Get partition meta for HBase data synchronization and Table batch single-partition operation

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

@@ -17,7 +17,9 @@
 
 package com.alipay.oceanbase.rpc;
 
+import com.alipay.oceanbase.rpc.location.model.partition.Partition;
 import com.alipay.oceanbase.rpc.mutation.Row;
+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.aggregation.ObTableAggregationType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHTableFilter;
@@ -152,6 +154,28 @@ public TableQuery addScanRange(Object[] start, boolean startEquals, Object[] end
         return this;
     }
 
+    @Override
+    public TableQuery addScanRange(Object start, Object end) {
+        if (start instanceof Partition) {
+            Long startPartitionId = ((Partition) start).getPartitionId();
+            Long endPartitionId = ((Partition) end).getPartitionId();
+            if (!startPartitionId.equals(endPartitionId)) {
+                throw new IllegalArgumentException(
+                    "The partition id must be the same for start and end partition in scan range");
+            }
+            Long startPartId = ((Partition) start).getPartId();
+            Long endPartId = ((Partition) end).getPartId();
+            if (!startPartId.equals(endPartId)) {
+                throw new IllegalArgumentException(
+                    "The logic part id must be the same for start and end partition in scan range");
+            }
+            tableClientQuery.setPartId(startPartId);
+            start = ObObj.getMin();
+            end = ObObj.getMax();
+        }
+        return addScanRange(new Object[] { start }, true, new Object[] { end }, true);
+    }
+
     /**
      * Add scan range starts with.
      */

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

@@ -23,6 +23,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.Pcodes;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.direct_load.ObTableDirectLoadResult;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableApiMove;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObFetchPartitionMetaResult;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableBatchOperationResult;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableLSOpResult;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperationResult;
@@ -106,6 +107,15 @@ public ObPayload newPayload(ObRpcPacketHeader header) {
             return new ObTableLSOpResult();
         }
     }, //
+    OB_TABLE_API_PART_META_QUERY(Pcodes.OB_TABLE_API_PART_META_QUERY) {
+        /**
+         * New payload.
+         */
+        @Override
+        public ObPayload newPayload(ObRpcPacketHeader header) {
+            return new ObFetchPartitionMetaResult();
+        }
+    }, //
     OB_TABLE_API_MOVE(Pcodes.OB_TABLE_API_MOVE) {
         /**
          * New payload.
@@ -159,6 +169,8 @@ public static ObTablePacketCode valueOf(short value) {
                 return OB_TABLE_API_DIRECT_LOAD;
             case Pcodes.OB_TABLE_API_LS_EXECUTE:
                 return OB_TABLE_API_LS_EXECUTE;
+            case Pcodes.OB_TABLE_API_PART_META_QUERY:
+                return OB_TABLE_API_PART_META_QUERY;
             case Pcodes.OB_TABLE_API_MOVE:
                 return OB_TABLE_API_MOVE;
             case Pcodes.OB_ERROR_PACKET:

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

@@ -122,7 +122,7 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
             ObRpcResultCode resultCode = new ObRpcResultCode();
             resultCode.decode(buf);
             // If response indicates the request is routed to wrong server, we should refresh the routing meta.
-            if (!conn.getObTable().getReRouting() &&response.getHeader().isRoutingWrong()) {
+            if (!conn.getObTable().isEnableRerouting() && response.getHeader().isRoutingWrong()) {
                 String errMessage = TraceUtil.formatTraceMessage(conn, request,
                     "routed to the wrong server: " + response.getMessage());
                 logger.warn(errMessage);
@@ -139,7 +139,8 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
                     throw new ObTableNeedFetchAllException(errMessage, resultCode.getRcode());
                 }
             }
-            if (resultCode.getRcode() != 0 && response.getHeader().getPcode() != Pcodes.OB_TABLE_API_MOVE) {
+            if (resultCode.getRcode() != 0
+                && response.getHeader().getPcode() != Pcodes.OB_TABLE_API_MOVE) {
                 String errMessage = TraceUtil.formatTraceMessage(conn, request,
                     "routed to the wrong server: " + response.getMessage());
                 logger.warn(errMessage);
@@ -193,23 +194,25 @@ private boolean needFetchAll(int errorCode, int pcode) {
                || errorCode == ResultCodes.OB_TABLE_NOT_EXIST.errorCode
                || errorCode == ResultCodes.OB_TABLET_NOT_EXIST.errorCode
                || errorCode == ResultCodes.OB_LS_NOT_EXIST.errorCode
+               || errorCode == ResultCodes.OB_MAPPING_BETWEEN_TABLET_AND_LS_NOT_EXIST.errorCode
+               || errorCode == ResultCodes.OB_SNAPSHOT_DISCARDED.errorCode
                || (pcode == Pcodes.OB_TABLE_API_LS_EXECUTE && errorCode == ResultCodes.OB_NOT_MASTER.errorCode);
     }
 
     private boolean needFetchPartial(int errorCode) {
         return errorCode == ResultCodes.OB_LOCATION_LEADER_NOT_EXIST.errorCode
-               || errorCode == ResultCodes.OB_NOT_MASTER.errorCode
-               || errorCode == ResultCodes.OB_RS_NOT_MASTER.errorCode
-               || errorCode == ResultCodes.OB_RS_SHUTDOWN.errorCode
-               || errorCode == ResultCodes.OB_RPC_SEND_ERROR.errorCode
-               || errorCode == ResultCodes.OB_RPC_POST_ERROR.errorCode
-               || errorCode == ResultCodes.OB_PARTITION_NOT_EXIST.errorCode
-               || errorCode == ResultCodes.OB_LOCATION_NOT_EXIST.errorCode
-               || errorCode == ResultCodes.OB_PARTITION_IS_STOPPED.errorCode
-               || errorCode == ResultCodes.OB_PARTITION_IS_BLOCKED.errorCode
-               || errorCode == ResultCodes.OB_SERVER_IS_INIT.errorCode
-               || errorCode == ResultCodes.OB_SERVER_IS_STOPPING.errorCode
-               || errorCode == ResultCodes.OB_TRANS_RPC_TIMEOUT.errorCode
-               || errorCode == ResultCodes.OB_NO_READABLE_REPLICA.errorCode;
+                || errorCode == ResultCodes.OB_NOT_MASTER.errorCode
+                || errorCode == ResultCodes.OB_RS_NOT_MASTER.errorCode
+                || errorCode == ResultCodes.OB_RS_SHUTDOWN.errorCode
+                || errorCode == ResultCodes.OB_RPC_SEND_ERROR.errorCode
+                || errorCode == ResultCodes.OB_RPC_POST_ERROR.errorCode
+                || errorCode == ResultCodes.OB_PARTITION_NOT_EXIST.errorCode
+                || errorCode == ResultCodes.OB_LOCATION_NOT_EXIST.errorCode
+                || errorCode == ResultCodes.OB_PARTITION_IS_STOPPED.errorCode
+                || errorCode == ResultCodes.OB_PARTITION_IS_BLOCKED.errorCode
+                || errorCode == ResultCodes.OB_SERVER_IS_INIT.errorCode
+                || errorCode == ResultCodes.OB_SERVER_IS_STOPPING.errorCode
+                || errorCode == ResultCodes.OB_TRANS_RPC_TIMEOUT.errorCode
+                || errorCode == ResultCodes.OB_NO_READABLE_REPLICA.errorCode;
     }
 }

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

@@ -81,6 +81,11 @@ public static ObTableException convertToObTableException(String host, int port,
                                                          + "]" + "[" + resultCodes.name() + "]"
                                                          + "[" + errMsg + "]" + "[" + server + "]"
                                                          + "[" + trace + "]", resultCodes.errorCode);
+        } else if (resultCodes.errorCode == OB_ERR_KV_ROUTE_ENTRY_EXPIRE.errorCode) {
+            return new ObTablePartitionChangeException("[" + String.valueOf(resultCodes.errorCode)
+                                                       + "]" + "[" + resultCodes.name() + "]" + "["
+                                                       + errMsg + "]" + "[" + server + "]" + "["
+                                                       + trace + "]", resultCodes.errorCode);
         } else {
             // [errCode][errCodeName][errMsg][server][trace]
             return new ObTableException("[" + String.valueOf(resultCodes.errorCode) + "]" + "["

src/main/java/com/alipay/oceanbase/rpc/location/LocationUtil.java

@@ -21,7 +21,6 @@
 import com.alibaba.fastjson.JSONException;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.parser.ParserConfig;
-import com.alipay.oceanbase.rpc.ObClusterTableBatchOps;
 import com.alipay.oceanbase.rpc.ObGlobal;
 import com.alipay.oceanbase.rpc.constant.Constants;
 import com.alipay.oceanbase.rpc.exception.*;

src/main/java/com/alipay/oceanbase/rpc/location/model/TableEntry.java

@@ -23,9 +23,10 @@
 import com.alipay.oceanbase.rpc.location.model.partition.ObPartitionLevel;
 import com.alipay.oceanbase.rpc.protocol.payload.Constants;
 
-import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
 
 import static com.google.common.base.Preconditions.checkArgument;
 
@@ -46,14 +47,17 @@ public class TableEntry {
     private ObPartitionInfo                  partitionInfo         = null;
     private volatile long                    refreshTimeMills;
     private volatile long                    refreshAllTimeMills;
+    private volatile long                    odpRefreshTimeMills;
     private Map<String, Integer>             rowKeyElement         = null;
 
     // table location
     private TableLocation                    tableLocation         = null;
     // partition location
     private TableEntryKey                    tableEntryKey         = null;
     private volatile ObPartitionEntry        partitionEntry        = null;
-
+
+    public ConcurrentHashMap<Long, Lock> refreshLockMap = new ConcurrentHashMap<>();
+
     /*
      * Is valid.
      */
@@ -156,6 +160,10 @@ public long getRefreshAllTimeMills() {
         return refreshAllTimeMills;
     }
 
+    public long getOdpRefreshTimeMills() {
+        return odpRefreshTimeMills;
+    }
+
     /*
      * Set refresh time mills.
      */
@@ -170,6 +178,10 @@ public void setRefreshAllTimeMills(long refreshAllTimeMills) {
         this.refreshAllTimeMills = refreshAllTimeMills;
     }
 
+    public void setOdpRefreshTimeMills(long odpRefreshTimeMills) {
+        this.odpRefreshTimeMills = odpRefreshTimeMills;
+    }
+
     public Map<String, Integer> getRowKeyElement() {
         return rowKeyElement;
     }
@@ -220,8 +232,6 @@ public void prepare() throws IllegalArgumentException {
             checkArgument(partitionInfo != null, "partition table partition info is not ready. key"
                                                  + tableEntryKey);
             partitionInfo.prepare();
-            checkArgument(partitionEntry != null,
-                "partition table partition entry is not ready. key" + tableEntryKey);
         }
     }
 

src/main/java/com/alipay/oceanbase/rpc/location/model/partition/ObHashPartDesc.java

@@ -24,6 +24,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjType;
 import com.alipay.oceanbase.rpc.util.RandomUtil;
+import com.alipay.oceanbase.rpc.util.Serialization;
 import com.alipay.oceanbase.rpc.util.TableClientLoggerFactory;
 import com.alipay.oceanbase.rpc.mutation.Row;
 import org.apache.commons.lang.builder.ToStringBuilder;

src/main/java/com/alipay/oceanbase/rpc/location/model/partition/ObKeyPartDesc.java

@@ -24,6 +24,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObColumn;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
 import com.alipay.oceanbase.rpc.util.ObHashUtils;
+import com.alipay.oceanbase.rpc.util.Serialization;
 import com.alipay.oceanbase.rpc.util.TableClientLoggerFactory;
 import org.apache.commons.lang.builder.ToStringBuilder;
 import org.slf4j.Logger;

src/main/java/com/alipay/oceanbase/rpc/location/model/partition/ObPartitionEntry.java

@@ -21,13 +21,23 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 
 public class ObPartitionEntry {
     private Map<Long, ObPartitionLocation> partitionLocation = new HashMap<Long, ObPartitionLocation>();
 
     // mapping from tablet id to ls id, and the part id to tablet id mapping is in ObPartitionInfo
     private Map<Long, Long> tabletLsIdMap = new HashMap<>();
+
+    // tabelt id -> (PartitionLocation, LsId)
+    private ConcurrentHashMap<Long, ObPartitionLocationInfo> partitionInfos = new ConcurrentHashMap<>();
+
 
+    public ObPartitionLocationInfo getPartitionInfo(long tabletId) {
+        return partitionInfos.computeIfAbsent(tabletId, id -> new ObPartitionLocationInfo());
+    }
+
     public Map<Long, ObPartitionLocation> getPartitionLocation() {
         return partitionLocation;
     }
@@ -39,6 +49,16 @@ public void setPartitionLocation(Map<Long, ObPartitionLocation> partitionLocatio
         this.partitionLocation = partitionLocation;
     }
 
+    public Map<Long, Long> getTabletLsIdMap() {
+        return tabletLsIdMap;
+    }
+
+    public void setTabletLsIdMap(Map<Long, Long> tabletLsIdMap) {
+        this.tabletLsIdMap = tabletLsIdMap;
+    }
+
+    public long getLsId(long tabletId) { return tabletLsIdMap.get(tabletId); }
+
     /*
      * Get partition location with part id.
      */
@@ -86,14 +106,4 @@ public void prepareForWeakRead(ObServerLdcLocation ldcLocation) {
     public String toString() {
         return "ObPartitionEntry{" + "partitionLocation=" + partitionLocation + '}';
     }
-
-    public Map<Long, Long> getTabletLsIdMap() {
-        return tabletLsIdMap;
-    }
-
-    public void setTabletLsIdMap(Map<Long, Long> tabletLsIdMap) {
-        this.tabletLsIdMap = tabletLsIdMap;
-    }
-
-    public long getLsId(long tabletId) { return tabletLsIdMap.get(tabletId); }
 }

src/main/java/com/alipay/oceanbase/rpc/location/model/partition/ObPartitionInfo.java

@@ -18,6 +18,7 @@
 package com.alipay.oceanbase.rpc.location.model.partition;
 
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObColumn;
+import com.alipay.oceanbase.rpc.util.Serialization;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -30,7 +31,7 @@ public class ObPartitionInfo {
     private ObPartDesc           firstPartDesc   = null;
     private ObPartDesc           subPartDesc     = null;
     private List<ObColumn>       partColumns     = new ArrayList<ObColumn>(1);
-    // mapping from part id to tablet id, and the tablet id to ls id mapping is in ObPartitionInfo
+    // mapping from part id to tablet id, and the tablet id to ls id mapping is in ObPartitionEntry
     private Map<Long, Long>      partTabletIdMap = null;
     private Map<String, Long>    partNameIdMap   = null;
     private Map<String, Integer> rowKeyElement   = null;
@@ -163,4 +164,5 @@ public Map<Long, Long> getPartTabletIdMap() {
     public void setPartTabletIdMap(Map<Long, Long> partTabletIdMap) {
         this.partTabletIdMap = partTabletIdMap;
     }
+
 }

src/main/java/com/alipay/oceanbase/rpc/location/model/partition/ObPartitionLocationInfo.java

@@ -0,0 +1,71 @@
+/*-
+ * #%L
+ * com.oceanbase:obkv-table-client
+ * %%
+ * Copyright (C) 2021 - 2024 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.location.model.partition;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import static com.alipay.oceanbase.rpc.protocol.payload.Constants.OB_INVALID_ID;
+
+public class ObPartitionLocationInfo {
+    private ObPartitionLocation   partitionLocation   = null;
+    private Long                  tabletLsId          = OB_INVALID_ID;
+    private Long                  lastUpdateTime      = 0L;
+    public ReentrantReadWriteLock rwLock              = new ReentrantReadWriteLock();
+    public AtomicBoolean          initialized         = new AtomicBoolean(false);
+    public final CountDownLatch   initializationLatch = new CountDownLatch(1);
+
+    public ObPartitionLocation getPartitionLocation() {
+        rwLock.readLock().lock();
+        try {
+            return partitionLocation;
+        } finally {
+            rwLock.readLock().unlock();
+        }
+    }
+
+    public void updateLocation(ObPartitionLocation newLocation, Long tabletLsId) {
+        rwLock.writeLock().lock();
+        try {
+            this.partitionLocation = newLocation;
+            this.tabletLsId = tabletLsId;
+            this.lastUpdateTime = System.currentTimeMillis();
+        } finally {
+            rwLock.writeLock().unlock();
+        }
+    }
+
+    public Long getTabletLsId() {
+        rwLock.readLock().lock();
+        try {
+            return tabletLsId;
+        } finally {
+            rwLock.readLock().unlock();
+        }
+    }
+
+    public Long getLastUpdateTime() {
+        rwLock.readLock().lock();
+        try {
+            return lastUpdateTime;
+        } finally {
+            rwLock.readLock().unlock();
+        }
+    }
+}

src/main/java/com/alipay/oceanbase/rpc/location/model/partition/ObRangePartDesc.java

@@ -26,6 +26,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.column.ObGeneratedColumn;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.column.ObSimpleColumn;
 import com.alipay.oceanbase.rpc.util.RandomUtil;
+import com.alipay.oceanbase.rpc.util.Serialization;
 import com.alipay.oceanbase.rpc.util.TableClientLoggerFactory;
 import org.apache.commons.lang.builder.ToStringBuilder;
 import org.slf4j.Logger;