#include <Planner/Planner.h>

#include <Core/Names.h>

#include <Core/ProtocolDefines.h>

#include <Core/ServerSettings.h>

#include <Core/Settings.h>

#include <Processors/QueryPlan/BlocksMarshallingStep.h>

#include <Common/Exception.h>

#include <Common/FieldVisitorToString.h>

#include <Common/FieldVisitors.h>

#include <Common/ProfileEvents.h>

#include <Common/logger_useful.h>

#include <Processors/QueryPlan/FractionalLimitStep.h>

#include <Processors/QueryPlan/FractionalOffsetStep.h>

#include <Processors/QueryPlan/MaterializingCTEStep.h>

#include <Processors/QueryPlan/QueryPlan.h>

#include <Processors/QueryPlan/ExpressionStep.h>

#include <Processors/QueryPlan/Optimizations/QueryPlanOptimizationSettings.h>

#include <Processors/QueryPlan/FilterStep.h>

#include <Processors/QueryPlan/UnionStep.h>

#include <Processors/QueryPlan/DistinctStep.h>

#include <Processors/QueryPlan/IntersectOrExceptStep.h>

#include <Processors/QueryPlan/CreatingSetsStep.h>

#include <Processors/QueryPlan/AggregatingStep.h>

#include <Processors/QueryPlan/MergingAggregatedStep.h>

#include <Processors/QueryPlan/SortingStep.h>

#include <Processors/QueryPlan/FillingStep.h>

#include <Processors/QueryPlan/LimitStep.h>

#include <Processors/QueryPlan/NegativeLimitStep.h>

#include <Processors/QueryPlan/OffsetStep.h>

#include <Processors/QueryPlan/NegativeOffsetStep.h>

#include <Processors/QueryPlan/ExtremesStep.h>

#include <Processors/QueryPlan/TotalsHavingStep.h>

#include <Processors/QueryPlan/RollupStep.h>

#include <Processors/QueryPlan/CubeStep.h>

#include <Processors/QueryPlan/LimitByStep.h>

#include <Processors/QueryPlan/WindowStep.h>

#include <Processors/QueryPlan/ReadFromRecursiveCTEStep.h>

#include <QueryPipeline/QueryPipelineBuilder.h>

#include <Interpreters/Context.h>

#include <Interpreters/convertFieldToType.h>

#include <Interpreters/HashTablesStatistics.h>

#include <Interpreters/StorageID.h>

#include <Storages/ColumnsDescription.h>

#include <Storages/IStorage.h>

#include <Storages/MergeTree/MergeTreeData.h>

#include <Storages/SelectQueryInfo.h>

#include <Storages/StorageDistributed.h>

#include <Storages/StorageDummy.h>

#include <Storages/StorageView.h>

#include <Storages/ObjectStorage/StorageObjectStorageCluster.h>

#include <AggregateFunctions/IAggregateFunction.h>

#include <Analyzer/Utils.h>

#include <Analyzer/ColumnNode.h>

#include <Analyzer/ConstantNode.h>

#include <Analyzer/FunctionNode.h>

#include <Analyzer/InterpolateNode.h>

#include <Analyzer/TableNode.h>

#include <Analyzer/TableFunctionNode.h>

#include <Analyzer/QueryNode.h>

#include <Analyzer/UnionNode.h>

#include <Analyzer/QueryTreeBuilder.h>

#include <Analyzer/AggregationUtils.h>

#include <Analyzer/WindowFunctionsUtils.h>

#include <Planner/CollectColumnIdentifiers.h>

#include <Planner/CollectMaterializedCTE.h>

#include <Planner/CollectSets.h>

#include <Planner/CollectTableExpressionData.h>

#include <Planner/findQueryForParallelReplicas.h>

#include <Planner/PlannerActionsVisitor.h>

#include <Planner/PlannerContext.h>

#include <Planner/PlannerCorrelatedSubqueries.h>

#include <Planner/PlannerExpressionAnalysis.h>

#include <Planner/PlannerJoins.h>

#include <Planner/PlannerJoinTree.h>

#include <Planner/PlannerQueryProcessingInfo.h>

#include <Planner/PlannerSorting.h>

#include <Planner/PlannerWindowFunctions.h>

#include <Planner/Utils.h>

#include <base/types.h>

namespace ProfileEvents

{

extern const Event SelectQueriesWithSubqueries;

extern const Event QueriesWithSubqueries;

}

namespace DB

