{

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());

}

{

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 (!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();

QueryPlanOptimizationSettings optimization_settings(query_context);

optimization_settings.build_sets = false; // no need to build sets 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;

{

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);

{

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());

}

}

{

// 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");

// We need to be careful because -Int64::min() is not representable as Int64

const UInt64 magnitude = (int_value == std::numeric_limits<Int64>::min())

? (static_cast<UInt64>(std::numeric_limits<Int64>::max()) + 1ULL)

: 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));

{

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;

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));

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));

QueryPlan & query_plan,

FilterAnalysisResult & filter_analysis_result,

const char (&step_description)[size])

{

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

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

std::move(actions),

filter_analysis_result.filter_column_name);

where_step->setStepDescription(step_description);

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

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 order_descr;

order_descr.reserve(names.size());

for (const auto & name : names)

order_descr.emplace_back(name, 1, 1);

return order_descr;

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));

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.

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));

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));

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));

}

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

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));

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));

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));

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");

const auto * expression_to_interpolate = expression_to_interpolate_expression_nodes[0];

const auto * interpolate_expression = interpolate_expression_nodes[0];

if (!interpolate_expression->result_type->equals(*expression_to_interpolate->result_type))

{

interpolate_expression = &interpolate_actions_dag.addCast(*interpolate_expression,

expression_to_interpolate->result_type,

interpolate_expression->result_name,

planner_context->getQueryContext());

}

const auto * alias_node = &interpolate_actions_dag.addAlias(*interpolate_expression, interpolate_node_typed.getExpressionName());

interpolate_actions_dag.getOutputs().push_back(alias_node);

}