| 1706 | } |
| 1707 | |
| 1708 | Status TmpFileGroup::WaitForAsyncRead( |
| 1709 | TmpWriteHandle* handle, MemRange buffer, const BufferPoolClientCounters* counters) { |
| 1710 | DCHECK(handle->read_range_ != nullptr); |
| 1711 | // Don't grab handle->write_state_lock_, it is safe to touch all of handle's state |
| 1712 | // since the write is not in flight. |
| 1713 | SCOPED_TIMER(disk_read_timer_); |
| 1714 | MemRange read_buffer = handle->is_compressed() ? |
| 1715 | MemRange{handle->compressed_.buffer(), handle->compressed_.Size()} : |
| 1716 | buffer; |
| 1717 | DCHECK(read_buffer.data() != nullptr); |
| 1718 | unique_ptr<BufferDescriptor> io_mgr_buffer; |
| 1719 | Status status = handle->read_range_->GetNext(&io_mgr_buffer); |
| 1720 | if (!status.ok()) goto exit; |
| 1721 | DCHECK(io_mgr_buffer != NULL); |
| 1722 | DCHECK(io_mgr_buffer->eosr()); |
| 1723 | DCHECK_LE(io_mgr_buffer->len(), read_buffer.len()); |
| 1724 | if (io_mgr_buffer->len() < read_buffer.len()) { |
| 1725 | // The read was truncated - this is an error. |
| 1726 | status = Status(TErrorCode::SCRATCH_READ_TRUNCATED, read_buffer.len(), |
| 1727 | handle->write_range_->file(), GetBackendString(), handle->write_range_->offset(), |
| 1728 | io_mgr_buffer->len()); |
| 1729 | goto exit; |
| 1730 | } |
| 1731 | DCHECK_EQ(io_mgr_buffer->buffer(), |
| 1732 | handle->is_compressed() ? handle->compressed_.buffer() : buffer.data()); |
| 1733 | |
| 1734 | // Decrypt and decompress in the reverse order that we compressed then encrypted the |
| 1735 | // data originally. |
| 1736 | if (FLAGS_disk_spill_encryption) { |
| 1737 | status = handle->CheckHashAndDecrypt(read_buffer, counters); |
| 1738 | if (!status.ok()) goto exit; |
| 1739 | } |
| 1740 | |
| 1741 | if (handle->is_compressed()) { |
| 1742 | SCOPED_TIMER2( |
| 1743 | compression_timer_, counters == nullptr ? nullptr : counters->compression_time); |
| 1744 | scoped_ptr<Codec> decompressor; |
| 1745 | status = Codec::CreateDecompressor( |
| 1746 | nullptr, false, tmp_file_mgr_->compression_codec(), &decompressor); |
| 1747 | if (status.ok()) { |
| 1748 | int64_t decompressed_len = buffer.len(); |
| 1749 | uint8_t* decompressed_buffer = buffer.data(); |
| 1750 | status = decompressor->ProcessBlock(true, read_buffer.len(), read_buffer.data(), |
| 1751 | &decompressed_len, &decompressed_buffer); |
| 1752 | } |
| 1753 | // Free the compressed data regardless of whether the read was successful. |
| 1754 | handle->FreeCompressedBuffer(); |
| 1755 | if (!status.ok()) goto exit; |
| 1756 | } |
| 1757 | exit: |
| 1758 | if (handle->file_ != nullptr && !handle->file_->is_local()) { |
| 1759 | auto tmp_file = static_cast<TmpFileRemote*>(handle->file_); |
| 1760 | // If all the pages of specific read buffer have been read, try delete the read |
| 1761 | // buffer. |
| 1762 | if (tmp_file_mgr()->IsRemoteBatchReadingEnabled()) { |
| 1763 | int buffer_idx = tmp_file->GetReadBufferIndex(handle->write_range_->offset()); |
| 1764 | bool all_read = tmp_file->IncrementReadPageCount(buffer_idx); |
| 1765 | if (all_read) tmp_file->TryDeleteMemReadBufferShared(buffer_idx); |
no test coverage detected