private <T> Optional<T> getResult(ISourceHandleSupplier<T> dataSupplier) throws IoTDBException {
    checkArgument(resultHandle != null, "ResultHandle in Coordinator should be init firstly.");
    // iterate until we get a non-nullable TsBlock or result is finished
    while (true) {
      try {
        if (resultHandle.isAborted()) {
          LOGGER.warn("[ResultHandleAborted]");
          stateMachine.transitionToAborted();
          if (stateMachine.getFailureStatus() != null) {
            throw new IoTDBException(
                stateMachine.getFailureStatus().getMessage(), stateMachine.getFailureStatus().code);
          } else {
            Throwable rootCause = stateMachine.getFailureException();
            if (rootCause instanceof IoTDBRuntimeException) {
              throw (IoTDBRuntimeException) rootCause;
            } else if (rootCause instanceof IoTDBException) {
              throw (IoTDBException) rootCause;
            }
            throw new IoTDBException(
                stateMachine.getFailureMessage(),
                TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
          }
        } else if (resultHandle.isFinished()) {
          LOGGER.debug("[ResultHandleFinished]");
          stateMachine.transitionToFinished();
          return Optional.empty();
        }

        long startTime = System.nanoTime();
        try {
          ListenableFuture<?> blocked = resultHandle.isBlocked();
          blocked.get();
        } finally {
          QUERY_EXECUTION_METRIC_SET.recordExecutionCost(
              WAIT_FOR_RESULT, System.nanoTime() - startTime);
        }

        if (!resultHandle.isFinished()) {
          // use the getSerializedTsBlock instead of receive to get ByteBuffer result
          T res = dataSupplier.get();
          if (res == null) {
            continue;
          }
          return Optional.of(res);
        } else {
          return Optional.empty();
        }
      } catch (ExecutionException | CancellationException e) {
        dealWithException(e.getCause() != null ? e.getCause() : e);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        dealWithException(e);
      } catch (Throwable t) {
        dealWithException(t);
      }
    }
  }
This method will be called by the request thread from client connection. This method will block
until one of these conditions occurs: 1. There is a batch of result 2. There is no more result
3. The query has been cancelled 4. The query is timeout This method will fetch the result from
DataStreamManager use the virtual ResultOperator's ID (This part will be designed and
implemented with DataStreamManager)
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java