private void dispatchLocally(final FragmentInstance instance)
      throws FragmentInstanceDispatchException {
    // deserialize ConsensusGroupId
    ConsensusGroupId groupId = null;
    if (instance.getExecutorType().isStorageExecutor()) {
      try {
        groupId =
            ConsensusGroupId.Factory.createFromTConsensusGroupId(
                instance.getRegionReplicaSet().getRegionId());
      } catch (final Throwable t) {
        LOGGER.warn("Deserialize ConsensusGroupId failed. ", t);
        throw new FragmentInstanceDispatchException(
            RpcUtils.getStatus(
                TSStatusCode.EXECUTE_STATEMENT_ERROR,
                "Deserialize ConsensusGroupId failed: " + t.getMessage()));
      }
    }

    switch (instance.getType()) {
      case READ:
        RegionExecutionResult executionResult = null;
        if (Objects.nonNull(groupId)) {
          // For schema query, there may be a "write" before "read"
          // the result is not null iff the "write" has failed
          executionResult = new TableSchemaQueryWriteVisitor().processFragment(instance, groupId);
        }
        if (Objects.isNull(executionResult)) {
          final RegionReadExecutor readExecutor = new RegionReadExecutor();
          executionResult =
              groupId == null
                  ? readExecutor.execute(instance)
                  : readExecutor.execute(groupId, instance);
        }
        if (!executionResult.isAccepted()) {
          LOGGER.warn(executionResult.getMessage());
          if (executionResult.isReadNeedRetry()) {
            if (executionResult.getStatus() != null
                && executionResult.getStatus().getCode()
                    == TSStatusCode.TOO_MANY_CONCURRENT_QUERIES_ERROR.getStatusCode()) {
              throw new FragmentInstanceDispatchException(executionResult.getStatus());
            }
            throw new FragmentInstanceDispatchException(
                RpcUtils.getStatus(TSStatusCode.DISPATCH_ERROR, executionResult.getMessage()));
          } else if (executionResult.getStatus() != null) {
            throw new FragmentInstanceDispatchException(executionResult.getStatus());
          } else {
            throw new FragmentInstanceDispatchException(
                RpcUtils.getStatus(
                    TSStatusCode.EXECUTE_STATEMENT_ERROR, executionResult.getMessage()));
          }
        } else if (Objects.nonNull(groupId)) {
          // Assume the consensus group casting will not generate any exceptions
          new TableSchemaQuerySuccessfulCallbackVisitor().processFragment(instance, groupId);
        }
        break;
      case WRITE:
        final PlanNode planNode = instance.getFragment().getPlanNodeTree();
        final RegionWriteExecutor writeExecutor = new RegionWriteExecutor();
        final RegionExecutionResult writeResult = writeExecutor.execute(groupId, planNode);
        if (!writeResult.isAccepted()) {
          if (writeResult.getStatus() == null) {
            throw new FragmentInstanceDispatchException(
                RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, writeResult.getMessage()));
          } else {
            // DO NOT LOG READ_ONLY ERROR
            if (writeResult.getStatus().getCode()
                != TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()) {
              LOGGER.warn(
                  "write locally failed. TSStatus: {}, message: {}",
                  writeResult.getStatus(),
                  writeResult.getMessage());
            }
            throw new FragmentInstanceDispatchException(writeResult.getStatus());
          }
        } else {
          // Some expected and accepted status except SUCCESS_STATUS need to be returned
          final TSStatus status = writeResult.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