Method DoFetch

be/src/runtime/io/disk-io-mgr.cc:413–485

}

Status RemoteOperRange::DoFetch() {
  hdfsFS hdfs_conn = disk_file_src_->hdfs_conn_;
  DCHECK(hdfs_conn != nullptr);
  // Fetch the data from the source file (remote) to the destination file (local).
  DCHECK(disk_file_dst_ != nullptr);
  DCHECK(disk_file_src_ != nullptr);
  int64_t buffer_idx = disk_file_dst_->GetReadBufferIndex(offset_);
  int64_t local_file_size = disk_file_dst_->GetReadBuffActualSize(buffer_idx);
  const string& remote_file_path = disk_file_src_->path();
  DiskQueue* queue = io_mgr_->disk_queues_[disk_id_];
  Status status = Status::OK();

  // Get the shared lock to prevent the physical files from deletion during the fetching.
  // The sequence is to get the local file lock, then remote file lock, or it might meet
  // deadlocks.
  shared_lock<shared_mutex> dstl(disk_file_dst_->physical_file_lock_);
  shared_lock<shared_mutex> srcl(disk_file_src_->physical_file_lock_);

  // Check if the remote file is deleted.
  auto src_status = disk_file_src_->GetFileStatus();
  if (src_status != io::DiskFileStatus::PERSISTED) {
    DCHECK(src_status == io::DiskFileStatus::DELETED);
    return Status(Substitute("File has been deleted, path: '$0'", remote_file_path));
  }

  unique_lock<SpinLock> read_buffer_lock(
      *(disk_file_dst_->GetBufferBlockLock(buffer_idx)));
  MemBlock* read_buffer_bloc = disk_file_dst_->GetBufferBlock(buffer_idx);
  if (disk_file_dst_->IsReadBufferBlockStatus(
          read_buffer_bloc, MemBlockStatus::DISABLED, dstl, &read_buffer_lock)) {
    // If the read block is disabled, the status doesn't allow any writes to
    // the block, probably the query ends or is cancelled.
    return Status(Substitute(
        "Mem block '$0' has been deleted, path: '$1'", buffer_idx, remote_file_path));
  }
  RETURN_IF_ERROR(disk_file_dst_->AllocReadBufferBlockLocked(
      read_buffer_bloc, local_file_size, dstl, read_buffer_lock));
  DCHECK(read_buffer_bloc->data() != nullptr);
  hdfsFile remote_hdfs_file =
      hdfsOpenFile(hdfs_conn, remote_file_path.c_str(), O_RDONLY, 0, 0, block_size_);
  if (remote_hdfs_file == nullptr) {
    status = Status(TErrorCode::DISK_IO_ERROR, GetBackendString(),
        Substitute("Could not open file: $0: $1", remote_file_path, GetStrErrMsg()));
  } else {
    int ret = [&]() {
      ScopedHistogramTimer read_timer(queue->read_latency());
      return hdfsPreadFully(hdfs_conn, remote_hdfs_file, offset_,
          read_buffer_bloc->data(), local_file_size);
    }();
    if (ret != -1) {
      queue->read_size()->Update(local_file_size);
      disk_file_dst_->SetReadBufferBlockStatus(
          read_buffer_bloc, MemBlockStatus::WRITTEN, dstl, &read_buffer_lock);
    } else {
      // The caller may need to handle the error, and deal with the read buffer block.
      status = Status(TErrorCode::DISK_IO_ERROR, GetBackendString(),
          GetHdfsErrorMsg("Error reading from HDFS file: ", remote_file_path));
    }