Method PartitionedOutput

bolt/exec/PartitionedOutput.cpp:125–168

} // namespace detail

PartitionedOutput::PartitionedOutput(
    int32_t operatorId,
    DriverCtx* ctx,
    const std::shared_ptr<const core::PartitionedOutputNode>& planNode,
    bool eagerFlush)
    : Operator(
          ctx,
          planNode->outputType(),
          operatorId,
          planNode->id(),
          "PartitionedOutput"),
      keyChannels_(toChannels(planNode->inputType(), planNode->keys())),
      numDestinations_(planNode->numPartitions()),
      replicateNullsAndAny_(planNode->isReplicateNullsAndAny()),
      partitionFunction_(
          numDestinations_ == 1
              ? nullptr
              : planNode->partitionFunctionSpec().create(numDestinations_)),
      outputChannels_(calculateOutputChannels(
          planNode->inputType(),
          planNode->outputType(),
          planNode->outputType())),
      bufferManager_(OutputBufferManager::getInstance()),
      // NOTE: 'bufferReleaseFn_' holds a reference on the associated task to
      // prevent it from deleting while there are output buffers being accessed
      // out of the partitioned output buffer manager such as in Prestissimo,
      // the http server holds the buffers while sending the data response.
      bufferReleaseFn_([task = operatorCtx_->task()]() {}),
      maxBufferedBytes_(ctx->task->queryCtx()
                            ->queryConfig()
                            .maxPartitionedOutputBufferSize()),
      eagerFlush_(eagerFlush),
      compressionKind_(
          ctx->task->queryCtx()->queryConfig().isExchangeCompressionEnabled()
              ? common::CompressionKind_ZSTD
              : common::CompressionKind_NONE) {
  if (!planNode->isPartitioned()) {
    BOLT_USER_CHECK_EQ(numDestinations_, 1);
  }
  if (numDestinations_ == 1) {
    BOLT_USER_CHECK(keyChannels_.empty());
    BOLT_USER_CHECK_NULL(partitionFunction_);
  }
}

void PartitionedOutput::initializeInput(RowVectorPtr input) {
  input_ = std::move(input);