Method ProcessBuildInputAndOpenProbe

be/src/exec/blocking-join-node.cc:236–323

}

Status BlockingJoinNode::ProcessBuildInputAndOpenProbe(
    RuntimeState* state, JoinBuilder* build_sink) {
  // This function implements various strategies for executing Open() on the left child
  // and doing or waiting for the join build. Some allow the two to proceed in parallel,
  // while others do them serially. Generally parallelism yields better performance
  // except inside a subplan. There we expect Open() to be called a number of times
  // proportional to the input data of the SubplanNode, so processing the build input
  // in the main thread can be more efficient, assuming that thread creation is expensive
  // relative to a single subplan iteration.
  //
  // In this block, we also compute the 'overlap' time for the left and right child. This
  // is the time (i.e. clock reads) when the right child stops overlapping with the left
  // child. For the single threaded case, the left and right child never overlap. For the
  // build side in a different thread, the overlap stops when the left child Open()
  // returns.
  if (UseSeparateBuild(state->query_options())) {
    // Open the left child in parallel before waiting for the build fragment to maximise
    // parallelism. The build execution is done concurrently by the build finstance's
    // thread.
    RETURN_IF_ERROR(child(0)->Open(state));
    // AcquireResourcesForBuild() opens the buffer pool client, so that probe reservation
    // can be transferred.
    RETURN_IF_ERROR(AcquireResourcesForBuild(state));
    {
      SCOPED_TIMER(runtime_profile_->inactive_timer());
      events_->MarkEvent("Waiting for initial build");
      RETURN_IF_ERROR(build_sink->WaitForInitialBuild(state));
      events_->MarkEvent("Initial build available");
    }
    waited_for_build_ = true;
  } else if (!IsInSubplan() && state->resource_pool()->TryAcquireThreadToken()) {
    // The build is integrated into the join node and we got a thread token. Do the hash
    // table build in a separate thread.
    Status build_side_status;
    runtime_profile()->AppendExecOption("Join Build-Side Prepared Asynchronously");
    string thread_name = Substitute("join-build-thread (finst:$0, plan-node-id:$1)",
        PrintId(state->fragment_instance_id()), id());
    unique_ptr<Thread> build_thread;
    Status thread_status = Thread::Create(FragmentInstanceState::FINST_THREAD_GROUP_NAME,
        thread_name, [this, state, build_sink, status=&build_side_status]() {
          ProcessBuildInputAsync(state, build_sink, status);
        }, &build_thread, true);
    if (!thread_status.ok()) {
      state->resource_pool()->ReleaseThreadToken(false);
      return thread_status;
    }
    // Open the left child so that it may perform any initialisation in parallel.
    // Don't exit even if we see an error, we still need to wait for the build thread
    // to finish.
    Status open_status = child(0)->Open(state);

    // The left/right child overlap stops here.
    built_probe_overlap_stop_watch_.SetTimeCeiling();

    // Blocks until ProcessBuildInput has returned, after which the build side structures
    // are fully constructed.
    build_thread->Join();
    RETURN_IF_ERROR(build_side_status);