You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2022/06/09 07:55:15 UTC
[iotdb] 02/03: optimize dispatching planNode
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch jira3385
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit fd97bc95bb011db40bbfe20181542f0c9d4c51c3
Author: LebronAl <TX...@gmail.com>
AuthorDate: Thu Jun 9 15:41:05 2022 +0800
optimize dispatching planNode
---
.../scheduler/FragmentInstanceDispatcherImpl.java | 30 +++++--
.../scheduler/SimpleFragInstanceDispatcher.java | 24 ++++--
.../service/thrift/impl/InternalServiceImpl.java | 97 +++++++++++-----------
.../iotdb/db/service/InternalServiceImplTest.java | 3 -
thrift/src/main/thrift/mpp.thrift | 23 ++++-
5 files changed, 107 insertions(+), 70 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
index e505fc58b4..459fe536da 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.mpp.plan.scheduler;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
@@ -39,8 +38,11 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstance;
+import org.apache.iotdb.mpp.rpc.thrift.TPlanNode;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp;
import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.SettableFuture;
@@ -49,7 +51,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
@@ -150,17 +151,28 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher {
throws FragmentInstanceDispatchException {
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
- ByteBuffer buffer = instance.serializeToByteBuffer();
- TConsensusGroupId groupId = instance.getRegionReplicaSet().getRegionId();
- TSendFragmentInstanceReq req =
- new TSendFragmentInstanceReq(
- new TFragmentInstance(buffer), groupId, instance.getType().toString());
- TSendFragmentInstanceResp resp = client.sendFragmentInstance(req);
- return resp.accepted;
+ switch (instance.getType()) {
+ case READ:
+ TSendFragmentInstanceReq sendFragmentInstanceReq =
+ new TSendFragmentInstanceReq(
+ new TFragmentInstance(instance.serializeToByteBuffer()),
+ instance.getRegionReplicaSet().getRegionId());
+ TSendFragmentInstanceResp sendFragmentInstanceResp =
+ client.sendFragmentInstance(sendFragmentInstanceReq);
+ return sendFragmentInstanceResp.accepted;
+ case WRITE:
+ TSendPlanNodeReq sendPlanNodeReq =
+ new TSendPlanNodeReq(
+ new TPlanNode(instance.getFragment().getRoot().serializeToByteBuffer()),
+ instance.getRegionReplicaSet().getRegionId());
+ TSendPlanNodeResp sendPlanNodeResp = client.sendPlanNode(sendPlanNodeReq);
+ return sendPlanNodeResp.accepted;
+ }
} catch (IOException | TException e) {
logger.error("can't connect to node {}", endPoint, e);
throw new FragmentInstanceDispatchException(e);
}
+ return false;
}
private boolean dispatchLocally(FragmentInstance instance)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java
index 62ef784c21..329c8601c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java
@@ -19,21 +19,21 @@
package org.apache.iotdb.db.mpp.plan.scheduler;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstance;
+import org.apache.iotdb.mpp.rpc.thrift.TPlanNode;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -62,13 +62,21 @@ public class SimpleFragInstanceDispatcher implements IFragInstanceDispatcher {
// TODO: (jackie tien) change the port
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
- ByteBuffer buffer = instance.serializeToByteBuffer();
- TConsensusGroupId groupId = instance.getRegionReplicaSet().getRegionId();
- TSendFragmentInstanceReq req =
- new TSendFragmentInstanceReq(
- new TFragmentInstance(buffer), groupId, instance.getType().toString());
LOGGER.info("send FragmentInstance[{}] to {}", instance.getId(), endPoint);
- resp = client.sendFragmentInstance(req);
+ switch (instance.getType()) {
+ case READ:
+ TSendFragmentInstanceReq sendFragmentInstanceReq =
+ new TSendFragmentInstanceReq(
+ new TFragmentInstance(instance.serializeToByteBuffer()),
+ instance.getRegionReplicaSet().getRegionId());
+ resp.setAccepted(client.sendFragmentInstance(sendFragmentInstanceReq).accepted);
+ case WRITE:
+ TSendPlanNodeReq sendPlanNodeReq =
+ new TSendPlanNodeReq(
+ new TPlanNode(instance.getFragment().getRoot().serializeToByteBuffer()),
+ instance.getRegionReplicaSet().getRegionId());
+ resp.setAccepted(client.sendPlanNode(sendPlanNodeReq).accepted);
+ }
} catch (IOException | TException e) {
LOGGER.error("can't connect to node {}", endPoint, e);
throw e;
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java
index 9bf7e3117c..5dc64a7f4a 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java
@@ -35,7 +35,6 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.consensus.common.Peer;
-import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse;
import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
@@ -53,12 +52,12 @@ import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceInfo;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceManager;
import org.apache.iotdb.db.mpp.plan.Coordinator;
import org.apache.iotdb.db.mpp.plan.analyze.ClusterPartitionFetcher;
-import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
import org.apache.iotdb.db.mpp.plan.analyze.SchemaValidator;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.DeleteRegionNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.service.metrics.MetricsService;
import org.apache.iotdb.db.service.metrics.enums.Metric;
@@ -85,6 +84,8 @@ import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchRequest;
import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchResponse;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
@@ -113,58 +114,56 @@ public class InternalServiceImpl implements InternalService.Iface {
@Override
public TSendFragmentInstanceResp sendFragmentInstance(TSendFragmentInstanceReq req) {
LOGGER.info("receive FragmentInstance to group[{}]", req.getConsensusGroupId());
- QueryType type = QueryType.valueOf(req.queryType);
ConsensusGroupId groupId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
- switch (type) {
- case READ:
- ConsensusReadResponse readResponse;
- if (groupId instanceof DataRegionId) {
- readResponse =
- DataRegionConsensusImpl.getInstance()
- .read(groupId, new ByteBufferConsensusRequest(req.fragmentInstance.body));
- } else {
- readResponse =
- SchemaRegionConsensusImpl.getInstance()
- .read(groupId, new ByteBufferConsensusRequest(req.fragmentInstance.body));
- }
- if (!readResponse.isSuccess()) {
- LOGGER.error(
- "execute FragmentInstance in ConsensusGroup {} failed because {}",
- req.getConsensusGroupId(),
- readResponse.getException());
- return new TSendFragmentInstanceResp(false);
- }
- FragmentInstanceInfo info = (FragmentInstanceInfo) readResponse.getDataset();
- return new TSendFragmentInstanceResp(!info.getState().isFailed());
- case WRITE:
- TSendFragmentInstanceResp response = new TSendFragmentInstanceResp();
- ConsensusWriteResponse writeResponse;
+ ConsensusReadResponse readResponse;
+ // We deserialize here instead of the underlying state machine because parallelism is possible
+ // here but not at the underlying state machine
+ FragmentInstance fragmentInstance = FragmentInstance.deserializeFrom(req.fragmentInstance.body);
+ if (groupId instanceof DataRegionId) {
+ readResponse = DataRegionConsensusImpl.getInstance().read(groupId, fragmentInstance);
+ } else {
+ readResponse = SchemaRegionConsensusImpl.getInstance().read(groupId, fragmentInstance);
+ }
+ if (!readResponse.isSuccess()) {
+ LOGGER.error(
+ "execute FragmentInstance in ConsensusGroup {} failed because {}",
+ req.getConsensusGroupId(),
+ readResponse.getException());
+ return new TSendFragmentInstanceResp(false);
+ }
+ FragmentInstanceInfo info = (FragmentInstanceInfo) readResponse.getDataset();
+ return new TSendFragmentInstanceResp(!info.getState().isFailed());
+ }
- FragmentInstance fragmentInstance =
- FragmentInstance.deserializeFrom(req.fragmentInstance.body);
- PlanNode planNode = fragmentInstance.getFragment().getRoot();
- if (planNode instanceof InsertNode) {
- try {
- SchemaValidator.validate((InsertNode) planNode);
- } catch (SemanticException e) {
- response.setAccepted(false);
- response.setMessage(e.getMessage());
- return response;
- }
- }
- if (groupId instanceof DataRegionId) {
- writeResponse = DataRegionConsensusImpl.getInstance().write(groupId, planNode);
- } else {
- writeResponse = SchemaRegionConsensusImpl.getInstance().write(groupId, planNode);
- }
- // TODO need consider more status
- response.setAccepted(
- TSStatusCode.SUCCESS_STATUS.getStatusCode() == writeResponse.getStatus().getCode());
- response.setMessage(writeResponse.getStatus().message);
+ @Override
+ public TSendPlanNodeResp sendPlanNode(TSendPlanNodeReq req) {
+ LOGGER.info("receive PlanNode to group[{}]", req.getConsensusGroupId());
+ ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+ TSendPlanNodeResp response = new TSendPlanNodeResp();
+ ConsensusWriteResponse writeResponse;
+
+ PlanNode planNode = PlanNodeType.deserialize(req.planNode.body);
+ if (planNode instanceof InsertNode) {
+ try {
+ SchemaValidator.validate((InsertNode) planNode);
+ } catch (SemanticException e) {
+ response.setAccepted(false);
+ response.setMessage(e.getMessage());
return response;
+ }
}
- return null;
+ if (groupId instanceof DataRegionId) {
+ writeResponse = DataRegionConsensusImpl.getInstance().write(groupId, planNode);
+ } else {
+ writeResponse = SchemaRegionConsensusImpl.getInstance().write(groupId, planNode);
+ }
+ // TODO need consider more status
+ response.setAccepted(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode() == writeResponse.getStatus().getCode());
+ response.setMessage(writeResponse.getStatus().message);
+ return response;
}
@Override
diff --git a/server/src/test/java/org/apache/iotdb/db/service/InternalServiceImplTest.java b/server/src/test/java/org/apache/iotdb/db/service/InternalServiceImplTest.java
index 96b55a90fc..62f511d7ad 100644
--- a/server/src/test/java/org/apache/iotdb/db/service/InternalServiceImplTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/service/InternalServiceImplTest.java
@@ -157,7 +157,6 @@ public class InternalServiceImplTest {
tFragmentInstance.setBody(byteBuffer);
request.setFragmentInstance(tFragmentInstance);
request.setConsensusGroupId(regionReplicaSet.getRegionId());
- request.setQueryType(QueryType.WRITE.toString());
// Use consensus layer to execute request
TSendFragmentInstanceResp response = internalServiceImpl.sendFragmentInstance(request);
@@ -244,7 +243,6 @@ public class InternalServiceImplTest {
tFragmentInstance.setBody(byteBuffer);
request.setFragmentInstance(tFragmentInstance);
request.setConsensusGroupId(regionReplicaSet.getRegionId());
- request.setQueryType(QueryType.WRITE.toString());
// Use consensus layer to execute request
TSendFragmentInstanceResp response = internalServiceImpl.sendFragmentInstance(request);
@@ -341,7 +339,6 @@ public class InternalServiceImplTest {
tFragmentInstance.setBody(byteBuffer);
request.setFragmentInstance(tFragmentInstance);
request.setConsensusGroupId(regionReplicaSet.getRegionId());
- request.setQueryType(QueryType.WRITE.toString());
// Use consensus layer to execute request
TSendFragmentInstanceResp response = internalServiceImpl.sendFragmentInstance(request);
diff --git a/thrift/src/main/thrift/mpp.thrift b/thrift/src/main/thrift/mpp.thrift
index 702050e540..f5ea85ca35 100644
--- a/thrift/src/main/thrift/mpp.thrift
+++ b/thrift/src/main/thrift/mpp.thrift
@@ -88,10 +88,13 @@ struct TFragmentInstance {
1: required binary body
}
+struct TPlanNode {
+ 1: required binary body
+}
+
struct TSendFragmentInstanceReq {
1: required TFragmentInstance fragmentInstance
2: required common.TConsensusGroupId consensusGroupId
- 3: required string queryType
}
struct TSendFragmentInstanceResp {
@@ -99,6 +102,16 @@ struct TSendFragmentInstanceResp {
2: optional string message
}
+struct TSendPlanNodeReq {
+ 1: required TPlanNode planNode
+ 2: required common.TConsensusGroupId consensusGroupId
+}
+
+struct TSendPlanNodeResp {
+ 1: required bool accepted
+ 2: optional string message
+}
+
struct TFetchFragmentInstanceStateReq {
1: required TFragmentInstanceId fragmentInstanceId
}
@@ -154,8 +167,16 @@ service InternalService {
// -----------------------------------For Data Node-----------------------------------------------
+ /**
+ * disptcher FragmentInstance to remote node for query request
+ */
TSendFragmentInstanceResp sendFragmentInstance(TSendFragmentInstanceReq req);
+ /**
+ * disptcher PlanNode to remote node for write request in order to saving resource
+ */
+ TSendPlanNodeResp sendPlanNode(TSendPlanNodeReq req);
+
TFragmentInstanceStateResp fetchFragmentInstanceState(TFetchFragmentInstanceStateReq req);
TCancelResp cancelQuery(TCancelQueryReq req);