You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/04/19 07:15:43 UTC
[iotdb] branch xingtanzjr/mpp_issues updated (fb7bf77ccc -> 946a69fd51)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a change to branch xingtanzjr/mpp_issues
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from fb7bf77ccc fix an issue that SourceHandle may become Blocked
add 83ffc19622 [IOTDB-2892] add PermissionInfoDataSet and test (#5492)
add a263e4c1a3 [IoTDB-2661] Add PlanNode visualizer tool (#5489)
add 016ed3d76c Combine InternalService with ManagementIService (#5578)
add e34854e20f add doc about about/Community(#5575)
add 37346fa551 [IOTDB-2619] Grafana Plugin: Alerting (#5366)
add 38364cf79d [IOTDB-2911] TransformOperator: operator implementation for nested expression evaluation (#5569)
add 77a33e9795 Remove mpp_mode property in iotdb-engine.properties (#5583)
add 3cc7eb2519 [IOTDB-2588] Triggers support any tree level, such as storage groups, devices, measurements, and so on (#5495)
add 8a00e48876 [IOTDB-2803][new cluster][mpp] refine show plan node ser/deserialize (#5581)
new bc62d7eb24 Merge branch 'master' into xingtanzjr/mpp_issues
new 946a69fd51 spotless
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/grafana-plugin.yml | 7 +-
.../iotdb/confignode/cli/TemporaryClient.java | 22 +-
.../consensus/response/PermissionInfoDataSet.java | 32 +-
.../iotdb/confignode/manager/ConfigManager.java | 19 +-
.../apache/iotdb/confignode/manager/Manager.java | 14 +-
.../confignode/manager/PermissionManager.java | 5 +
.../persistence/AuthorInfoPersistence.java | 129 +-
.../confignode/service/executor/PlanExecutor.java | 4 +-
.../server/ConfigNodeRPCServerProcessor.java | 36 +-
.../apache/iotdb/confignode/auth/AuthorTest.java | 44 -
.../server/ConfigNodeRPCServerProcessorTest.java | 375 ++
docs/Community/About.md | 79 +
grafana-plugin/.gitignore | 1 +
grafana-plugin/Magefile.go | 32 +
grafana-plugin/README.md | 38 +-
.../backend-compile.sh | 37 +-
grafana-plugin/go.mod | 23 +
grafana-plugin/go.sum | 659 +++
grafana-plugin/package.json | 7 +-
grafana-plugin/pkg/main.go | 40 +
grafana-plugin/pkg/plugin/plugin.go | 291 ++
grafana-plugin/pom.xml | 21 +
grafana-plugin/src/QueryEditor.tsx | 2 +-
grafana-plugin/src/datasource.ts | 130 +-
grafana-plugin/src/plugin.json | 3 +
grafana-plugin/yarn.lock | 4726 ++++++++++----------
.../db/integration/IoTDBTriggerExecutionIT.java | 188 +-
.../resources/conf/iotdb-engine.properties | 6 -
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 -
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 -
.../db/engine/trigger/executor/TriggerEngine.java | 20 +-
.../engine/trigger/executor/TriggerExecutor.java | 29 +-
.../service/TriggerRegistrationService.java | 52 +-
.../iotdb/db/metadata/LocalSchemaProcessor.java | 11 +-
.../idtable/entry/InsertMeasurementMNode.java | 20 +
.../org/apache/iotdb/db/metadata/mnode/IMNode.java | 8 +
.../org/apache/iotdb/db/metadata/mnode/MNode.java | 29 +
.../iotdb/db/metadata/mnode/MeasurementMNode.java | 2 -
.../mnode/estimator/BasicMNodSizeEstimator.java | 6 +-
.../db/metadata/schemaregion/ISchemaRegion.java | 5 +-
.../db/metadata/schemaregion/SchemaRegion.java | 11 +-
.../schemaregion/rocksdb/RSchemaRegion.java | 6 +-
.../schemaregion/rocksdb/mnode/RMNode.java | 20 +
.../iotdb/db/mpp/buffer/DataBlockManager.java | 25 +-
.../org/apache/iotdb/db/mpp/buffer/SinkHandle.java | 17 +-
.../apache/iotdb/db/mpp/buffer/SourceHandle.java | 31 +-
.../org/apache/iotdb/db/mpp/common/QueryId.java | 2 +-
.../apache/iotdb/db/mpp/execution/Coordinator.java | 2 +-
.../iotdb/db/mpp/execution/QueryExecution.java | 13 +-
...ageGroupTask.java => AuthorizerConfigTask.java} | 50 +-
.../scheduler/AbstractFragInsStateTracker.java | 5 +-
.../mpp/execution/scheduler/ClusterScheduler.java | 5 +-
.../scheduler/InternalServiceClientFactory.java | 7 +-
.../scheduler/SimpleFragInstanceDispatcher.java | 5 +-
.../execution/scheduler/SimpleQueryTerminator.java | 5 +-
.../db/mpp/operator/process/TransformOperator.java | 152 +-
.../operator/schema/DevicesSchemaScanOperator.java | 12 +-
.../db/mpp/operator/schema/SchemaScanOperator.java | 17 +-
.../schema/TimeSeriesSchemaScanOperator.java | 12 +-
.../db/mpp/operator/source/SeriesScanOperator.java | 4 +-
.../apache/iotdb/db/mpp/sql/analyze/Analyzer.java | 8 +-
.../db/mpp/sql/planner/LocalExecutionPlanner.java | 8 +-
.../iotdb/db/mpp/sql/planner/QueryPlanBuilder.java | 9 +-
.../sql/planner/plan/node/PlanGraphPrinter.java | 309 ++
.../db/mpp/sql/planner/plan/node/PlanNodeType.java | 12 +-
.../node/metedata/read/DevicesSchemaScanNode.java | 47 +-
.../plan/node/metedata/read/SchemaFetchNode.java | 6 -
.../plan/node/metedata/read/SchemaScanNode.java | 27 +-
.../plan/node/metedata/read/ShowDevicesNode.java | 67 -
.../planner/plan/node/metedata/read/ShowNode.java | 36 -
.../metedata/read/TimeSeriesSchemaScanNode.java | 51 +-
.../plan/node/metedata/write/AuthorNode.java | 2 +-
.../metadata/ShowStorageGroupStatement.java | 7 +
.../java/org/apache/iotdb/db/service/DataNode.java | 6 +-
.../iotdb/db/service/DataNodeManagementServer.java | 102 -
.../db/service/DataNodeManagementServerMBean.java | 22 -
.../apache/iotdb/db/service/InternalService.java | 4 +-
.../iotdb/db/service/InternalServiceImpl.java | 112 -
.../handler/DataNodeManagementServiceHandler.java | 58 -
.../handler}/InternalServiceThriftHandler.java | 2 +-
...ntServiceImpl.java => InternalServiceImpl.java} | 105 +-
.../apache/iotdb/db/utils/QueryDataSetUtils.java | 2 +-
.../apache/iotdb/db/mpp/buffer/SinkHandleTest.java | 16 +-
.../iotdb/db/mpp/buffer/SourceHandleTest.java | 30 +-
.../{ => schema}/SchemaScanOperatorTest.java | 20 +-
.../db/mpp/sql/plan/DistributionPlannerTest.java | 1 -
.../read/DeviceSchemaScanNodeSerdeTest.java | 76 +
.../metadata/read/ShowDevicesNodeSerdeTest.java | 42 -
.../read/TimeSeriesSchemaScanNodeSerdeTest.java | 79 +
.../plan/node/process/AggregateNodeSerdeTest.java | 15 +-
.../node/process/DeviceMergeNodeSerdeTest.java | 17 +-
.../plan/node/process/ExchangeNodeSerdeTest.java | 19 +-
.../sql/plan/node/process/FillNodeSerdeTest.java | 17 +-
.../sql/plan/node/process/FilterNodeSerdeTest.java | 17 +-
.../plan/node/process/FilterNullNodeSerdeTest.java | 17 +-
.../node/process/GroupByLevelNodeSerdeTest.java | 17 +-
.../sql/plan/node/process/LimitNodeSerdeTest.java | 17 +-
.../sql/plan/node/process/OffsetNodeSerdeTest.java | 15 +-
.../sql/plan/node/process/SortNodeSerdeTest.java | 18 +-
.../plan/node/process/TimeJoinNodeSerdeTest.java | 19 +-
.../plan/node/sink/FragmentSinkNodeSerdeTest.java | 7 +-
.../iotdb/db/service/InternalServiceImplTest.java | 1 +
.../src/main/thrift/confignode.thrift | 7 +
thrift/src/main/thrift/management.thrift | 77 -
thrift/src/main/thrift/mpp.thrift | 92 +-
.../tsfile/read/common/block/TsBlockBuilder.java | 15 +-
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 12 +
107 files changed, 5604 insertions(+), 3633 deletions(-)
delete mode 100644 confignode/src/test/java/org/apache/iotdb/confignode/auth/AuthorTest.java
create mode 100644 grafana-plugin/Magefile.go
copy cluster/src/assembly/resources/sbin/stop-node.sh => grafana-plugin/backend-compile.sh (58%)
mode change 100644 => 100755
create mode 100644 grafana-plugin/go.mod
create mode 100644 grafana-plugin/go.sum
create mode 100644 grafana-plugin/pkg/main.go
create mode 100644 grafana-plugin/pkg/plugin/plugin.go
copy server/src/main/java/org/apache/iotdb/db/mpp/execution/config/{SetStorageGroupTask.java => AuthorizerConfigTask.java} (55%)
create mode 100644 server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanGraphPrinter.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/ShowDevicesNode.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/ShowNode.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/service/DataNodeManagementServer.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/service/DataNodeManagementServerMBean.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/service/InternalServiceImpl.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/service/thrift/handler/DataNodeManagementServiceHandler.java
rename server/src/main/java/org/apache/iotdb/db/service/{ => thrift/handler}/InternalServiceThriftHandler.java (96%)
rename server/src/main/java/org/apache/iotdb/db/service/thrift/impl/{DataNodeManagementServiceImpl.java => InternalServiceImpl.java} (57%)
rename server/src/test/java/org/apache/iotdb/db/mpp/operator/{ => schema}/SchemaScanOperatorTest.java (95%)
create mode 100644 server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/node/metadata/read/DeviceSchemaScanNodeSerdeTest.java
delete mode 100644 server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/node/metadata/read/ShowDevicesNodeSerdeTest.java
create mode 100644 server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/node/metadata/read/TimeSeriesSchemaScanNodeSerdeTest.java
delete mode 100644 thrift/src/main/thrift/management.thrift
[iotdb] 01/02: Merge branch 'master' into xingtanzjr/mpp_issues
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/mpp_issues
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit bc62d7eb24482345bc30545025b39754d87b0f8d
Merge: fb7bf77ccc 8a00e48876
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Apr 19 15:12:54 2022 +0800
Merge branch 'master' into xingtanzjr/mpp_issues
.github/workflows/grafana-plugin.yml | 7 +-
.../iotdb/confignode/cli/TemporaryClient.java | 22 +-
.../consensus/response/PermissionInfoDataSet.java | 32 +-
.../iotdb/confignode/manager/ConfigManager.java | 19 +-
.../apache/iotdb/confignode/manager/Manager.java | 14 +-
.../confignode/manager/PermissionManager.java | 5 +
.../persistence/AuthorInfoPersistence.java | 129 +-
.../confignode/service/executor/PlanExecutor.java | 4 +-
.../server/ConfigNodeRPCServerProcessor.java | 36 +-
.../apache/iotdb/confignode/auth/AuthorTest.java | 44 -
.../server/ConfigNodeRPCServerProcessorTest.java | 375 ++
docs/Community/About.md | 79 +
grafana-plugin/.gitignore | 1 +
grafana-plugin/Magefile.go | 32 +
grafana-plugin/README.md | 38 +-
grafana-plugin/backend-compile.sh | 36 +
grafana-plugin/go.mod | 23 +
grafana-plugin/go.sum | 659 +++
grafana-plugin/package.json | 7 +-
grafana-plugin/pkg/main.go | 40 +
grafana-plugin/pkg/plugin/plugin.go | 291 ++
grafana-plugin/pom.xml | 21 +
grafana-plugin/src/QueryEditor.tsx | 2 +-
grafana-plugin/src/datasource.ts | 130 +-
grafana-plugin/src/plugin.json | 3 +
grafana-plugin/yarn.lock | 4726 ++++++++++----------
.../db/integration/IoTDBTriggerExecutionIT.java | 188 +-
.../resources/conf/iotdb-engine.properties | 6 -
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 -
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 -
.../db/engine/trigger/executor/TriggerEngine.java | 20 +-
.../engine/trigger/executor/TriggerExecutor.java | 29 +-
.../service/TriggerRegistrationService.java | 52 +-
.../iotdb/db/metadata/LocalSchemaProcessor.java | 11 +-
.../idtable/entry/InsertMeasurementMNode.java | 20 +
.../org/apache/iotdb/db/metadata/mnode/IMNode.java | 8 +
.../org/apache/iotdb/db/metadata/mnode/MNode.java | 29 +
.../iotdb/db/metadata/mnode/MeasurementMNode.java | 2 -
.../mnode/estimator/BasicMNodSizeEstimator.java | 6 +-
.../db/metadata/schemaregion/ISchemaRegion.java | 5 +-
.../db/metadata/schemaregion/SchemaRegion.java | 11 +-
.../schemaregion/rocksdb/RSchemaRegion.java | 6 +-
.../schemaregion/rocksdb/mnode/RMNode.java | 20 +
.../iotdb/db/mpp/buffer/DataBlockManager.java | 22 +-
.../org/apache/iotdb/db/mpp/buffer/SinkHandle.java | 17 +-
.../apache/iotdb/db/mpp/buffer/SourceHandle.java | 16 +-
.../org/apache/iotdb/db/mpp/common/QueryId.java | 2 +-
.../apache/iotdb/db/mpp/execution/Coordinator.java | 2 +-
.../mpp/execution/config/AuthorizerConfigTask.java | 95 +
.../scheduler/AbstractFragInsStateTracker.java | 5 +-
.../scheduler/InternalServiceClientFactory.java | 7 +-
.../scheduler/SimpleFragInstanceDispatcher.java | 4 +-
.../execution/scheduler/SimpleQueryTerminator.java | 5 +-
.../db/mpp/operator/process/TransformOperator.java | 152 +-
.../operator/schema/DevicesSchemaScanOperator.java | 12 +-
.../db/mpp/operator/schema/SchemaScanOperator.java | 17 +-
.../schema/TimeSeriesSchemaScanOperator.java | 12 +-
.../apache/iotdb/db/mpp/sql/analyze/Analyzer.java | 8 +-
.../db/mpp/sql/planner/LocalExecutionPlanner.java | 8 +-
.../iotdb/db/mpp/sql/planner/QueryPlanBuilder.java | 9 +-
.../sql/planner/plan/node/PlanGraphPrinter.java | 309 ++
.../db/mpp/sql/planner/plan/node/PlanNodeType.java | 12 +-
.../node/metedata/read/DevicesSchemaScanNode.java | 47 +-
.../plan/node/metedata/read/SchemaFetchNode.java | 6 -
.../plan/node/metedata/read/SchemaScanNode.java | 27 +-
.../plan/node/metedata/read/ShowDevicesNode.java | 67 -
.../planner/plan/node/metedata/read/ShowNode.java | 36 -
.../metedata/read/TimeSeriesSchemaScanNode.java | 51 +-
.../plan/node/metedata/write/AuthorNode.java | 2 +-
.../metadata/ShowStorageGroupStatement.java | 7 +
.../java/org/apache/iotdb/db/service/DataNode.java | 6 +-
.../iotdb/db/service/DataNodeManagementServer.java | 102 -
.../db/service/DataNodeManagementServerMBean.java | 22 -
.../apache/iotdb/db/service/InternalService.java | 4 +-
.../iotdb/db/service/InternalServiceImpl.java | 112 -
.../handler/DataNodeManagementServiceHandler.java | 58 -
.../handler}/InternalServiceThriftHandler.java | 2 +-
...ntServiceImpl.java => InternalServiceImpl.java} | 105 +-
.../apache/iotdb/db/mpp/buffer/SinkHandleTest.java | 16 +-
.../iotdb/db/mpp/buffer/SourceHandleTest.java | 30 +-
.../{ => schema}/SchemaScanOperatorTest.java | 20 +-
.../db/mpp/sql/plan/DistributionPlannerTest.java | 1 -
.../read/DeviceSchemaScanNodeSerdeTest.java | 76 +
.../metadata/read/ShowDevicesNodeSerdeTest.java | 42 -
.../read/TimeSeriesSchemaScanNodeSerdeTest.java | 79 +
.../plan/node/process/AggregateNodeSerdeTest.java | 15 +-
.../node/process/DeviceMergeNodeSerdeTest.java | 17 +-
.../plan/node/process/ExchangeNodeSerdeTest.java | 19 +-
.../sql/plan/node/process/FillNodeSerdeTest.java | 17 +-
.../sql/plan/node/process/FilterNodeSerdeTest.java | 17 +-
.../plan/node/process/FilterNullNodeSerdeTest.java | 17 +-
.../node/process/GroupByLevelNodeSerdeTest.java | 17 +-
.../sql/plan/node/process/LimitNodeSerdeTest.java | 17 +-
.../sql/plan/node/process/OffsetNodeSerdeTest.java | 15 +-
.../sql/plan/node/process/SortNodeSerdeTest.java | 18 +-
.../plan/node/process/TimeJoinNodeSerdeTest.java | 19 +-
.../plan/node/sink/FragmentSinkNodeSerdeTest.java | 7 +-
.../iotdb/db/service/InternalServiceImplTest.java | 1 +
.../src/main/thrift/confignode.thrift | 7 +
thrift/src/main/thrift/management.thrift | 77 -
thrift/src/main/thrift/mpp.thrift | 92 +-
.../tsfile/read/common/block/TsBlockBuilder.java | 15 +-
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 12 +
103 files changed, 5652 insertions(+), 3586 deletions(-)
diff --cc server/src/main/java/org/apache/iotdb/db/mpp/buffer/SinkHandle.java
index fc3cd825fe,af4ecf089d..c569c8d04e
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SinkHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SinkHandle.java
@@@ -160,10 -160,14 +160,10 @@@ public class SinkHandle implements ISin
}
private void sendEndOfDataBlockEvent() throws TException {
- logger.debug(
- "Send end of data block event to plan node {} of {}. {}",
- remotePlanNodeId,
- remoteFragmentInstanceId,
- Thread.currentThread().getName());
+ logger.info("[SinkHandle {}]: send end of data block event.", this.getRemotePlanNodeId());
int attempt = 0;
- EndOfDataBlockEvent endOfDataBlockEvent =
- new EndOfDataBlockEvent(
+ TEndOfDataBlockEvent endOfDataBlockEvent =
+ new TEndOfDataBlockEvent(
remoteFragmentInstanceId,
remotePlanNodeId,
localFragmentInstanceId,
diff --cc server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
index 01dc2f0fb1,aece366b82..d044be172e
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
@@@ -316,14 -307,15 +316,14 @@@ public class SourceHandle implements IS
@Override
public void run() {
- logger.debug(
- "Get data blocks [{}, {}) from {} for plan node {} of {}.",
+ logger.info(
+ "[SourceHandle-{}]: Get data blocks [{}, {}) from {}",
+ localPlanNodeId,
startSequenceId,
endSequenceId,
- remoteFragmentInstanceId,
- localPlanNodeId,
- localFragmentInstanceId);
+ remoteFragmentInstanceId);
- GetDataBlockRequest req =
- new GetDataBlockRequest(remoteFragmentInstanceId, startSequenceId, endSequenceId);
+ TGetDataBlockRequest req =
+ new TGetDataBlockRequest(remoteFragmentInstanceId, startSequenceId, endSequenceId);
int attempt = 0;
while (attempt < MAX_ATTEMPT_TIMES) {
attempt += 1;
diff --cc server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java
index b39ec04406,17a193d9b4..40b4df4a63
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java
@@@ -48,29 -46,33 +48,29 @@@ public class SimpleFragInstanceDispatch
return executor.submit(
() -> {
TSendFragmentInstanceResp resp = new TSendFragmentInstanceResp(false);
- try {
- for (FragmentInstance instance : instances) {
- // TODO: (jackie tien) change the port
- InternalService.Iface client =
- InternalServiceClientFactory.getInternalServiceClient(
- new Endpoint(
- instance.getHostEndpoint().getIp(),
- IoTDBDescriptor.getInstance().getConfig().getInternalPort()));
- // TODO: (xingtanzjr) consider how to handle the buffer here
- ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
- instance.serializeRequest(buffer);
- buffer.flip();
- TConsensusGroupId groupId =
- new TConsensusGroupId(
- instance.getRegionReplicaSet().getConsensusGroupId().getId(),
- instance.getRegionReplicaSet().getConsensusGroupId().getType().toString());
- TSendFragmentInstanceReq req =
- new TSendFragmentInstanceReq(
- new TFragmentInstance(buffer), groupId, instance.getType().toString());
- resp = client.sendFragmentInstance(req);
- if (!resp.accepted) {
- break;
- }
+ for (FragmentInstance instance : instances) {
+ InternalService.Iface client =
- InternalServiceClientFactory.getMppServiceClient(
++ InternalServiceClientFactory.getInternalServiceClient(
+ new Endpoint(
+ instance.getHostEndpoint().getIp(),
- IoTDBDescriptor.getInstance().getConfig().getMppPort()));
++ IoTDBDescriptor.getInstance().getConfig().getInternalPort()));
+ // TODO: (xingtanzjr) consider how to handle the buffer here
+ ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
+ instance.serializeRequest(buffer);
+ buffer.flip();
+ TConsensusGroupId groupId =
+ new TConsensusGroupId(
+ instance.getRegionReplicaSet().getConsensusGroupId().getId(),
+ instance.getRegionReplicaSet().getConsensusGroupId().getType().toString());
+ TSendFragmentInstanceReq req =
+ new TSendFragmentInstanceReq(
+ new TFragmentInstance(buffer), groupId, instance.getType().toString());
+ LOGGER.info("start to dispatch fragment instance: {}", instance.getId());
+ resp = client.sendFragmentInstance(req);
+ LOGGER.info("dispatch complete: {}", instance.getId());
+ if (!resp.accepted) {
+ break;
}
- } catch (Exception e) {
- // TODO: (xingtanzjr) add more details
- return new FragInstanceDispatchResult(false);
}
return new FragInstanceDispatchResult(resp.accepted);
});
[iotdb] 02/02: spotless
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/mpp_issues
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 946a69fd516b24f7578e63171fb4487890879fd3
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Apr 19 15:13:30 2022 +0800
spotless
---
.../org/apache/iotdb/db/mpp/buffer/DataBlockManager.java | 3 ++-
.../java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java | 15 ++++++++++-----
.../org/apache/iotdb/db/mpp/execution/QueryExecution.java | 13 ++++++++++---
.../db/mpp/execution/scheduler/ClusterScheduler.java | 5 ++++-
.../execution/scheduler/SimpleFragInstanceDispatcher.java | 1 +
.../iotdb/db/mpp/operator/source/SeriesScanOperator.java | 4 +++-
.../java/org/apache/iotdb/db/utils/QueryDataSetUtils.java | 2 +-
7 files changed, 31 insertions(+), 12 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java
index b5bf3f65a0..573771fe81 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java
@@ -180,7 +180,8 @@ public class DataBlockManager implements IDataBlockManager {
.get(sourceHandle.getLocalFragmentInstanceId())
.remove(sourceHandle.getLocalPlanNodeId());
}
- if (sourceHandles.containsKey(sourceHandle.getLocalFragmentInstanceId()) && sourceHandles.get(sourceHandle.getLocalFragmentInstanceId()).isEmpty()) {
+ if (sourceHandles.containsKey(sourceHandle.getLocalFragmentInstanceId())
+ && sourceHandles.get(sourceHandle.getLocalFragmentInstanceId()).isEmpty()) {
sourceHandles.remove(sourceHandle.getLocalFragmentInstanceId());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
index d044be172e..9a38876954 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
@@ -211,13 +211,19 @@ public class SourceHandle implements ISourceHandle {
}
synchronized void setNoMoreTsBlocks(int lastSequenceId) {
- logger.info("[SourceHandle {}]: No more TsBlock. {} ", localPlanNodeId, remoteFragmentInstanceId);
+ logger.info(
+ "[SourceHandle {}]: No more TsBlock. {} ", localPlanNodeId, remoteFragmentInstanceId);
this.lastSequenceId = lastSequenceId;
if (!blocked.isDone() && currSequenceId - 1 == lastSequenceId) {
- logger.info("[SourceHandle {}]: all blocks are consumed. set blocked to null.", localPlanNodeId);
+ logger.info(
+ "[SourceHandle {}]: all blocks are consumed. set blocked to null.", localPlanNodeId);
blocked.set(null);
} else {
- logger.info("[SourceHandle {}]: No need to set blocked. Blocked: {}, Consumed: {} ", localPlanNodeId, blocked.isDone(), currSequenceId - 1 == lastSequenceId);
+ logger.info(
+ "[SourceHandle {}]: No need to set blocked. Blocked: {}, Consumed: {} ",
+ localPlanNodeId,
+ blocked.isDone(),
+ currSequenceId - 1 == lastSequenceId);
}
}
@@ -251,8 +257,7 @@ public class SourceHandle implements ISourceHandle {
@Override
public boolean isFinished() {
- return throwable == null
- && currSequenceId - 1 == lastSequenceId;
+ return throwable == null && currSequenceId - 1 == lastSequenceId;
}
String getRemoteHostname() {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
index 380361c3fe..2b2ea1f762 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
@@ -110,8 +110,11 @@ public class QueryExecution implements IQueryExecution {
return;
}
this.stop();
- // TODO: (xingtanzjr) If the query is in abnormal state, the releaseResource() should be invoked
- if (state == QueryState.FAILED || state == QueryState.ABORTED || state == QueryState.CANCELED) {
+ // TODO: (xingtanzjr) If the query is in abnormal state, the releaseResource() should be
+ // invoked
+ if (state == QueryState.FAILED
+ || state == QueryState.ABORTED
+ || state == QueryState.CANCELED) {
releaseResource();
}
});
@@ -206,7 +209,11 @@ public class QueryExecution implements IQueryExecution {
LOG.info("[QueryExecution {}]: try to get result.", context.getQueryId());
ListenableFuture<Void> blocked = resultHandle.isBlocked();
blocked.get();
- LOG.info("[QueryExecution {}]: unblock. Cancelled: {}, Done: {}", context.getQueryId(), blocked.isCancelled(), blocked.isDone());
+ LOG.info(
+ "[QueryExecution {}]: unblock. Cancelled: {}, Done: {}",
+ context.getQueryId(),
+ blocked.isCancelled(),
+ blocked.isDone());
if (resultHandle.isFinished()) {
LOG.info("[QueryExecution {}]: result is null", context.getQueryId());
releaseResource();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java
index 47c57c873d..694a82738f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java
@@ -83,7 +83,10 @@ public class ClusterScheduler implements IScheduler {
@Override
public void start() {
- LOGGER.info("[{}] start to dispatch fragment instance. size: {}", queryContext.getQueryId(), instances.size());
+ LOGGER.info(
+ "[{}] start to dispatch fragment instance. size: {}",
+ queryContext.getQueryId(),
+ instances.size());
stateMachine.transitionToDispatching();
Future<FragInstanceDispatchResult> dispatchResultFuture = dispatcher.dispatch(instances);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java
index 40b4df4a63..8745dfe95b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/SimpleFragInstanceDispatcher.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstance;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java
index b11f643810..2927cf0d77 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java
@@ -107,7 +107,9 @@ public class SeriesScanOperator implements DataSourceOperator {
return true;
}
}
- System.out.println(String.format("[SeriesScanOperator-%s]: hasNext returned: %s", sourceId, hasCachedTsBlock));
+ System.out.println(
+ String.format(
+ "[SeriesScanOperator-%s]: hasNext returned: %s", sourceId, hasCachedTsBlock));
return hasCachedTsBlock;
} catch (IOException e) {
throw new RuntimeException("Error happened while scanning the file", e);
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
index aca0b37906..d255f03ddd 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.db.utils;
-import org.apache.iotdb.db.mpp.execution.Coordinator;
import org.apache.iotdb.db.mpp.execution.IQueryExecution;
import org.apache.iotdb.db.tools.watermark.WatermarkEncoder;
import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
@@ -32,6 +31,7 @@ import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.BitMap;
import org.apache.iotdb.tsfile.utils.BytesUtils;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;