Method noMoreData

bolt/exec/LocalPartition.cpp:202–235

}

void LocalExchangeQueue::noMoreData() {
  auto abort = aborted_.rlock();
  if (*abort) {
    return;
  }

  std::vector<ContinuePromise> consumerPromises;
  queue_.withWLock([&](auto& queue) {
    BOLT_CHECK_GT(pendingProducers_, 0);
    --pendingProducers_;
    if (noMoreProducers_ && pendingProducers_ == 0) {
      // [morsel-driven] optionally notify the PartitionedOutputBuffer of the
      // task one creator is finished
      // Start the last driver using all remaining data chunks in the queue,
      // even if the queue is empty because if the last morsel-driven driver
      // finishes before the last driver creator finishes, the query might hang
      // because Task::allPeersFinished will return false and since there is no
      // more new driver, then the HashTable will never finish.
      if (driverDispatcher_) {
        uint64_t reducedBytes = 0;
        std::shared_ptr<LocalExchangeQueue> primedQueue =
            spawnPrimedQueueLocked(queue, reducedBytes);
        if (driverDispatcher_->schedule(primedQueue, partition_, true)) {
          memoryManager_->decreaseMemoryUsage(reducedBytes);
        } else {
          VLOG(1) << "Dispatcher already finished. No new driver created.";
        }
      }

      consumerPromises = std::move(consumerPromises_);
    }
  });
  notify(consumerPromises);
}

BlockingReason LocalExchangeQueue::next(
    ContinueFuture* future,