| 289 | namespace { |
| 290 | |
| 291 | Status ReadFieldsSubset(int64_t offset, int32_t metadata_length, |
| 292 | io::RandomAccessFile* file, |
| 293 | const FieldsLoaderFunction& fields_loader, |
| 294 | const std::shared_ptr<Buffer>& metadata, int64_t required_size, |
| 295 | std::shared_ptr<Buffer>& body) { |
| 296 | const flatbuf::Message* message = nullptr; |
| 297 | uint8_t continuation_metadata_size = sizeof(int32_t) + sizeof(int32_t); |
| 298 | // skip 8 bytes (32-bit continuation indicator + 32-bit little-endian length prefix) |
| 299 | RETURN_NOT_OK(internal::VerifyMessage(metadata->data() + continuation_metadata_size, |
| 300 | metadata->size() - continuation_metadata_size, |
| 301 | &message)); |
| 302 | auto batch = message->header_as_RecordBatch(); |
| 303 | if (batch == nullptr) { |
| 304 | return Status::IOError( |
| 305 | "Header-type of flatbuffer-encoded Message is not RecordBatch."); |
| 306 | } |
| 307 | internal::IoRecordedRandomAccessFile io_recorded_random_access_file(required_size); |
| 308 | RETURN_NOT_OK(fields_loader(batch, &io_recorded_random_access_file)); |
| 309 | const auto& read_ranges = io_recorded_random_access_file.GetReadRanges(); |
| 310 | for (const auto& range : read_ranges) { |
| 311 | auto read_result = file->ReadAt(offset + metadata_length + range.offset, range.length, |
| 312 | body->mutable_data() + range.offset); |
| 313 | if (!read_result.ok()) { |
| 314 | return Status::IOError("Failed to read message body, error ", |
| 315 | read_result.status().ToString()); |
| 316 | } |
| 317 | } |
| 318 | return Status::OK(); |
| 319 | } |
| 320 | |
| 321 | } // namespace |
| 322 |
no test coverage detected