Method TmpFileRemote

be/src/runtime/tmp-file-mgr.cc:1015–1059

}

TmpFileRemote::TmpFileRemote(TmpFileGroup* file_group, TmpFileMgr::DeviceId device_id,
    const std::string& path, const std::string& local_buffer_path, bool expected_local,
    const char* hdfs_url)
  : TmpFile(file_group, device_id, path, expected_local) {
  DCHECK(hdfs_url != nullptr);
  hdfs_conn_ = nullptr;
  const HdfsFsCache::HdfsConnOptions* options = nullptr;
  if (IsHdfsPath(hdfs_url, false)) {
    disk_type_ = io::DiskFileType::DFS;
    disk_id_ = file_group->io_mgr_->RemoteDfsDiskId();
    disk_id_file_op_ = file_group->io_mgr_->RemoteDfsDiskFileOperId();
  } else if (IsOzonePath(hdfs_url, false)) {
    disk_type_ = io::DiskFileType::DFS;
    disk_id_ = file_group->io_mgr_->RemoteOzoneDiskId();
    disk_id_file_op_ = file_group->io_mgr_->RemoteDfsDiskFileOperId();
  } else if (IsS3APath(hdfs_url, false)) {
    disk_type_ = io::DiskFileType::S3;
    disk_id_ = file_group->io_mgr_->RemoteS3DiskId();
    disk_id_file_op_ = file_group->io_mgr_->RemoteS3DiskFileOperId();
    options = file_group_->tmp_file_mgr_->s3a_options();
  }
  Status status = HdfsFsCache::instance()->GetConnection(
      hdfs_url, &hdfs_conn_, &file_group_->tmp_file_mgr_->hdfs_conns_, options);
  file_size_ = file_group_->tmp_file_mgr_->GetRemoteTmpFileSize();
  local_buffer_path_ = local_buffer_path;
  disk_file_ = make_unique<io::DiskFile>(path_, file_group->io_mgr_,
      file_group_->tmp_file_mgr_->GetRemoteTmpFileSize(), disk_type_, &hdfs_conn_);
  if (file_group_->tmp_file_mgr_->IsRemoteBatchReadingEnabled()) {
    read_buffer_block_size_ = file_group_->tmp_file_mgr_->GetReadBufferBlockSize();
    int num_of_read_buffers = file_group_->tmp_file_mgr_->GetNumReadBuffersPerFile();
    disk_buffer_file_ = make_unique<io::DiskFile>(local_buffer_path_,
        file_group_->io_mgr_, file_group_->tmp_file_mgr_->GetRemoteTmpFileSize(),
        io::DiskFileType::LOCAL_BUFFER, read_buffer_block_size_, num_of_read_buffers);
    disk_read_page_cnts_ = std::make_unique<int64_t[]>(num_of_read_buffers);
    DCHECK(disk_read_page_cnts_.get() != nullptr);
    memset(disk_read_page_cnts_.get(), 0, num_of_read_buffers * sizeof(int64_t));
    for (int i = 0; i < num_of_read_buffers; i++) {
      fetch_ranges_.emplace_back(nullptr);
    }
  } else {
    disk_buffer_file_ = make_unique<io::DiskFile>(local_buffer_path_,
        file_group_->io_mgr_, file_group_->tmp_file_mgr_->GetRemoteTmpFileSize(),
        io::DiskFileType::LOCAL_BUFFER);
  }
}

TmpFileRemote::~TmpFileRemote() {
  // Need to return the buffer before deconstruction if buffer space is reserved.