fix scan/get more result after meta refreshing

Author: JackShi148

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

@@ -46,6 +46,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
 
+import static com.alipay.oceanbase.rpc.protocol.payload.Constants.INVALID_TABLET_ID;
 import static com.alipay.oceanbase.rpc.util.TableClientLoggerFactory.RUNTIME;
 
 public abstract class AbstractQueryStreamResult extends AbstractPayload implements
@@ -412,7 +413,7 @@ public boolean next() throws Exception {
         }
     }
 
-    protected Map<Long, ObPair<Long, ObTableParam>> buildPartitions(ObTableClient client, ObTableQuery tableQuery, String tableName) throws Exception {
+    protected Map<Long, ObPair<Long, ObTableParam>> buildAllPartitions(ObTableClient client, ObTableQuery tableQuery, String tableName) throws Exception {
         Map<Long, ObPair<Long, ObTableParam>> partitionObTables = new LinkedHashMap<>();
         String indexName = tableQuery.getIndexName();
         if (!client.isOdpMode()) {
@@ -456,6 +457,40 @@ protected Map<Long, ObPair<Long, ObTableParam>> buildPartitions(ObTableClient cl
         return partitionObTables;
     }
 
+    protected Map<Long, ObPair<Long, ObTableParam>> buildFirstPartitions(ObTableClient client, ObTableQuery tableQuery, String tableName) throws Exception {
+        Map<Long, ObPair<Long, ObTableParam>> partitionObTables = new LinkedHashMap<>();
+        String indexName = tableQuery.getIndexName();
+
+        if (!this.client.isOdpMode()) {
+            indexTableName = client.getIndexTableName(tableName, indexName, tableQuery.getScanRangeColumns(), false);
+        }
+
+        if (tableQuery.getKeyRanges().isEmpty()) {
+            throw new IllegalArgumentException("query ranges is empty");
+        } else {
+            ObNewRange range = tableQuery.getKeyRanges().get(0);
+            ObRowKey startKey = range.getStartKey();
+            int startKeySize = startKey.getObjs().size();
+            Object[] start = new Object[startKeySize];
+
+            for (int i = 0; i < startKeySize; i++) {
+                start[i] = startKey.getObj(i).isMinObj() || startKey.getObj(i).isMaxObj() ?
+                        startKey.getObj(i) : startKey.getObj(i).getValue();
+            }
+
+            if (this.entityType == ObTableEntityType.HKV && client.isTableGroupName(tableName)) {
+                indexTableName = client.tryGetTableNameFromTableGroupCache(tableName, false);
+            }
+            ObBorderFlag borderFlag = range.getBorderFlag();
+            List<ObTableParam> params = this.client.getTableParams(indexTableName, tableQuery, start,
+                    borderFlag.isInclusiveStart(), start, borderFlag.isInclusiveEnd());
+
+            partitionObTables.put(INVALID_TABLET_ID, new ObPair<>(params.get(0).getPartId(), params.get(0)));
+        }
+
+        return partitionObTables;
+    }
+
     protected void nextRow() {
         rowIndex = rowIndex + 1;
         row = cacheRows.poll();

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

@@ -278,7 +278,11 @@ public boolean queryNewStreamResultInNext() throws Exception {
     protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery,
                                                                      String tableName)
                                                                                       throws Exception {
-        return buildPartitions(client, tableQuery, tableName);
+        if (isDistributeScan()) {
+            return buildFirstPartitions(client, tableQuery, tableName);
+        } else {
+            return buildAllPartitions(client, tableQuery, tableName);
+        }
     }
 
     // This function is designed for HBase-type requests.

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

@@ -17,7 +17,6 @@
 
 package com.alipay.oceanbase.rpc.stream;
 
-import com.alipay.oceanbase.rpc.ObTableClient;
 import com.alipay.oceanbase.rpc.bolt.transport.ObTableConnection;
 import com.alipay.oceanbase.rpc.exception.ObTableEntryRefreshException;
 import com.alipay.oceanbase.rpc.exception.ObTableRetryExhaustedException;
@@ -114,7 +113,11 @@ protected ObTableQueryAsyncResult executeAsync(ObPair<Long, ObTableParam> partId
     protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery tableQuery,
                                                                      String tableName)
                                                                                       throws Exception {
-        return buildPartitions(client, tableQuery, tableName);
+        if (client.getServerCapacity().isSupportDistributedExecute()) {
+            return buildFirstPartitions(client, tableQuery, tableName);
+        } else {
+            return buildAllPartitions(client, tableQuery, tableName);
+        }
     }
 
     private boolean needTabletId(ObTableQueryRequest request) {