fix scan_to_next_partition_test route error; fix scan_during_split_test result error

Author: maochongxin

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

@@ -65,6 +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;
+    // ObRowKey objs: [startKey, MIN_OBJECT, MIN_OBJECT]
     public List<ObObj>                                                         currentStartKey;
 
     /*

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

@@ -77,17 +77,18 @@ public void adjustStartKey(List<ObObj> key) throws IllegalArgumentException {
         List<ObNewRange> keyRanges = getKeyRanges();
         for (ObNewRange range : keyRanges) {
             if (key != null && isKeyInRange(range, key)) {
-                byte[] bytes = parseStartKeyToBytes(key);
                 ObRowKey newStartKey;
                 if (getScanOrder() == ObScanOrder.Forward) {
-                    newStartKey = ObRowKey.getInstance(new Object[]{incrementByteArray(bytes), ObObj.getMin(), ObObj.getMin()});
+                    // get the real rowkey
+                    newStartKey = ObRowKey.getInstance(new Object[]{key.get(0).getValue(), ObObj.getMax(), ObObj.getMax()});
                 } else {
-                    newStartKey = ObRowKey.getInstance(new Object[]{decrementByteArray(bytes), ObObj.getMax(), ObObj.getMax()});
+                    newStartKey = ObRowKey.getInstance(new Object[]{key.get(0).getValue(), ObObj.getMax(), ObObj.getMax()});
                 }
                 range.setStartKey(newStartKey);
                 return;
             }
         }
+        System.out.println("noting changed");
         /* keyRanges not changed */
     }
 

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

@@ -29,6 +29,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.syncquery.ObTableQueryAsyncRequest;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.syncquery.ObTableQueryAsyncResult;
 import com.alipay.oceanbase.rpc.table.ObTableParam;
+import com.alipay.oceanbase.rpc.util.ByteUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -166,10 +167,22 @@ public boolean next() throws Exception {
             if (!isEnd() && !expectant.isEmpty()) {
                 Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet()
                     .iterator();
-                Map.Entry<Long, ObPair<Long, ObTableParam>> lastEntry = it.next();
-                // try access new partition, async will not remove useless expectant
-                referToLastStreamResult(lastEntry.getValue());
 
+                Map.Entry<Long, ObPair<Long, ObTableParam>> lastEntry = it.next();
+                try {
+                    // try access new partition, async will not remove useless expectant
+                    referToLastStreamResult(lastEntry.getValue());
+                } catch (Exception e) {
+                    if (e instanceof ObTableNeedFetchAllException) {
+                        this.asyncRequest.getObTableQueryRequest().getTableQuery()
+                            .adjustStartKey(currentStartKey);
+                        setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest()
+                            .getTableQuery(), tableName));
+                        setEnd(true);
+                    } else {
+                        throw e;
+                    }
+                }
                 // remove useless expectant if it is end
                 if (isEnd())
                     it.remove();
@@ -191,8 +204,10 @@ public boolean next() throws Exception {
                     referToNewPartition(entry.getValue());
                 } catch (Exception e) {
                     if (e instanceof ObTableNeedFetchAllException) {
-                        this.tableQuery.adjustStartKey(currentStartKey);
-                        setExpectant(refreshPartition(tableQuery, tableName));
+                        this.asyncRequest.getObTableQueryRequest().getTableQuery()
+                            .adjustStartKey(currentStartKey);
+                        setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest()
+                            .getTableQuery(), tableName));
                         it = expectant.entrySet().iterator();
                         continue;
                     } else {