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