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/21 03:48:39 UTC
[iotdb] branch xingtanzjr/write_opt_version1 updated (5887294c79 -> b29b8bb165)
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
from 5887294c79 Merge branch 'master' into xingtanzjr/write_opt_version1
new feb198b2f8 remove some test codes
new ef3d5abba7 re-format code
new b29b8bb165 spotless
The 3 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:
.../iotdb/confignode/manager/PartitionManager.java | 8 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +-
.../iotdb/db/mpp/execution/StateMachine.java | 7 +-
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 47 ++++---
.../db/mpp/plan/analyze/ClusterSchemaFetcher.java | 11 --
.../db/mpp/plan/execution/QueryExecution.java | 135 +++++++++------------
.../db/mpp/plan/scheduler/ClusterScheduler.java | 2 -
.../mpp/plan/scheduler/SimpleQueryTerminator.java | 1 -
8 files changed, 84 insertions(+), 129 deletions(-)
[iotdb] 02/03: re-format code
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 ef3d5abba7e835217e8e3895c9329226bb7cd173
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Jun 21 11:47:05 2022 +0800
re-format code
---
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 11 +-
.../db/mpp/plan/execution/QueryExecution.java | 173 +++++++++------------
.../db/mpp/plan/scheduler/ClusterScheduler.java | 2 -
.../mpp/plan/scheduler/SimpleQueryTerminator.java | 1 -
4 files changed, 80 insertions(+), 107 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 f9a66b7ec1..dc7f6b485e 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,13 +109,11 @@ public class Coordinator {
String sql,
IPartitionFetcher partitionFetcher,
ISchemaFetcher schemaFetcher) {
- 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);
- // }
+ if (sql != null) {
+ LOGGER.info("start executing sql: {}", sql);
+ }
IQueryExecution execution =
createQueryExecution(
statement,
@@ -134,9 +132,6 @@ public class Coordinator {
return execution.getStatus();
}
- } finally {
- StepTracker.trace("statementExec", startTime, System.nanoTime());
- }
}
public IQueryExecution getQueryExecution(Long queryId) {
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 d2d8b1c3dd..24a7537fa1 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
@@ -47,6 +47,7 @@ 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,8 +114,6 @@ public class QueryExecution implements IQueryExecution {
private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
internalServiceClientManager;
- private long queryExecutionStartTime;
-
public QueryExecution(
Statement statement,
MPPQueryContext context,
@@ -124,15 +123,12 @@ 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;
@@ -193,60 +189,49 @@ public class QueryExecution implements IQueryExecution {
MPPQueryContext context,
IPartitionFetcher partitionFetcher,
ISchemaFetcher schemaFetcher) {
- // initialize the variable `analysis`
- // logger.info("start to analyze query");
return new Analyzer(context, partitionFetcher, schemaFetcher).analyze(statement);
}
private void schedule() {
- 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());
- }
+ // 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();
}
// Use LogicalPlanner to do the logical query plan and logical optimization
public void doLogicalPlan() {
- 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());
- }
+ 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()));
}
// Generate the distributed plan and split it into fragments
public void doDistributedPlan() {
- long startTime = System.nanoTime();
- try {
- DistributionPlanner planner = new DistributionPlanner(this.analysis, this.logicalPlan);
- this.distributedPlan = planner.planFragments();
- } finally {
- StepTracker.trace("doDistributionPlan", startTime, System.nanoTime());
- }
+ 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()));
}
private String printFragmentInstances(List<FragmentInstance> instances) {
@@ -270,7 +255,9 @@ public class QueryExecution implements IQueryExecution {
releaseResource();
}
- /** Release the resources that current QueryExecution hold. */
+ /**
+ * Release the resources that current QueryExecution hold.
+ */
private void releaseResource() {
// close ResultHandle to unblock client's getResult request
// Actually, we should not close the ResultHandle when the QueryExecution is Finished.
@@ -320,13 +307,17 @@ public class QueryExecution implements IQueryExecution {
}
}
- /** @return true if there is more tsblocks, otherwise false */
+ /**
+ * @return true if there is more tsblocks, otherwise false
+ */
@Override
public boolean hasNextResult() {
return resultHandle != null && !resultHandle.isFinished();
}
- /** return the result column count without the time column */
+ /**
+ * return the result column count without the time column
+ */
@Override
public int getOutputValueColumnCount() {
return analysis.getRespDatasetHeader().getOutputValueColumnCount();
@@ -344,41 +335,31 @@ public class QueryExecution implements IQueryExecution {
* @return ExecutionStatus. Contains the QueryId and the TSStatus.
*/
public ExecutionResult getStatus() {
- long startTime = System.nanoTime();
+ // Although we monitor the state to transition to RUNNING, the future will return if any
+ // Terminated state is triggered
try {
- // 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);
- }
- });
- 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()));
+ if (stateMachine.getState() == QueryState.FINISHED) {
+ return getExecutionResult(QueryState.FINISHED);
+ }
+ SettableFuture<QueryState> future = SettableFuture.create();
+ stateMachine.addStateChangeListener(
+ state -> {
+ if (state == QueryState.RUNNING || state.isDone()) {
+ future.set(state);
+ }
+ });
+ 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();
}
- } finally {
- StepTracker.trace("getStatus", startTime, System.nanoTime());
- StepTracker.trace("QueryExecutionLifeCycle", this.queryExecutionStartTime, System.nanoTime());
+ return new ExecutionResult(
+ context.getQueryId(),
+ RpcUtils.getStatus(
+ TSStatusCode.INTERNAL_SERVER_ERROR, stateMachine.getFailureMessage()));
}
}
@@ -389,20 +370,20 @@ public class QueryExecution implements IQueryExecution {
this.resultHandle =
isSameNode(upstreamEndPoint)
? DataBlockService.getInstance()
- .getDataBlockManager()
- .createLocalSourceHandle(
- context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
- context.getResultNodeContext().getVirtualResultNodeId().getId(),
- context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
- stateMachine::transitionToFailed)
+ .getDataBlockManager()
+ .createLocalSourceHandle(
+ context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
+ context.getResultNodeContext().getVirtualResultNodeId().getId(),
+ context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
+ stateMachine::transitionToFailed)
: DataBlockService.getInstance()
- .getDataBlockManager()
- .createSourceHandle(
- context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
- context.getResultNodeContext().getVirtualResultNodeId().getId(),
- upstreamEndPoint,
- context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
- stateMachine::transitionToFailed);
+ .getDataBlockManager()
+ .createSourceHandle(
+ context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
+ context.getResultNodeContext().getVirtualResultNodeId().getId(),
+ upstreamEndPoint,
+ context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
+ stateMachine::transitionToFailed);
}
}
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 989b025660..054c2e4a51 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
@@ -92,7 +92,6 @@ public class ClusterScheduler implements IScheduler {
@Override
public void start() {
stateMachine.transitionToDispatching();
- // logger.info("transit to DISPATCHING");
Future<FragInstanceDispatchResult> dispatchResultFuture = dispatcher.dispatch(instances);
// NOTICE: the FragmentInstance may be dispatched to another Host due to consensus redirect.
@@ -122,7 +121,6 @@ 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");
instances.forEach(
instance -> {
stateMachine.initialFragInstanceState(instance.getId(), FragmentInstanceState.RUNNING);
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 55e6047555..0a459f2e0f 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
@@ -82,7 +82,6 @@ public class SimpleQueryTerminator implements IQueryTerminator {
public Boolean syncTerminate() {
for (TEndPoint endPoint : relatedHost) {
- // TODO (jackie tien) change the port
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
client.cancelQuery(
[iotdb] 03/03: 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/write_opt_version1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b29b8bb16539f2cfea925a9873720540953be463
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Jun 21 11:48:28 2022 +0800
spotless
---
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 44 ++++++-------
.../db/mpp/plan/execution/QueryExecution.java | 72 ++++++++++------------
2 files changed, 54 insertions(+), 62 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 dc7f6b485e..743c9a396c 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,29 @@ 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);
- }
- execution.start();
-
- return execution.getStatus();
+ 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();
+ }
}
public IQueryExecution getQueryExecution(Long queryId) {
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 24a7537fa1..7f338b3631 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,7 +31,6 @@ 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;
@@ -197,22 +196,22 @@ public class QueryExecution implements IQueryExecution {
this.scheduler =
config.isClusterMode()
? new ClusterScheduler(
- context,
- stateMachine,
- distributedPlan.getInstances(),
- context.getQueryType(),
- executor,
- writeOperationExecutor,
- scheduledExecutor,
- internalServiceClientManager)
+ context,
+ stateMachine,
+ distributedPlan.getInstances(),
+ context.getQueryType(),
+ executor,
+ writeOperationExecutor,
+ scheduledExecutor,
+ internalServiceClientManager)
: new StandaloneScheduler(
- context,
- stateMachine,
- distributedPlan.getInstances(),
- context.getQueryType(),
- executor,
- scheduledExecutor,
- internalServiceClientManager);
+ context,
+ stateMachine,
+ distributedPlan.getInstances(),
+ context.getQueryType(),
+ executor,
+ scheduledExecutor,
+ internalServiceClientManager);
this.scheduler.start();
}
@@ -255,9 +254,7 @@ public class QueryExecution implements IQueryExecution {
releaseResource();
}
- /**
- * Release the resources that current QueryExecution hold.
- */
+ /** Release the resources that current QueryExecution hold. */
private void releaseResource() {
// close ResultHandle to unblock client's getResult request
// Actually, we should not close the ResultHandle when the QueryExecution is Finished.
@@ -307,17 +304,13 @@ public class QueryExecution implements IQueryExecution {
}
}
- /**
- * @return true if there is more tsblocks, otherwise false
- */
+ /** @return true if there is more tsblocks, otherwise false */
@Override
public boolean hasNextResult() {
return resultHandle != null && !resultHandle.isFinished();
}
- /**
- * return the result column count without the time column
- */
+ /** return the result column count without the time column */
@Override
public int getOutputValueColumnCount() {
return analysis.getRespDatasetHeader().getOutputValueColumnCount();
@@ -358,8 +351,7 @@ public class QueryExecution implements IQueryExecution {
}
return new ExecutionResult(
context.getQueryId(),
- RpcUtils.getStatus(
- TSStatusCode.INTERNAL_SERVER_ERROR, stateMachine.getFailureMessage()));
+ RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, stateMachine.getFailureMessage()));
}
}
@@ -370,20 +362,20 @@ public class QueryExecution implements IQueryExecution {
this.resultHandle =
isSameNode(upstreamEndPoint)
? DataBlockService.getInstance()
- .getDataBlockManager()
- .createLocalSourceHandle(
- context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
- context.getResultNodeContext().getVirtualResultNodeId().getId(),
- context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
- stateMachine::transitionToFailed)
+ .getDataBlockManager()
+ .createLocalSourceHandle(
+ context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
+ context.getResultNodeContext().getVirtualResultNodeId().getId(),
+ context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
+ stateMachine::transitionToFailed)
: DataBlockService.getInstance()
- .getDataBlockManager()
- .createSourceHandle(
- context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
- context.getResultNodeContext().getVirtualResultNodeId().getId(),
- upstreamEndPoint,
- context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
- stateMachine::transitionToFailed);
+ .getDataBlockManager()
+ .createSourceHandle(
+ context.getResultNodeContext().getVirtualFragmentInstanceId().toThrift(),
+ context.getResultNodeContext().getVirtualResultNodeId().getId(),
+ upstreamEndPoint,
+ context.getResultNodeContext().getUpStreamFragmentInstanceId().toThrift(),
+ stateMachine::transitionToFailed);
}
}
[iotdb] 01/03: remove some test codes
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 feb198b2f8f06c31385c95b3d943494fe5a5eb7d
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Jun 21 11:38:10 2022 +0800
remove some test codes
---
.../org/apache/iotdb/confignode/manager/PartitionManager.java | 8 +++++++-
.../src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +-
.../java/org/apache/iotdb/db/mpp/execution/StateMachine.java | 7 +------
.../iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java | 11 -----------
4 files changed, 9 insertions(+), 19 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
index 9b0d07a03c..ac1ae22eea 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
@@ -325,7 +325,13 @@ public class PartitionManager {
getNodeManager().getOnlineDataNodeCount()
/ ConfigNodeDescriptor.getInstance().getConf().getSchemaReplicationFactor());
} else {
- regionNum = 1;
+ regionNum =
+ Math.max(
+ 2,
+ (int)
+ (getNodeManager().getTotalCpuCoreCount()
+ * 0.3
+ / ConfigNodeDescriptor.getInstance().getConf().getDataReplicationFactor()));
}
// Do Region allocation for those StorageGroups who get the particle
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 91492a28ac..bd1e33a41a 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
@@ -2926,7 +2926,7 @@ public class IoTDBConfig {
public void setTriggerForwardMQTTPoolSize(int triggerForwardMQTTPoolSize) {
this.triggerForwardMQTTPoolSize = triggerForwardMQTTPoolSize;
}
-
+
public int getCoordinatorReadExecutorSize() {
return coordinatorReadExecutorSize;
}
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 e53c2cb02d..c1b7e76360 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,8 +13,6 @@
*/
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;
@@ -272,7 +270,6 @@ 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;
@@ -284,7 +281,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));
@@ -313,8 +310,6 @@ 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/analyze/ClusterSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
index 9f72dc4802..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
@@ -96,21 +96,10 @@ 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 {