| 30 | } |
| 31 | |
| 32 | void ExplainAnalyzeTransform::transform(Chunk & chunk) |
| 33 | { |
| 34 | chunk.clear(); |
| 35 | if (!input.isFinished()) |
| 36 | return; |
| 37 | |
| 38 | // If the information of segment0 cannot be accepted |
| 39 | ProcessorsSet processors_set; |
| 40 | ProcessorProfiles profiles; |
| 41 | getProcessorProfiles(processors_set, profiles, this); |
| 42 | auto segment0_profile = GroupedProcessorProfile::getGroupedProfiles(profiles); |
| 43 | |
| 44 | auto scheduler = context->getSegmentScheduler(); |
| 45 | UInt64 time_out = context->getSettingsRef().operator_profile_receive_timeout; |
| 46 | auto time_start = std::chrono::system_clock::now(); |
| 47 | while (!scheduler->alreadyReceivedAllSegmentStatus(context->getCurrentQueryId())) |
| 48 | { |
| 49 | auto now = std::chrono::system_clock::now(); |
| 50 | UInt64 elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - time_start).count(); |
| 51 | if (elapsed >= time_out) |
| 52 | break; |
| 53 | } |
| 54 | |
| 55 | auto profiles_map = scheduler->getSegmentsProfile(context->getCurrentQueryId()); |
| 56 | String explain; |
| 57 | if ((kind == ASTExplainQuery::ExplainKind::LogicalAnalyze || kind == ASTExplainQuery::ExplainKind::DistributedAnalyze)) |
| 58 | { |
| 59 | AddressToStepProfile addr_to_step_profile; |
| 60 | for (auto & [segment_id, segment_profiles] : profiles_map) |
| 61 | { |
| 62 | for (auto & segment_profile : segment_profiles) |
| 63 | { |
| 64 | for (auto & [step_id, profile] : segment_profile->profiles) |
| 65 | addr_to_step_profile[segment_profile->worker_address][step_id] = profile; |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | auto segment0_steps_profiles = GroupedProcessorProfile::aggregateOperatorProfileToStepLevel(segment0_profile); |
| 70 | for (auto & [step_id, profile] : segment0_steps_profiles) |
| 71 | addr_to_step_profile[coordinator_address][step_id] = profile; |
| 72 | |
| 73 | CardinalityEstimator::estimate(*query_plan_ptr, context); |
| 74 | std::unordered_map<PlanNodeId, double> costs = CostCalculator::calculate(*query_plan_ptr, *context); |
| 75 | auto step_agg_operator_profiles = ProfileMetric::aggregateStepProfileBetweenWorkers(addr_to_step_profile); |
| 76 | if (kind == ASTExplainQuery::ExplainKind::LogicalAnalyze) |
| 77 | { |
| 78 | if (settings.json) |
| 79 | { |
| 80 | auto plan_cost = CostCalculator::calculatePlanCost(*query_plan_ptr, *context); |
| 81 | explain = PlanPrinter::jsonLogicalPlan(*query_plan_ptr, plan_cost, step_agg_operator_profiles, costs, settings); |
| 82 | } |
| 83 | else |
| 84 | explain = PlanPrinter::textLogicalPlan(*query_plan_ptr, context, costs, step_agg_operator_profiles, settings); |
| 85 | } |
| 86 | else if (kind == ASTExplainQuery::ExplainKind::DistributedAnalyze && !segment_descriptions.empty()) |
| 87 | { |
| 88 | if (settings.json) |
| 89 | explain = PlanPrinter::jsonDistributedPlan(segment_descriptions, step_agg_operator_profiles); |
nothing calls this directly
no test coverage detected