{

namespace Setting

{

extern const SettingsUInt64 aggregation_in_order_max_block_bytes;

extern const SettingsUInt64 aggregation_memory_efficient_merge_threads;

extern const SettingsUInt64 allow_experimental_parallel_reading_from_replicas;

extern const SettingsBool collect_hash_table_stats_during_aggregation;

extern const SettingsOverflowMode distinct_overflow_mode;

extern const SettingsBool distributed_aggregation_memory_efficient;

extern const SettingsBool enable_memory_bound_merging_of_aggregation_results;

extern const SettingsBool empty_result_for_aggregation_by_constant_keys_on_empty_set;

extern const SettingsBool empty_result_for_aggregation_by_empty_set;

extern const SettingsBool exact_rows_before_limit;

extern const SettingsBool extremes;

extern const SettingsBool force_aggregation_in_order;

extern const SettingsUInt64 group_by_two_level_threshold;

extern const SettingsUInt64 group_by_two_level_threshold_bytes;

extern const SettingsBool group_by_use_nulls;

extern const SettingsUInt64 max_bytes_in_distinct;

extern const SettingsNonZeroUInt64 max_block_size;

extern const SettingsUInt64 max_size_to_preallocate_for_aggregation;

extern const SettingsUInt64 max_subquery_depth;

extern const SettingsUInt64 max_rows_in_distinct;

extern const SettingsMaxThreads max_threads;

extern const SettingsBool parallel_replicas_allow_in_with_subquery;

extern const SettingsString parallel_replicas_custom_key;

extern const SettingsUInt64 parallel_replicas_min_number_of_rows_per_replica;

extern const SettingsBool query_plan_enable_multithreading_after_window_functions;

extern const SettingsBool throw_on_unsupported_query_inside_transaction;

extern const SettingsFloat totals_auto_threshold;

extern const SettingsTotalsMode totals_mode;

extern const SettingsBool use_with_fill_by_sorting_prefix;

extern const SettingsFloat min_hit_rate_to_use_consecutive_keys_optimization;

extern const SettingsUInt64 max_rows_to_group_by;

extern const SettingsOverflowModeGroupBy group_by_overflow_mode;

extern const SettingsUInt64 max_bytes_before_external_group_by;

extern const SettingsDouble max_bytes_ratio_before_external_group_by;

extern const SettingsUInt64 min_free_disk_space_for_temporary_data;

extern const SettingsBool compile_aggregate_expressions;

extern const SettingsUInt64 min_count_to_compile_aggregate_expression;

extern const SettingsBool enable_software_prefetch_in_aggregation;

extern const SettingsBool optimize_group_by_constant_keys;

extern const SettingsUInt64 max_bytes_to_transfer;

extern const SettingsUInt64 max_rows_to_transfer;

extern const SettingsOverflowMode transfer_overflow_mode;

extern const SettingsBool enable_producing_buckets_out_of_order_in_aggregation;

extern const SettingsBool enable_parallel_blocks_marshalling;

extern const SettingsBool use_variant_as_common_type;

extern const SettingsBool serialize_string_in_memory_with_zero_byte;

extern const SettingsString temporary_files_codec;

extern const SettingsNonZeroUInt64 temporary_files_buffer_size;

}

namespace ServerSetting

{

extern const ServerSettingsUInt64 max_entries_for_hash_table_stats;

}

namespace ErrorCodes

{

extern const int UNSUPPORTED_METHOD;

extern const int LOGICAL_ERROR;

extern const int BAD_ARGUMENTS;

extern const int TOO_DEEP_SUBQUERIES;

extern const int NOT_IMPLEMENTED;

extern const int SUPPORT_IS_DISABLED;

extern const int INVALID_LIMIT_EXPRESSION;

}

namespace

{

/** Check that table and table function table expressions from planner context support transactions.

*

* There is precondition that table expression data for table expression nodes is collected in planner context.

*/

void checkStoragesSupportTransactions(const PlannerContextPtr & planner_context)

{

const auto & query_context = planner_context->getQueryContext();

if (!query_context->getSettingsRef()[Setting::throw_on_unsupported_query_inside_transaction])

return;

if (!query_context->getCurrentTransaction())

return;

for (const auto & [table_expression, _] : planner_context->getTableExpressionNodeToData())

{

StoragePtr storage;

if (auto * table_node = table_expression->as<TableNode>())

storage = table_node->getStorage();

else if (auto * table_function_node = table_expression->as<TableFunctionNode>())

storage = table_function_node->getStorage();

if (storage && !storage->supportsTransactions())

throw Exception(ErrorCodes::NOT_IMPLEMENTED,

"Storage {} (table {}) does not support transactions",

storage->getName(),

storage->getStorageID().getNameForLogs());

}

}

/** Storages can rely that filters that for storage will be available for analysis before

* getQueryProcessingStage method will be called.

*

* StorageDistributed skip unused shards optimization relies on this.

* Parallel replicas estimation relies on this too.

* StorageMerge common header calculation relies on this too.

*

* To collect filters that will be applied to specific table in case we have JOINs requires

* to run query plan optimization pipeline.

*

* Algorithm:

* 1. Replace all table expressions in query tree with dummy tables.

* 2. Build query plan.

* 3. Optimize query plan.

* 4. Extract filters from ReadFromDummy query plan steps from query plan leaf nodes.

*/

FiltersForTableExpressionMap collectFiltersForAnalysis(const QueryTreeNodePtr & query_tree, const QueryTreeNodes & table_nodes, const ContextPtr & query_context, const ActionsDAG * post_filter)

{

bool collect_filters = false;

const auto & settings = query_context->getSettingsRef();

bool parallel_replicas_estimation_enabled

= query_context->canUseParallelReplicasOnInitiator() && settings[Setting::parallel_replicas_min_number_of_rows_per_replica] > 0;

for (const auto & table_expression : table_nodes)

{

auto * table_node = table_expression->as<TableNode>();

auto * table_function_node = table_expression->as<TableFunctionNode>();

if (!table_node && !table_function_node)

continue;

const auto & storage = table_node ? table_node->getStorage() : table_function_node->getStorage();

if (typeid_cast<const StorageDistributed *>(storage.get())

|| (parallel_replicas_estimation_enabled && std::dynamic_pointer_cast<MergeTreeData>(storage)))

{

collect_filters = true;

break;

}

if (typeid_cast<const StorageObjectStorageCluster *>(storage.get()))

{

collect_filters = true;

break;

}

if (typeid_cast<const StorageView *>(storage.get()))

{

collect_filters = true;

break;

}

}

if (!collect_filters)

return {};

ResultReplacementMap replacement_map;

auto updated_query_tree = replaceTableExpressionsWithDummyTables(query_tree, table_nodes, query_context, &replacement_map);

/// disable parallel replicas to collect filters

{

if (auto * query_node = updated_query_tree->as<QueryNode>())

{

auto new_context = Context::createCopy(query_node->getMutableContext());

new_context->setSetting("enable_parallel_replicas", false);

query_node->getMutableContext() = new_context;

}

else if (auto * union_node = updated_query_tree->as<UnionNode>())

{

auto new_context = Context::createCopy(union_node->getMutableContext());

new_context->setSetting("enable_parallel_replicas", false);

union_node->getMutableContext() = new_context;

}

}

std::unordered_map<const IStorage *, QueryTreeNodePtr> dummy_storage_to_table;

for (auto & [from_table_expression, dummy_table_expression] : replacement_map)

{

auto * dummy_storage = dummy_table_expression->as<TableNode &>().getStorage().get();

dummy_storage_to_table.emplace(dummy_storage, from_table_expression);

}

SelectQueryOptions select_query_options;

Planner planner(updated_query_tree, select_query_options);

planner.buildQueryPlanIfNeeded();

auto & result_query_plan = planner.getQueryPlan();

/// This part is needed to support "skip unused shards" for the View over Distributed.

/// When reading through a View, the outer WHERE clause is not visible to the inner query.

/// Inject it as a FilterStep on top of the dummy plan so that predicate-push-down works.

if (post_filter)

{

const auto & header = *result_query_plan.getCurrentHeader();

bool all_inputs_present = true;

for (const auto & input : post_filter->getInputs())

{

if (!header.has(input->result_name))

{

all_inputs_present = false;

break;

}

}

if (all_inputs_present)

{

auto filter_dag = post_filter->clone();

// filter_dag.appendInputsForUnusedColumns(header);

String filter_column_name = filter_dag.getOutputs().at(0)->result_name;

auto filter_step = std::make_unique<FilterStep>(

result_query_plan.getCurrentHeader(),

std::move(filter_dag),

filter_column_name,

true /* remove_filter_column */);

result_query_plan.addStep(std::move(filter_step));

}

}

QueryPlanOptimizationSettings optimization_settings(query_context);

optimization_settings.build_sets = false; // no need to build sets to collect filters

optimization_settings.materialize_ctes = false; // no need to materialize CTEs to collect filters

result_query_plan.optimize(optimization_settings);

FiltersForTableExpressionMap res;

std::vector<QueryPlan::Node *> nodes_to_process;

nodes_to_process.push_back(result_query_plan.getRootNode());

while (!nodes_to_process.empty())

{

const auto * node_to_process = nodes_to_process.back();

nodes_to_process.pop_back();

nodes_to_process.insert(nodes_to_process.end(), node_to_process->children.begin(), node_to_process->children.end());

auto * read_from_dummy = typeid_cast<ReadFromDummy *>(node_to_process->step.get());

if (!read_from_dummy)

continue;

if (auto filter_actions = read_from_dummy->detachFilterActionsDAG())

{

const auto & table_node = dummy_storage_to_table.at(&read_from_dummy->getStorage());

res[table_node] = FiltersForTableExpression{filter_actions, read_from_dummy->getPrewhereInfo()};

}

}

return res;

}

FiltersForTableExpressionMap collectFiltersForAnalysis(const QueryTreeNodePtr & query_tree_node, const SelectQueryOptions & select_query_options, const ActionsDAG * post_filter)

{

if (select_query_options.only_analyze)

return {};

auto * query_node = query_tree_node->as<QueryNode>();

auto * union_node = query_tree_node->as<UnionNode>();

if (!query_node && !union_node)

throw Exception(ErrorCodes::UNSUPPORTED_METHOD,

"Expected QUERY or UNION node. Actual {}",

query_tree_node->formatASTForErrorMessage());

auto context = query_node ? query_node->getContext() : union_node->getContext();

auto table_expressions_nodes

= extractTableExpressions(query_tree_node, false /* add_array_join */, true /* recursive */);

return collectFiltersForAnalysis(query_tree_node, table_expressions_nodes, context, post_filter);

}

/// Extend lifetime of query context, storages, and table locks

void extendQueryContextAndStoragesLifetime(QueryPlan & query_plan, const PlannerContextPtr & planner_context)

{

query_plan.addInterpreterContext(planner_context->getQueryContext());

for (const auto & [table_expression, _] : planner_context->getTableExpressionNodeToData())

{

if (auto * table_node = table_expression->as<TableNode>())

{

query_plan.addStorageHolder(table_node->getStorage());

query_plan.addTableLock(table_node->getStorageLock());

}

else if (auto * table_function_node = table_expression->as<TableFunctionNode>())

{

query_plan.addStorageHolder(table_function_node->getStorage());

}

}

}

/// The LIMIT/OFFSET expression value can be either UInt64 or Float64, negative or positive.

std::tuple<UInt64, Float64, bool> getLimitOffsetValue(const Field & field)

{

// First check if it is nonnegative limit since they are more common

const Field converted_value_uint = convertFieldToType(field, DataTypeUInt64());

if (!converted_value_uint.isNull())

return {converted_value_uint.safeGet<UInt64>(), 0, false};

const Field converted_value_int = convertFieldToType(field, DataTypeInt64());

if (!converted_value_int.isNull())

{

Int64 int_value = converted_value_int.safeGet<Int64>();

assert(int_value < 0 && "nonnegative limit/offset values should be handled with UInt64");

const UInt64 magnitude = -static_cast<UInt64>(int_value);

return {magnitude, 0, true};

}

Field converted_value_float = convertFieldToType(field, DataTypeFloat64());

if (!converted_value_float.isNull())

return {0, converted_value_float.safeGet<Float64>(), false};

throw Exception(

ErrorCodes::INVALID_LIMIT_EXPRESSION,

"The value {} of LIMIT/OFFSET expression is not representable as UInt64 or Int64 or Float64 in the range (0, 1)",

applyVisitor(FieldVisitorToString(), field));

}

class QueryAnalysisResult

{

public:

QueryAnalysisResult(const QueryTreeNodePtr & query_tree,

const PlannerQueryProcessingInfo & query_processing_info,

const PlannerContextPtr & planner_context)

{

const auto & query_node = query_tree->as<QueryNode &>();

const auto & query_context = planner_context->getQueryContext();

const auto & settings = query_context->getSettingsRef();

aggregate_overflow_row = query_node.isGroupByWithTotals() && settings[Setting::max_rows_to_group_by]

&& settings[Setting::group_by_overflow_mode] == OverflowMode::ANY && settings[Setting::totals_mode] != TotalsMode::AFTER_HAVING_EXCLUSIVE;

aggregate_final = query_processing_info.getToStage() > QueryProcessingStage::WithMergeableState

&& !query_node.isGroupByWithTotals() && !query_node.isGroupByWithRollup() && !query_node.isGroupByWithCube();

aggregation_with_rollup_or_cube_or_grouping_sets = query_node.isGroupByWithRollup() || query_node.isGroupByWithCube() ||

query_node.isGroupByWithGroupingSets();

aggregation_should_produce_results_in_order_of_bucket_number

= query_processing_info.getToStage() == QueryProcessingStage::WithMergeableState

&& (settings[Setting::distributed_aggregation_memory_efficient] || settings[Setting::enable_memory_bound_merging_of_aggregation_results]);

query_has_array_join_in_join_tree = queryHasArrayJoinInJoinTree(query_tree);

query_has_with_totals_in_any_subquery_in_join_tree = queryHasWithTotalsInAnySubqueryInJoinTree(query_tree);

sort_description = extractSortDescription(query_node.getOrderByNode(), *planner_context);

if (query_node.hasLimit())

{

/// Constness of limit is validated during query analysis stage

std::tie(limit_length, fractional_limit, is_limit_length_negative)

= getLimitOffsetValue(query_node.getLimit()->as<ConstantNode &>().getValue());

if (query_node.hasOffset() && (limit_length || fractional_limit > 0))

{

/// Constness of offset is validated during query analysis stage

std::tie(limit_offset, fractional_offset, is_limit_offset_negative)

= getLimitOffsetValue(query_node.getOffset()->as<ConstantNode &>().getValue());

}

}

else if (query_node.hasOffset())

{

/// Constness of offset is validated during query analysis stage

std::tie(limit_offset, fractional_offset, is_limit_offset_negative)

= getLimitOffsetValue(query_node.getOffset()->as<ConstantNode &>().getValue());

}

/// Partial sort can be done if there is LIMIT, but no DISTINCT, LIMIT WITH TIES, LIMIT BY, ARRAY JOIN, NEGATIVE LIMIT, FRACTIONAL LIMIT/OFFSET

if (limit_length != 0 &&

!query_node.isDistinct() &&

!query_node.isLimitWithTies() &&

!query_node.hasLimitBy() &&

!query_has_array_join_in_join_tree &&

fractional_offset == 0 &&

fractional_limit == 0 &&

limit_length <= std::numeric_limits<UInt64>::max() - limit_offset &&

!is_limit_length_negative)

{

partial_sorting_limit = limit_length + limit_offset;

}

if (query_node.hasLimitBy())

{

bool is_limitby_limit_negative = false;

Float64 fractional_limitby_limit = 0;

std::tie(std::ignore, fractional_limitby_limit, is_limitby_limit_negative)

= getLimitOffsetValue(query_node.getLimitByLimit()->as<ConstantNode &>().getValue());

bool is_limitby_offset_negative = false;

Float64 fractional_limitby_offset = 0;

if (query_node.hasLimitByOffset())

std::tie(std::ignore, fractional_limitby_offset, is_limitby_offset_negative)

= getLimitOffsetValue(query_node.getLimitByOffset()->as<ConstantNode &>().getValue());

if (is_limitby_limit_negative || is_limitby_offset_negative)

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Negative LIMIT/OFFSET with LIMIT BY is not supported yet");

if (fractional_limitby_limit > 0 || fractional_limitby_offset > 0)

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Fractional LIMIT/OFFSET with LIMIT BY is not supported yet");

}

if (query_node.isLimitWithTies())

{

if (is_limit_length_negative)

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Negative LIMIT WITH TIES is not supported yet");

}

}

bool aggregate_overflow_row = false;

bool aggregate_final = false;

bool aggregation_with_rollup_or_cube_or_grouping_sets = false;

bool aggregation_should_produce_results_in_order_of_bucket_number = false;

bool query_has_array_join_in_join_tree = false;

bool query_has_with_totals_in_any_subquery_in_join_tree = false;

SortDescription sort_description;

UInt64 limit_length = 0;

UInt64 limit_offset = 0;

Float64 fractional_limit = 0;

Float64 fractional_offset = 0;

UInt64 partial_sorting_limit = 0;

bool is_limit_length_negative = false;

bool is_limit_offset_negative = false;

};

template <size_t size>

ALWAYS_INLINE void addExpressionStep(

const PlannerContextPtr & planner_context,

QueryPlan & query_plan,

ActionsAndProjectInputsFlagPtr & expression_actions,

const CorrelatedSubtrees & correlated_subtrees,

const SelectQueryOptions & select_query_options,

const char (&step_description)[size],

UsefulSets & useful_sets)

{

NameSet input_columns_set;

for (const auto & column : query_plan.getCurrentHeader()->getColumnsWithTypeAndName())

input_columns_set.insert(column.name);

for (const auto & correlated_subquery : correlated_subtrees.subqueries)

{

for (const auto & identifier : correlated_subquery.correlated_column_identifiers)

{

if (!input_columns_set.contains(identifier))

throw Exception(

ErrorCodes::NOT_IMPLEMENTED,

"Current query is not supported yet, because can't find correlated column '{}' in current header: {}",

identifier,

query_plan.getCurrentHeader()->dumpNames());

}

buildQueryPlanForCorrelatedSubquery(planner_context, query_plan, correlated_subquery, select_query_options);

}

auto actions = std::move(expression_actions->dag);

if (expression_actions->project_input)

actions.appendInputsForUnusedColumns(*query_plan.getCurrentHeader());

auto expression_step = std::make_unique<ExpressionStep>(query_plan.getCurrentHeader(), std::move(actions));

appendSetsFromActionsDAG(expression_step->getExpression(), useful_sets);

expression_step->setStepDescription(step_description);

query_plan.addStep(std::move(expression_step));

}

template <size_t size>

ALWAYS_INLINE void addFilterStep(

const PlannerContextPtr & planner_context,

QueryPlan & query_plan,

FilterAnalysisResult & filter_analysis_result,

const SelectQueryOptions & select_query_options,

const char (&step_description)[size],

UsefulSets & useful_sets)

{

for (const auto & correlated_subquery : filter_analysis_result.correlated_subtrees.subqueries)

{

buildQueryPlanForCorrelatedSubquery(planner_context, query_plan, correlated_subquery, select_query_options);

}

auto actions = std::move(filter_analysis_result.filter_actions->dag);

if (filter_analysis_result.filter_actions->project_input)

actions.appendInputsForUnusedColumns(*query_plan.getCurrentHeader());

auto where_step = std::make_unique<FilterStep>(query_plan.getCurrentHeader(),

std::move(actions),

filter_analysis_result.filter_column_name,

filter_analysis_result.remove_filter_column);

appendSetsFromActionsDAG(where_step->getExpression(), useful_sets);

where_step->setStepDescription(step_description);

query_plan.addStep(std::move(where_step));

}

Aggregator::Params getAggregatorParams(const PlannerContextPtr & planner_context,

const AggregationAnalysisResult & aggregation_analysis_result,

const QueryAnalysisResult & query_analysis_result,

const SelectQueryInfo & select_query_info,

bool aggregate_descriptions_remove_arguments = false)

{

const auto & query_context = planner_context->getQueryContext();

const Settings & settings = query_context->getSettingsRef();

const auto stats_collecting_params = StatsCollectingParams(

calculateCacheKey(select_query_info.query),

settings[Setting::collect_hash_table_stats_during_aggregation],

query_context->getServerSettings()[ServerSetting::max_entries_for_hash_table_stats],

settings[Setting::max_size_to_preallocate_for_aggregation]);

auto aggregate_descriptions = aggregation_analysis_result.aggregate_descriptions;

if (aggregate_descriptions_remove_arguments)

{

for (auto & aggregate_description : aggregate_descriptions)

aggregate_description.argument_names.clear();

}

auto tmp_data_scope = query_context->getTempDataOnDisk();

if (tmp_data_scope)

tmp_data_scope = tmp_data_scope->childScope(/* metrics */{}, settings[Setting::temporary_files_buffer_size], settings[Setting::temporary_files_codec]);

Aggregator::Params aggregator_params = Aggregator::Params(

aggregation_analysis_result.aggregation_keys,

aggregate_descriptions,

query_analysis_result.aggregate_overflow_row,

settings[Setting::max_rows_to_group_by],

settings[Setting::group_by_overflow_mode],

settings[Setting::group_by_two_level_threshold],

settings[Setting::group_by_two_level_threshold_bytes],

Aggregator::Params::getMaxBytesBeforeExternalGroupBy(

settings[Setting::max_bytes_before_external_group_by], settings[Setting::max_bytes_ratio_before_external_group_by]),

settings[Setting::empty_result_for_aggregation_by_empty_set]

|| (settings[Setting::empty_result_for_aggregation_by_constant_keys_on_empty_set]

&& aggregation_analysis_result.aggregation_keys.empty() && aggregation_analysis_result.group_by_with_constant_keys),

tmp_data_scope,

settings[Setting::max_threads],

settings[Setting::min_free_disk_space_for_temporary_data],

settings[Setting::compile_aggregate_expressions],

settings[Setting::min_count_to_compile_aggregate_expression],

settings[Setting::max_block_size],

settings[Setting::enable_software_prefetch_in_aggregation],

/* only_merge */ false,

settings[Setting::optimize_group_by_constant_keys],

settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization],

stats_collecting_params,

settings[Setting::enable_producing_buckets_out_of_order_in_aggregation],

settings[Setting::serialize_string_in_memory_with_zero_byte]);

return aggregator_params;

}

SortDescription getSortDescriptionFromNames(const Names & names)

{

SortDescription order_descr;

order_descr.reserve(names.size());

for (const auto & name : names)

order_descr.emplace_back(name, 1, 1);

return order_descr;

}

void addAggregationStep(QueryPlan & query_plan,

const AggregationAnalysisResult & aggregation_analysis_result,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

const SelectQueryInfo & select_query_info)

{

const Settings & settings = planner_context->getQueryContext()->getSettingsRef();

auto aggregator_params = getAggregatorParams(planner_context, aggregation_analysis_result, query_analysis_result, select_query_info);

SortDescription sort_description_for_merging;

SortDescription group_by_sort_description;

if (settings[Setting::force_aggregation_in_order])

{

group_by_sort_description = getSortDescriptionFromNames(aggregation_analysis_result.aggregation_keys);

sort_description_for_merging = group_by_sort_description;

}

auto merge_threads = settings[Setting::max_threads];

auto temporary_data_merge_threads = settings[Setting::aggregation_memory_efficient_merge_threads]

? static_cast<size_t>(settings[Setting::aggregation_memory_efficient_merge_threads])

: static_cast<size_t>(settings[Setting::max_threads]);

bool storage_has_evenly_distributed_read = false;

const auto & table_expression_node_to_data = planner_context->getTableExpressionNodeToData();

if (table_expression_node_to_data.size() == 1)

{

auto it = table_expression_node_to_data.begin();

const auto & table_expression_node = it->first;

if (const auto * table_node = table_expression_node->as<TableNode>())

storage_has_evenly_distributed_read = table_node->getStorage()->hasEvenlyDistributedRead();

else if (const auto * table_function_node = table_expression_node->as<TableFunctionNode>())

storage_has_evenly_distributed_read = table_function_node->getStorageOrThrow()->hasEvenlyDistributedRead();

}

auto aggregating_step = std::make_unique<AggregatingStep>(

query_plan.getCurrentHeader(),

aggregator_params,

aggregation_analysis_result.grouping_sets_parameters_list,

query_analysis_result.aggregate_final,

settings[Setting::max_block_size],

settings[Setting::aggregation_in_order_max_block_bytes],

merge_threads,

temporary_data_merge_threads,

storage_has_evenly_distributed_read,

settings[Setting::group_by_use_nulls],

std::move(sort_description_for_merging),

std::move(group_by_sort_description),

query_analysis_result.aggregation_should_produce_results_in_order_of_bucket_number,

settings[Setting::enable_memory_bound_merging_of_aggregation_results],

settings[Setting::force_aggregation_in_order]);

query_plan.addStep(std::move(aggregating_step));

}

void addMergingAggregatedStep(QueryPlan & query_plan,

const AggregationAnalysisResult & aggregation_analysis_result,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context)

{

const auto & query_context = planner_context->getQueryContext();

const auto & settings = query_context->getSettingsRef();

/** There are two modes of distributed aggregation.

*

* 1. In different threads read from the remote servers blocks.

* Save all the blocks in the RAM. Merge blocks.

* If the aggregation is two-level - parallelize to the number of buckets.

*

* 2. In one thread, read blocks from different servers in order.

* RAM stores only one block from each server.

* If the aggregation is a two-level aggregation, we consistently merge the blocks of each next level.

*

* The second option consumes less memory (up to 256 times less)

* in the case of two-level aggregation, which is used for large results after GROUP BY,

* but it can work more slowly.

*/

const auto & keys = aggregation_analysis_result.aggregation_keys;

/// For count() without parameters try to use just one thread

/// Typically this will either be a trivial count or a really small number of states

size_t max_threads = settings[Setting::max_threads];

if (keys.empty() && aggregation_analysis_result.aggregate_descriptions.size() == 1

&& aggregation_analysis_result.aggregate_descriptions[0].function->getName() == String{"count"}

&& aggregation_analysis_result.grouping_sets_parameters_list.empty())

max_threads = 1;

Aggregator::Params params(

keys,

aggregation_analysis_result.aggregate_descriptions,

query_analysis_result.aggregate_overflow_row,

max_threads,

settings[Setting::max_block_size],

settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization],

settings[Setting::serialize_string_in_memory_with_zero_byte]);

bool is_remote_storage = false;

bool parallel_replicas_from_merge_tree = false;

const auto & table_expression_node_to_data = planner_context->getTableExpressionNodeToData();

if (table_expression_node_to_data.size() == 1)

{

auto it = table_expression_node_to_data.begin();

is_remote_storage = it->second.isRemote();

parallel_replicas_from_merge_tree = it->second.isMergeTree() && query_context->canUseParallelReplicasOnInitiator();

}

auto merging_aggregated = std::make_unique<MergingAggregatedStep>(

query_plan.getCurrentHeader(),

params,

aggregation_analysis_result.grouping_sets_parameters_list,

query_analysis_result.aggregate_final,

/// Grouping sets don't work with distributed_aggregation_memory_efficient enabled (#43989)

settings[Setting::distributed_aggregation_memory_efficient] && (is_remote_storage || parallel_replicas_from_merge_tree)

&& !query_analysis_result.aggregation_with_rollup_or_cube_or_grouping_sets,

settings[Setting::aggregation_memory_efficient_merge_threads],

query_analysis_result.aggregation_should_produce_results_in_order_of_bucket_number,

settings[Setting::max_block_size],

settings[Setting::aggregation_in_order_max_block_bytes],

settings[Setting::enable_memory_bound_merging_of_aggregation_results]);

query_plan.addStep(std::move(merging_aggregated));

}

