Method ExecLoadDataRequestImpl

be/src/service/client-request-state.cc:941–1007

}

void ClientRequestState::ExecLoadDataRequestImpl(bool exec_in_worker_thread) {
  const TExecRequest& exec_req = exec_request();
  if (exec_in_worker_thread) {
    VLOG_QUERY << "Running in worker thread";
    DCHECK(exec_state() == ExecState::PENDING);
    UpdateNonErrorExecState(ExecState::RUNNING);
  }
  DebugActionNoFail(
      exec_req.query_options, "CRS_DELAY_BEFORE_LOAD_DATA");

  TLoadDataResp response;
  Status status = frontend_->LoadData(exec_req.load_data_request, &response);
  if (exec_req.load_data_request.iceberg_tbl) {
    ExecLoadIcebergDataRequestImpl(response);
  }
  {
    lock_guard<mutex> l(lock_);
    RETURN_VOID_IF_ERROR(UpdateQueryStatus(status, exec_in_worker_thread));
  }

  request_result_set_.reset(new vector<TResultRow>);
  request_result_set_->push_back(response.load_summary);

  // We use TUpdateCatalogRequest to refresh the table metadata so that it will
  // fire an insert event just like an insert statement.
  TUpdatedPartition updatedPartition;
  updatedPartition.files.insert(updatedPartition.files.end(),
    response.loaded_files.begin(), response.loaded_files.end());
  TUpdateCatalogRequest catalog_update;
  // The partition_name is an empty string for unpartitioned tables.
  catalog_update.updated_partitions[response.partition_name] = updatedPartition;

  catalog_update.__set_sync_ddl(exec_req.query_options.sync_ddl);
  catalog_update.__set_header(GetCatalogServiceRequestHeader());
  catalog_update.target_table = exec_req.load_data_request.table_name.table_name;
  catalog_update.db_name = exec_req.load_data_request.table_name.db_name;
  catalog_update.is_overwrite = exec_req.load_data_request.overwrite;

  CatalogServiceConnection client(ExecEnv::GetInstance()->catalogd_client_cache(),
      *ExecEnv::GetInstance()->GetCatalogdAddress().get(), &status);
  {
    lock_guard<mutex> l(lock_);
    RETURN_VOID_IF_ERROR(UpdateQueryStatus(status, exec_in_worker_thread));
  }

  TUpdateCatalogResponse resp;
  status = client.DoRpc(
      &CatalogServiceClientWrapper::UpdateCatalog, catalog_update, &resp);
  query_events_->MarkEvent("UpdateCatalog finished");
  if (resp.__isset.profile) {
    for (const TEventSequence& catalog_timeline : resp.profile.event_sequences) {
      summary_profile_->AddEventSequence(catalog_timeline.name, catalog_timeline);
    }
  }
  {
    lock_guard<mutex> l(lock_);
    RETURN_VOID_IF_ERROR(UpdateQueryStatus(status, exec_in_worker_thread));
  }