Method ReadAsync

be/src/runtime/tmp-file-mgr.cc:1649–1706

}

Status TmpFileGroup::ReadAsync(TmpWriteHandle* handle, MemRange buffer) {
  DCHECK(handle->write_range_ != nullptr);
  DCHECK(!handle->is_cancelled_);
  DCHECK_EQ(buffer.len(), handle->data_len());
  Status status;
  VLOG(3) << "ReadAsync " << handle->TmpFilePath() << " "
          << handle->write_range_->offset() << " " << handle->on_disk_len();
  // Don't grab 'write_state_lock_' in this method - it is not necessary because we
  // don't touch any members that it protects and could block other threads for the
  // duration of the synchronous read.
  DCHECK(!handle->write_in_flight_);
  DCHECK(handle->read_range_ == nullptr);
  DCHECK(handle->write_range_ != nullptr);

  MemRange read_buffer = buffer;
  if (handle->is_compressed()) {
    int64_t compressed_len = handle->compressed_len_;
    if (!handle->compressed_.TryAllocate(compressed_len)) {
      return tmp_file_mgr_->compressed_buffer_tracker()->MemLimitExceeded(
          nullptr, "Failed to decompress spilled data", compressed_len);
    }
    DCHECK_EQ(compressed_len, handle->write_range_->len());
    read_buffer = MemRange(handle->compressed_.buffer(), compressed_len);
  }

  // Don't grab handle->write_state_lock_, it is safe to touch all of handle's state
  // since the write is not in flight.
  handle->read_range_ = scan_range_pool_.Add(new ScanRange);
  int64_t offset = handle->write_range_->offset();
  if (handle->file_ != nullptr && !handle->file_->is_local()) {
    TmpFileRemote* tmp_file = static_cast<TmpFileRemote*>(handle->file_);
    DiskFile* local_read_buffer_file = tmp_file->GetReadBufferFile(offset);
    DiskFile* remote_file = tmp_file->DiskFile();
    // Reset the read_range, use the remote filesystem's disk id.
    handle->read_range_->Reset(
        ScanRange::FileInfo{
            remote_file->path().c_str(), tmp_file->hdfs_conn_, tmp_file->mtime_},
        handle->write_range_->len(), offset, tmp_file->disk_id(), false,
        BufferOpts::ReadInto(
            read_buffer.data(), read_buffer.len(), BufferOpts::NO_CACHING),
        nullptr, remote_file, local_read_buffer_file);
  } else {
    // Read from local.
    handle->read_range_->Reset(
        ScanRange::FileInfo{handle->write_range_->file()},
        handle->write_range_->len(), offset, handle->write_range_->disk_id(), false,
        BufferOpts::ReadInto(
            read_buffer.data(), read_buffer.len(), BufferOpts::NO_CACHING));
  }

  read_counter_->Add(1);
  bytes_read_counter_->Add(read_buffer.len());

  bool needs_buffers;
  RETURN_IF_ERROR(io_ctx_->StartScanRange(handle->read_range_, &needs_buffers));
  DCHECK(!needs_buffers) << "Already provided a buffer";
  return Status::OK();
}