Function ReadFieldsSubset

cpp/src/arrow/ipc/message.cc:287–319

namespace {

Status ReadFieldsSubset(int64_t offset, io::RandomAccessFile* file,
                        const FieldsLoaderFunction& fields_loader,
                        const std::shared_ptr<Buffer>& metadata, int64_t required_size,
                        std::shared_ptr<Buffer>& body) {
  DCHECK_GE(static_cast<size_t>(metadata->size()), sizeof(int32_t));
  const auto continuation = util::SafeLoadAs<int32_t>(metadata->data());
  // Either 8 bytes (32-bit continuation indicator + 32-bit little-endian length prefix)
  // or 4 bytes for legacy IPC without continuation indicator
  const auto continuation_size = (continuation == internal::kIpcContinuationToken)
                                     ? 2 * sizeof(int32_t)
                                     : sizeof(int32_t);

  const flatbuf::Message* message = nullptr;
  RETURN_NOT_OK(internal::VerifyMessage(metadata->data() + continuation_size,
                                        metadata->size() - continuation_size, &message));
  auto batch = message->header_as_RecordBatch();
  if (batch == nullptr) {
    return Status::IOError(
        "Header-type of flatbuffer-encoded Message is not RecordBatch.");
  }
  internal::IoRecordedRandomAccessFile io_recorded_random_access_file(required_size);
  RETURN_NOT_OK(fields_loader(batch, &io_recorded_random_access_file));
  const auto& read_ranges = io_recorded_random_access_file.GetReadRanges();
  for (const auto& range : read_ranges) {
    auto read_result = file->ReadAt(offset + metadata->size() + range.offset,
                                    range.length, body->mutable_data() + range.offset);
    if (!read_result.ok()) {
      return Status::IOError("Failed to read message body, error ",
                             read_result.status().ToString());
    }
  }
  return Status::OK();
}

struct ReadMessageState {
  std::unique_ptr<Message> result;