Method addInput

bolt/exec/LocalPartition.cpp:498–611

}

void LocalPartition::addInput(RowVectorPtr input) {
  {
    auto lockedStats = stats_.wlock();
    lockedStats->addOutputVector(input->estimateFlatSize(), input->size());
  }

  // Lazy vectors must be loaded or processed.
  for (auto& child : input->children()) {
    child->loadedVector();
  }

  if (numPartitions_ == 1) {
    ContinueFuture future;
    auto blockingReason = queues_[0]->enqueue(input, &future);
    if (FOLLY_UNLIKELY(blockingReason == BlockingReason::kYield)) {
      // [morsel] when enqueue to any localExchangeQueue returns yield, it means
      // the LocalExchangeQueues have been aborted (possibly due to
      // skipProbeOnEmptyBuild, in this case, we need to issue noMoreInput to
      // halt the driver.

      // @zj: find the correct way to yield producing pipeline when consuming
      // pipeline is aborted in skipProbeOnEmptyBuild
      //      notifyPeersToTerminate();
      //      noMoreInput();
    } else if (blockingReason != BlockingReason::kNotBlocked) {
      blockingReasons_.push_back(blockingReason);
      futures_.push_back(std::move(future));
    }
    return;
  }

  const auto singlePartition =
      partitionFunction_->partition(*input, partitions_);
  if (singlePartition.has_value()) {
    ContinueFuture future;
    auto blockingReason =
        queues_[singlePartition.value()]->enqueue(input, &future);
    if (FOLLY_UNLIKELY(blockingReason == BlockingReason::kYield)) {
      // [morsel] when enqueue to any localExchangeQueue returns yield, it means
      // the LocalExchangeQueues have been aborted (possibly due to
      // skipProbeOnEmptyBuild, in this case, we need to issue noMoreInput to
      // halt the driver.

      // @zj: find the correct way to yield producing pipeline when consuming
      // pipeline is aborted in skipProbeOnEmptyBuild
      //      notifyPeersToTerminate();
      //      noMoreInput();
    } else if (blockingReason != BlockingReason::kNotBlocked) {
      blockingReasons_.push_back(blockingReason);
      futures_.push_back(std::move(future));
    }
    return;
  }

  const auto numInput = input->size();
  std::vector<vector_size_t> maxIndex(numPartitions_, 0);
  for (auto i = 0; i < numInput; ++i) {
    ++maxIndex[partitions_[i]];