| 521 | } |
| 522 | |
| 523 | void logStatsIfNeeded(const OperatorStats& stats, int64_t lastBatchSize) { |
| 524 | constexpr int64_t kLogFreq = 1000 * 1000; |
| 525 | if ((stats.outputPositions / kLogFreq) != |
| 526 | (stats.outputPositions - lastBatchSize) / kLogFreq) { |
| 527 | auto operatorName = fmt::format( |
| 528 | "{}#{}[pipelineId={}]", |
| 529 | stats.operatorType, |
| 530 | stats.operatorId, |
| 531 | stats.pipelineId); |
| 532 | LOG(INFO) << "Operator " << operatorName << " stats: " |
| 533 | << "outputPositions=" << stats.outputPositions |
| 534 | << ", inputPositions=" << stats.inputPositions |
| 535 | << ", outputBytes=" << stats.outputBytes |
| 536 | << ", inputBytes=" << stats.inputBytes |
| 537 | << ", outputVectors=" << stats.outputVectors |
| 538 | << ", inputVectors=" << stats.inputVectors; |
| 539 | } |
| 540 | } |
| 541 | |
| 542 | bool Driver::checkUnderArbitration(ContinueFuture* future) { |
| 543 | return task()->queryCtx()->checkUnderArbitration(future); |