[iotdb] branch master updated: Fix the issue that the QueryExecution may not be released (#6318)

[xi...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new ed68cb5d9b Fix the issue that the QueryExecution may not be released (#6318)
ed68cb5d9b is described below

commit ed68cb5d9b3707fea0d31a03fa019e75eb13662f
Author: Zhang.Jinrui <xi...@gmail.com>
AuthorDate: Fri Jun 17 12:23:59 2022 +0800

    Fix the issue that the QueryExecution may not be released (#6318)
---
 .../execution/datatransfer/DataBlockManager.java   |  8 ++--
 .../db/mpp/plan/analyze/ClusterSchemaFetcher.java  | 55 ++++++++++++----------
 2 files changed, 33 insertions(+), 30 deletions(-)

diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/DataBlockManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/DataBlockManager.java
index 0db8bb2328..5b9743a07f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/DataBlockManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/DataBlockManager.java
@@ -177,10 +177,10 @@ public class DataBlockManager implements IDataBlockManager {
                 .get(e.getTargetFragmentInstanceId())
                 .get(e.getTargetPlanNodeId())
                 .isAborted()) {
-          throw new TException(
-              "Target fragment instance not found. Fragment instance ID: "
-                  + e.getTargetFragmentInstanceId()
-                  + ".");
+          logger.warn(
+              "received onEndOfDataBlockEvent but the downstream FragmentInstance[{}] is not found",
+              e.getTargetFragmentInstanceId());
+          return;
         }
         SourceHandle sourceHandle =
             (SourceHandle)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
index 2d2d06e2f1..84db6df7f7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
@@ -102,35 +102,38 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
 
   private SchemaTree executeSchemaFetchQuery(SchemaFetchStatement schemaFetchStatement) {
     long queryId = SessionManager.getInstance().requestQueryId(false);
-    ExecutionResult executionResult =
-        coordinator.execute(schemaFetchStatement, queryId, null, "", partitionFetcher, this);
-    // TODO: (xingtanzjr) throw exception
-    if (executionResult.status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      throw new RuntimeException(
-          String.format(
-              "cannot fetch schema, status is: %s, msg is: %s",
-              executionResult.status.getCode(), executionResult.status.getMessage()));
-    }
-    try (SetThreadName threadName = new SetThreadName(executionResult.queryId.getId())) {
-      SchemaTree result = new SchemaTree();
-      while (coordinator.getQueryExecution(queryId).hasNextResult()) {
-        // The query will be transited to FINISHED when invoking getBatchResult() at the last time
-        // So we don't need to clean up it manually
-        Optional<TsBlock> tsBlock = coordinator.getQueryExecution(queryId).getBatchResult();
-        if (!tsBlock.isPresent() || tsBlock.get().isEmpty()) {
-          break;
-        }
-        Binary binary;
-        SchemaTree fetchedSchemaTree;
-        Column column = tsBlock.get().getColumn(0);
-        for (int i = 0; i < column.getPositionCount(); i++) {
-          binary = column.getBinary(i);
-          fetchedSchemaTree = SchemaTree.deserialize(ByteBuffer.wrap(binary.getValues()));
-          result.mergeSchemaTree(fetchedSchemaTree);
+    try {
+      ExecutionResult executionResult =
+          coordinator.execute(schemaFetchStatement, queryId, null, "", partitionFetcher, this);
+      // TODO: (xingtanzjr) throw exception
+      if (executionResult.status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+        throw new RuntimeException(
+            String.format(
+                "cannot fetch schema, status is: %s, msg is: %s",
+                executionResult.status.getCode(), executionResult.status.getMessage()));
+      }
+      try (SetThreadName threadName = new SetThreadName(executionResult.queryId.getId())) {
+        SchemaTree result = new SchemaTree();
+        while (coordinator.getQueryExecution(queryId).hasNextResult()) {
+          // The query will be transited to FINISHED when invoking getBatchResult() at the last time
+          // So we don't need to clean up it manually
+          Optional<TsBlock> tsBlock = coordinator.getQueryExecution(queryId).getBatchResult();
+          if (!tsBlock.isPresent() || tsBlock.get().isEmpty()) {
+            break;
+          }
+          Binary binary;
+          SchemaTree fetchedSchemaTree;
+          Column column = tsBlock.get().getColumn(0);
+          for (int i = 0; i < column.getPositionCount(); i++) {
+            binary = column.getBinary(i);
+            fetchedSchemaTree = SchemaTree.deserialize(ByteBuffer.wrap(binary.getValues()));
+            result.mergeSchemaTree(fetchedSchemaTree);
+          }
         }
+        return result;
       }
+    } finally {
       coordinator.removeQueryExecution(queryId);
-      return result;
     }
   }