This is an automated email from the ASF dual-hosted git repository. xingtanzjr pushed a commit to branch xingtanzjr/fix_potential_memory in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f854a60ba5a9fd8e0c4afeaf56123f46a7d81723 Author: Jinrui.Zhang <[email protected]> AuthorDate: Thu Jun 16 22:53:39 2022 +0800 Fix the issue that the QueryExecution may not be released --- .../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 05748f7519..c57203082a 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; } }
