| 499 | } |
| 500 | |
| 501 | Status Scheduler::CheckEffectiveInstanceCount( |
| 502 | const FragmentScheduleState* fragment_state, ScheduleState* state) { |
| 503 | // These checks are only intended if COMPUTE_PROCESSING_COST=true. |
| 504 | if (!state->query_options().compute_processing_cost) return Status::OK(); |
| 505 | |
| 506 | int effective_instance_count = fragment_state->fragment.effective_instance_count; |
| 507 | DCHECK_GT(effective_instance_count, 0); |
| 508 | if (effective_instance_count < fragment_state->instance_states.size()) { |
| 509 | initializeSchedulerWarning(state->summary_profile()); |
| 510 | string warn_message = Substitute( |
| 511 | "$0 scheduled instance count ($1) is higher than its effective count ($2)", |
| 512 | fragment_state->fragment.display_name, fragment_state->instance_states.size(), |
| 513 | effective_instance_count); |
| 514 | state->summary_profile()->AppendInfoString(SCHEDULER_WARNING_KEY, warn_message); |
| 515 | LOG(WARNING) << warn_message; |
| 516 | } |
| 517 | |
| 518 | DCHECK(!fragment_state->instance_states.empty()); |
| 519 | // Find host with the largest instance assignment. |
| 520 | // Initialize to the first fragment instance. |
| 521 | int largest_inst_idx = 0; |
| 522 | int largest_inst_per_host = 1; |
| 523 | int this_inst_count = 1; |
| 524 | int num_host = 1; |
| 525 | for (int i = 1; i < fragment_state->instance_states.size(); i++) { |
| 526 | if (fragment_state->instance_states[i].host |
| 527 | == fragment_state->instance_states[i - 1].host) { |
| 528 | this_inst_count++; |
| 529 | } else { |
| 530 | if (largest_inst_per_host < this_inst_count) { |
| 531 | largest_inst_per_host = this_inst_count; |
| 532 | largest_inst_idx = i - 1; |
| 533 | } |
| 534 | this_inst_count = 1; |
| 535 | num_host++; |
| 536 | } |
| 537 | } |
| 538 | if (largest_inst_per_host < this_inst_count) { |
| 539 | largest_inst_per_host = this_inst_count; |
| 540 | largest_inst_idx = fragment_state->instance_states.size() - 1; |
| 541 | } |
| 542 | |
| 543 | QueryConstants qc; |
| 544 | if (largest_inst_per_host > qc.MAX_FRAGMENT_INSTANCES_PER_NODE) { |
| 545 | return Status(Substitute( |
| 546 | "$0 scheduled instance count ($1) is higher than maximum instances per node" |
| 547 | " ($2), indicating a planner bug. Consider running the query with" |
| 548 | " COMPUTE_PROCESSING_COST=false. Scheduler see $3 hosts for this fragment" |
| 549 | " with at most $4 fragment instance assignment at one host.", |
| 550 | fragment_state->fragment.display_name, largest_inst_per_host, |
| 551 | qc.MAX_FRAGMENT_INSTANCES_PER_NODE, num_host, largest_inst_per_host)); |
| 552 | } |
| 553 | |
| 554 | int planned_inst_per_host = ceil((float)effective_instance_count / num_host); |
| 555 | if (largest_inst_per_host > planned_inst_per_host) { |
| 556 | LOG(WARNING) << fragment_state->fragment.display_name |
| 557 | << " has imbalance number of instance to host assignment." |
| 558 | << " Consider running the query with COMPUTE_PROCESSING_COST=false." |
nothing calls this directly
no test coverage detected