| 692 | } |
| 693 | |
| 694 | void addAggregationStep(QueryPlan & query_plan, |
| 695 | const AggregationAnalysisResult & aggregation_analysis_result, |
| 696 | const QueryAnalysisResult & query_analysis_result, |
| 697 | const PlannerContextPtr & planner_context, |
| 698 | const SelectQueryInfo & select_query_info) |
| 699 | { |
| 700 | const Settings & settings = planner_context->getQueryContext()->getSettingsRef(); |
| 701 | auto aggregator_params = getAggregatorParams(planner_context, aggregation_analysis_result, query_analysis_result, select_query_info); |
| 702 | |
| 703 | SortDescription sort_description_for_merging; |
| 704 | SortDescription group_by_sort_description; |
| 705 | |
| 706 | if (settings[Setting::force_aggregation_in_order]) |
| 707 | { |
| 708 | group_by_sort_description = getSortDescriptionFromNames(aggregation_analysis_result.aggregation_keys); |
| 709 | sort_description_for_merging = group_by_sort_description; |
| 710 | } |
| 711 | |
| 712 | const size_t memory_limited_max_threads = getMaxThreadsForAvailableMemory( |
| 713 | settings[Setting::max_threads], settings[Setting::max_threads_min_free_memory_per_thread]); |
| 714 | auto merge_threads = memory_limited_max_threads; |
| 715 | auto temporary_data_merge_threads = settings[Setting::aggregation_memory_efficient_merge_threads] |
| 716 | ? static_cast<size_t>(settings[Setting::aggregation_memory_efficient_merge_threads]) |
| 717 | : memory_limited_max_threads; |
| 718 | |
| 719 | bool storage_has_evenly_distributed_read = false; |
| 720 | const auto & table_expression_node_to_data = planner_context->getTableExpressionNodeToData(); |
| 721 | |
| 722 | if (table_expression_node_to_data.size() == 1) |
| 723 | { |
| 724 | auto it = table_expression_node_to_data.begin(); |
| 725 | const auto & table_expression_node = it->first; |
| 726 | if (const auto * table_node = table_expression_node->as<TableNode>()) |
| 727 | storage_has_evenly_distributed_read = table_node->getStorage()->hasEvenlyDistributedRead(); |
| 728 | else if (const auto * table_function_node = table_expression_node->as<TableFunctionNode>()) |
| 729 | storage_has_evenly_distributed_read = table_function_node->getStorageOrThrow()->hasEvenlyDistributedRead(); |
| 730 | } |
| 731 | |
| 732 | auto aggregating_step = std::make_unique<AggregatingStep>( |
| 733 | query_plan.getCurrentHeader(), |
| 734 | aggregator_params, |
| 735 | aggregation_analysis_result.grouping_sets_parameters_list, |
| 736 | query_analysis_result.aggregate_final, |
| 737 | settings[Setting::max_block_size], |
| 738 | settings[Setting::aggregation_in_order_max_block_bytes], |
| 739 | merge_threads, |
| 740 | temporary_data_merge_threads, |
| 741 | storage_has_evenly_distributed_read, |
| 742 | settings[Setting::group_by_use_nulls], |
| 743 | std::move(sort_description_for_merging), |
| 744 | std::move(group_by_sort_description), |
| 745 | query_analysis_result.aggregation_should_produce_results_in_order_of_bucket_number, |
| 746 | settings[Setting::enable_memory_bound_merging_of_aggregation_results], |
| 747 | settings[Setting::force_aggregation_in_order], |
| 748 | settings[Setting::enable_sharding_aggregator]); |
| 749 | query_plan.addStep(std::move(aggregating_step)); |
| 750 | } |
| 751 |
no test coverage detected