void addTotalsHavingStep(QueryPlan & query_plan,

PlannerExpressionsAnalysisResult & expression_analysis_result,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

const QueryNode & query_node,

UsefulSets & useful_sets)

{

const auto & query_context = planner_context->getQueryContext();

const auto & settings = query_context->getSettingsRef();

auto & aggregation_analysis_result = expression_analysis_result.getAggregation();

auto & having_analysis_result = expression_analysis_result.getHaving();

bool need_finalize = !query_node.isGroupByWithRollup() && !query_node.isGroupByWithCube();

std::optional<ActionsDAG> actions;

if (having_analysis_result.filter_actions)

{

actions = std::move(having_analysis_result.filter_actions->dag);

if (having_analysis_result.filter_actions->project_input)

actions->appendInputsForUnusedColumns(*query_plan.getCurrentHeader());

}

auto totals_having_step = std::make_unique<TotalsHavingStep>(

query_plan.getCurrentHeader(),

aggregation_analysis_result.aggregate_descriptions,

query_analysis_result.aggregate_overflow_row,

std::move(actions),

having_analysis_result.filter_column_name,

having_analysis_result.remove_filter_column,

settings[Setting::totals_mode],

settings[Setting::totals_auto_threshold],

need_finalize);

if (having_analysis_result.filter_actions)

appendSetsFromActionsDAG(*totals_having_step->getActions(), useful_sets);

query_plan.addStep(std::move(totals_having_step));

}

