Method probeAdmissionExtraReservationBytes

bolt/exec/HashBuild.cpp:1288–1334

}

uint64_t HashBuild::probeAdmissionExtraReservationBytes(
    uint64_t numRows) const {
  const bool notReadyForProbeAdmission =
      !operatorCtx_->driverCtx()
           ->queryConfig()
           .hashBuildProbeAdmissionUnderMemoryPressureEnabled() ||
      !spillEnabled() || isInputFromSpill() || numRows == 0;
  const bool spillUnavailable = spiller_ == nullptr ||
      spiller_->isAllSpilled() || exceededMaxSpillLevelLimit_;
  if (notReadyForProbeAdmission || spillUnavailable) {
    return 0;
  }

  const auto currentBytes = pool()->currentBytes();
  const auto* task = operatorCtx_->task().get();
  const auto pressure = task->memoryPressureSnapshot();
  const auto pressureWatermarkBytes = pressure.admissionWatermarkBytes();
  // Reclaim records a task-level pressure watermark. Borrow-from-RSS records
  // the current task's execution memory usage at the point dynamic memory
  // management is triggered. Taking the minimum non-zero watermark makes the
  // earliest pressure signal drive a stronger pre-probe admission reservation
  // while the build side can still spill.
  LOG(INFO) << name() << " probeAdmissionExtraReservationBytes: currentBytes="
            << succinctBytes(currentBytes) << ", reclaimWatermarkBytes="
            << succinctBytes(pressure.reclaimWatermarkBytes)
            << ", borrowFromRssWatermarkBytes="
            << succinctBytes(pressure.borrowFromRssWatermarkBytes)
            << ", configuredTaskMemoryQuotaBytes="
            << succinctBytes(pressure.configuredTaskMemoryQuotaBytes)
            << ", pressureWatermarkBytes="
            << succinctBytes(pressureWatermarkBytes) << ", details "
            << task->memoryPressureDetails();
  if (pressureWatermarkBytes != 0 &&
      currentBytes > pressureWatermarkBytes *
              operatorCtx_->driverCtx()
                  ->queryConfig()
                  .memoryPressureWatermarkRatio()) {
    return pressureWatermarkBytes;
  }

  const uint64_t probeReservationLimit =
      operatorCtx_->driverCtx()->queryConfig().preferredOutputBatchBytes() *
      operatorCtx_->driverCtx()
          ->queryConfig()
          .outputBatchMemoryReservationMultiple();
  return std::min<uint64_t>(currentBytes / 2, probeReservationLimit);
}

void HashBuild::recordSpillStats() {
  recordSpillStats(spiller_.get());