{

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

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

}

}

{

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

limit_length = query_node.getLimit()->as<ConstantNode &>().getValue().safeGet<UInt64>();

if (query_node.hasOffset() && limit_length)

{

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

limit_offset = query_node.getOffset()->as<ConstantNode &>().getValue().safeGet<UInt64>();

}

}

else if (query_node.hasOffset())

{

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

limit_offset = query_node.getOffset()->as<ConstantNode &>().getValue().safeGet<UInt64>();

}

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

if (limit_length != 0 &&

!query_node.isDistinct() &&

!query_node.isLimitWithTies() &&

!query_node.hasLimitBy() &&

!query_has_array_join_in_join_tree &&

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

{

partial_sorting_limit = limit_length + limit_offset;

}

}

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;

UInt64 partial_sorting_limit = 0;

const PlannerContextPtr & planner_context,

QueryPlan & query_plan,

ActionsAndProjectInputsFlagPtr & expression_actions,

const CorrelatedSubtrees & correlated_subtrees,

const SelectQueryOptions & select_query_options,

const std::string & step_description,

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 std::string & step_description,

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

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

}

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

query_context->getTempDataOnDisk(),

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

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

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

distinct_step->setStepDescription(pre_distinct ? "Preliminary DISTINCT" : "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 std::string & description)

{

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("Merge sorted streams " + description);

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

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

const QueryNode & query_node)

{

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

{

ActionsDAG interpolate_actions_dag;

auto query_plan_columns = header->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 & expression_to_interpolate_name = expression_to_interpolate->result_name;

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

}

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

interpolate_actions_dag.getOutputs().push_back(alias_node);

/// Here we fix INTERPOLATE by constant expression.

/// Example from 02336_sort_optimization_with_fill:

///

/// SELECT 5 AS x, 'Hello' AS s ORDER BY x WITH FILL FROM 1 TO 10 INTERPOLATE (s AS s||'A')

///

/// For this query, INTERPOLATE_EXPRESSION would be : s AS concat(s, 'A'),

/// so that interpolate_actions_dag would have INPUT `s`.

///

/// However, INPUT `s` does not exist. Instead, we have a constant with execution name 'Hello'_String.

/// To fix this, we prepend a rename : 'Hello'_String -> s

if (const auto * /*constant_node*/ _ = interpolate_node_typed.getExpression()->as<const ConstantNode>())

{

const auto & name = interpolate_node_typed.getExpressionName();

const auto * node = &rename_dag.addInput(alias_node->result_name, alias_node->result_type);

node = &rename_dag.addAlias(*node, name);

rename_dag.getOutputs().push_back(node);

/// Interpolate DAG should contain INPUT with same name to ensure a proper merging

const auto & inputs = interpolate_actions_dag.getInputs();

if (std::ranges::find_if(inputs, [&name](const auto & input){ return input->result_name == name; }) == inputs.end())

interpolate_actions_dag.addInput(name, interpolate_node_typed.getExpression()->getResultType());

}

}

if (!rename_dag.getOutputs().empty())

interpolate_actions_dag = ActionsDAG::merge(std::move(rename_dag), std::move(interpolate_actions_dag));

interpolate_actions_dag.removeUnusedActions();

}

Aliases empty_aliases;

interpolate_description = std::make_shared<InterpolateDescription>(std::move(interpolate_actions_dag), empty_aliases);

}

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

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

auto filling_step = std::make_unique<FillingStep>(

header,

query_analysis_result.sort_description,

std::move(fill_description),

interpolate_description,

settings[Setting::use_with_fill_by_sorting_prefix]);

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

QueryPlan & query_plan, const LimitByAnalysisResult & limit_by_analysis_result, const QueryNode & query_node, bool do_not_skip_offset)

{

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

UInt64 limit_by_limit = query_node.getLimitByLimit()->as<ConstantNode &>().getValue().safeGet<UInt64>();

UInt64 limit_by_offset = 0;

if (query_node.hasLimitByOffset())

{

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

limit_by_offset = query_node.getLimitByOffset()->as<ConstantNode &>().getValue().safeGet<UInt64>();

}

if (do_not_skip_offset)

{

if (limit_by_limit > std::numeric_limits<UInt64>::max() - limit_by_offset)

return;

limit_by_limit += limit_by_offset;

limit_by_offset = 0;

}

auto limit_by_step = std::make_unique<LimitByStep>(query_plan.getCurrentHeader(),

limit_by_limit,

limit_by_offset,

limit_by_analysis_result.limit_by_column_names);

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

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr & planner_context,

bool do_not_skip_offset)

{

UInt64 limit_offset = query_analysis_result.limit_offset;

UInt64 limit_length = query_analysis_result.limit_length;

if (do_not_skip_offset)

{

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

return;

limit_length += limit_offset;

limit_offset = 0;

}

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

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

auto limit

= std::make_unique<LimitStep>(query_plan.getCurrentHeader(), limit_length, limit_offset, settings[Setting::exact_rows_before_limit]);

limit->setStepDescription(do_not_skip_offset ? "preliminary LIMIT (with OFFSET)" : "preliminary LIMIT (without OFFSET)");

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

const QueryAnalysisResult & query_analysis_result,

const PlannerContextPtr planner_context,

const PlannerQueryProcessingInfo & query_processing_info,

const QueryTreeNodePtr & query_tree)

{

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

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

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

const auto & sort_description = query_analysis_result.sort_description;

bool has_withfill = false;

for (const auto & desc : sort_description)

{

if (desc.with_fill)

{

has_withfill = true;

break;

}

}

bool apply_limit = query_processing_info.getToStage() != QueryProcessingStage::WithMergeableStateAfterAggregation;