reset cached result in sync query init if need to refresh table meta; refresh tablet location in sync query and normal batch if needed

Author: JackShi148

src/main/java/com/alipay/oceanbase/rpc/location/model/TableRoute.java

@@ -603,7 +603,7 @@ private Long[] getTabletsFromTableEntry(TableEntry tableEntry) {
             tablets = tableEntry.getPartitionInfo().getPartTabletIdMap().values()
                     .toArray(new Long[0]);
         } else {
-            tablets = new Long[1];
+            tablets = new Long[]{0L};
         }
         return tablets;
     }

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

@@ -250,10 +250,9 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
                                 // if the tableEntry is not cached in this client, tableEntry will refresh meta information and its tablet-ls information
                                 // if the server has the capacity to distribute ls request, no need to refresh ls_id and so tablet location
                                 if (e instanceof ObTableNeedFetchMetaException) {
-                                    client.getOrRefreshTableEntry(indexTableName, true);
-                                    if (((ObTableNeedFetchMetaException) e).isNeedRefreshMetaAndLocation()
-                                        && !client.getServerCapacity().isSupportDistributedExecute()) {
-                                        long tabletId = partIdWithIndex.getRight().getTabletId();
+                                    TableEntry tableEntry = client.getOrRefreshTableEntry(indexTableName, true);
+                                    if (((ObTableNeedFetchMetaException) e).isNeedRefreshMetaAndLocation()) {
+                                        long tabletId = client.getTabletIdByPartId(tableEntry, partIdWithIndex.getLeft());
                                         client.refreshTableLocationByTabletId(indexTableName,
                                             tabletId);
                                     }
@@ -369,6 +368,7 @@ public boolean next() throws Exception {
 
                 } catch (Exception e) {
                     if (e instanceof ObTableNeedFetchMetaException) {
+                        // TODO: need to skip over the partitions that have been scanned
                         setExpectant(refreshPartition(tableQuery, tableName));
                         // Reset the iterator to start over  
                         it = expectant.entrySet().iterator();
@@ -536,6 +536,13 @@ protected abstract Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTabl
                                                                               String tableName)
                                                                                                throws Exception;
 
+
+    private void resetCachedResult() {
+        cacheRows.clear();
+        cacheProperties.clear();
+        partitionLastResult.clear();
+    }
+
     protected void cacheResultRows(ObTableQueryResult tableQueryResult) {
         cacheRows.addAll(tableQueryResult.getPropertiesRows());
         cacheProperties = tableQueryResult.getPropertiesNames();
@@ -604,6 +611,7 @@ public void init() throws Exception {
                         referToNewPartition(entry.getValue());
                     } catch (Exception e) {
                         if (e instanceof ObTableNeedFetchMetaException) {
+                            resetCachedResult();
                             setExpectant(refreshPartition(tableQuery, tableName));
                             it = expectant.entrySet().iterator();
                             retryTimes++;

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

@@ -267,7 +267,7 @@ public Map<Long, ObPair<ObTableParam, List<ObPair<Integer, ObTableOperation>>>>
             ObTableParam tableParam = obTableClient.getTableParamWithRoute(
                     tableName, row, obTableClient.getRoute(batchOperation.isReadOnly()));
             ObPair<ObTableParam, List<ObPair<Integer, ObTableOperation>>> obTableOperations = partitionOperationsMap
-                    .computeIfAbsent(tableParam.getTabletId(), k -> new ObPair<>(
+                    .computeIfAbsent(tableParam.getPartId(), k -> new ObPair<>(
                             tableParam, new ArrayList<>()));
             obTableOperations.getRight().add(new ObPair<>(i, operation));
         }
@@ -420,7 +420,11 @@ public void partitionExecute(ObTableOperationResult[] results,
                         if (obTableClient.isRetryOnChangeMasterTimes()) {
                             if (ex instanceof ObTableNeedFetchMetaException) {
                                 // refresh table info
-                                obTableClient.getOrRefreshTableEntry(tableName, true);
+                                TableEntry entry = obTableClient.getOrRefreshTableEntry(tableName, true);
+                                if (((ObTableNeedFetchMetaException) ex).isNeedRefreshMetaAndLocation()) {
+                                    long tabletId = obTableClient.getTabletIdByPartId(entry, originPartId);
+                                    obTableClient.refreshTableLocationByTabletId(tableName, tabletId);
+                                }
                                 throw ex;
                             }
                         } else {