Method Load

be/src/exec/blob-reader.h:92–130

template <typename OutputType, typename Deserializer>
Status BlobReader<OutputType, Deserializer>::Load(io::RequestContext* request_context,
    MemTracker* mem_tracker, ObjectPool* obj_pool, const std::string& path,
    int64_t content_offset, int64_t content_length, OutputType* output) {
  DCHECK(request_context != nullptr);
  DCHECK(mem_tracker != nullptr);
  DCHECK(obj_pool != nullptr);
  DCHECK(output != nullptr);

  // Allocate buffer for reading the blob
  ScopedBuffer buffer(mem_tracker);
  if (!buffer.TryAllocate(content_length)) {
    return Status(strings::Substitute(
        "Could not allocate buffer of $0 bytes for blob file '$1'.",
        content_length, path));
  }

  // Get HDFS connection
  io::ScanRange::FileInfo file_info;
  file_info.mtime = 1;
  RETURN_IF_ERROR(HdfsFsCache::instance()->GetConnection(
      path, &file_info.fs, &fs_cache_));
  file_info.filename = path.c_str();

  // Create scan range for the blob content
  io::ScanRange* scan_range = io::ScanRange::AllocateScanRange(obj_pool, file_info,
      content_length, content_offset, {}, nullptr, -1, false,
      io::BufferOpts::ReadInto(buffer.buffer(), buffer.Size(),
          io::BufferOpts::USE_DATA_CACHE));

  // Read the blob data
  std::unique_ptr<io::BufferDescriptor> io_buffer;
  bool needs_buffers;
  RETURN_IF_ERROR(request_context->StartScanRange(scan_range, &needs_buffers));
  DCHECK(!needs_buffers) << "Already provided a buffer";
  RETURN_IF_ERROR(scan_range->GetNext(&io_buffer));
  scan_range->ReturnBuffer(std::move(io_buffer));

  return Deserializer::Deserialize(buffer.buffer(), content_length, output);
}

/// Size of the blob header (length + magic) for deletion vector blobs
static constexpr int DELETION_VECTOR_BLOB_HEADER_SIZE = 8;