fix tablet_id error bug (#302)

* fix sub expr bug

* fix scan bug

Author: WeiXinChan

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

@@ -81,15 +81,17 @@ public void init() throws Exception {
                 } catch (Exception e) {
                     if (e instanceof ObTableNeedFetchAllException) {
                         setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest()
-                            .getTableQuery(), client.getPhyTableNameFromTableGroup(entityType, tableName)));
+                            .getTableQuery(), client.getPhyTableNameFromTableGroup(entityType,
+                            tableName)));
                         it = expectant.entrySet().iterator();
                         retryTimes++;
                         if (retryTimes > maxRetries) {
                             RUNTIME.error("Fail to get refresh table entry response after {}",
                                 retryTimes);
                             throw new ObTableRetryExhaustedException(
                                 "Fail to get refresh table entry response after " + retryTimes
-                                + "errorCode:" + ((ObTableNeedFetchAllException) e).getErrorCode());
+                                        + "errorCode:"
+                                        + ((ObTableNeedFetchAllException) e).getErrorCode());
 
                         }
                     } else {
@@ -113,8 +115,8 @@ protected ObTableQueryAsyncResult referToNewPartition(ObPair<Long, ObTableParam>
                                                                                                        throws Exception {
         ObTableParam obTableParam = partIdWithObTable.getRight();
         ObTableQueryRequest queryRequest = asyncRequest.getObTableQueryRequest();
-        long partitionId = client.getServerCapacity().isSupportSecondaryPartition() ?
-                INVALID_TABLET_ID : obTableParam.getPartitionId();
+        long partitionId = client.getServerCapacity().isSupportDistributedExecute() ? INVALID_TABLET_ID
+            : obTableParam.getPartitionId();
         // refresh request info
         queryRequest.setPartitionId(partitionId);
         queryRequest.setTableId(obTableParam.getTableId());
@@ -141,7 +143,9 @@ protected ObTableQueryAsyncResult referToLastStreamResult(ObPair<Long, ObTablePa
         ObTableQueryRequest queryRequest = asyncRequest.getObTableQueryRequest();
 
         // refresh request info
-        queryRequest.setPartitionId(obTableParam.getPartitionId());
+        long partitionId = client.getServerCapacity().isSupportDistributedExecute() ? INVALID_TABLET_ID
+            : obTableParam.getPartitionId();
+        queryRequest.setPartitionId(partitionId);
         queryRequest.setTableId(obTableParam.getTableId());
 
         // refresh async query request
@@ -160,7 +164,9 @@ protected void closeLastStreamResult(ObPair<Long, ObTableParam> partIdWithObTabl
         ObTableQueryRequest queryRequest = asyncRequest.getObTableQueryRequest();
 
         // refresh request info
-        queryRequest.setPartitionId(obTableParam.getPartitionId());
+        long partitionId = client.getServerCapacity().isSupportDistributedExecute() ? INVALID_TABLET_ID
+            : obTableParam.getPartitionId();
+        queryRequest.setPartitionId(partitionId);
         queryRequest.setTableId(obTableParam.getTableId());
 
         // set end async query
@@ -188,15 +194,17 @@ protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery ta
     public void renewLease() throws Exception {
         if (!isEnd() && !expectant.isEmpty()) {
             Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet()
-                    .iterator();
+                .iterator();
             Map.Entry<Long, ObPair<Long, ObTableParam>> lastEntry = it.next();
             ObPair<Long, ObTableParam> partIdWithObTable = lastEntry.getValue();
             // try access new partition, async will not remove useless expectant
             ObTableParam obTableParam = partIdWithObTable.getRight();
             ObTableQueryRequest queryRequest = asyncRequest.getObTableQueryRequest();
 
             // refresh request info
-            queryRequest.setPartitionId(obTableParam.getPartitionId());
+            long partitionId = client.getServerCapacity().isSupportDistributedExecute() ? INVALID_TABLET_ID
+                : obTableParam.getPartitionId();
+            queryRequest.setPartitionId(partitionId);
             queryRequest.setTableId(obTableParam.getTableId());
 
             // refresh async query request
@@ -230,20 +238,23 @@ public boolean next() throws Exception {
                     referToLastStreamResult(lastEntry.getValue());
                 } catch (Exception e) {
                     if (e instanceof ObTableNeedFetchAllException) {
-                        String realTableName = client.getPhyTableNameFromTableGroup(entityType, tableName);
-                        TableEntry entry = client.getOrRefreshTableEntry(realTableName, false, false, false);
+                        String realTableName = client.getPhyTableNameFromTableGroup(entityType,
+                            tableName);
+                        TableEntry entry = client.getOrRefreshTableEntry(realTableName, false,
+                            false, false);
                         // Calculate the next partition only when the range partition is affected by a split, based on the keys already scanned.
                         if (ObGlobal.obVsnMajor() >= 4
-                                && entry.isPartitionTable()
-                                && entry.getPartitionInfo().getFirstPartDesc().getPartFuncType().isRangePart()) {
+                            && entry.isPartitionTable()
+                            && entry.getPartitionInfo().getFirstPartDesc().getPartFuncType()
+                                .isRangePart()) {
                             this.asyncRequest.getObTableQueryRequest().getTableQuery()
-                                    .adjustStartKey(currentStartKey);
-                            setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest()
-                                    .getTableQuery(), realTableName));
+                                .adjustStartKey(currentStartKey);
+                            setExpectant(refreshPartition(this.asyncRequest
+                                .getObTableQueryRequest().getTableQuery(), realTableName));
                             setEnd(true);
                         } else {
-                            setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest()
-                                    .getTableQuery(), realTableName));
+                            setExpectant(refreshPartition(this.asyncRequest
+                                .getObTableQueryRequest().getTableQuery(), realTableName));
                         }
                     } else {
                         throw e;
@@ -271,15 +282,18 @@ public boolean next() throws Exception {
                     referToNewPartition(entry.getValue());
                 } catch (Exception e) {
                     if (e instanceof ObTableNeedFetchAllException) {
-                        String realTableName = client.getPhyTableNameFromTableGroup(entityType, tableName);
-                        TableEntry tableEntry = client.getOrRefreshTableEntry(realTableName, false, false, false);
+                        String realTableName = client.getPhyTableNameFromTableGroup(entityType,
+                            tableName);
+                        TableEntry tableEntry = client.getOrRefreshTableEntry(realTableName, false,
+                            false, false);
                         if (ObGlobal.obVsnMajor() >= 4
                             && tableEntry.isPartitionTable()
-                            && tableEntry.getPartitionInfo().getFirstPartDesc().getPartFuncType().isRangePart()) {
+                            && tableEntry.getPartitionInfo().getFirstPartDesc().getPartFuncType()
+                                .isRangePart()) {
                             this.asyncRequest.getObTableQueryRequest().getTableQuery()
-                                    .adjustStartKey(currentStartKey);
-                            setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest()
-                                    .getTableQuery(), realTableName));
+                                .adjustStartKey(currentStartKey);
+                            setExpectant(refreshPartition(this.asyncRequest
+                                .getObTableQueryRequest().getTableQuery(), realTableName));
                         }
                         it = expectant.entrySet().iterator();
                         retryTimes++;
@@ -366,7 +380,7 @@ public void close() throws Exception {
             closeLastStreamResult(lastEntry.getValue());
         }
     }
-    
+
     public boolean isEnd() {
         return isEnd;
     }