Method ExecInternal

be/src/runtime/fragment-instance-state.cc:483–535

}

Status FragmentInstanceState::ExecInternal() {
  DCHECK_EQ(current_state_.Load(), FInstanceExecStatePB::WAITING_FOR_OPEN);
  // Inject failure if debug actions are enabled.
  RETURN_IF_ERROR(DebugAction(query_state_->query_options(), "FIS_IN_EXEC_INTERNAL"));

  RuntimeProfile::Counter* plan_exec_timer =
      ADD_CHILD_TIMER(timings_profile_, "ExecTreeExecTime", EXEC_TIMER_NAME);
  RuntimeProfile::SummaryStatsCounter* row_batch_mem_alloc_duration =
      PROFILE_RowBatchMemPoolAllocDuration.Instantiate(profile());
  RuntimeProfile::SummaryStatsCounter* row_batch_mem_alloc_bytes =
      PROFILE_RowBatchMemPoolAllocBytes.Instantiate(profile());
  RuntimeProfile::SummaryStatsCounter* row_batch_mem_free_duration =
      PROFILE_RowBatchMemPoolFreeDuration.Instantiate(profile());
  RuntimeProfile::SummaryStatsCounter* row_batch_mem_free_bytes =
      PROFILE_RowBatchMemPoolFreeBytes.Instantiate(profile());
  auto update_counters = MakeScopeExitTrigger([&]() {
    MemPoolCounters mem_counters = row_batch_->GetMemPoolCounters();
    row_batch_mem_alloc_duration->Merge(mem_counters.sys_alloc_duration);
    row_batch_mem_alloc_bytes->Merge(mem_counters.allocated_bytes);
    row_batch_mem_free_duration->Merge(mem_counters.sys_free_duration);
    row_batch_mem_free_bytes->Merge(mem_counters.freed_bytes);
  });
  SCOPED_THREAD_COUNTER_MEASUREMENT(runtime_state_->total_thread_statistics());
  bool exec_tree_complete = false;
  UpdateState(StateEvent::WAITING_FOR_FIRST_BATCH);
  do {
    Status status;
    row_batch_->Reset();
    {
      SCOPED_TIMER(plan_exec_timer);
      RETURN_IF_ERROR(
          exec_tree_->GetNext(runtime_state_, row_batch_.get(), &exec_tree_complete));
    }
    UpdateState(StateEvent::BATCH_PRODUCED);
    if (VLOG_ROW_IS_ON) row_batch_->VLogRows("FragmentInstanceState::ExecInternal()");
    COUNTER_ADD(rows_produced_counter_, row_batch_->num_rows());
    RETURN_IF_ERROR(sink_->Send(runtime_state_, row_batch_.get()));
    UpdateState(StateEvent::BATCH_SENT);
  } while (!exec_tree_complete);
  // Release resources from final row batch.
  row_batch_->Reset();

  UpdateState(StateEvent::LAST_BATCH_SENT);

  // Close the tree before the sink is flushed to release 'exec_tree_' resources.
  // This can significantly reduce resource consumption if 'sink_' is a join
  // build, where FlushFinal() blocks until the consuming fragment is finished.
  exec_tree_->Close(runtime_state_);

  // Flush the sink as a final step.
  RETURN_IF_ERROR(sink_->FlushFinal(runtime_state()));
  return Status::OK();
}

void FragmentInstanceState::Close() {
  DCHECK(runtime_state_ != nullptr);