| 498 | } |
| 499 | |
| 500 | int32_t Operator::getMaxReadBatchSize( |
| 501 | int32_t minMaxReadBatchSize, |
| 502 | int32_t batchSizeAlignment, |
| 503 | int32_t complexTypeNumberThreshold) { |
| 504 | const auto& queryConfig = operatorCtx_->task()->queryCtx()->queryConfig(); |
| 505 | int32_t maxReadBatchSize = queryConfig.maxOutputBatchRows(); |
| 506 | if (minMaxReadBatchSize > maxReadBatchSize) { |
| 507 | // if maxReadBatchSize from config is less than minMaxReadBatchSize, |
| 508 | // return minMaxReadBatchSize directly |
| 509 | LOG(INFO) << name() << " max read batch size: " << maxReadBatchSize; |
| 510 | return maxReadBatchSize; |
| 511 | } |
| 512 | int32_t fixWidthLength = 0; |
| 513 | int32_t complexTypeNumber = 0; |
| 514 | for (const auto& type : asRowType(outputType_)->children()) { |
| 515 | fixWidthLength += type->isFixedWidth() |
| 516 | ? type->cppSizeInBytes() |
| 517 | : ((type->isVarchar() || type->isVarbinary()) ? sizeof(StringView) : 0); |
| 518 | complexTypeNumber += |
| 519 | (type->isArray() || type->isMap() || type->isRow()) ? 1 : 0; |
| 520 | } |
| 521 | |
| 522 | uint64_t maxBatchBytes = queryConfig.preferredOutputBatchBytes(); |
| 523 | if (fixWidthLength * maxReadBatchSize > maxBatchBytes) { |
| 524 | // if fix length is too large, reduce maxReadBatchSize correspondingly |
| 525 | maxReadBatchSize = maxBatchBytes / fixWidthLength; |
| 526 | } |
| 527 | if (maxReadBatchSize > minMaxReadBatchSize) { |
| 528 | // if contains complex type, reduce maxReadBatchSize correspondingly |
| 529 | if (complexTypeNumber >= complexTypeNumberThreshold) { |
| 530 | maxReadBatchSize = minMaxReadBatchSize; |
| 531 | } else { |
| 532 | maxReadBatchSize -= (maxReadBatchSize - minMaxReadBatchSize) * |
| 533 | complexTypeNumber / complexTypeNumberThreshold; |
| 534 | } |
| 535 | } |
| 536 | // align maxReadBatchSize to batchSizeAlignment |
| 537 | maxReadBatchSize = std::max( |
| 538 | maxReadBatchSize / batchSizeAlignment * batchSizeAlignment, |
| 539 | minMaxReadBatchSize); |
| 540 | |
| 541 | LOG(INFO) << name() << " max read batch size: " << maxReadBatchSize |
| 542 | << ", fix width length: " << fixWidthLength |
| 543 | << ", complex type number: " << complexTypeNumber; |
| 544 | return maxReadBatchSize; |
| 545 | } |
| 546 | |
| 547 | void Operator::recordBlockingTime(uint64_t start, BlockingReason reason) { |
| 548 | uint64_t now = |
nothing calls this directly
no test coverage detected