Method getOutput

bolt/exec/Exchange.cpp:126–160

}

RowVectorPtr Exchange::getOutput() {
  if (currentPages_.empty()) {
    return nullptr;
  }

  uint64_t rawInputBytes{0};
  vector_size_t resultOffset = 0;
  for (const auto& page : currentPages_) {
    rawInputBytes += page->size();

    auto inputStream = page->prepareStreamForDeserialize();

    while (!inputStream->atEnd()) {
      getSerde()->deserialize(
          inputStream.get(),
          pool(),
          outputType_,
          &result_,
          resultOffset,
          &options_);
      resultOffset = result_->size();
    }
  }

  currentPages_.clear();

  {
    auto lockedStats = stats_.wlock();
    lockedStats->rawInputBytes += rawInputBytes;
    lockedStats->rawInputPositions += result_->size();
    lockedStats->addInputVector(result_->estimateFlatSize(), result_->size());
  }

  return result_;
}

void Exchange::close() {
  SourceOperator::close();