| 1657 | } |
| 1658 | |
| 1659 | void HashBuild::reclaim( |
| 1660 | uint64_t /*unused*/, |
| 1661 | memory::MemoryReclaimer::Stats& stats) { |
| 1662 | BOLT_CHECK(canReclaim()); |
| 1663 | auto* driver = operatorCtx_->driver(); |
| 1664 | BOLT_CHECK_NOT_NULL(driver); |
| 1665 | BOLT_CHECK(!nonReclaimableSection_); |
| 1666 | |
| 1667 | BOLT_TEST_ADJUST("bytedance::bolt::exec::HashBuild::reclaim", this); |
| 1668 | |
| 1669 | // can another thread call close() while hashbuild is in arbitration and |
| 1670 | // reclaim is called on it? |
| 1671 | if (exceededMaxSpillLevelLimit_) { |
| 1672 | // NOTE: we might have reached to the max spill limit. |
| 1673 | return; |
| 1674 | } |
| 1675 | |
| 1676 | // NOTE: a hash build operator is reclaimable if it is in the middle of table |
| 1677 | // build processing and is not under non-reclaimable execution section. |
| 1678 | if (nonReclaimableState()) { |
| 1679 | // TODO: reduce the log frequency if it is too verbose. |
| 1680 | RECORD_METRIC_VALUE(kMetricMemoryNonReclaimableCount); |
| 1681 | ++stats.numNonReclaimableAttempts; |
| 1682 | LOG(WARNING) << name() << " can't reclaim from hash build operator, state_[" |
| 1683 | << stateName(state_) << "], nonReclaimableSection_[" |
| 1684 | << nonReclaimableSection_ << "], spiller_[" |
| 1685 | << (intermediateStateCleared_ || spiller_->finalized() |
| 1686 | ? "finalized" |
| 1687 | : "non-finalized") |
| 1688 | << "] " << pool()->name() |
| 1689 | << ", usage: " << succinctBytes(pool()->currentBytes()); |
| 1690 | return; |
| 1691 | } |
| 1692 | |
| 1693 | const auto& task = driver->task(); |
| 1694 | BOLT_CHECK(task->pauseRequested()); |
| 1695 | const std::vector<Operator*> operators = |
| 1696 | task->findPeerOperators(operatorCtx_->driverCtx()->pipelineId, this); |
| 1697 | |
| 1698 | if (isFirstSpill_ && operators.size() == 1) { |
| 1699 | isFirstSpill_ = false; |
| 1700 | LOG(INFO) << name() << " calculateJoinBits triggered by " << __FUNCTION__; |
| 1701 | if (calculateJoinBits<false>()) { |
| 1702 | spiller_.reset(); |
| 1703 | setupSpiller(); |
| 1704 | } |
| 1705 | spiller_->setMemoryUsedTriggered(pool()->currentBytes()); |
| 1706 | auto* spillConf = const_cast<common::SpillConfig*>(spillConfig()); |
| 1707 | operatorCtx_->adjustSpillCompressionKind(spillConf); |
| 1708 | } |
| 1709 | |
| 1710 | for (auto* op : operators) { |
| 1711 | HashBuild* buildOp = dynamic_cast<HashBuild*>(op); |
| 1712 | BOLT_CHECK_NOT_NULL(buildOp); |
| 1713 | BOLT_CHECK(buildOp->canReclaim()); |
| 1714 | if (buildOp->nonReclaimableState()) { |
| 1715 | // TODO: reduce the log frequency if it is too verbose. |
| 1716 | RECORD_METRIC_VALUE(kMetricMemoryNonReclaimableCount); |
nothing calls this directly
no test coverage detected