Method Init

be/src/runtime/runtime-state.cc:133–179

}

void RuntimeState::Init() {
  SCOPED_TIMER(profile_->total_time_counter());

  // Register with the thread mgr
  resource_pool_ = ExecEnv::GetInstance()->thread_mgr()->CreatePool();
  DCHECK(resource_pool_ != nullptr);
  if (fragment_ != nullptr) {
    // Ensure that the planner correctly determined the required threads.
    resource_pool_->set_max_required_threads(fragment_->thread_reservation);
  }

  total_thread_statistics_ = ADD_THREAD_COUNTERS(runtime_profile(), "TotalThreads");
  total_storage_wait_timer_ = ADD_TIMER(runtime_profile(), "TotalStorageWaitTime");
  total_network_send_timer_ = ADD_TIMER(runtime_profile(), "TotalNetworkSendTime");
  total_network_receive_timer_ = ADD_TIMER(runtime_profile(), "TotalNetworkReceiveTime");

  instance_mem_tracker_ = obj_pool()->Add(new MemTracker(
      runtime_profile(), -1, runtime_profile()->name(), query_mem_tracker()));

  if (instance_buffer_reservation_ != nullptr) {
    instance_buffer_reservation_->InitChildTracker(profile_,
        query_state_->buffer_reservation(), instance_mem_tracker_,
        numeric_limits<int64_t>::max());
  }

  // Find local timezone.
  // (For FE tests leave 'local_time_zone_' as default. FE tests don't load the timezone
  // db since they don't need any timezone information.)
  if (!TestInfo::is_fe_test()) {
    const Timezone* tz = TimezoneDatabase::FindTimezone(query_ctx().local_time_zone);
    if (tz != nullptr) {
      // Use UTCPTR (actually a nullptr) if the timezone is UTC. This makes it easy to
      // optimize many code paths for the UTC case.
      local_time_zone_ = tz == &TimezoneDatabase::GetUtcTimezone() ? UTCPTR : tz;
    } else {
      const string msg = Substitute(
          "Failed to find local timezone '$0'.Falling back to UTC.",
          query_ctx().local_time_zone);
      LOG(WARNING) << msg;
      LogError(ErrorMsg(TErrorCode::GENERAL, msg));
      local_time_zone_ = UTCPTR;
    }
    if (query_options().use_local_tz_for_unix_timestamp_conversions) {
      time_zone_for_unix_time_conversions_ = local_time_zone_;
    }
  }
}

Status RuntimeState::StartSpilling(MemTracker* mem_tracker) {
  return query_state_->StartSpilling(this, mem_tracker);