void addCubeOrRollupStepIfNeeded(QueryPlan & query_plan,

const AggregationAnalysisResult & aggregation_analysis_result,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

const SelectQueryInfo & select_query_info,

const QueryNode & query_node)

{

if (!query_node.isGroupByWithCube() && !query_node.isGroupByWithRollup())

return;

const auto & query_context = planner_context->getQueryContext();

const auto & settings = query_context->getSettingsRef();

auto aggregator_params = getAggregatorParams(planner_context,

aggregation_analysis_result,

query_analysis_result,

select_query_info,

true /*aggregate_descriptions_remove_arguments*/);

if (query_node.isGroupByWithRollup())

{

auto rollup_step = std::make_unique<RollupStep>(

query_plan.getCurrentHeader(), std::move(aggregator_params), true /*final*/, settings[Setting::group_by_use_nulls]);

query_plan.addStep(std::move(rollup_step));

}

else if (query_node.isGroupByWithCube())

{

auto cube_step = std::make_unique<CubeStep>(

query_plan.getCurrentHeader(), std::move(aggregator_params), true /*final*/, settings[Setting::group_by_use_nulls]);

query_plan.addStep(std::move(cube_step));

}

}

void addDistinctStep(QueryPlan & query_plan,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

const Names & column_names,

const QueryNode & query_node,

bool before_order,

bool pre_distinct)

