Method next

bolt/exec/MergeSource.cpp:232–280

  }

  BlockingReason next(RowVectorPtr& data, ContinueFuture* future) override {
    data.reset();

    if (atEnd_ && !currentPage_) {
      return BlockingReason::kNotBlocked;
    }

    if (!currentPage_) {
      auto pages = client_->next(1, &atEnd_, future);
      BOLT_CHECK_LE(pages.size(), 1);
      currentPage_ = pages.empty() ? nullptr : std::move(pages.front());

      if (!currentPage_) {
        if (atEnd_) {
          return BlockingReason::kNotBlocked;
        }
        return BlockingReason::kWaitForProducer;
      }
    }
    if (inputStream_ == nullptr) {
      mergeExchange_->stats().wlock()->rawInputBytes += currentPage_->size();
      inputStream_ = currentPage_->prepareStreamForDeserialize();
    }

    if (!inputStream_->atEnd()) {
      static const serializer::presto::PrestoVectorSerde::PrestoOptions options(
          false, common::CompressionKind::CompressionKind_ZSTD);
      VectorStreamGroup::read(
          inputStream_.get(),
          mergeExchange_->pool(),
          mergeExchange_->outputType(),
          &data,
          &options);

      auto lockedStats = mergeExchange_->stats().wlock();
      lockedStats->addInputVector(data->estimateFlatSize(), data->size());
      lockedStats->rawInputPositions += data->size();
    }

    // Since VectorStreamGroup::read() may cause inputStream to be at end,
    // check again and reset currentPage_ and inputStream_ here.
    if (inputStream_->atEnd()) {
      // Reached end of the stream.
      currentPage_ = nullptr;
      inputStream_.reset();
    }

    return BlockingReason::kNotBlocked;
  }

  void close() override {
    if (client_) {