Method TaskCursorBase

bolt/exec/tests/utils/Cursor.cpp:154–215

  // Unblock consumers.
  if (consumerBlocked_) {
    consumerBlocked_ = false;
    consumerPromise_.setValue();
  }
}

bool TaskQueue::hasNext() {
  std::lock_guard<std::mutex> l(mutex_);
  return !queue_.empty();
}

class TaskCursorBase : public TaskCursor {
 public:
  TaskCursorBase(
      const CursorParameters& params,
      const std::shared_ptr<folly::Executor>& executor) {
    static std::atomic<int32_t> cursorId;
    if (params.taskId.has_value()) {
      taskId_ = *params.taskId;
    } else {
      taskId_ = fmt::format("test_cursor_{}", ++cursorId);
    }

    if (params.queryCtx) {
      queryCtx_ = params.queryCtx;
    } else {
      // NOTE: the destructor of 'executor_' will wait for all the async task
      // activities to finish on TaskCursor destruction.
      executor_ = executor;
      static std::atomic<uint64_t> cursorQueryId{0};
      queryCtx_ = core::QueryCtx::create(
          executor_.get(),
          core::QueryConfig({}),
          std::
              unordered_map<std::string, std::shared_ptr<config::ConfigBase>>{},
          cache::AsyncDataCache::getInstance(),
          nullptr,
          nullptr,
          fmt::format("TaskCursorQuery_{}", cursorQueryId++));
    }

    if (!params.queryConfigs.empty()) {
      auto configCopy = params.queryConfigs;
      queryCtx_->testingOverrideConfigUnsafe(std::move(configCopy));
    }

    planFragment_ = {
        params.planNode,
        params.executionStrategy,
        params.numSplitGroups,
        params.groupedExecutionLeafNodeIds};

    if (!params.spillDirectory.empty()) {
      taskSpillDirectory_ = params.spillDirectory + "/" + taskId_;
      taskSpillDirectoryCb_ = params.spillDirectoryCallback;
      if (taskSpillDirectoryCb_ == nullptr) {
        auto fileSystem =
            bolt::filesystems::getFileSystem(taskSpillDirectory_, nullptr);