| 540 | }; |
| 541 | |
| 542 | Result<std::shared_ptr<Buffer>> DecompressBuffer(const std::shared_ptr<Buffer>& buf, |
| 543 | const IpcReadOptions& options, |
| 544 | util::Codec* codec) { |
| 545 | if (buf == nullptr || buf->size() == 0) { |
| 546 | return buf; |
| 547 | } |
| 548 | |
| 549 | if (buf->size() < 8) { |
| 550 | return Status::Invalid( |
| 551 | "Likely corrupted message, compressed buffers " |
| 552 | "are larger than 8 bytes by construction"); |
| 553 | } |
| 554 | |
| 555 | const uint8_t* data = buf->data(); |
| 556 | int64_t compressed_size = buf->size() - sizeof(int64_t); |
| 557 | int64_t uncompressed_size = bit_util::FromLittleEndian(util::SafeLoadAs<int64_t>(data)); |
| 558 | |
| 559 | if (uncompressed_size == -1) { |
| 560 | return SliceBuffer(buf, sizeof(int64_t), compressed_size); |
| 561 | } |
| 562 | |
| 563 | ARROW_ASSIGN_OR_RAISE(auto uncompressed, |
| 564 | AllocateBuffer(uncompressed_size, options.memory_pool)); |
| 565 | |
| 566 | ARROW_ASSIGN_OR_RAISE( |
| 567 | int64_t actual_decompressed, |
| 568 | codec->Decompress(compressed_size, data + sizeof(int64_t), uncompressed_size, |
| 569 | uncompressed->mutable_data())); |
| 570 | if (actual_decompressed != uncompressed_size) { |
| 571 | return Status::Invalid("Failed to fully decompress buffer, expected ", |
| 572 | uncompressed_size, " bytes but decompressed ", |
| 573 | actual_decompressed); |
| 574 | } |
| 575 | |
| 576 | // R build with openSUSE155 requires an explicit shared_ptr construction |
| 577 | return std::shared_ptr<Buffer>(std::move(uncompressed)); |
| 578 | } |
| 579 | |
| 580 | Status DecompressBuffers(Compression::type compression, const IpcReadOptions& options, |
| 581 | ArrayDataVector* fields) { |
no test coverage detected