| 95 | } |
| 96 | |
| 97 | BlockingReason LocalExchangeQueue::enqueue( |
| 98 | RowVectorPtr input, |
| 99 | ContinueFuture* future) { |
| 100 | auto abort = aborted_.rlock(); |
| 101 | if (*abort) { |
| 102 | // [morsel] LocalExchangeQueue has been aborted, we need to notify the |
| 103 | // LocalPartition operator to stop fueling the queue with more data. |
| 104 | BOLT_CHECK_NOT_NULL( |
| 105 | driverDispatcher_, |
| 106 | "LocalExchangeQueue cannot be aborted when it does not have 'morsel-driven' driverDispatcher"); |
| 107 | return BlockingReason::kYield; |
| 108 | } |
| 109 | |
| 110 | std::vector<RowVectorPtr> morsels; |
| 111 | |
| 112 | if (morselSliced_ && input->size() > morselSize_) { |
| 113 | BOLT_CHECK_GE( |
| 114 | morselSize_, |
| 115 | 0, |
| 116 | "Morsel size cannot be < 1. Current morselSize={}", |
| 117 | morselSize_); |
| 118 | for (auto offset = 0; offset < input->size(); offset += morselSize_) { |
| 119 | auto length = std::min(input->size() - offset, morselSize_); |
| 120 | auto sliced = |
| 121 | std::dynamic_pointer_cast<RowVector>(input->slice(offset, length)); |
| 122 | morsels.emplace_back(std::move(sliced)); |
| 123 | } |
| 124 | } |
| 125 | // Enqueue last piece enqueue |
| 126 | else if (input->size() > 0) { |
| 127 | morsels.emplace_back(std::move(input)); |
| 128 | } |
| 129 | |
| 130 | std::vector<ContinuePromise> consumerPromises; |
| 131 | std::vector<ContinuePromise> producerPromises; |
| 132 | |
| 133 | bool blockedOnConsumer = false; |
| 134 | bool isClosed = queue_.withWLock([&](auto& queue) { |
| 135 | if (closed_) { |
| 136 | return true; |
| 137 | } |
| 138 | |
| 139 | uint64_t inputBytesActual = 0; |
| 140 | for (auto morsel : morsels) { |
| 141 | inputBytesActual += morsel->estimateFlatSize(); |
| 142 | queue.push(std::move(morsel)); |
| 143 | } |
| 144 | consumerPromises = std::move(consumerPromises_); |
| 145 | |
| 146 | // When size reaches the limit, make the current LocalExchangeQueue |
| 147 | // spawn one or more child queue named "primedQueue" and then start a driver |
| 148 | // for the consuming pipeline reading from each queue. |
| 149 | if (driverDispatcher_) { |
| 150 | BOLT_CHECK_GE( |
| 151 | primedQueueSize_, |
| 152 | 0, |
| 153 | "Primed queue size cannot be < 1. Current primedQueueSize_={}", |
| 154 | primedQueueSize_); |
no test coverage detected