Method getOutput

bolt/exec/Window.cpp:917–984

}

RowVectorPtr Window::getOutput() {
  MicrosecondTimer timer(&computeWindowFunctionTimeUs_);
  if (needSort_ && !noMoreInput_) {
    return nullptr;
  }
  if (isFinished()) {
    recordWindowStats();
    if (numRows_ == 0 || numRows_ - numProcessedRows_ == 0) {
      windowBuild_->finish();
    }
  }

  if (numRows_ == 0) {
    return nullptr;
  }

  auto numRowsLeft = numRows_ - numProcessedRows_;
  if (numRowsLeft == 0) {
    return nullptr;
  }

  if (!currentPartition_) {
    callResetPartition();
    if (!currentPartition_) {
      // WindowBuild doesn't have a partition to output.
      if (isFinished()) {
        windowBuild_->finish();
      }
      return nullptr;
    }
  }

  if (isSpillableWindowBuild_ && currentPartition_->isSpilled()) {
    return getOutputFromSpilledPartition();
  }

  if (currentPartition_->supportRowsStreaming() &&
      partitionOffset_ == currentPartition_->numRows()) {
    if (!currentPartition_->buildNextRows()) {
      windowBuild_->loadNextPartialPartitionFromSpill();
      return nullptr;
    }
  }

  auto numOutputRows = std::min(numRowsPerOutput_, numRowsLeft);
  auto usedBytes = operatorCtx_->pool()->currentBytes();
  auto result = BaseVector::create<RowVector>(
      outputType_, numOutputRows, operatorCtx_->pool());

  // Compute the output values of window functions.
  auto numResultRows = callApplyLoop(numOutputRows, result);
  if (numRows_ == 0 || numRows_ - numProcessedRows_ == 0) {
    windowBuild_->finish();
  }

  auto resultBytes = operatorCtx_->pool()->currentBytes() - usedBytes;
  std::optional<uint64_t> resultRowSize = resultBytes / numResultRows;