Method ReadFromCache

be/src/runtime/io/scan-range.cc:621–678

}

Status ScanRange::ReadFromCache(
    const unique_lock<mutex>& reader_lock, bool* read_succeeded) {
  DCHECK(reader_lock.mutex() == &reader_->lock_ && reader_lock.owns_lock());
  DCHECK(UseHdfsCache());
  DCHECK_EQ(bytes_read_, 0);
  *read_succeeded = false;
  Status status = file_reader_->Open();
  if (!status.ok()) return status;

  // Check cancel status.
  {
    unique_lock<mutex> lock(lock_);
    RETURN_IF_ERROR(cancel_status_);
  }

  file_reader_->CachedFile(&cache_.data, &cache_.len);
  // Data was not cached, caller will fall back to normal read path.
  if (cache_.data == nullptr) {
    VLOG_QUERY << "Cache read failed for scan range: " << DebugString()
               << ". Switching to disk read path.";
    // Clean up the scan range state before re-issuing it.
    file_reader_->Close();
    return Status::OK();
  }
  // A partial read can happen when files are truncated.
  // TODO: If HDFS ever supports partially cached blocks, we'll have to distinguish
  // between errors and partially cached blocks here.
  if (cache_.len < len()) {
    VLOG_QUERY << "Error reading file from HDFS cache: " << file_ << ". Expected "
      << len() << " bytes, but read " << cache_.len << ". Switching to disk read path.";
    // Null out the cache buffer to avoid any interactions when this falls
    // back to the regular read path.
    cache_.len = 0;
    cache_.data = nullptr;
    // Close the scan range. 'read_succeeded' is still false, so the caller will fall back
    // to non-cached read of this scan range.
    file_reader_->Close();
    return Status::OK();
  }

  *read_succeeded = true;
  // If there are sub-ranges, then we need to memcpy() them from the cached buffer.
  if (HasSubRanges()) return Status::OK();

  DCHECK(!buffer_manager_->is_client_buffer());
  buffer_manager_->set_cached_buffer();
  bytes_read_ = cache_.len;

  // Create a single buffer desc for the entire scan range and enqueue that.
  // The memory is owned by the HDFS java client, not the Impala backend.
  unique_ptr<BufferDescriptor> desc = unique_ptr<BufferDescriptor>(new BufferDescriptor(
      this, cache_.data, 0));
  desc->len_ = cache_.len;
  desc->eosr_ = true;
  EnqueueReadyBuffer(move(desc));
  COUNTER_ADD_IF_NOT_NULL(reader_->bytes_read_counter_, cache_.len);
  return Status::OK();
}