{

const Settings & settings = planner_context->getQueryContext()->getSettingsRef();

UInt64 limit_offset = query_analysis_result.limit_offset;

UInt64 limit_length = query_analysis_result.limit_length;

UInt64 limit_hint_for_distinct = 0;

/** If after this stage of DISTINCT

* 1. ORDER BY is not executed.

* 2. There is no LIMIT BY.

* Then you can get no more than limit_length + limit_offset of different rows.

*/

if ((!query_node.hasOrderBy() || !before_order) && !query_node.hasLimitBy())

{

if (limit_length <= std::numeric_limits<UInt64>::max() - limit_offset)

limit_hint_for_distinct = limit_length + limit_offset;

}

SizeLimits limits(settings[Setting::max_rows_in_distinct], settings[Setting::max_bytes_in_distinct], settings[Setting::distinct_overflow_mode]);

auto distinct_step = std::make_unique<DistinctStep>(

query_plan.getCurrentHeader(),

limits,

limit_hint_for_distinct,

column_names,

pre_distinct);

if (pre_distinct)

distinct_step->setStepDescription("Preliminary DISTINCT");

else

distinct_step->setStepDescription("DISTINCT");

query_plan.addStep(std::move(distinct_step));

}

void addSortingStep(QueryPlan & query_plan,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context)

