fix(scan): prevent infinite retries on illegal partitions during splits

Author: maochongxin

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java

@@ -65,7 +65,7 @@ public abstract class AbstractQueryStreamResult extends AbstractPayload implemen
     protected LinkedList<List<ObObj>>                                          cacheRows           = new LinkedList<List<ObObj>>();
     private LinkedList<ObPair<ObPair<Long, ObTableParam>, ObTableQueryResult>> partitionLastResult = new LinkedList<ObPair<ObPair<Long, ObTableParam>, ObTableQueryResult>>();
     private ObReadConsistency                                                  readConsistency     = ObReadConsistency.STRONG;
-    public ObRowKey                                                        currentStartKey;
+    public List<ObObj>                                                         currentStartKey;
 
     /*
      * Get pcode.
@@ -302,7 +302,8 @@ public boolean next() throws Exception {
             // lastly, refer to the new partition
             boolean hasNext = false;
             List<Map.Entry<Long, ObPair<Long, ObTableParam>>> referPartition = new ArrayList<Map.Entry<Long, ObPair<Long, ObTableParam>>>();
-            Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet().iterator();
+            Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet()
+                .iterator();
             while (it.hasNext()) {
                 Map.Entry<Long, ObPair<Long, ObTableParam>> entry = it.next();
                 referPartition.add(entry);
@@ -311,7 +312,8 @@ public boolean next() throws Exception {
                     referPartition.add(entry);
 
                     // Try accessing the new partition  
-                    ObTableQueryResult tableQueryResult = (ObTableQueryResult) referToNewPartition(entry.getValue());
+                    ObTableQueryResult tableQueryResult = (ObTableQueryResult) referToNewPartition(entry
+                        .getValue());
 
                     if (tableQueryResult.getRowCount() == 0) {
                         continue;
@@ -329,13 +331,13 @@ public boolean next() throws Exception {
 
                         // Reset the iterator to start over  
                         it = expectant.entrySet().iterator();
-                        referPartition.clear(); // Clear the referPartition if needed  
+                        referPartition.clear(); // Clear the referPartition if needed
                     } else {
                         throw e;
                     }
                 }
             }
-            
+
             for (Map.Entry<Long, ObPair<Long, ObTableParam>> entry : expectant.entrySet()) {
                 // mark the refer partition
                 referPartition.add(entry);
@@ -411,7 +413,7 @@ protected void nextRow() {
         rowIndex = rowIndex + 1;
         row = cacheRows.poll();
         if (row != null) {
-            currentStartKey = (ObRowKey) row.get(0).getValue();
+            currentStartKey = row;
         }
     }
 
@@ -490,7 +492,10 @@ protected abstract ObTableQueryResult execute(ObPair<Long, ObTableParam> partIdW
     protected abstract ObTableQueryAsyncResult executeAsync(ObPair<Long, ObTableParam> partIdWithObTable,
                                                             ObPayload streamRequest)
                                                                                     throws Exception;
-    protected abstract Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery, String tableName) throws Exception;
+
+    protected abstract Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery,
+                                                                              String tableName)
+                                                                                               throws Exception;
 
     protected void cacheResultRows(ObTableQueryResult tableQueryResult) {
         cacheRows.addAll(tableQueryResult.getPropertiesRows());

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQuery.java

@@ -28,9 +28,11 @@
 import com.alipay.oceanbase.rpc.util.Serialization;
 import io.netty.buffer.ByteBuf;
 
+import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 
+import static com.alipay.oceanbase.rpc.util.ByteUtil.*;
 import static com.alipay.oceanbase.rpc.util.Serialization.encodeObUniVersionHeader;
 import static com.alipay.oceanbase.rpc.util.Serialization.getObUniVersionHeaderLength;
 
@@ -70,86 +72,44 @@ public class ObTableQuery extends AbstractPayload {
     private List<ObTableAggregationSingle>    aggregations       = new LinkedList<>();
 
     private ObKVParams obKVParams;
-
-    public void adjustStartKey(ObRowKey key) throws IllegalArgumentException {
+    
+    public void adjustStartKey(List<ObObj> key) throws IllegalArgumentException {
         List<ObNewRange> keyRanges = getKeyRanges();
         for (ObNewRange range : keyRanges) {
             if (isKeyInRange(range, key)) {
                 byte[] bytes = parseStartKeyToBytes(key);
-                ObRowKey newStartKey = ObRowKey.getInstance(new Object[]{incrementByteArray(bytes), ObObj.getMin(), ObObj.getMin()});
-                range.setStartKey(newStartKey);
+                if (getScanOrder() == ObScanOrder.Forward) {
+                    ObRowKey newStartKey = ObRowKey.getInstance(new Object[]{incrementByteArray(bytes), ObObj.getMin(), ObObj.getMin()});
+                    range.setStartKey(newStartKey);
+                } else {
+                    ObRowKey newStartKey = ObRowKey.getInstance(new Object[]{decrementByteArray(bytes), ObObj.getMax(), ObObj.getMax()});
+                    range.setEndKey(newStartKey);
+                }
                 return;
             }
         }
         throw new IllegalArgumentException("Key not found in any KeyRange.");
     }
 
-    private byte[] parseStartKeyToBytes(ObRowKey key) {
-        if (key == null || key.getObjs() == null || key.getObjs().isEmpty()) {
-            return new byte[0];
-        }
-
-        ObObj obObjKey = key.getObjs().get(0);
-        if (obObjKey.getValue() instanceof byte[]) {
-            return (byte[]) obObjKey.getValue();
-        }
-
-        throw new IllegalArgumentException("The start key does not contain a byte[] value.");
+    private byte[] parseStartKeyToBytes(List<ObObj> key) {
+        ObObj obObjKey = key.get(0);
+        return obObjKey.encode();
     }
 
-    private boolean isKeyInRange(ObNewRange range, ObRowKey key) {
-        byte[] startKeyBytes = parseStartKeyToBytes(range.getStartKey());
-        byte[] endKeyBytes = parseStartKeyToBytes(range.getEndKey());
+    private boolean isKeyInRange(ObNewRange range, List<ObObj> key) {
+        byte[] startKeyBytes = parseStartKeyToBytes(range.getStartKey().getObjs());
+        byte[] endKeyBytes = parseStartKeyToBytes(range.getEndKey().getObjs());
         byte[] keyBytes = parseStartKeyToBytes(key);
 
         int startComparison = compareByteArrays(startKeyBytes, keyBytes);
         int endComparison = compareByteArrays(endKeyBytes, keyBytes);
-
-        // 假设范围是 [start, end)，那么需要：  
-        boolean withinStart = startComparison <= 0;  // key >= start  
-        boolean withinEnd = endComparison > 0;       // key < end  
+        
+        boolean withinStart = startComparison <= 0;
+        boolean withinEnd = endComparison > 0;
 
         return withinStart && withinEnd;
     }
 
-    private int compareByteArrays(byte[] array1, byte[] array2) {
-        for (int i = 0; i < Math.min(array1.length, array2.length); i++) {
-            int a = (array1[i] & 0xFF);
-            int b = (array2[i] & 0xFF);
-            if (a != b) {
-                return a - b;
-            }
-        }
-        return array1.length - array2.length;
-    }
-
-    private byte[] incrementByteArray(byte[] input) {
-        if (input == null || input.length == 0) {
-            return new byte[]{1};  // 仅处理空输入情况  
-        }
-
-        byte[] result = input.clone();  // 复制数组以保持原数据不变  
-
-        for (int i = result.length - 1; i >= 0; i--) {
-            // 增加该字节1  
-            result[i] += 1;
-
-            // 检查是否出现溢出  
-            if ((result[i] & 0xFF) != 0) {
-                // 没有溢出，直接返回结果，不必再加了  
-                return result;
-            }
-
-            // 若溢出，需要下一个字节处理进位（当前字节已0）  
-            result[i] = 0;
-        }
-
-        // 如果循环结束还没有返回，说明最高位有进位  
-        byte[] extendedResult = new byte[result.length + 1];
-        extendedResult[0] = 1;  // 设置第一个字节，代表进位  
-
-        return extendedResult;
-    }
 
     /*
      * Check filter.

src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java

@@ -145,10 +145,12 @@ protected void closeLastStreamResult(ObPair<Long, ObTableParam> partIdWithObTabl
     }
 
     @Override
-    protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery, String tableName) throws Exception {
+    protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery,
+                                                                     String tableName)
+                                                                                      throws Exception {
         return buildPartitions(client, tableQuery, tableName);
     }
-    
+
     @Override
     public boolean next() throws Exception {
         checkStatus();
@@ -192,6 +194,7 @@ public boolean next() throws Exception {
                         this.tableQuery.adjustStartKey(currentStartKey);
                         setExpectant(refreshPartition(tableQuery, tableName));
                         it = expectant.entrySet().iterator();
+                        continue;
                     } else {
                         throw e;
                     }

src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryStreamResult.java

@@ -79,10 +79,12 @@ protected ObTableQueryAsyncResult executeAsync(ObPair<Long, ObTableParam> partId
     }
 
     @Override
-    protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery, String tableName) throws Exception {
+    protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery,
+                                                                     String tableName)
+                                                                                      throws Exception {
         return buildPartitions(client, tableQuery, tableName);
     }
-    
+
     /**
      * Get client.
      * @return client

src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java

@@ -675,7 +675,7 @@ public ObTableSingleOpResult[] executeInternal() throws Exception {
         long getTableTime = System.currentTimeMillis();
         final Map<Object, Object> context = ThreadLocalMap.getContextMap();
         final int maxRetries = obTableClient.getRuntimeRetryTimes();
-        
+
         if (executorService != null && !executorService.isShutdown() && lsOperations.size() > 1) {
             // execute sub-batch operation in parallel
             final ConcurrentTaskExecutor executor = new ConcurrentTaskExecutor(executorService,

src/main/java/com/alipay/oceanbase/rpc/util/ByteUtil.java

@@ -0,0 +1,132 @@
+/*-
+ * #%L
+ * com.oceanbase:obkv-table-client
+ * %%
+ * Copyright (C) 2021 - 2023 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.util;
+
+public class ByteUtil {
+    /**
+     * Compares two byte arrays lexicographically.  
+     *
+     * This method interprets each byte as an unsigned value between 0 and 255   
+     * and performs a comparison on the arrays byte-by-byte. If two corresponding  
+     * bytes differ, it returns the difference between the first non-equal byte   
+     * values (treated as unsigned). If one array is a prefix of the other,   
+     * the shorter array is considered to be less than the longer one.  
+     *
+     * @param array1 the first byte array to be compared  
+     * @param array2 the second byte array to be compared  
+     * @return a negative integer if array1 comes before array2,   
+     *         a positive integer if array1 comes after array2,   
+     *         or zero if they are equal  
+     */
+    static public int compareByteArrays(byte[] array1, byte[] array2) {
+        for (int i = 0; i < Math.min(array1.length, array2.length); i++) {
+            int a = (array1[i] & 0xFF);
+            int b = (array2[i] & 0xFF);
+            if (a != b) {
+                return a - b;
+            }
+        }
+        return array1.length - array2.length;
+    }
+
+    /**
+     * Increments a byte array treated as an unsigned integer by one.  
+     *
+     * This method treats the input byte array as a non-negative big-endian   
+     * integer, and increments its value by 1. The most significant byte is at   
+     * the beginning of the array. If the increment operation causes an overflow,  
+     * the resulting array will be extended to accommodate the carry.  
+     *
+     * @param input the byte array representing a non-negative integer  
+     * @return a new byte array representing the incremented value  
+     */
+    static public byte[] incrementByteArray(byte[] input) {
+        if (input == null || input.length == 0) {
+            return new byte[] { 1 };
+        }
+        byte[] result = input.clone();
+        for (int i = result.length - 1; i >= 0; i--) {
+            result[i] += 1;
+
+            if ((result[i] & 0xFF) != 0) {
+                return result;
+            }
+            result[i] = 0;
+        }
+
+        byte[] extendedResult = new byte[result.length + 1];
+        extendedResult[0] = 1;
+        return extendedResult;
+    }
+
+    /**
+     * Decrements a byte array treated as an unsigned integer by one.  
+     *
+     * This method treats the input byte array as a non-negative big-endian   
+     * integer, and decrements its value by 1. If the entire array represents  
+     * zero, the function should handle it appropriately, typically returning   
+     * an array representing zero.  
+     *
+     * @param input the byte array representing a non-negative integer  
+     * @return a new byte array representing the decremented value  
+     * @throws IllegalArgumentException if the input represents zero  
+     */
+    static public byte[] decrementByteArray(byte[] input) {
+        if (input == null || input.length == 0 || isZero(input)) {
+            throw new IllegalArgumentException("Input array must represent a positive integer.");
+        }
+
+        byte[] result = input.clone();
+
+        // Traverse the array from the least significant byte to the most  
+        for (int i = result.length - 1; i >= 0; i--) {
+            if ((result[i] & 0xFF) > 0) {
+                result[i] -= 1;
+                return result;
+            }
+
+            // If the byte is zero, we need to borrow from the next  
+            result[i] = (byte) 0xFF; // Set current byte to 255 after borrow  
+        }
+
+        // Handle cases where we subtract from a number like 1000 -> 999  
+        if (result[0] == 0) {
+            byte[] shortenedResult = new byte[result.length - 1];
+            System.arraycopy(result, 1, shortenedResult, 0, shortenedResult.length);
+            return shortenedResult;
+        }
+
+        return result;
+    }
+
+    /**
+     * Helper method to check if a byte array represents zero.  
+     *
+     * @param array the byte array to check  
+     * @return true if the byte array is zero, false otherwise  
+     */
+    static private boolean isZero(byte[] array) {
+        for (byte b : array) {
+            if (b != 0) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+}