| 83 | } |
| 84 | |
| 85 | BlockingReason Destination::flush( |
| 86 | OutputBufferManager& bufferManager, |
| 87 | const std::function<void()>& bufferReleaseFn, |
| 88 | ContinueFuture* future) { |
| 89 | if (!current_) { |
| 90 | return BlockingReason::kNotBlocked; |
| 91 | } |
| 92 | |
| 93 | // Upper limit of message size with no columns. |
| 94 | constexpr int32_t kMinMessageSize = 128; |
| 95 | auto listener = bufferManager.newListener(); |
| 96 | IOBufOutputStream stream( |
| 97 | *current_->pool(), |
| 98 | listener.get(), |
| 99 | std::max<int64_t>(kMinMessageSize, current_->size())); |
| 100 | const int64_t flushedRows = rowsInCurrent_; |
| 101 | |
| 102 | current_->flush(&stream); |
| 103 | current_.reset(); |
| 104 | |
| 105 | const int64_t flushedBytes = stream.tellp(); |
| 106 | |
| 107 | bytesInCurrent_ = 0; |
| 108 | rowsInCurrent_ = 0; |
| 109 | setTargetSizePct(); |
| 110 | |
| 111 | bool blocked = bufferManager.enqueue( |
| 112 | taskId_, |
| 113 | destination_, |
| 114 | std::make_unique<SerializedPage>( |
| 115 | stream.getIOBuf(bufferReleaseFn), nullptr, flushedRows), |
| 116 | future); |
| 117 | |
| 118 | recordEnqueued_(flushedBytes, flushedRows); |
| 119 | |
| 120 | return blocked ? BlockingReason::kWaitForConsumer |
| 121 | : BlockingReason::kNotBlocked; |
| 122 | } |
| 123 | } // namespace detail |
| 124 | |
| 125 | PartitionedOutput::PartitionedOutput( |