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/06/20 14:58:48 UTC
[iotdb] branch xingtanzjr/write_opt_version1 created (now 22ee02fe43)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a change to branch xingtanzjr/write_opt_version1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 22ee02fe43 fix the issue in queryTerminator
This branch includes the following new commits:
new e84d20e81b stable version but has getStatus() issue
new 22ee02fe43 fix the issue in queryTerminator
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.
[iotdb] 01/02: stable version but has getStatus() issue
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/write_opt_version1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e84d20e81b33d5bb28951488bd5a95488b88afc3
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Mon Jun 20 20:29:20 2022 +0800
stable version but has getStatus() issue
---
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 48 ++++----
.../org/apache/iotdb/db/mpp/plan/StepTracker.java | 73 ++++++++++++
.../db/mpp/plan/execution/QueryExecution.java | 130 ++++++++++++---------
.../planner/distribution/DistributionPlanner.java | 3 -
.../db/mpp/plan/scheduler/ClusterScheduler.java | 4 +-
.../db/mpp/plan/scheduler/StandaloneScheduler.java | 1 -
6 files changed, 176 insertions(+), 83 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
index 111a9222a6..495af1f3e0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
@@ -109,29 +109,33 @@ public class Coordinator {
String sql,
IPartitionFetcher partitionFetcher,
ISchemaFetcher schemaFetcher) {
-
- QueryId globalQueryId = queryIdGenerator.createNextQueryId();
- try (SetThreadName queryName = new SetThreadName(globalQueryId.getId())) {
- if (sql != null) {
- LOGGER.info("start executing sql: {}", sql);
- }
- IQueryExecution execution =
- createQueryExecution(
- statement,
- new MPPQueryContext(
- sql,
- globalQueryId,
- session,
- DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT,
- DataNodeEndPoints.LOCAL_HOST_INTERNAL_ENDPOINT),
- partitionFetcher,
- schemaFetcher);
- if (execution.isQuery()) {
- queryExecutionMap.put(queryId, execution);
+ long startTime = System.nanoTime();
+ try {
+ QueryId globalQueryId = queryIdGenerator.createNextQueryId();
+ try (SetThreadName queryName = new SetThreadName(globalQueryId.getId())) {
+ // if (sql != null) {
+ // LOGGER.info("start executing sql: {}", sql);
+ // }
+ IQueryExecution execution =
+ createQueryExecution(
+ statement,
+ new MPPQueryContext(
+ sql,
+ globalQueryId,
+ session,
+ DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT,
+ DataNodeEndPoints.LOCAL_HOST_INTERNAL_ENDPOINT),
+ partitionFetcher,
+ schemaFetcher);
+ if (execution.isQuery()) {
+ queryExecutionMap.put(queryId, execution);
+ }
+ execution.start();
+
+ return execution.getStatus();
}
- execution.start();
-
- return execution.getStatus();
+ } finally {
+ StepTracker.trace("statementExec", startTime, System.nanoTime());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
new file mode 100644
index 0000000000..1fe7df7be0
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.plan;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+class Metric {
+ private static final Logger logger = LoggerFactory.getLogger(Metric.class);
+ public String stepName;
+ public long invokeCount;
+ public long totalTime;
+ public long last100Time;
+
+ public Metric(String stepName) {
+ this.stepName = stepName;
+ this.invokeCount = 0;
+ this.totalTime = 0;
+ this.last100Time = 0;
+ }
+
+ public void trace(long startTime, long endTime) {
+ this.invokeCount++;
+ this.totalTime += (endTime - startTime);
+ }
+
+ public void tryPrint() {
+ if (invokeCount % 100 == 0) {
+ logger.info(
+ String.format(
+ "step metrics [%d]-[%s] - Total: %d, SUM: %.2fms, AVG: %fms, Last100AVG: %fms",
+ Thread.currentThread().getId(),
+ stepName,
+ invokeCount,
+ totalTime * 1.0 / 1000000,
+ totalTime * 1.0 / 1000000 / invokeCount,
+ (totalTime * 1.0 - last100Time) / 1000000 / 100));
+ last100Time = totalTime;
+ }
+ }
+}
+
+public class StepTracker {
+ private static final ThreadLocal<Map<String, Metric>> metrics = new ThreadLocal<>();
+
+ public static void trace(String stepName, long startTime, long endTime) {
+ if (metrics.get() == null) {
+ metrics.set(new HashMap<>());
+ }
+ metrics.get().computeIfAbsent(stepName, Metric::new).trace(startTime, endTime);
+ metrics.get().get(stepName).tryPrint();
+ }
+}
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 5f4ce1ae3f..a7a39564ef 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
@@ -31,6 +31,7 @@ import org.apache.iotdb.db.mpp.execution.QueryState;
import org.apache.iotdb.db.mpp.execution.QueryStateMachine;
import org.apache.iotdb.db.mpp.execution.datatransfer.DataBlockService;
import org.apache.iotdb.db.mpp.execution.datatransfer.ISourceHandle;
+import org.apache.iotdb.db.mpp.plan.StepTracker;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.analyze.Analyzer;
import org.apache.iotdb.db.mpp.plan.analyze.IPartitionFetcher;
@@ -46,7 +47,6 @@ import org.apache.iotdb.db.mpp.plan.planner.distribution.DistributionPlanner;
import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeUtil;
import org.apache.iotdb.db.mpp.plan.scheduler.ClusterScheduler;
import org.apache.iotdb.db.mpp.plan.scheduler.IScheduler;
import org.apache.iotdb.db.mpp.plan.scheduler.StandaloneScheduler;
@@ -113,6 +113,8 @@ public class QueryExecution implements IQueryExecution {
private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
internalServiceClientManager;
+ private long queryExecutionStartTime;
+
public QueryExecution(
Statement statement,
MPPQueryContext context,
@@ -122,12 +124,15 @@ public class QueryExecution implements IQueryExecution {
IPartitionFetcher partitionFetcher,
ISchemaFetcher schemaFetcher,
IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> internalServiceClientManager) {
+ this.queryExecutionStartTime = System.nanoTime();
this.executor = executor;
this.writeOperationExecutor = writeOperationExecutor;
this.scheduledExecutor = scheduledExecutor;
this.context = context;
this.planOptimizers = new ArrayList<>();
+ long startTime = System.nanoTime();
this.analysis = analyze(statement, context, partitionFetcher, schemaFetcher);
+ StepTracker.trace("analyze", startTime, System.nanoTime());
this.stateMachine = new QueryStateMachine(context.getQueryId(), executor);
this.partitionFetcher = partitionFetcher;
this.schemaFetcher = schemaFetcher;
@@ -189,52 +194,59 @@ public class QueryExecution implements IQueryExecution {
IPartitionFetcher partitionFetcher,
ISchemaFetcher schemaFetcher) {
// initialize the variable `analysis`
- logger.info("start to analyze query");
+ // logger.info("start to analyze query");
return new Analyzer(context, partitionFetcher, schemaFetcher).analyze(statement);
}
private void schedule() {
- // TODO: (xingtanzjr) initialize the query scheduler according to configuration
- this.scheduler =
- config.isClusterMode()
- ? new ClusterScheduler(
- context,
- stateMachine,
- distributedPlan.getInstances(),
- context.getQueryType(),
- executor,
- writeOperationExecutor,
- scheduledExecutor,
- internalServiceClientManager)
- : new StandaloneScheduler(
- context,
- stateMachine,
- distributedPlan.getInstances(),
- context.getQueryType(),
- executor,
- scheduledExecutor,
- internalServiceClientManager);
- this.scheduler.start();
+ long startTime = System.nanoTime();
+ try {
+ // TODO: (xingtanzjr) initialize the query scheduler according to configuration
+ this.scheduler =
+ config.isClusterMode()
+ ? new ClusterScheduler(
+ context,
+ stateMachine,
+ distributedPlan.getInstances(),
+ context.getQueryType(),
+ executor,
+ writeOperationExecutor,
+ scheduledExecutor,
+ internalServiceClientManager)
+ : new StandaloneScheduler(
+ context,
+ stateMachine,
+ distributedPlan.getInstances(),
+ context.getQueryType(),
+ executor,
+ scheduledExecutor,
+ internalServiceClientManager);
+ this.scheduler.start();
+ } finally {
+ StepTracker.trace("dispatch", startTime, System.nanoTime());
+ }
}
// Use LogicalPlanner to do the logical query plan and logical optimization
public void doLogicalPlan() {
- logger.info("do logical plan...");
- LogicalPlanner planner = new LogicalPlanner(this.context, this.planOptimizers);
- this.logicalPlan = planner.plan(this.analysis);
- logger.info(
- "logical plan is: \n {}", PlanNodeUtil.nodeToString(this.logicalPlan.getRootNode()));
+ long startTime = System.nanoTime();
+ try {
+ LogicalPlanner planner = new LogicalPlanner(this.context, this.planOptimizers);
+ this.logicalPlan = planner.plan(this.analysis);
+ } finally {
+ StepTracker.trace("doLogicalPlan", startTime, System.nanoTime());
+ }
}
// Generate the distributed plan and split it into fragments
public void doDistributedPlan() {
- logger.info("do distribution plan...");
- DistributionPlanner planner = new DistributionPlanner(this.analysis, this.logicalPlan);
- this.distributedPlan = planner.planFragments();
- logger.info(
- "distribution plan done. Fragment instance count is {}, details is: \n {}",
- distributedPlan.getInstances().size(),
- printFragmentInstances(distributedPlan.getInstances()));
+ long startTime = System.nanoTime();
+ try {
+ DistributionPlanner planner = new DistributionPlanner(this.analysis, this.logicalPlan);
+ this.distributedPlan = planner.planFragments();
+ } finally {
+ StepTracker.trace("doDistributionPlan", startTime, System.nanoTime());
+ }
}
private String printFragmentInstances(List<FragmentInstance> instances) {
@@ -332,28 +344,36 @@ public class QueryExecution implements IQueryExecution {
* @return ExecutionStatus. Contains the QueryId and the TSStatus.
*/
public ExecutionResult getStatus() {
- // Although we monitor the state to transition to RUNNING, the future will return if any
- // Terminated state is triggered
- SettableFuture<QueryState> future = SettableFuture.create();
- stateMachine.addStateChangeListener(
- state -> {
- if (state == QueryState.RUNNING || state.isDone()) {
- future.set(state);
- }
- });
-
+ long startTime = System.nanoTime();
try {
- QueryState state = future.get();
- // TODO: (xingtanzjr) use more TSStatusCode if the QueryState isn't FINISHED
- return getExecutionResult(state);
- } catch (InterruptedException | ExecutionException e) {
- // TODO: (xingtanzjr) use more accurate error handling
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
+ // Although we monitor the state to transition to RUNNING, the future will return if any
+ // Terminated state is triggered
+ try {
+ SettableFuture<QueryState> future = SettableFuture.create();
+ stateMachine.addStateChangeListener(
+ state -> {
+ if (state == QueryState.RUNNING || state.isDone()) {
+ future.set(state);
+ }
+ });
+ long startTimeGet = System.nanoTime();
+ QueryState state = future.get();
+ StepTracker.trace("get()", startTimeGet, System.nanoTime());
+ // TODO: (xingtanzjr) use more TSStatusCode if the QueryState isn't FINISHED
+ return getExecutionResult(state);
+ } catch (InterruptedException | ExecutionException e) {
+ // TODO: (xingtanzjr) use more accurate error handling
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ return new ExecutionResult(
+ context.getQueryId(),
+ RpcUtils.getStatus(
+ TSStatusCode.INTERNAL_SERVER_ERROR, stateMachine.getFailureMessage()));
}
- return new ExecutionResult(
- context.getQueryId(),
- RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, stateMachine.getFailureMessage()));
+ } finally {
+ StepTracker.trace("getStatus", startTime, System.nanoTime());
+ StepTracker.trace("QueryExecutionLifeCycle", this.queryExecutionStartTime, System.nanoTime());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
index 421de275da..4dfe1b968f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
@@ -29,7 +29,6 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.mpp.plan.planner.plan.PlanFragment;
import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeUtil;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
@@ -67,9 +66,7 @@ public class DistributionPlanner {
public DistributedQueryPlan planFragments() {
PlanNode rootAfterRewrite = rewriteSource();
- System.out.println(PlanNodeUtil.nodeToString(rootAfterRewrite));
PlanNode rootWithExchange = addExchangeNode(rootAfterRewrite);
- System.out.println(PlanNodeUtil.nodeToString(rootWithExchange));
if (analysis.getStatement() instanceof QueryStatement) {
analysis
.getRespDatasetHeader()
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java
index 9437c22536..8390ac6fe5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java
@@ -86,7 +86,7 @@ public class ClusterScheduler implements IScheduler {
@Override
public void start() {
stateMachine.transitionToDispatching();
- logger.info("transit to DISPATCHING");
+ // logger.info("transit to DISPATCHING");
Future<FragInstanceDispatchResult> dispatchResultFuture = dispatcher.dispatch(instances);
// NOTICE: the FragmentInstance may be dispatched to another Host due to consensus redirect.
@@ -116,7 +116,7 @@ public class ClusterScheduler implements IScheduler {
// The FragmentInstances has been dispatched successfully to corresponding host, we mark the
// QueryState to Running
stateMachine.transitionToRunning();
- logger.info("transit to RUNNING");
+ // logger.info("transit to RUNNING");
instances.forEach(
instance -> {
stateMachine.initialFragInstanceState(instance.getId(), FragmentInstanceState.RUNNING);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java
index 89bacfb562..5433b07c7c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java
@@ -98,7 +98,6 @@ public class StandaloneScheduler implements IScheduler {
@Override
public void start() {
stateMachine.transitionToDispatching();
- LOGGER.info("{} transit to DISPATCHING", getLogHeader());
// For the FragmentInstance of WRITE, it will be executed directly when dispatching.
// TODO: Other QueryTypes
switch (queryType) {
[iotdb] 02/02: fix the issue in queryTerminator
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/write_opt_version1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 22ee02fe43b9deeb0c668e03a2e22024be4eed66
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Mon Jun 20 22:58:35 2022 +0800
fix the issue in queryTerminator
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 4 +-
.../iotdb/db/mpp/execution/StateMachine.java | 7 ++-
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 2 +-
.../org/apache/iotdb/db/mpp/plan/StepTracker.java | 15 +++--
.../db/mpp/plan/analyze/ClusterSchemaFetcher.java | 11 ++++
.../db/mpp/plan/execution/QueryExecution.java | 5 ++
.../db/mpp/plan/scheduler/ClusterScheduler.java | 30 ++++++----
.../mpp/plan/scheduler/SimpleQueryTerminator.java | 65 +++++++++++++---------
.../db/mpp/plan/scheduler/StandaloneScheduler.java | 6 +-
9 files changed, 98 insertions(+), 47 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 5de782e678..1ca1c2c794 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -928,10 +928,10 @@ public class IoTDBConfig {
private int triggerForwardMQTTPoolSize = 4;
/** ThreadPool size for read operation in coordinator */
- private int coordinatorReadExecutorSize = 100;
+ private int coordinatorReadExecutorSize = 50;
/** ThreadPool size for write operation in coordinator */
- private int coordinatorWriteExecutorSize = 100;
+ private int coordinatorWriteExecutorSize = 50;
IoTDBConfig() {}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/StateMachine.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/StateMachine.java
index c1b7e76360..e53c2cb02d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/StateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/StateMachine.java
@@ -13,6 +13,8 @@
*/
package org.apache.iotdb.db.mpp.execution;
+import org.apache.iotdb.db.mpp.plan.StepTracker;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -270,6 +272,7 @@ public class StateMachine<T> {
* execution. The listener is notified immediately of the current state.
*/
public void addStateChangeListener(StateChangeListener<T> stateChangeListener) {
+ long startTime = System.nanoTime();
requireNonNull(stateChangeListener, "stateChangeListener is null");
boolean inTerminalState;
@@ -281,7 +284,7 @@ public class StateMachine<T> {
stateChangeListeners.add(stateChangeListener);
}
}
-
+ StepTracker.trace("addStateListener", startTime, System.nanoTime());
// fire state change listener with the current state
// always fire listener callbacks from a different thread
safeExecute(() -> stateChangeListener.stateChanged(currentState));
@@ -310,6 +313,8 @@ public class StateMachine<T> {
private void safeExecute(Runnable command) {
try {
+ // new Thread(command).start();
+ // command.run();
executor.execute(command);
} catch (RejectedExecutionException e) {
if ((executor instanceof ExecutorService) && ((ExecutorService) executor).isShutdown()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
index 495af1f3e0..f9a66b7ec1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
@@ -57,7 +57,7 @@ public class Coordinator {
private static final String COORDINATOR_EXECUTOR_NAME = "MPPCoordinator";
private static final String COORDINATOR_WRITE_EXECUTOR_NAME = "MPPCoordinatorWrite";
private static final String COORDINATOR_SCHEDULED_EXECUTOR_NAME = "MPPCoordinatorScheduled";
- private static final int COORDINATOR_SCHEDULED_EXECUTOR_SIZE = 1;
+ private static final int COORDINATOR_SCHEDULED_EXECUTOR_SIZE = 10;
private static final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
INTERNAL_SERVICE_CLIENT_MANAGER =
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
index 1fe7df7be0..9d3da35f72 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
@@ -27,16 +27,18 @@ import java.util.Map;
class Metric {
private static final Logger logger = LoggerFactory.getLogger(Metric.class);
+ private static final int PRINT_RATE = 100;
+
public String stepName;
public long invokeCount;
public long totalTime;
- public long last100Time;
+ public long lastCycleTime;
public Metric(String stepName) {
this.stepName = stepName;
this.invokeCount = 0;
this.totalTime = 0;
- this.last100Time = 0;
+ this.lastCycleTime = 0;
}
public void trace(long startTime, long endTime) {
@@ -45,17 +47,18 @@ class Metric {
}
public void tryPrint() {
- if (invokeCount % 100 == 0) {
+ if (invokeCount % PRINT_RATE == 0) {
logger.info(
String.format(
- "step metrics [%d]-[%s] - Total: %d, SUM: %.2fms, AVG: %fms, Last100AVG: %fms",
+ "step metrics [%d]-[%s] - Total: %d, SUM: %.2fms, AVG: %fms, Last%dAVG: %fms",
Thread.currentThread().getId(),
stepName,
invokeCount,
totalTime * 1.0 / 1000000,
totalTime * 1.0 / 1000000 / invokeCount,
- (totalTime * 1.0 - last100Time) / 1000000 / 100));
- last100Time = totalTime;
+ PRINT_RATE,
+ (totalTime * 1.0 - lastCycleTime) / 1000000 / PRINT_RATE));
+ lastCycleTime = totalTime;
}
}
}
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 84db6df7f7..9f72dc4802 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
@@ -96,10 +96,21 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
schemaFetchStatement.setSchemaPartition(schemaPartition);
SchemaTree result = executeSchemaFetchQuery(schemaFetchStatement);
+ // SchemaTree result = mockFetch(patternTree);
result.setStorageGroups(storageGroups);
return result;
}
+ public SchemaTree mockFetch(PathPatternTree patternTree) {
+ SchemaTree tree = new SchemaTree();
+ List<PartialPath> paths = patternTree.getAllPathPatterns();
+ for (PartialPath path : paths) {
+ tree.appendSingleMeasurement(
+ path, new MeasurementSchema(path.getMeasurement(), TSDataType.FLOAT), null, false);
+ }
+ return tree;
+ }
+
private SchemaTree executeSchemaFetchQuery(SchemaFetchStatement schemaFetchStatement) {
long queryId = SessionManager.getInstance().requestQueryId(false);
try {
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 a7a39564ef..d2d8b1c3dd 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
@@ -349,9 +349,14 @@ public class QueryExecution implements IQueryExecution {
// Although we monitor the state to transition to RUNNING, the future will return if any
// Terminated state is triggered
try {
+ if (stateMachine.getState() == QueryState.FINISHED) {
+ return getExecutionResult(QueryState.FINISHED);
+ }
SettableFuture<QueryState> future = SettableFuture.create();
+ final long addStart = System.nanoTime();
stateMachine.addStateChangeListener(
state -> {
+ StepTracker.trace("stateQueue", addStart, System.nanoTime());
if (state == QueryState.RUNNING || state.isDone()) {
future.set(state);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java
index 8390ac6fe5..989b025660 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/ClusterScheduler.java
@@ -57,8 +57,8 @@ public class ClusterScheduler implements IScheduler {
private final List<FragmentInstance> instances;
private final IFragInstanceDispatcher dispatcher;
- private final IFragInstanceStateTracker stateTracker;
- private final IQueryTerminator queryTerminator;
+ private IFragInstanceStateTracker stateTracker;
+ private IQueryTerminator queryTerminator;
public ClusterScheduler(
MPPQueryContext queryContext,
@@ -75,12 +75,18 @@ public class ClusterScheduler implements IScheduler {
this.dispatcher =
new FragmentInstanceDispatcherImpl(
queryType, executor, writeOperationExecutor, internalServiceClientManager);
- this.stateTracker =
- new FixedRateFragInsStateTracker(
- stateMachine, executor, scheduledExecutor, instances, internalServiceClientManager);
- this.queryTerminator =
- new SimpleQueryTerminator(
- executor, queryContext.getQueryId(), instances, internalServiceClientManager);
+ if (queryType == QueryType.READ) {
+ this.stateTracker =
+ new FixedRateFragInsStateTracker(
+ stateMachine, executor, scheduledExecutor, instances, internalServiceClientManager);
+ this.queryTerminator =
+ new SimpleQueryTerminator(
+ executor,
+ scheduledExecutor,
+ queryContext.getQueryId(),
+ instances,
+ internalServiceClientManager);
+ }
}
@Override
@@ -132,9 +138,13 @@ public class ClusterScheduler implements IScheduler {
// TODO: It seems that it is unnecessary to check whether they are null or not. Is it a best
// practice ?
dispatcher.abort();
- stateTracker.abort();
+ if (stateTracker != null) {
+ stateTracker.abort();
+ }
// TODO: (xingtanzjr) handle the exception when the termination cannot succeed
- queryTerminator.terminate();
+ if (queryTerminator != null) {
+ queryTerminator.terminate();
+ }
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java
index 357c26fcf5..55e6047555 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java
@@ -32,67 +32,80 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class SimpleQueryTerminator implements IQueryTerminator {
private static final Logger logger = LoggerFactory.getLogger(SimpleQueryTerminator.class);
private static final long TERMINATION_GRACE_PERIOD_IN_MS = 1000L;
private final ExecutorService executor;
+ protected ScheduledExecutorService scheduledExecutor;
private final QueryId queryId;
- private final List<FragmentInstance> fragmentInstances;
+ private List<TEndPoint> relatedHost;
+ private Map<TEndPoint, List<TFragmentInstanceId>> ownedFragmentInstance;
private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
internalServiceClientManager;
public SimpleQueryTerminator(
ExecutorService executor,
+ ScheduledExecutorService scheduledExecutor,
QueryId queryId,
List<FragmentInstance> fragmentInstances,
IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> internalServiceClientManager) {
this.executor = executor;
+ this.scheduledExecutor = scheduledExecutor;
this.queryId = queryId;
- this.fragmentInstances = fragmentInstances;
this.internalServiceClientManager = internalServiceClientManager;
+ calculateParameter(fragmentInstances);
+ }
+
+ private void calculateParameter(List<FragmentInstance> instances) {
+ this.relatedHost = getRelatedHost(instances);
+ this.ownedFragmentInstance = new HashMap<>();
+ for (TEndPoint endPoint : relatedHost) {
+ ownedFragmentInstance.put(endPoint, getRelatedFragmentInstances(endPoint, instances));
+ }
}
@Override
public Future<Boolean> terminate() {
- List<TEndPoint> relatedHost = getRelatedHost();
- try {
- Thread.sleep(TERMINATION_GRACE_PERIOD_IN_MS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ return scheduledExecutor.schedule(
+ this::syncTerminate, TERMINATION_GRACE_PERIOD_IN_MS, TimeUnit.MILLISECONDS);
+ }
+
+ public Boolean syncTerminate() {
+ for (TEndPoint endPoint : relatedHost) {
+ // TODO (jackie tien) change the port
+ try (SyncDataNodeInternalServiceClient client =
+ internalServiceClientManager.borrowClient(endPoint)) {
+ client.cancelQuery(
+ new TCancelQueryReq(queryId.getId(), ownedFragmentInstance.get(endPoint)));
+ } catch (IOException e) {
+ logger.error("can't connect to node {}", endPoint, e);
+ return false;
+ } catch (TException e) {
+ return false;
+ }
}
- return executor.submit(
- () -> {
- for (TEndPoint endPoint : relatedHost) {
- // TODO (jackie tien) change the port
- try (SyncDataNodeInternalServiceClient client =
- internalServiceClientManager.borrowClient(endPoint)) {
- client.cancelQuery(
- new TCancelQueryReq(queryId.getId(), getRelatedFragmentInstances(endPoint)));
- } catch (IOException e) {
- logger.error("can't connect to node {}", endPoint, e);
- return false;
- } catch (TException e) {
- return false;
- }
- }
- return true;
- });
+ return true;
}
- private List<TEndPoint> getRelatedHost() {
+ private List<TEndPoint> getRelatedHost(List<FragmentInstance> fragmentInstances) {
return fragmentInstances.stream()
.map(instance -> instance.getHostDataNode().internalEndPoint)
.distinct()
.collect(Collectors.toList());
}
- private List<TFragmentInstanceId> getRelatedFragmentInstances(TEndPoint endPoint) {
+ private List<TFragmentInstanceId> getRelatedFragmentInstances(
+ TEndPoint endPoint, List<FragmentInstance> fragmentInstances) {
return fragmentInstances.stream()
.filter(instance -> instance.getHostDataNode().internalEndPoint.equals(endPoint))
.map(instance -> instance.getId().toThrift())
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java
index 5433b07c7c..f4b7664c57 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/StandaloneScheduler.java
@@ -91,7 +91,11 @@ public class StandaloneScheduler implements IScheduler {
stateMachine, executor, scheduledExecutor, instances, internalServiceClientManager);
this.queryTerminator =
new SimpleQueryTerminator(
- executor, queryContext.getQueryId(), instances, internalServiceClientManager);
+ executor,
+ scheduledExecutor,
+ queryContext.getQueryId(),
+ instances,
+ internalServiceClientManager);
}
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning