Method Window

bolt/exec/Window.cpp:54–103

}

Window::Window(
    int32_t operatorId,
    DriverCtx* driverCtx,
    const std::shared_ptr<const core::WindowNode>& windowNode)
    : Operator(
          driverCtx,
          windowNode->outputType(),
          operatorId,
          windowNode->id(),
          "Window",
          windowNode->inputsSorted()
              ? (windowNode->canSpill(driverCtx->queryConfig())
                     ? driverCtx->makeSpillConfig(operatorId)
                     : std::nullopt)
              : (windowNode->canSpill(driverCtx->queryConfig())
                     ? driverCtx->makeSpillConfig(
                           operatorId,
                           driverCtx->queryConfig().rowBasedSpillMode())
                     : std::nullopt)),
      numInputColumns_(windowNode->inputType()->size()),
      stringAllocator_(pool()),
      windowNode_(windowNode),
      currentPartition_(nullptr) {
  enableJit_ = driverCtx->queryConfig().enableJitRowCmpRow();
  const auto numPartitionKeys = windowNode->partitionKeys().size();
  const auto numSortingKeys = windowNode->sortingKeys().size();
  for (auto i = 0; i < numSortingKeys; ++i) {
    sortKeyInfo_.push_back(
        std::make_pair(numPartitionKeys + i, windowNode->sortingOrders()[i]));
  }

  auto* spillConfig =
      spillConfig_.has_value() ? &spillConfig_.value() : nullptr;
  needSort_ = !(windowNode->inputsSorted());
  const auto windowBuildType = getWindowBuildType();
  const bool ignore = ignorePeer();
  const auto maxBatchRows = driverCtx->queryConfig().maxOutputBatchRows();
  const auto preferredBatchBytes =
      driverCtx->queryConfig().preferredOutputBatchBytes();
  const auto spillThreshold =
      operatorCtx_->driverCtx()->queryConfig().orderBySpillMemoryThreshold();

  setWindowBuild(
      windowBuildType,
      spillConfig,
      spillThreshold,
      ignore,
      maxBatchRows,
      preferredBatchBytes);
}

void Window::setRowsStreamingWindowBuild(
    bool ignorePeer,