{

const auto & sort_description = query_analysis_result.sort_description;

const auto & query_context = planner_context->getQueryContext();

SortingStep::Settings sort_settings(query_context->getSettingsRef());

auto sorting_step = std::make_unique<SortingStep>(

query_plan.getCurrentHeader(),

sort_description,

query_analysis_result.partial_sorting_limit,

sort_settings);

sorting_step->setStepDescription("Sorting for ORDER BY");

query_plan.addStep(std::move(sorting_step));

}

template<size_t size>

ALWAYS_INLINE void addMergeSortingStep(QueryPlan & query_plan,

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

const char (&description)[size])

{

const auto & query_context = planner_context->getQueryContext();

const auto & settings = query_context->getSettingsRef();

const auto & sort_description = query_analysis_result.sort_description;

auto merging_sorted = std::make_unique<SortingStep>(

query_plan.getCurrentHeader(),

sort_description,

settings[Setting::max_block_size],

query_analysis_result.partial_sorting_limit,

settings[Setting::exact_rows_before_limit]);

merging_sorted->setStepDescription(description);

query_plan.addStep(std::move(merging_sorted));

}

void addWithFillStepIfNeeded(QueryPlan & query_plan,

const QueryAnalysisResult & query_analysis_result,

SortAnalysisResult & sort_analysis_result,

const PlannerContextPtr & planner_context,

const QueryNode & query_node,

const SelectQueryOptions & select_query_options,

UsefulSets & useful_sets)

