query refresh partitionTables

Author: maochongxin

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

@@ -129,7 +129,6 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
             }
             tryTimes++;
             try {
-                // 重试时重新 getTable
                 if (tryTimes > 1) {
                     if (client.isOdpMode()) {
                         subObTable = client.getOdpTable();
@@ -273,7 +272,6 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
     /*
      * Next.
      */
-    // 这个函数中，任何的计算结果都会填充到cacheRows
     public boolean next() throws Exception {
         checkStatus();
         lock.lock();
@@ -325,7 +323,7 @@ public boolean next() throws Exception {
 
                 } catch (Exception e) {
                     if (e instanceof ObTableNeedFetchAllException) {
-                        // Adjust the start key and refresh the expectant  
+                        // Adjust the start key and refresh the expectant
                         this.tableQuery.adjustStartKey(currentStartKey);
                         setExpectant(refreshPartition(tableQuery, tableName));
 
@@ -408,7 +406,6 @@ protected Map<Long, ObPair<Long, ObTableParam>> buildPartitions(ObTableClient cl
         return partitionObTables;
     }
 
-    // 上层会不断的调getRow， 也就是不断的取出缓存，所以可以在这里给row赋值的时候顺便把这个row记录下来，用来作为最后拿到的key
     protected void nextRow() {
         rowIndex = rowIndex + 1;
         row = cacheRows.poll();

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

@@ -76,24 +76,27 @@ public class ObTableQuery extends AbstractPayload {
     public void adjustStartKey(List<ObObj> key) throws IllegalArgumentException {
         List<ObNewRange> keyRanges = getKeyRanges();
         for (ObNewRange range : keyRanges) {
-            if (isKeyInRange(range, key)) {
+            if (key != null && isKeyInRange(range, key)) {
                 byte[] bytes = parseStartKeyToBytes(key);
+                ObRowKey newStartKey;
                 if (getScanOrder() == ObScanOrder.Forward) {
-                    ObRowKey newStartKey = ObRowKey.getInstance(new Object[]{incrementByteArray(bytes), ObObj.getMin(), ObObj.getMin()});
-                    range.setStartKey(newStartKey);
+                    newStartKey = ObRowKey.getInstance(new Object[]{incrementByteArray(bytes), ObObj.getMin(), ObObj.getMin()});
                 } else {
-                    ObRowKey newStartKey = ObRowKey.getInstance(new Object[]{decrementByteArray(bytes), ObObj.getMax(), ObObj.getMax()});
-                    range.setEndKey(newStartKey);
+                    newStartKey = ObRowKey.getInstance(new Object[]{decrementByteArray(bytes), ObObj.getMax(), ObObj.getMax()});
                 }
+                range.setStartKey(newStartKey);
                 return;
             }
         }
-        throw new IllegalArgumentException("Key not found in any KeyRange.");
+        /* keyRanges not changed */
     }
 
     private byte[] parseStartKeyToBytes(List<ObObj> key) {
-        ObObj obObjKey = key.get(0);
-        return obObjKey.encode();
+        if (key != null) {
+            ObObj obObjKey = key.get(0);
+            return obObjKey.encode();
+        }
+        return new byte[0];
     }
 
     private boolean isKeyInRange(ObNewRange range, List<ObObj> key) {