@Override
  public void submitDrivers(
      QueryId queryId, List<IDriver> drivers, long timeOut, SessionInfo sessionInfo)
      throws CpuNotEnoughException, MemoryNotEnoughException {
    DriverTaskHandle driverTaskHandle =
        new DriverTaskHandle(
            getNextDriverTaskHandleId(),
            (MultilevelPriorityQueue) readyQueue,
            OptionalInt.of(Integer.MAX_VALUE));
    List<DriverTask> tasks = new ArrayList<>();
    drivers.forEach(
        driver ->
            tasks.add(
                new DriverTask(
                    driver,
                    timeOut > 0 ? timeOut : QUERY_TIMEOUT_MS,
                    DriverTaskStatus.READY,
                    driverTaskHandle,
                    driver.getEstimatedMemorySize(),
                    driver.isHighestPriority())));

    List<DriverTask> submittedTasks = new ArrayList<>();
    for (DriverTask task : tasks) {
      IDriver driver = task.getDriver();
      int dependencyDriverIndex = driver.getDriverContext().getDependencyDriverIndex();
      if (dependencyDriverIndex != -1) {
        SettableFuture<?> blockedDependencyFuture =
            tasks.get(dependencyDriverIndex).getBlockedDependencyDriver();
        blockedDependencyFuture.addListener(
            () -> {
              // Only if query is alive, we can submit this task
              queryMap.computeIfPresent(
                  queryId,
                  (k1, queryRelatedTasks) -> {
                    queryRelatedTasks.computeIfPresent(
                        task.getDriverTaskId().getFragmentInstanceId(),
                        (k2, instanceRelatedTasks) -> {
                          instanceRelatedTasks.add(task);
                          submitTaskToReadyQueue(task);
                          return instanceRelatedTasks;
                        });
                    return queryRelatedTasks;
                  });
            },
            MoreExecutors.directExecutor());
      } else {
        submittedTasks.add(task);
      }
    }

    if (IoTDBDescriptor.getInstance().getConfig().isQuotaEnable()
        && sessionInfo != null
        && !sessionInfo.getUserName().equals(IoTDBConstant.PATH_ROOT)) {
      AtomicInteger usedCpu = new AtomicInteger();
      AtomicLong estimatedMemory = new AtomicLong();
      queryMap
          .get(queryId)
          .values()
          .forEach(
              driverTasks ->
                  driverTasks.forEach(
                      driverTask -> {
                        if (driverTask.getStatus().equals(DriverTaskStatus.RUNNING)
                            && driverTask.getDriver() instanceof DataDriver) {
                          usedCpu.addAndGet(1);
                          estimatedMemory.addAndGet(driverTask.getEstimatedMemorySize());
                        }
                      }));
      if (!DataNodeThrottleQuotaManager.getInstance()
          .getThrottleQuotaLimit()
          .checkCpu(sessionInfo.getUserName(), usedCpu.get())) {
        throw new CpuNotEnoughException(
            "There is not enough cpu to execute current fragment instance");
      }
      if (!DataNodeThrottleQuotaManager.getInstance()
          .getThrottleQuotaLimit()
          .checkMemory(sessionInfo.getUserName(), estimatedMemory.get())) {
        throw new MemoryNotEnoughException(
            "There is no enough memory to execute current fragment instance");
      }
    }

    for (DriverTask task : submittedTasks) {
      registerTaskToQueryMap(queryId, task);
    }
    scheduler.enforceTimeLimit(submittedTasks.get(submittedTasks.size() - 1));
    for (DriverTask task : submittedTasks) {
      submitTaskToReadyQueue(task);
    }
  }
The manager of fragment instances scheduling.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java