This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch rel/1.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.1 by this push:
new a2c2d45c52b [To rel/1.1][IOTDB-5840] Enhance the problem that the
insertRecords interface may cause the number of threads to balloon when there
are too many data regions (#9769)
a2c2d45c52b is described below
commit a2c2d45c52b566e6de96f9a021f9f754fa64c590
Author: Potato <[email protected]>
AuthorDate: Sat May 6 18:03:32 2023 +0800
[To rel/1.1][IOTDB-5840] Enhance the problem that the insertRecords
interface may cause the number of threads to balloon when there are too many
data regions (#9769)
---
.../db/mpp/plan/scheduler/AsyncPlanNodeSender.java | 74 ++++++++++++++++------
.../plan/scheduler/AsyncSendPlanNodeHandler.java | 27 ++++----
.../scheduler/FragmentInstanceDispatcherImpl.java | 20 ++++--
.../impl/DataNodeInternalRPCServiceImpl.java | 36 +++++++----
.../DataNodeInternalRPCServiceImplTest.java | 31 +++++----
thrift/src/main/thrift/datanode.thrift | 16 +++--
6 files changed, 139 insertions(+), 65 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java
index 29a5f80f628..69bb8d7e35c 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java
@@ -25,8 +25,9 @@ import org.apache.iotdb.commons.client.IClientManager;
import
org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.mpp.rpc.thrift.TPlanNode;
-import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq;
-import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -34,6 +35,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -43,38 +45,50 @@ import java.util.concurrent.atomic.AtomicLong;
import static com.google.common.util.concurrent.Futures.immediateFuture;
public class AsyncPlanNodeSender {
+
private static final Logger logger =
LoggerFactory.getLogger(AsyncPlanNodeSender.class);
private final IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient>
asyncInternalServiceClientManager;
private final List<FragmentInstance> instances;
- private final Map<Integer, TSendPlanNodeResp> instanceId2RespMap;
+
+ private final Map<TEndPoint, BatchRequestWithIndex> batchRequests;
+ private final Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap;
private final AtomicLong pendingNumber;
+ private final long startSendTime;
public AsyncPlanNodeSender(
IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient>
asyncInternalServiceClientManager,
List<FragmentInstance> instances) {
+ this.startSendTime = System.nanoTime();
this.asyncInternalServiceClientManager = asyncInternalServiceClientManager;
this.instances = instances;
- this.instanceId2RespMap = new ConcurrentHashMap<>();
- this.pendingNumber = new AtomicLong(instances.size());
+ this.batchRequests = new HashMap<>();
+ for (int i = 0; i < instances.size(); i++) {
+ this.batchRequests
+ .computeIfAbsent(
+ instances.get(i).getHostDataNode().getInternalEndPoint(),
+ x -> new BatchRequestWithIndex())
+ .addSinglePlanNodeReq(
+ i,
+ new TSendSinglePlanNodeReq(
+ new TPlanNode(
+
instances.get(i).getFragment().getPlanNodeTree().serializeToByteBuffer()),
+ instances.get(i).getRegionReplicaSet().getRegionId()));
+ }
+ this.instanceId2RespMap = new ConcurrentHashMap<>(instances.size() + 1, 1);
+ this.pendingNumber = new AtomicLong(batchRequests.keySet().size());
}
public void sendAll() {
- long startSendTime = System.nanoTime();
- for (int i = 0; i < instances.size(); ++i) {
- FragmentInstance instance = instances.get(i);
+ for (Map.Entry<TEndPoint, BatchRequestWithIndex> entry :
batchRequests.entrySet()) {
AsyncSendPlanNodeHandler handler =
- new AsyncSendPlanNodeHandler(i, pendingNumber, instanceId2RespMap,
startSendTime);
+ new AsyncSendPlanNodeHandler(
+ entry.getValue().getIndexes(), pendingNumber,
instanceId2RespMap, startSendTime);
try {
- TSendPlanNodeReq sendPlanNodeReq =
- new TSendPlanNodeReq(
- new
TPlanNode(instance.getFragment().getPlanNodeTree().serializeToByteBuffer()),
- instance.getRegionReplicaSet().getRegionId());
AsyncDataNodeInternalServiceClient client =
- asyncInternalServiceClientManager.borrowClient(
- instance.getHostDataNode().getInternalEndPoint());
- client.sendPlanNode(sendPlanNodeReq, handler);
+ asyncInternalServiceClientManager.borrowClient(entry.getKey());
+ client.sendBatchPlanNode(entry.getValue().getBatchRequest(), handler);
} catch (Exception e) {
handler.onError(e);
}
@@ -92,7 +106,7 @@ public class AsyncPlanNodeSender {
public List<TSStatus> getFailureStatusList() {
List<TSStatus> failureStatusList = new ArrayList<>();
TSStatus status;
- for (Map.Entry<Integer, TSendPlanNodeResp> entry :
instanceId2RespMap.entrySet()) {
+ for (Map.Entry<Integer, TSendSinglePlanNodeResp> entry :
instanceId2RespMap.entrySet()) {
status = entry.getValue().getStatus();
if (!entry.getValue().accepted) {
if (status == null) {
@@ -122,7 +136,7 @@ public class AsyncPlanNodeSender {
}
public Future<FragInstanceDispatchResult> getResult() {
- for (Map.Entry<Integer, TSendPlanNodeResp> entry :
instanceId2RespMap.entrySet()) {
+ for (Map.Entry<Integer, TSendSinglePlanNodeResp> entry :
instanceId2RespMap.entrySet()) {
if (!entry.getValue().accepted) {
logger.warn(
"dispatch write failed. status: {}, code: {}, message: {}, node
{}",
@@ -142,4 +156,28 @@ public class AsyncPlanNodeSender {
}
return immediateFuture(new FragInstanceDispatchResult(true));
}
+
+ /**
+ * This class is used to aggregate PlanNode of the same datanode into one
rpc. In order to ensure
+ * the one-to-one correspondence between response and request, the
corresponding index needs to be
+ * recorded.
+ */
+ static class BatchRequestWithIndex {
+
+ private final List<Integer> indexes = new ArrayList<>();
+ private final TSendBatchPlanNodeReq batchRequest = new
TSendBatchPlanNodeReq();
+
+ void addSinglePlanNodeReq(int index, TSendSinglePlanNodeReq singleRequest)
{
+ indexes.add(index);
+ batchRequest.addToRequests(singleRequest);
+ }
+
+ public List<Integer> getIndexes() {
+ return indexes;
+ }
+
+ public TSendBatchPlanNodeReq getBatchRequest() {
+ return batchRequest;
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java
index 2bd50a6c7ed..f47ad999e12 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java
@@ -19,37 +19,42 @@
package org.apache.iotdb.db.mpp.plan.scheduler;
import
org.apache.iotdb.commons.service.metric.enums.PerformanceOverviewMetrics;
-import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.async.AsyncMethodCallback;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
-public class AsyncSendPlanNodeHandler implements
AsyncMethodCallback<TSendPlanNodeResp> {
- private final int instanceId;
+public class AsyncSendPlanNodeHandler implements
AsyncMethodCallback<TSendBatchPlanNodeResp> {
+
+ private final List<Integer> instanceIds;
private final AtomicLong pendingNumber;
- private final Map<Integer, TSendPlanNodeResp> instanceId2RespMap;
+ private final Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap;
private final long sendTime;
private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS
=
PerformanceOverviewMetrics.getInstance();
public AsyncSendPlanNodeHandler(
- int instanceId,
+ List<Integer> instanceIds,
AtomicLong pendingNumber,
- Map<Integer, TSendPlanNodeResp> instanceId2RespMap,
+ Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap,
long sendTime) {
- this.instanceId = instanceId;
+ this.instanceIds = instanceIds;
this.pendingNumber = pendingNumber;
this.instanceId2RespMap = instanceId2RespMap;
this.sendTime = sendTime;
}
@Override
- public void onComplete(TSendPlanNodeResp tSendPlanNodeResp) {
- instanceId2RespMap.put(instanceId, tSendPlanNodeResp);
+ public void onComplete(TSendBatchPlanNodeResp sendBatchPlanNodeResp) {
+ for (int i = 0; i < sendBatchPlanNodeResp.getResponses().size(); i++) {
+ instanceId2RespMap.put(instanceIds.get(i),
sendBatchPlanNodeResp.getResponses().get(i));
+ }
if (pendingNumber.decrementAndGet() == 0) {
PERFORMANCE_OVERVIEW_METRICS.recordScheduleRemoteCost(System.nanoTime()
- sendTime);
synchronized (pendingNumber) {
@@ -60,13 +65,13 @@ public class AsyncSendPlanNodeHandler implements
AsyncMethodCallback<TSendPlanNo
@Override
public void onError(Exception e) {
- TSendPlanNodeResp resp = new TSendPlanNodeResp();
+ TSendSinglePlanNodeResp resp = new TSendSinglePlanNodeResp();
String errorMsg = String.format("Fail to send plan node, exception
message: %s", e);
resp.setAccepted(false);
resp.setMessage(errorMsg);
resp.setStatus(
RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
errorMsg));
- instanceId2RespMap.put(instanceId, resp);
+ instanceIds.forEach(instanceId -> instanceId2RespMap.put(instanceId,
resp));
if (pendingNumber.decrementAndGet() == 0) {
PERFORMANCE_OVERVIEW_METRICS.recordScheduleRemoteCost(System.nanoTime()
- sendTime);
synchronized (pendingNumber) {
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 a471e9256aa..cb64acaf6bd 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
@@ -40,10 +40,11 @@ import
org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.utils.SetThreadName;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstance;
import org.apache.iotdb.mpp.rpc.thrift.TPlanNode;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq;
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.mpp.rpc.thrift.TSendSinglePlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -52,6 +53,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -268,11 +270,15 @@ public class FragmentInstanceDispatcherImpl implements
IFragInstanceDispatcher {
}
break;
case WRITE:
- TSendPlanNodeReq sendPlanNodeReq =
- new TSendPlanNodeReq(
- new
TPlanNode(instance.getFragment().getPlanNodeTree().serializeToByteBuffer()),
- instance.getRegionReplicaSet().getRegionId());
- TSendPlanNodeResp sendPlanNodeResp =
client.sendPlanNode(sendPlanNodeReq);
+ TSendBatchPlanNodeReq sendPlanNodeReq =
+ new TSendBatchPlanNodeReq(
+ Collections.singletonList(
+ new TSendSinglePlanNodeReq(
+ new TPlanNode(
+
instance.getFragment().getPlanNodeTree().serializeToByteBuffer()),
+ instance.getRegionReplicaSet().getRegionId())));
+ TSendSinglePlanNodeResp sendPlanNodeResp =
+ client.sendBatchPlanNode(sendPlanNodeReq).getResponses().get(0);
if (!sendPlanNodeResp.accepted) {
logger.warn(
"dispatch write failed. status: {}, code: {}, message: {},
node {}",
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 76a03e517b1..80fc5c1c640 100644
---
a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -166,10 +166,11 @@ import
org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListWithTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchRequest;
import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchResponse;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp;
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.mpp.rpc.thrift.TSendSinglePlanNodeResp;
import org.apache.iotdb.mpp.rpc.thrift.TTsFilePieceReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateConfigNodeGroupReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
@@ -272,18 +273,25 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
}
@Override
- public TSendPlanNodeResp sendPlanNode(TSendPlanNodeReq req) {
- LOGGER.debug("receive PlanNode to group[{}]", req.getConsensusGroupId());
- ConsensusGroupId groupId =
-
ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
- PlanNode planNode = PlanNodeType.deserialize(req.planNode.body);
- RegionWriteExecutor executor = new RegionWriteExecutor();
- TSendPlanNodeResp resp = new TSendPlanNodeResp();
- RegionExecutionResult executionResult = executor.execute(groupId,
planNode);
- resp.setAccepted(executionResult.isAccepted());
- resp.setMessage(executionResult.getMessage());
- resp.setStatus(executionResult.getStatus());
- return resp;
+ public TSendBatchPlanNodeResp sendBatchPlanNode(TSendBatchPlanNodeReq req) {
+ List<TSendSinglePlanNodeResp> responses =
+ req.getRequests().stream()
+ .map(
+ request -> {
+ ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromTConsensusGroupId(
+ request.getConsensusGroupId());
+ PlanNode planNode =
PlanNodeType.deserialize(request.planNode.body);
+ RegionWriteExecutor executor = new RegionWriteExecutor();
+ TSendSinglePlanNodeResp resp = new TSendSinglePlanNodeResp();
+ RegionExecutionResult executionResult =
executor.execute(groupId, planNode);
+ resp.setAccepted(executionResult.isAccepted());
+ resp.setMessage(executionResult.getMessage());
+ resp.setStatus(executionResult.getStatus());
+ return resp;
+ })
+ .collect(Collectors.toList());
+ return new TSendBatchPlanNodeResp(responses);
}
@Override
diff --git
a/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java
b/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java
index c06fcf60078..6e183699289 100644
---
a/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java
@@ -43,8 +43,9 @@ import
org.apache.iotdb.db.service.thrift.impl.DataNodeInternalRPCServiceImpl;
import org.apache.iotdb.db.service.thrift.impl.DataNodeRegionManager;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.mpp.rpc.thrift.TPlanNode;
-import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq;
-import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp;
+import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeReq;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
@@ -61,11 +62,13 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class DataNodeInternalRPCServiceImplTest {
+
private static final IoTDBConfig conf =
IoTDBDescriptor.getInstance().getConfig();
DataNodeInternalRPCServiceImpl dataNodeInternalRPCServiceImpl;
private static final int dataNodeId = 0;
@@ -145,16 +148,18 @@ public class DataNodeInternalRPCServiceImplTest {
ByteBuffer byteBuffer = createTimeSeriesNode.serializeToByteBuffer();
// put serialized planNode to TSendPlanNodeReq
- TSendPlanNodeReq request = new TSendPlanNodeReq();
+ TSendSinglePlanNodeReq request = new TSendSinglePlanNodeReq();
TPlanNode tPlanNode = new TPlanNode();
tPlanNode.setBody(byteBuffer);
request.setPlanNode(tPlanNode);
request.setConsensusGroupId(regionReplicaSet.getRegionId());
// Use consensus layer to execute request
- TSendPlanNodeResp response =
dataNodeInternalRPCServiceImpl.sendPlanNode(request);
+ TSendBatchPlanNodeResp response =
+ dataNodeInternalRPCServiceImpl.sendBatchPlanNode(
+ new TSendBatchPlanNodeReq(Collections.singletonList(request)));
- Assert.assertTrue(response.accepted);
+ Assert.assertTrue(response.getResponses().get(0).accepted);
}
@Test
@@ -221,16 +226,18 @@ public class DataNodeInternalRPCServiceImplTest {
ByteBuffer byteBuffer =
createAlignedTimeSeriesNode.serializeToByteBuffer();
// put serialized planNode to TSendPlanNodeReq
- TSendPlanNodeReq request = new TSendPlanNodeReq();
+ TSendSinglePlanNodeReq request = new TSendSinglePlanNodeReq();
TPlanNode tPlanNode = new TPlanNode();
tPlanNode.setBody(byteBuffer);
request.setPlanNode(tPlanNode);
request.setConsensusGroupId(regionReplicaSet.getRegionId());
// Use consensus layer to execute request
- TSendPlanNodeResp response =
dataNodeInternalRPCServiceImpl.sendPlanNode(request);
+ TSendBatchPlanNodeResp response =
+ dataNodeInternalRPCServiceImpl.sendBatchPlanNode(
+ new TSendBatchPlanNodeReq(Collections.singletonList(request)));
- Assert.assertTrue(response.accepted);
+ Assert.assertTrue(response.getResponses().get(0).accepted);
}
@Test
@@ -308,16 +315,18 @@ public class DataNodeInternalRPCServiceImplTest {
ByteBuffer byteBuffer = createMultiTimeSeriesNode.serializeToByteBuffer();
// put serialized planNode to TSendPlanNodeReq
- TSendPlanNodeReq request = new TSendPlanNodeReq();
+ TSendSinglePlanNodeReq request = new TSendSinglePlanNodeReq();
TPlanNode tPlanNode = new TPlanNode();
tPlanNode.setBody(byteBuffer);
request.setPlanNode(tPlanNode);
request.setConsensusGroupId(regionReplicaSet.getRegionId());
// Use consensus layer to execute request
- TSendPlanNodeResp response =
dataNodeInternalRPCServiceImpl.sendPlanNode(request);
+ TSendBatchPlanNodeResp response =
+ dataNodeInternalRPCServiceImpl.sendBatchPlanNode(
+ new TSendBatchPlanNodeReq(Collections.singletonList(request)));
- Assert.assertTrue(response.accepted);
+ Assert.assertTrue(response.getResponses().get(0).accepted);
}
private TRegionReplicaSet genRegionReplicaSet() {
diff --git a/thrift/src/main/thrift/datanode.thrift
b/thrift/src/main/thrift/datanode.thrift
index aa4cdc030fb..769dbb697c5 100644
--- a/thrift/src/main/thrift/datanode.thrift
+++ b/thrift/src/main/thrift/datanode.thrift
@@ -118,17 +118,25 @@ struct TSendFragmentInstanceResp {
2: optional string message
}
-struct TSendPlanNodeReq {
+struct TSendSinglePlanNodeReq {
1: required TPlanNode planNode
2: required common.TConsensusGroupId consensusGroupId
}
-struct TSendPlanNodeResp {
+struct TSendSinglePlanNodeResp {
1: required bool accepted
2: optional string message
3: optional common.TSStatus status
}
+struct TSendBatchPlanNodeReq {
+ 1: required list<TSendSinglePlanNodeReq> requests;
+}
+
+struct TSendBatchPlanNodeResp {
+ 1: required list<TSendSinglePlanNodeResp> responses;
+}
+
struct TFetchFragmentInstanceInfoReq {
1: required TFragmentInstanceId fragmentInstanceId
}
@@ -437,9 +445,9 @@ service IDataNodeRPCService {
TSendFragmentInstanceResp sendFragmentInstance(TSendFragmentInstanceReq req);
/**
- * dispatch PlanNode to remote node for write request in order to save
resource
+ * dispatch PlanNodes in batches to remote node for write request in order to
save resource
*/
- TSendPlanNodeResp sendPlanNode(TSendPlanNodeReq req);
+ TSendBatchPlanNodeResp sendBatchPlanNode(TSendBatchPlanNodeReq req);
TFragmentInstanceInfoResp
fetchFragmentInstanceInfo(TFetchFragmentInstanceInfoReq req);
