private void dispatchRemoteHelper(final FragmentInstance instance, final TEndPoint endPoint)
      throws FragmentInstanceDispatchException,
          TException,
          ClientManagerException,
          RatisReadUnavailableException,
          ConsensusGroupNotExistException {
    try (final SyncDataNodeInternalServiceClient client =
        syncInternalServiceClientManager.borrowClient(endPoint)) {
      switch (instance.getType()) {
        case READ:
          final TSendFragmentInstanceReq sendFragmentInstanceReq =
              new TSendFragmentInstanceReq(new TFragmentInstance(instance.serializeToByteBuffer()));
          if (instance.getExecutorType().isStorageExecutor()) {
            sendFragmentInstanceReq.setConsensusGroupId(
                instance.getRegionReplicaSet().getRegionId());
          }
          final TSendFragmentInstanceResp sendFragmentInstanceResp =
              client.sendFragmentInstance(sendFragmentInstanceReq);
          if (!sendFragmentInstanceResp.accepted) {
            LOGGER.warn(sendFragmentInstanceResp.message);
            if (sendFragmentInstanceResp.isSetNeedRetry()
                && sendFragmentInstanceResp.isNeedRetry()
                && sendFragmentInstanceResp.status != null) {
              if (sendFragmentInstanceResp.status.getCode()
                  == TSStatusCode.RATIS_READ_UNAVAILABLE.getStatusCode()) {
                throw new RatisReadUnavailableException(sendFragmentInstanceResp.message);
              } else if (sendFragmentInstanceResp.status.getCode()
                  == TSStatusCode.CONSENSUS_GROUP_NOT_EXIST.getStatusCode()) {
                throw new ConsensusGroupNotExistException(sendFragmentInstanceResp.message);
              } else if (sendFragmentInstanceResp.status.getCode()
                  == TSStatusCode.TOO_MANY_CONCURRENT_QUERIES_ERROR.getStatusCode()) {
                throw new FragmentInstanceDispatchException(sendFragmentInstanceResp.status);
              }
            } else if (sendFragmentInstanceResp.status != null) {
              throw new FragmentInstanceDispatchException(sendFragmentInstanceResp.status);
            }
            throw new FragmentInstanceDispatchException(
                RpcUtils.getStatus(
                    TSStatusCode.EXECUTE_STATEMENT_ERROR, sendFragmentInstanceResp.message));
          } else if (Objects.nonNull(sendFragmentInstanceReq.getConsensusGroupId())) {
            // Assume the consensus group casting will not generate any exceptions
            new TableSchemaQuerySuccessfulCallbackVisitor()
                .processFragment(
                    instance,
                    ConsensusGroupId.Factory.createFromTConsensusGroupId(
                        sendFragmentInstanceReq.getConsensusGroupId()));
          }
          break;
        case WRITE:
          final TSendBatchPlanNodeReq sendPlanNodeReq =
              new TSendBatchPlanNodeReq(
                  Collections.singletonList(
                      new TSendSinglePlanNodeReq(
                          new TPlanNode(
                              instance.getFragment().getPlanNodeTree().serializeToByteBuffer()),
                          instance.getRegionReplicaSet().getRegionId())));
          final TSendSinglePlanNodeResp sendPlanNodeResp =
              client.sendBatchPlanNode(sendPlanNodeReq).getResponses().get(0);
          if (!sendPlanNodeResp.accepted) {
            if (sendPlanNodeResp.getStatus() == null) {
              throw new FragmentInstanceDispatchException(
                  RpcUtils.getStatus(
                      TSStatusCode.WRITE_PROCESS_ERROR, sendPlanNodeResp.getMessage()));
            } else {
              // DO NOT LOG READ_ONLY ERROR
              if (sendPlanNodeResp.getStatus().getCode()
                  != TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()) {
                LOGGER.warn(
                    "Dispatch write failed. status: {}, code: {}, message: {}, node {}",
                    sendPlanNodeResp.status,
                    TSStatusCode.representOf(sendPlanNodeResp.status.code),
                    sendPlanNodeResp.message,
                    endPoint);
              }
              throw new FragmentInstanceDispatchException(sendPlanNodeResp.getStatus());
            }
          } else {
            // some expected and accepted status except SUCCESS_STATUS need to be returned
            final TSStatus status = sendPlanNodeResp.getStatus();
            if (status != null && status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
              throw new FragmentInstanceDispatchException(status);
            }
          }
          break;
        default:
          throw new FragmentInstanceDispatchException(
              RpcUtils.getStatus(
                  TSStatusCode.EXECUTE_STATEMENT_ERROR,
                  String.format("unknown read type [%s]", instance.getType())));
      }
    }
  }
Dispatch the given write instances once. It will dispatch the given instances locally or
remotely, give the host datanode.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java