| 3373 | } |
| 3374 | |
| 3375 | void flushCompressed( |
| 3376 | const std::vector<std::unique_ptr<VectorStream>>& streams, |
| 3377 | const StreamArena& arena, |
| 3378 | folly::io::Codec& codec, |
| 3379 | int32_t numRows, |
| 3380 | OutputStream* output, |
| 3381 | PrestoOutputStreamListener* listener) { |
| 3382 | char codecMask = kCompressedBitMask; |
| 3383 | #ifdef BOLT_ENABLE_CRC |
| 3384 | if (listener) { |
| 3385 | codecMask |= kCheckSumBitMask; |
| 3386 | } |
| 3387 | |
| 3388 | // Reset CRC computation |
| 3389 | if (listener) { |
| 3390 | listener->pause(); |
| 3391 | } |
| 3392 | #endif |
| 3393 | |
| 3394 | writeInt32(output, numRows); |
| 3395 | |
| 3396 | IOBufOutputStream out(*(arena.pool()), nullptr, arena.size()); |
| 3397 | writeInt32(&out, streams.size()); |
| 3398 | |
| 3399 | for (auto& stream : streams) { |
| 3400 | stream->flush(&out); |
| 3401 | } |
| 3402 | |
| 3403 | const auto uncompressedSize64 = out.tellp(); |
| 3404 | BOLT_CHECK( |
| 3405 | uncompressedSize64 > 0, |
| 3406 | "uncompressedSize {} should be positive, numRows {}", |
| 3407 | uncompressedSize64, |
| 3408 | numRows); |
| 3409 | BOLT_CHECK( |
| 3410 | uncompressedSize64 <= std::numeric_limits<int32_t>::max(), |
| 3411 | "uncompressedSize {} exceeds INT32_MAX, numRows {}", |
| 3412 | uncompressedSize64, |
| 3413 | numRows); |
| 3414 | const auto uncompressedSize = static_cast<int32_t>(uncompressedSize64); |
| 3415 | auto iobuf = out.getIOBuf(); |
| 3416 | |
| 3417 | // compress |
| 3418 | if (uncompressedSize <= codec.maxUncompressedLength()) { |
| 3419 | std::unique_ptr<folly::IOBuf> compressed; |
| 3420 | try { |
| 3421 | compressed = codec.compress(iobuf.get()); |
| 3422 | } catch (const std::exception& e) { |
| 3423 | BOLT_FAIL( |
| 3424 | "got exception in compression, uncompressedSize: {}, maxUncompressedLength: {}, detail: {}", |
| 3425 | uncompressedSize, |
| 3426 | codec.maxUncompressedLength(), |
| 3427 | e.what()); |
| 3428 | } |
| 3429 | const int32_t& compressedSize = compressed->computeChainDataLength(); |
| 3430 | if (compressedSize < uncompressedSize) { |
| 3431 | flushSerialization( |
| 3432 | numRows, |
no test coverage detected