Method Init

cpp/src/arrow/acero/hash_join_node.cc:943–994

  }

  Status Init() override {
    QueryContext* ctx = plan_->query_context();
    if (ctx->options().use_legacy_batching) {
      return Status::Invalid(
          "The plan was configured to use legacy batching but contained a join node "
          "which is incompatible with legacy batching");
    }

    bool use_sync_execution = ctx->executor()->GetCapacity() == 1;
    // TODO(ARROW-15732)
    // Each side of join might have an IO thread being called from. Once this is fixed
    // we will change it back to just the CPU's thread pool capacity.
    size_t num_threads = (GetCpuThreadPoolCapacity() + io::GetIOThreadPoolCapacity() + 1);

    RETURN_NOT_OK(pushdown_context_.Init(
        this, num_threads,
        [ctx](std::function<Status(size_t, int64_t)> fn,
              std::function<Status(size_t)> on_finished) {
          return ctx->RegisterTaskGroup(std::move(fn), std::move(on_finished));
        },
        [ctx](int task_group_id, int64_t num_tasks) {
          return ctx->StartTaskGroup(task_group_id, num_tasks);
        },
        [this](size_t thread_index) { return OnFiltersReceived(thread_index); },
        disable_bloom_filter_, use_sync_execution));

    RETURN_NOT_OK(impl_->Init(
        ctx, join_type_, num_threads, &(schema_mgr_->proj_maps[0]),
        &(schema_mgr_->proj_maps[1]), key_cmp_, filter_,
        [ctx](std::function<Status(size_t, int64_t)> fn,
              std::function<Status(size_t)> on_finished) {
          return ctx->RegisterTaskGroup(std::move(fn), std::move(on_finished));
        },
        [ctx](int task_group_id, int64_t num_tasks) {
          return ctx->StartTaskGroup(task_group_id, num_tasks);
        },
        [this](int64_t, ExecBatch batch) { return this->OutputBatchCallback(batch); },
        [this](int64_t total_num_batches) {
          return this->FinishedCallback(total_num_batches);
        }));

    task_group_probe_ = ctx->RegisterTaskGroup(
        [this](size_t thread_index, int64_t task_id) -> Status {
          return impl_->ProbeSingleBatch(thread_index,
                                         std::move(queued_batches_to_probe_[task_id]));
        },
        [this](size_t thread_index) -> Status {
          return OnQueuedBatchesProbed(thread_index);
        });

    return Status::OK();
  }

  Status StartProducing() override {
    NoteStartProducing(ToStringExtra());