Method init

bolt/exec/Task.cpp:393–440

}

void Task::init(std::optional<common::SpillDiskOptions>&& spillDiskOpts) {
  BOLT_CHECK(driverFactories_.empty());
  initTaskPool();

  setSpillDiskConfig(std::move(spillDiskOpts));

  if (mode_ != Task::ExecutionMode::kSerial) {
    return;
  }

  // Create drivers.
  BOLT_CHECK_NULL(
      consumerSupplier_,
      "Serial execution mode doesn't support delivering results to a "
      "callback");

  taskStats_.executionStartTimeMs = getCurrentTimeMs();
  LocalPlanner::plan(
      planFragment_,
      nullptr,
      &driverFactories_,
      queryCtx_->queryConfig(),
      1,
      queryCtx_->queryConfig().isMultiDriverEnabled());
  exchangeClients_.resize(driverFactories_.size());

  // In Task::next() we always assume ungrouped execution.
  for (const auto& factory : driverFactories_) {
    BOLT_CHECK(factory->supportsSerialExecution());
    numDriversUngrouped_ += factory->numDrivers;
    numTotalDrivers_ += factory->numTotalDrivers;
    taskStats_.pipelineStats.emplace_back(
        factory->inputDriver, factory->outputDriver, factory->isMorselDriven());
  }

  // Create drivers.
  createSplitGroupStateLocked(kUngroupedGroupId);
  std::vector<std::shared_ptr<Driver>> drivers =
      createDriversLocked(kUngroupedGroupId);
  if (pool_->reservedBytes() != 0) {
    BOLT_FAIL(
        "Unexpected memory pool allocations during task[{}] driver initialization: {}",
        taskId_,
        pool_->treeMemoryUsage());
  }

  drivers_ = std::move(drivers);
}

void Task::setSpillDiskConfig(
    std::optional<common::SpillDiskOptions>&& spillDiskOpts) {