Method AllocateRemoteSpace

be/src/runtime/tmp-file-mgr.cc:1430–1516

}

Status TmpFileGroup::AllocateRemoteSpace(int64_t num_bytes, TmpFile** tmp_file,
    int64_t* file_offset, vector<int>* at_capacity_dirs) {
  // Only one remote dir supported currently.
  string dir = tmp_file_mgr_->tmp_dirs_remote_->path();
  // It is not supposed to have a remote directory other than HDFS, Ozone, or S3.
  DCHECK(IsHdfsPath(dir.c_str(), false) || IsOzonePath(dir.c_str(), false)
      || IsS3APath(dir.c_str(), false));

  // Look for the space from a previous created file.
  if (!tmp_files_remote_.empty()) {
    TmpFile* tmp_file_cur = tmp_files_remote_.back().get();
    // If the file is blocklisted or is at capacity, we will create a new file instead.
    if (!tmp_file_cur->is_blacklisted()) {
      if (tmp_file_cur->AllocateSpace(num_bytes, file_offset)) {
        *tmp_file = tmp_file_cur;
        return Status::OK();
      }
    }
  }

  // Return an error if the new bytes is over the bytes limit of the query or the remote
  // directory.
  int64_t new_bytes =
      current_bytes_allocated_.Load() + tmp_file_mgr_->GetRemoteTmpFileSize();
  if (bytes_limit_ != -1 && new_bytes > bytes_limit_) {
    return Status(TErrorCode::SCRATCH_LIMIT_EXCEEDED, bytes_limit_, GetBackendString());
  }

  int64_t remote_dir_bytes_limit = tmp_file_mgr_->tmp_dirs_remote_->bytes_limit();
  if (remote_dir_bytes_limit != -1 && new_bytes > remote_dir_bytes_limit) {
    return Status(
        TErrorCode::SCRATCH_LIMIT_EXCEEDED, remote_dir_bytes_limit, GetBackendString());
  }

  // The device id of remote directory is defined as the max local device id
  // plus the index of the remote dir. Since we only support one remote dir now,
  // the id is the max local device id plus one.
  DeviceId dev_id = tmp_file_mgr_->tmp_dirs_.size();
  string unique_name = lexical_cast<string>(random_generator()());
  stringstream file_name;
  dir = ConstructRemoteDirPath(dir,
      ExecEnv::GetInstance()->configured_backend_address().hostname,
      PrintId(ExecEnv::GetInstance()->backend_id(), "_"), PrintId(unique_id(), "_"));

  string new_file_path = GenerateNewPath(dir, unique_name);
  const string& local_buffer_dir = tmp_file_mgr_->local_buff_dir_->path();
  string new_file_path_local = GenerateNewPath(local_buffer_dir, unique_name);

  unique_ptr<TmpFileRemote> tmp_file_r = make_unique<TmpFileRemote>(
      this, dev_id, new_file_path, new_file_path_local, false, dir.c_str());
  if (tmp_file_r == nullptr) {
    return Status("Failed to allocate temporary file object.");
  }
  if (tmp_file_r->hdfs_conn_ == nullptr) {
    return Status(Substitute("Failed to connect to FS: $0.", dir));
  }
  shared_ptr<TmpFile> tmp_file_remote(move(tmp_file_r));
  int64_t file_size = tmp_file_mgr_->GetRemoteTmpFileSize();