| 43 | } |
| 44 | |
| 45 | Status OutboundRowBatch::TryCompress(TrackedString* compression_scratch, |
| 46 | bool* is_compressed) { |
| 47 | DCHECK(compression_scratch != nullptr); |
| 48 | Lz4Compressor compressor(nullptr, false); |
| 49 | RETURN_IF_ERROR(compressor.Init()); |
| 50 | auto compressor_cleanup = |
| 51 | MakeScopeExitTrigger([&compressor]() { compressor.Close(); }); |
| 52 | |
| 53 | *is_compressed = false; |
| 54 | int64_t uncompressed_size = tuple_data_.size(); |
| 55 | // If the input size is too large for LZ4 to compress, MaxOutputLen() will return 0. |
| 56 | int64_t compressed_size = compressor.MaxOutputLen(uncompressed_size); |
| 57 | if (compressed_size == 0) { |
| 58 | return Status(TErrorCode::LZ4_COMPRESSION_INPUT_TOO_LARGE, uncompressed_size); |
| 59 | } |
| 60 | DCHECK_GT(compressed_size, 0); |
| 61 | if (compression_scratch->size() < compressed_size) { |
| 62 | compression_scratch->resize(compressed_size); |
| 63 | } |
| 64 | |
| 65 | uint8_t* input = reinterpret_cast<uint8_t*>(tuple_data_.data()); |
| 66 | uint8_t* compressed_output = const_cast<uint8_t*>( |
| 67 | reinterpret_cast<const uint8_t*>(compression_scratch->data())); |
| 68 | RETURN_IF_ERROR(compressor.ProcessBlock( |
| 69 | true, uncompressed_size, input, &compressed_size, &compressed_output)); |
| 70 | if (LIKELY(compressed_size < uncompressed_size)) { |
| 71 | compression_scratch->resize(compressed_size); |
| 72 | tuple_data_.swap(*compression_scratch); |
| 73 | *is_compressed = true; |
| 74 | // TODO: could copy to a smaller buffer if compressed data is much smaller to |
| 75 | // save memory |
| 76 | } |
| 77 | VLOG_ROW << "uncompressed size: " << uncompressed_size << ", compressed size: " |
| 78 | << compressed_size; |
| 79 | return Status::OK(); |
| 80 | } |
| 81 | |
| 82 | void OutboundRowBatch::SetHeader(int num_rows, int num_tuples_per_row, |
| 83 | int64_t uncompressed_size, bool is_compressed) { |
nothing calls this directly
no test coverage detected