Method StartFInstances

be/src/runtime/query-state.cc:861–967

}

bool QueryState::StartFInstances() {
  VLOG(2) << "StartFInstances(): query_id=" << PrintId(query_id())
          << " #instances=" << fragment_info_.fragment_instance_ctxs.size();
  DCHECK_GT(refcnt_.Load(), 0);
  DCHECK_GT(backend_resource_refcnt_.Load(), 0) << "Should have been taken in Init()";

  DCHECK_GT(fragment_info_.fragments.size(), 0);
  vector<unique_ptr<Thread>> codegen_threads;
  int num_unstarted_instances = fragment_info_.fragment_instance_ctxs.size();

  // set up desc tbl
  DCHECK(query_ctx().__isset.desc_tbl_serialized);
  Status start_finstances_status =
      DescriptorTbl::Create(&obj_pool_, query_ctx().desc_tbl_serialized, &desc_tbl_);
  if (UNLIKELY(!start_finstances_status.ok())) goto error;
  VLOG(2) << "descriptor table for query=" << PrintId(query_id())
          << "\n" << desc_tbl_->DebugString();
  // IMPALA-13378: Verify that tuple ids in all PlanNode exist in the descriptor table.
  for (TPlanFragment f : fragment_info_.fragments) {
    for (TPlanNode node : f.plan.nodes) {
      for (TTupleId tuple_id : node.row_tuples) {
        if (UNLIKELY(desc_tbl_->GetTupleDescriptor(tuple_id) == nullptr)) {
          string msg = Substitute(
              "Tuple id $0 of PlanNode $1 not found in descriptor table",
              tuple_id, node.node_id);
          // It'd be helpful to also print 'fragment_info_' but it might lead to crash
          // if it's corrupt.
          LOG(ERROR) << msg << ": " << desc_tbl_->DebugString();
          start_finstances_status = Status(msg);
          goto error;
        }
      }
    }
  }

  start_finstances_status = FragmentState::CreateFragmentStateMap(
      fragment_info_, exec_rpc_params_, this, fragment_state_map_);
  if (UNLIKELY(!start_finstances_status.ok())) goto error;

  fragment_events_start_time_ = MonotonicStopWatch::Now();
  for (auto& fragment : fragment_state_map_) {
    FragmentState* fragment_state = fragment.second;
    for (int i = 0; i < fragment_state->instance_ctxs().size(); ++i) {
      const TPlanFragmentInstanceCtx* instance_ctx = fragment_state->instance_ctxs()[i];
      const PlanFragmentInstanceCtxPB* instance_ctx_pb =
          fragment_state->instance_ctx_pbs()[i];
      DCHECK_EQ(instance_ctx->fragment_idx, instance_ctx_pb->fragment_idx());
      FragmentInstanceState* fis = obj_pool_.Add(new FragmentInstanceState(
          this, fragment_state, *instance_ctx, *instance_ctx_pb));

      // start new thread to execute instance
      refcnt_.Add(1); // decremented in ExecFInstance()
      AcquireBackendResourceRefcount(); // decremented in ExecFInstance()

      // Add the fragment instance ID to the 'fis_map_'. Has to happen before the thread
      // is spawned or we may race with users of 'fis_map_'.
      fis_map_.emplace(fis->instance_id(), fis);