| 56 | } |
| 57 | |
| 58 | bool CachedCompressedReadBuffer::nextImpl() |
| 59 | { |
| 60 | |
| 61 | /// It represents the end of file when the position exceeds the limit in hdfs shared storage or handling implicit column data in compact impl. |
| 62 | /// TODO: handle hdfs case |
| 63 | if (/*(storage_type == StorageType::Hdfs || */is_limit /*)*/ && file_pos >= static_cast<size_t>(limit_offset_in_file)) { |
| 64 | owned_cell = nullptr; |
| 65 | |
| 66 | return false; |
| 67 | } |
| 68 | |
| 69 | /// Let's check for the presence of a decompressed block in the cache, grab the ownership of this block, if it exists. |
| 70 | UInt128 key = cache->hash(path, file_pos); |
| 71 | |
| 72 | owned_cell = cache->getOrSet(key, [&]() |
| 73 | { |
| 74 | initInput(); |
| 75 | file_in->seek(file_pos, SEEK_SET); |
| 76 | |
| 77 | auto cell = std::make_shared<UncompressedCacheCell>(); |
| 78 | |
| 79 | size_t size_decompressed; |
| 80 | size_t size_compressed_without_checksum; |
| 81 | cell->compressed_size = readCompressedData(size_decompressed, size_compressed_without_checksum, false); |
| 82 | |
| 83 | if (cell->compressed_size) |
| 84 | { |
| 85 | // * a little bit hack here for reducing memory copy |
| 86 | // * allocate 12 more bytes to store {size_decompressed} and {size_decompressed}, padding at the end of the data |
| 87 | cell->additional_bytes = codec->getAdditionalSizeAtTheEndOfBuffer(); |
| 88 | auto buffer = HybridCache::Buffer{size_decompressed + cell->additional_bytes + sizeof(cell->compressed_size) + sizeof(cell->additional_bytes)}; |
| 89 | cell->data = std::move(buffer); |
| 90 | cell->data.shrink(size_decompressed + cell->additional_bytes); |
| 91 | decompressTo(reinterpret_cast<char *>(cell->data.data()), size_decompressed, size_compressed_without_checksum); |
| 92 | } |
| 93 | |
| 94 | return cell; |
| 95 | }); |
| 96 | |
| 97 | if (owned_cell->data.size() == 0) |
| 98 | return false; |
| 99 | |
| 100 | working_buffer = Buffer(reinterpret_cast<char *>(owned_cell->data.data()), reinterpret_cast<char *>(owned_cell->data.data()) + owned_cell->data.size() - owned_cell->additional_bytes); |
| 101 | |
| 102 | /// nextimpl_working_buffer_offset is set in the seek function (lazy seek). So we have to |
| 103 | /// check that we are not seeking beyond working buffer. |
| 104 | if (nextimpl_working_buffer_offset > working_buffer.size()) |
| 105 | throw Exception(ErrorCodes::SEEK_POSITION_OUT_OF_BOUND, "Seek position is beyond the decompressed block (pos: " |
| 106 | "{}, block size: {})", nextimpl_working_buffer_offset, toString(working_buffer.size())); |
| 107 | |
| 108 | file_pos += owned_cell->compressed_size; |
| 109 | |
| 110 | return true; |
| 111 | } |
| 112 | |
| 113 | CachedCompressedReadBuffer::CachedCompressedReadBuffer( |
| 114 | const std::string & path_, |