Method ArchiveQuery

be/src/service/impala-server.cc:1203–1268

}

void ImpalaServer::ArchiveQuery(const QueryHandle& query_handle) {
  vector<uint8_t> compressed_profile;
  Status status = query_handle->profile()->Compress(&compressed_profile);
  if (!status.ok()) {
    // Didn't serialize the string. Continue with empty string.
    LOG_EVERY_N(WARNING, 1000) << "Could not serialize profile to archive string "
                               << status.GetDetail();
    return;
  }

  // If there was an error initialising archival (e.g. directory is not writeable),
  // FLAGS_log_query_to_file will have been set to false
  if (FLAGS_log_query_to_file) {
    stringstream ss;
    ss << UnixMillis() << " " << PrintId(query_handle->query_id()) << " ";
    Base64Encode(compressed_profile, &ss);
    status = profile_logger_->AppendEntry(ss.str());
    if (!status.ok()) {
      LOG_EVERY_N(WARNING, 1000) << "Could not write to profile log file file ("
                                 << google::COUNTER << " attempts failed): "
                                 << status.GetDetail();
      LOG_EVERY_N(WARNING, 1000)
          << "Disable query logging with --log_query_to_file=false";
    }
  }

  // 'fetch_rows_lock()' protects several fields in ClientRequestState that are read
  // during QueryStateRecord creation. There should be no contention on this lock because
  // the query has already been closed (e.g. no more results can be fetched).
  shared_ptr<QueryStateRecord> record = nullptr;
  {
    lock_guard<mutex> l(*query_handle->fetch_rows_lock());
    record = make_shared<QueryStateRecord>(*query_handle, move(compressed_profile));
  }
  if (query_handle->GetCoordinator() != nullptr) {
    query_handle->GetCoordinator()->GetTExecSummary(&record->exec_summary);
  }

  EnqueueCompletedQuery(query_handle, record);

  if (FLAGS_query_log_size != 0 && FLAGS_query_log_size_in_bytes != 0) {
    int64_t record_size = EstimateSize(record.get());
    VLOG(3) << "QueryStateRecord of " << PrintId(query_handle->query_id()) << " is "
            << record_size << " bytes";

    lock_guard<mutex> l(query_log_lock_);
    // Add record to the beginning of the log, and to the lookup index.
    ImpaladMetrics::QUERY_LOG_EST_TOTAL_BYTES->Increment(record_size);
    query_log_est_sizes_.push_front(record_size);
    query_log_.push_front(move(record));
    query_log_index_[query_handle->query_id()] = &query_log_.front();

    while (!query_log_.empty()
        && ((FLAGS_query_log_size > -1 && FLAGS_query_log_size < query_log_.size())
            || (FLAGS_query_log_size_in_bytes > -1
                && FLAGS_query_log_size_in_bytes
                    < ImpaladMetrics::QUERY_LOG_EST_TOTAL_BYTES->GetValue()))) {
      query_log_index_.erase(query_log_.back()->id);