[iotdb] branch handle_redirection_during_dispatching updated: do not modify an existing redirection handle local redirection

[ji...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jiangtian pushed a commit to branch handle_redirection_during_dispatching
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/handle_redirection_during_dispatching by this push:
     new 6b813ee094 do not modify an existing redirection handle local redirection
6b813ee094 is described below

commit 6b813ee0941ef95d15bf6a852331789639b2b60e
Author: Tian Jiang <jt...@163.com>
AuthorDate: Tue May 16 11:07:15 2023 +0800

    do not modify an existing redirection
    handle local redirection
---
 .../iotdb/db/mpp/plan/execution/QueryExecution.java |  4 +++-
 .../scheduler/FragmentInstanceDispatcherImpl.java   | 21 ++++++++++++++-------
 2 files changed, 17 insertions(+), 8 deletions(-)

diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java
index 715cf29c71..0cac209f97 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java
@@ -660,7 +660,9 @@ public class QueryExecution implements IQueryExecution {
     if (analysis.getStatement() instanceof InsertBaseStatement
         && !analysis.isFinishQueryAfterAnalyze()
         && (!config.isEnable13DataInsertAdapt()
-            || IoTDBConstant.ClientVersion.V_1_0.equals(context.getSession().getVersion()))) {
+            || IoTDBConstant.ClientVersion.V_1_0.equals(context.getSession().getVersion()))
+        // if the status is already a redirection, do not modify it
+        && tsstatus.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
       InsertBaseStatement insertStatement = (InsertBaseStatement) analysis.getStatement();
       List<TEndPoint> redirectNodeList = analysis.getRedirectNodeList();
       if (insertStatement instanceof InsertRowsStatement
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
index d362d1fbfc..724cbf8ae2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -370,20 +370,27 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher {
         PlanNode planNode = instance.getFragment().getPlanNodeTree();
         RegionWriteExecutor writeExecutor = new RegionWriteExecutor();
         RegionExecutionResult writeResult = writeExecutor.execute(groupId, planNode);
+
+        TSStatus status = writeResult.getStatus();
         if (!writeResult.isAccepted()) {
-          logger.warn(
-              "write locally failed. TSStatus: {}, message: {}",
-              writeResult.getStatus(),
-              writeResult.getMessage());
-          if (writeResult.getStatus() == null) {
+          if (status == null || status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
+            logger.warn(
+                "write locally failed. TSStatus: {}, message: {}",
+                status,
+                writeResult.getMessage());
+          } else {
+            // status code == REDIRECTION_RECOMMEND
+            instance.getExecutorType().updatePreferredLocation(status.getRedirectNode());
+          }
+
+          if (status == null) {
             throw new FragmentInstanceDispatchException(
                 RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, writeResult.getMessage()));
           } else {
-            throw new FragmentInstanceDispatchException(writeResult.getStatus());
+            throw new FragmentInstanceDispatchException(status);
           }
         } else {
           // some expected and accepted status except SUCCESS_STATUS need to be returned
-          TSStatus status = writeResult.getStatus();
           if (status != null && status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
             throw new FragmentInstanceDispatchException(status);
           }