{

NameSet column_names_with_fill;

SortDescription fill_description;

const auto & header = query_plan.getCurrentHeader();

for (const auto & description : query_analysis_result.sort_description)

{

if (description.with_fill)

{

if (!header->findByName(description.column_name))

throw Exception(ErrorCodes::LOGICAL_ERROR, "Filling column {} is not present in the block {}", description.column_name, header->dumpNames());

fill_description.push_back(description);

column_names_with_fill.insert(description.column_name);

}

}

if (fill_description.empty())

return;

InterpolateDescriptionPtr interpolate_description;

if (query_node.hasInterpolate())

{

auto & before_interpolate_actions = sort_analysis_result.before_interpolate_actions;

addExpressionStep(planner_context, query_plan, before_interpolate_actions, CorrelatedSubtrees(), select_query_options, "Before INTERPOLATE", useful_sets);

ActionsDAG interpolate_actions_dag;

auto query_plan_columns = query_plan.getCurrentHeader()->getColumnsWithTypeAndName();

for (auto & query_plan_column : query_plan_columns)

{

/// INTERPOLATE actions dag input columns must be non constant

query_plan_column.column = nullptr;

interpolate_actions_dag.addInput(query_plan_column);

}

auto & interpolate_list_node = query_node.getInterpolate()->as<ListNode &>();

auto & interpolate_list_nodes = interpolate_list_node.getNodes();

if (interpolate_list_nodes.empty())

{

for (const auto * input_node : interpolate_actions_dag.getInputs())

{

if (column_names_with_fill.contains(input_node->result_name))

continue;

interpolate_actions_dag.getOutputs().push_back(input_node);

}

}

else

{

ActionsDAG rename_dag;

for (auto & interpolate_node : interpolate_list_nodes)

{

auto & interpolate_node_typed = interpolate_node->as<InterpolateNode &>();

ColumnNodePtrWithHashSet empty_correlated_columns_set;

PlannerActionsVisitor planner_actions_visitor(planner_context, empty_correlated_columns_set);

auto [expression_to_interpolate_expression_nodes, expression_to_interpolate_correlated_subtrees] = planner_actions_visitor.visit(interpolate_actions_dag,

interpolate_node_typed.getExpression());

expression_to_interpolate_correlated_subtrees.assertEmpty("in expression to interpolate");

if (expression_to_interpolate_expression_nodes.size() != 1)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expression to interpolate expected to have single action node");

auto [interpolate_expression_nodes, interpolate_correlated_subtrees] = planner_actions_visitor.visit(interpolate_actions_dag,

interpolate_node_typed.getInterpolateExpression());

interpolate_correlated_subtrees.assertEmpty("in interpolate expression");

if (interpolate_expression_nodes.size() != 1)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Interpolate expression expected to have single action node");