thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-22 23:52:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update ExpressionActions constructur

Browse Source
This commit is contained in:
Nikolai Kochetov 2020-11-10 19:27:55 +03:00
parent 1db8e77371
commit c6575c9032
10 changed files with 226 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/Interpreters/ActionsVisitor.cpp
View File

@ -895,7 +895,7 @@ void ActionsMatcher::visit(const ASTFunction & node, const ASTPtr & ast, Data &
String result_name = lambda->arguments->children.at(1)->getColumnName();
lambda_dag->removeUnusedActions(Names(1, result_name));
auto lambda_actions = lambda_dag->buildExpressions();
auto lambda_actions = std::make_shared<ExpressionActions>(lambda_dag);
DataTypePtr result_type = lambda_actions->getSampleBlock().getByName(result_name).type;

380
src/Interpreters/ExpressionActions.cpp
View File

@ -53,8 +53,164 @@ namespace ErrorCodes
extern const int TYPE_MISMATCH;
}
/// Read comment near usage
/// static constexpr auto DUMMY_COLUMN_NAME = "_dummy";
ExpressionActions::~ExpressionActions() = default;
ExpressionActions::ExpressionActions(ActionsDAGPtr actions_dag_)
{
actions_dag = actions_dag_->clone();
actions_dag->compileExpressions();
linearizeActions();
const auto & settings = actions_dag->getSettings();
if (settings.max_temporary_columns && num_columns > settings.max_temporary_columns)
throw Exception(ErrorCodes::TOO_MANY_TEMPORARY_COLUMNS,
"Too many temporary columns: {}. Maximum: {}",
actions_dag->dumpNames(), std::to_string(settings.max_temporary_columns));
max_temporary_non_const_columns = settings.max_temporary_non_const_columns;
project_input = settings.project_input;
}
ExpressionActionsPtr ExpressionActions::clone() const
{
auto expressions = std::make_shared<ExpressionActions>(*this);
}
void ExpressionActions::linearizeActions()
{
struct Data
{
const Node * node = nullptr;
size_t num_created_children = 0;
std::vector<const Node *> parents;
ssize_t position = -1;
size_t num_created_parents = 0;
bool used_in_result = false;
};
const auto & nodes = getNodes();
const auto & index = actions_dag->getIndex();
std::vector<Data> data(nodes.size());
std::unordered_map<const Node *, size_t> reverse_index;
for (const auto & node : nodes)
{
size_t id = reverse_index.size();
data[id].node = &node;
reverse_index[&node] = id;
}
std::queue<const Node *> ready_nodes;
std::queue<const Node *> ready_array_joins;
for (const auto * node : index)
data[reverse_index[node]].used_in_result = true;
for (const auto & node : nodes)
{
for (const auto & child : node.children)
data[reverse_index[child]].parents.emplace_back(&node);
}
for (const auto & node : nodes)
{
if (node.children.empty())
ready_nodes.emplace(&node);
}
std::stack<size_t> free_positions;
while (!ready_nodes.empty() || !ready_array_joins.empty())
{
auto & stack = ready_nodes.empty() ? ready_array_joins : ready_nodes;
const Node * node = stack.front();
stack.pop();
Names argument_names;
for (const auto & child : node->children)
argument_names.emplace_back(child->result_name);
auto & cur = data[reverse_index[node]];
size_t free_position = num_columns;
if (free_positions.empty())
++num_columns;
else
{
free_position = free_positions.top();
free_positions.pop();
}
cur.position = free_position;
ExpressionActions::Arguments arguments;
arguments.reserve(cur.node->children.size());
for (auto * child : cur.node->children)
{
auto & arg = data[reverse_index[child]];
if (arg.position < 0)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Argument was not calculated for {}", child->result_name);
++arg.num_created_parents;
ExpressionActions::Argument argument;
argument.pos = arg.position;
argument.needed_later = arg.used_in_result || arg.num_created_parents != arg.parents.size();
if (!argument.needed_later)
free_positions.push(argument.pos);
arguments.emplace_back(argument);
}
if (node->type == ActionsDAG::ActionType::INPUT)
{
/// Argument for input is special. It contains the position from required columns.
ExpressionActions::Argument argument;
argument.pos = required_columns.size();
argument.needed_later = !cur.parents.empty();
arguments.emplace_back(argument);
required_columns.push_back({node->result_name, node->result_type});
}
actions.push_back({node, arguments, free_position});
for (const auto & parent : cur.parents)
{
auto & parent_data = data[reverse_index[parent]];
++parent_data.num_created_children;
if (parent_data.num_created_children == parent->children.size())
{
auto & push_stack = parent->type == ActionsDAG::ActionType::ARRAY_JOIN ? ready_array_joins : ready_nodes;
push_stack.push(parent);
}
}
}
result_positions.reserve(index.size());
for (const auto & node : index)
{
auto pos = data[reverse_index[node]].position;
if (pos < 0)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Action for {} was not calculated", node->result_name);
result_positions.push_back(pos);
ColumnWithTypeAndName col{node->column, node->result_type, node->result_name};
sample_block.insert(std::move(col));
}
}
static std::ostream & operator << (std::ostream & out, const ExpressionActions::Argument & argument)
{
@ -101,8 +257,6 @@ std::string ExpressionActions::Action::toString() const
return out.str();
}
ExpressionActions::~ExpressionActions() = default;
void ExpressionActions::checkLimits(ExecutionContext & execution_context) const
{
if (max_temporary_non_const_columns)
@ -371,19 +525,6 @@ std::string ExpressionActions::dumpActions() const
return ss.str();
}
//static std::string getUniqueNameForIndex(ActionsDAG::Index & index, std::string name)
//{
// if (index.contains(name))
// return name;
//
// size_t next_id = 0;
// std::string res;
// do
// res = name + "_" + std::to_string(next_id);
// while (index.contains(res));
//
// return res;
//}
bool ActionsDAG::hasArrayJoin() const
{
@ -879,13 +1020,13 @@ const ActionsDAG::Node & ActionsDAG::addFunction(
std::string result_name,
const Context & context [[maybe_unused]])
{
const auto & settings = context.getSettingsRef();
max_temporary_columns = settings.max_temporary_columns;
max_temporary_non_const_columns = settings.max_temporary_non_const_columns;
const auto & all_settings = context.getSettingsRef();
settings.max_temporary_columns = all_settings.max_temporary_columns;
settings.max_temporary_non_const_columns = all_settings.max_temporary_non_const_columns;
#if USE_EMBEDDED_COMPILER
compile_expressions = settings.compile_expressions;
min_count_to_compile_expression = settings.min_count_to_compile_expression;
settings.compile_expressions = settings.compile_expressions;
settings.min_count_to_compile_expression = settings.min_count_to_compile_expression;
if (!compilation_cache)
compilation_cache = context.getCompiledExpressionCache();
@ -925,7 +1066,8 @@ const ActionsDAG::Node & ActionsDAG::addFunction(
/// If all arguments are constants, and function is suitable to be executed in 'prepare' stage - execute function.
/// But if we compile expressions compiled version of this function maybe placed in cache,
/// so we don't want to unfold non deterministic functions
if (all_const && node.function_base->isSuitableForConstantFolding() && (!compile_expressions || node.function_base->isDeterministic()))
if (all_const && node.function_base->isSuitableForConstantFolding()
&& (!settings.compile_expressions || node.function_base->isDeterministic()))
{
size_t num_rows = arguments.empty() ? 0 : arguments.front().column->size();
auto col = node.function->execute(arguments, node.result_type, num_rows, true);
@ -1141,7 +1283,7 @@ void ActionsDAG::project(const NamesWithAliases & projection)
addAliases(projection, result_nodes);
removeUnusedActions(result_nodes);
projectInput();
projected_output = true;
settings.projected_output = true;
}
void ActionsDAG::removeColumn(const std::string & column_name)
@ -1190,197 +1332,15 @@ ActionsDAGPtr ActionsDAG::clone() const
return actions;
}
ExpressionActionsPtr ExpressionActions::clone() const
void ActionsDAG::compileExpressions()
{
auto expressions = std::make_shared<ExpressionActions>();
expressions->actions = actions;
expressions->num_columns = num_columns;
expressions->required_columns = required_columns;
expressions->result_positions = result_positions;
expressions->sample_block = sample_block;
expressions->project_input = project_input;
expressions->max_temporary_non_const_columns = max_temporary_non_const_columns;
std::unordered_map<const Node *, Node *> copy_map;
for (const auto & node : nodes)
{
auto & copy_node = expressions->nodes.emplace_back(node);
copy_map[&node] = &copy_node;
}
for (auto & node : expressions->nodes)
for (auto & child : node.children)
child = copy_map[child];
for (auto & action : expressions->actions)
action.node = copy_map[action.node];
return expressions;
}
ExpressionActionsPtr ActionsDAG::linearizeActions() const
{
struct Data
{
const Node * node = nullptr;
size_t num_created_children = 0;
std::vector<const Node *> parents;
ssize_t position = -1;
size_t num_created_parents = 0;
bool used_in_result = false;
};
std::vector<Data> data(nodes.size());
std::unordered_map<const Node *, size_t> reverse_index;
for (const auto & node : nodes)
{
size_t id = reverse_index.size();
data[id].node = &node;
reverse_index[&node] = id;
}
std::queue<const Node *> ready_nodes;
std::queue<const Node *> ready_array_joins;
for (const auto * node : index)
data[reverse_index[node]].used_in_result = true;
for (const auto & node : nodes)
{
for (const auto & child : node.children)
data[reverse_index[child]].parents.emplace_back(&node);
}
for (const auto & node : nodes)
{
if (node.children.empty())
ready_nodes.emplace(&node);
}
auto expressions = std::make_shared<ExpressionActions>();
std::stack<size_t> free_positions;
while (!ready_nodes.empty() || !ready_array_joins.empty())
{
auto & stack = ready_nodes.empty() ? ready_array_joins : ready_nodes;
const Node * node = stack.front();
stack.pop();
Names argument_names;
for (const auto & child : node->children)
argument_names.emplace_back(child->result_name);
auto & cur = data[reverse_index[node]];
size_t free_position = expressions->num_columns;
if (free_positions.empty())
++expressions->num_columns;
else
{
free_position = free_positions.top();
free_positions.pop();
}
cur.position = free_position;
ExpressionActions::Arguments arguments;
arguments.reserve(cur.node->children.size());
for (auto * child : cur.node->children)
{
auto & arg = data[reverse_index[child]];
if (arg.position < 0)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Argument was not calculated for {}", child->result_name);
++arg.num_created_parents;
ExpressionActions::Argument argument;
argument.pos = arg.position;
argument.needed_later = arg.used_in_result || arg.num_created_parents != arg.parents.size();
if (!argument.needed_later)
free_positions.push(argument.pos);
arguments.emplace_back(argument);
}
if (node->type == ActionType::INPUT)
{
/// Argument for input is special. It contains the position from required columns.
ExpressionActions::Argument argument;
argument.pos = expressions->required_columns.size();
argument.needed_later = !cur.parents.empty();
arguments.emplace_back(argument);
expressions->required_columns.push_back({node->result_name, node->result_type});
}
expressions->actions.push_back({node, arguments, free_position});
for (const auto & parent : cur.parents)
{
auto & parent_data = data[reverse_index[parent]];
++parent_data.num_created_children;
if (parent_data.num_created_children == parent->children.size())
{
auto & push_stack = parent->type == ActionType::ARRAY_JOIN ? ready_array_joins : ready_nodes;
push_stack.push(parent);
}
}
}
expressions->result_positions.reserve(index.size());
for (const auto & node : index)
{
auto pos = data[reverse_index[node]].position;
if (pos < 0)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Action for {} was not calculated", node->result_name);
expressions->result_positions.push_back(pos);
ColumnWithTypeAndName col{node->column, node->result_type, node->result_name};
expressions->sample_block.insert(std::move(col));
}
return expressions;
}
ExpressionActionsPtr ActionsDAG::buildExpressions() const
{
auto cloned = clone();
#if USE_EMBEDDED_COMPILER
if (compile_expressions)
if (settings.compile_expressions)
{
cloned->compileFunctions();
cloned->removeUnusedActions();
compileFunctions();
removeUnusedActions();
}
#endif
auto expressions = cloned->linearizeActions();
expressions->nodes.swap(cloned->nodes);
if (max_temporary_columns && expressions->num_columns > max_temporary_columns)
throw Exception(ErrorCodes::TOO_MANY_TEMPORARY_COLUMNS,
"Too many temporary columns: {}. Maximum: {}",
dumpNames(), std::to_string(max_temporary_columns));
expressions->max_temporary_non_const_columns = max_temporary_non_const_columns;
expressions->project_input = project_input;
return expressions;
}
std::string ActionsDAG::dump() const
{
return linearizeActions()->dumpActions();
}
std::string ActionsDAG::dumpDAG() const

51
src/Interpreters/ExpressionActions.h
View File

@ -175,22 +175,26 @@ public:
using Nodes = std::list<Node>;
private:
Nodes nodes;
Index index;
struct ActionsSettings
{
size_t max_temporary_columns = 0;
size_t max_temporary_non_const_columns = 0;
size_t min_count_to_compile_expression = 0;
bool compile_expressions = false;
bool project_input = false;
bool projected_output = false;
};
private:
Nodes nodes;
Index index;
ActionsSettings settings;
#if USE_EMBEDDED_COMPILER
std::shared_ptr<CompiledExpressionCache> compilation_cache;
#endif
bool project_input = false;
bool projected_output = false;
public:
ActionsDAG() = default;
ActionsDAG(const ActionsDAG &) = delete;
@ -207,7 +211,6 @@ public:
Names getNames() const;
std::string dumpNames() const;
std::string dump() const;
std::string dumpDAG() const;
const Node & addInput(std::string name, DataTypePtr type);
@ -231,9 +234,8 @@ public:
/// If column is not in index, try to find it in nodes and insert back into index.
bool tryRestoreColumn(const std::string & column_name);
void projectInput() { project_input = true; }
void projectInput() { settings.project_input = true; }
void removeUnusedActions(const Names & required_names);
ExpressionActionsPtr buildExpressions() const;
/// Splits actions into two parts. Returned half may be swapped with ARRAY JOIN.
/// Returns nullptr if no actions may be moved before ARRAY JOIN.
@ -241,7 +243,10 @@ public:
bool hasArrayJoin() const;
bool empty() const; /// If actions only contain inputs.
bool projectedOutput() const { return projected_output; } /// Remove all columns which are not in inputs from block.
const ActionsSettings & getSettings() const { return settings; }
void compileExpressions();
ActionsDAGPtr clone() const;
@ -252,10 +257,7 @@ private:
ActionsDAGPtr cloneEmpty() const
{
auto actions = std::make_shared<ActionsDAG>();
actions->max_temporary_columns = max_temporary_columns;
actions->max_temporary_non_const_columns = max_temporary_non_const_columns;
actions->min_count_to_compile_expression = min_count_to_compile_expression;
actions->compile_expressions = compile_expressions;
actions->settings = settings;
#if USE_EMBEDDED_COMPILER
actions->compilation_cache = compilation_cache;
@ -263,7 +265,6 @@ private:
return actions;
}
ExpressionActionsPtr linearizeActions() const;
void removeUnusedActions(const std::vector<Node *> & required_nodes);
void removeUnusedActions();
void addAliases(const NamesWithAliases & aliases, std::vector<Node *> & result_nodes);
@ -311,7 +312,7 @@ private:
size_t num_rows;
};
std::list<Node> nodes;
ActionsDAGPtr actions_dag;
Actions actions;
size_t num_columns;
@ -328,12 +329,13 @@ private:
public:
~ExpressionActions();
ExpressionActions() = default;
ExpressionActions(const ExpressionActions &) = delete;
ExpressionActions & operator=(const ExpressionActions &) = delete;
explicit ExpressionActions(ActionsDAGPtr actions_dag_);
ExpressionActions(const ExpressionActions &) = default;
ExpressionActions & operator=(const ExpressionActions &) = default;
const Actions & getActions() const { return actions; }
const std::list<Node> & getNodes() const { return nodes; }
const std::list<Node> & getNodes() const { return actions_dag->getNodes(); }
const ActionsDAG & getActionsDAG() const { return *actions_dag; }
/// Adds to the beginning the removal of all extra columns.
void projectInput() { project_input = true; }
@ -363,10 +365,13 @@ public:
ExpressionActionsPtr clone() const;
private:
ExpressionActions() = default;
void checkLimits(ExecutionContext & execution_context) const;
static void executeAction(const Action & action, ExecutionContext & execution_context, bool dry_run);
void linearizeActions();
};
@ -433,7 +438,7 @@ struct ExpressionActionsChain
void finalize(const Names & required_output_) override
{
if (!actions->projectedOutput())
if (!actions->getSettings().projected_output)
actions->removeUnusedActions(required_output_);
}
@ -444,7 +449,7 @@ struct ExpressionActionsChain
std::string dump() const override
{
return actions->dump();
return actions->dumpDAG();
}
};

14
src/Interpreters/ExpressionAnalyzer.cpp
View File

@ -673,7 +673,7 @@ ActionsDAGPtr SelectQueryExpressionAnalyzer::appendPrewhere(
auto tmp_actions_dag = std::make_shared<ActionsDAG>(sourceColumns());
getRootActions(select_query->prewhere(), only_types, tmp_actions_dag);
tmp_actions_dag->removeUnusedActions({prewhere_column_name});
auto tmp_actions = tmp_actions_dag->buildExpressions();
auto tmp_actions = std::make_shared<ExpressionActions>(tmp_actions_dag);
auto required_columns = tmp_actions->getRequiredColumns();
NameSet required_source_columns(required_columns.begin(), required_columns.end());
@ -794,7 +794,7 @@ bool SelectQueryExpressionAnalyzer::appendGroupBy(ExpressionActionsChain & chain
{
auto actions_dag = std::make_shared<ActionsDAG>(columns_after_join);
getRootActions(child, only_types, actions_dag);
group_by_elements_actions.emplace_back(actions_dag->buildExpressions());
group_by_elements_actions.emplace_back(std::make_shared<ExpressionActions>(actions_dag));
}
}
@ -892,7 +892,7 @@ ActionsDAGPtr SelectQueryExpressionAnalyzer::appendOrderBy(ExpressionActionsChai
{
auto actions_dag = std::make_shared<ActionsDAG>(columns_after_join);
getRootActions(child, only_types, actions_dag);
order_by_elements_actions.emplace_back(actions_dag->buildExpressions());
order_by_elements_actions.emplace_back(std::make_shared<ExpressionActions>(actions_dag));
}
}
@ -1044,7 +1044,7 @@ ActionsDAGPtr ExpressionAnalyzer::getActionsDAG(bool add_aliases, bool project_r
ExpressionActionsPtr ExpressionAnalyzer::getActions(bool add_aliases, bool project_result)
{
return getActionsDAG(add_aliases, project_result)->buildExpressions();
return std::make_shared<ExpressionActions>(getActionsDAG(add_aliases, project_result));
}
@ -1053,7 +1053,7 @@ ExpressionActionsPtr ExpressionAnalyzer::getConstActions()
auto actions = std::make_shared<ActionsDAG>(NamesAndTypesList());
getRootActions(query, true, actions, true);
return actions->buildExpressions();
return std::make_shared<ExpressionActions>(actions);
}
ActionsDAGPtr SelectQueryExpressionAnalyzer::simpleSelectActions()
@ -1143,7 +1143,7 @@ ExpressionAnalysisResult::ExpressionAnalysisResult(
Block before_prewhere_sample = source_header;
if (sanitizeBlock(before_prewhere_sample))
{
prewhere_info->prewhere_actions->buildExpressions()->execute(before_prewhere_sample);
ExpressionActions(prewhere_info->prewhere_actions).execute(before_prewhere_sample);
auto & column_elem = before_prewhere_sample.getByName(query.prewhere()->getColumnName());
/// If the filter column is a constant, record it.
if (column_elem.column)
@ -1176,7 +1176,7 @@ ExpressionAnalysisResult::ExpressionAnalysisResult(
before_where_sample = source_header;
if (sanitizeBlock(before_where_sample))
{
before_where->buildExpressions()->execute(before_where_sample);
ExpressionActions(before_where).execute(before_where_sample);
auto & column_elem = before_where_sample.getByName(query.where()->getColumnName());
/// If the filter column is a constant, record it.
if (column_elem.column)

8
src/Interpreters/InterpreterSelectQuery.cpp
View File

@ -522,7 +522,7 @@ Block InterpreterSelectQuery::getSampleBlockImpl()
if (analysis_result.prewhere_info)
{
analysis_result.prewhere_info->prewhere_actions->buildExpressions()->execute(header);
ExpressionActions(analysis_result.prewhere_info->prewhere_actions).execute(header);
header = materializeBlock(header);
if (analysis_result.prewhere_info->remove_prewhere_column)
header.erase(analysis_result.prewhere_info->prewhere_column_name);
@ -1435,13 +1435,13 @@ void InterpreterSelectQuery::executeFetchColumns(
if (prewhere_info)
{
query_info.prewhere_info = std::make_shared<PrewhereInfo>(
prewhere_info->prewhere_actions->buildExpressions(),
std::make_shared<ExpressionActions>(prewhere_info->prewhere_actions),
prewhere_info->prewhere_column_name);
if (prewhere_info->alias_actions)
query_info.prewhere_info->alias_actions = prewhere_info->alias_actions->buildExpressions();
query_info.prewhere_info->alias_actions = std::make_shared<ExpressionActions>(prewhere_info->alias_actions);
if (prewhere_info->remove_columns_actions)
query_info.prewhere_info->remove_columns_actions = prewhere_info->remove_columns_actions->buildExpressions();
query_info.prewhere_info->remove_columns_actions = std::make_shared<ExpressionActions>(prewhere_info->remove_columns_actions);
query_info.prewhere_info->remove_prewhere_column = prewhere_info->remove_prewhere_column;
query_info.prewhere_info->need_filter = prewhere_info->need_filter;

8
src/Processors/QueryPlan/ExpressionStep.cpp
View File

@ -44,7 +44,7 @@ static ITransformingStep::Traits getJoinTraits()
ExpressionStep::ExpressionStep(const DataStream & input_stream_, ActionsDAGPtr actions_)
: ITransformingStep(
input_stream_,
Transform::transformHeader(input_stream_.header, actions_->buildExpressions()),
Transform::transformHeader(input_stream_.header, std::make_shared<ExpressionActions>(actions_)),
getTraits(actions_))
, actions(std::move(actions_))
{
@ -55,7 +55,7 @@ ExpressionStep::ExpressionStep(const DataStream & input_stream_, ActionsDAGPtr a
void ExpressionStep::updateInputStream(DataStream input_stream, bool keep_header)
{
Block out_header = keep_header ? std::move(output_stream->header)
: Transform::transformHeader(input_stream.header, actions->buildExpressions());
: Transform::transformHeader(input_stream.header, std::make_shared<ExpressionActions>(actions));
output_stream = createOutputStream(
input_stream,
std::move(out_header),
@ -67,7 +67,7 @@ void ExpressionStep::updateInputStream(DataStream input_stream, bool keep_header
void ExpressionStep::transformPipeline(QueryPipeline & pipeline)
{
auto expression = actions->buildExpressions();
auto expression = std::make_shared<ExpressionActions>(actions);
pipeline.addSimpleTransform([&](const Block & header)
{
return std::make_shared<Transform>(header, expression);
@ -88,7 +88,7 @@ void ExpressionStep::describeActions(FormatSettings & settings) const
String prefix(settings.offset, ' ');
bool first = true;
auto expression = actions->buildExpressions();
auto expression = std::make_shared<ExpressionActions>(actions);
for (const auto & action : expression->getActions())
{
settings.out << prefix << (first ? "Actions: "

8
src/Processors/QueryPlan/FilterStep.cpp
View File

@ -31,7 +31,7 @@ FilterStep::FilterStep(
bool remove_filter_column_)
: ITransformingStep(
input_stream_,
FilterTransform::transformHeader(input_stream_.header, actions_->buildExpressions(), filter_column_name_, remove_filter_column_),
FilterTransform::transformHeader(input_stream_.header, std::make_shared<ExpressionActions>(actions_), filter_column_name_, remove_filter_column_),
getTraits(actions_))
, actions(std::move(actions_))
, filter_column_name(std::move(filter_column_name_))
@ -45,7 +45,7 @@ void FilterStep::updateInputStream(DataStream input_stream, bool keep_header)
{
Block out_header = std::move(output_stream->header);
if (keep_header)
out_header = FilterTransform::transformHeader(input_stream.header, actions->buildExpressions(), filter_column_name, remove_filter_column);
out_header = FilterTransform::transformHeader(input_stream.header, std::make_shared<ExpressionActions>(actions), filter_column_name, remove_filter_column);
output_stream = createOutputStream(
input_stream,
@ -58,7 +58,7 @@ void FilterStep::updateInputStream(DataStream input_stream, bool keep_header)
void FilterStep::transformPipeline(QueryPipeline & pipeline)
{
auto expression = actions->buildExpressions();
auto expression = std::make_shared<ExpressionActions>(actions);
pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type)
{
bool on_totals = stream_type == QueryPipeline::StreamType::Totals;
@ -80,7 +80,7 @@ void FilterStep::describeActions(FormatSettings & settings) const
settings.out << prefix << "Filter column: " << filter_column_name << '\n';
bool first = true;
auto expression = actions->buildExpressions();
auto expression = std::make_shared<ExpressionActions>(actions);
for (const auto & action : expression->getActions())
{
settings.out << prefix << (first ? "Actions: "

9
src/Processors/QueryPlan/TotalsHavingStep.cpp
View File

@ -34,7 +34,10 @@ TotalsHavingStep::TotalsHavingStep(
bool final_)
: ITransformingStep(
input_stream_,
TotalsHavingTransform::transformHeader(input_stream_.header, (actions_ ? actions_->buildExpressions() : nullptr), final_),
TotalsHavingTransform::transformHeader(
input_stream_.header,
(actions_ ? std::make_shared<ExpressionActions>(actions_) : nullptr),
final_),
getTraits(!filter_column_.empty()))
, overflow_row(overflow_row_)
, actions(actions_)
@ -48,7 +51,7 @@ TotalsHavingStep::TotalsHavingStep(
void TotalsHavingStep::transformPipeline(QueryPipeline & pipeline)
{
auto totals_having = std::make_shared<TotalsHavingTransform>(
pipeline.getHeader(), overflow_row, (actions ? actions->buildExpressions() : nullptr),
pipeline.getHeader(), overflow_row, (actions ? std::make_shared<ExpressionActions>(actions) : nullptr),
filter_column_name, totals_mode, auto_include_threshold, final);
pipeline.addTotalsHavingTransform(std::move(totals_having));
@ -78,7 +81,7 @@ void TotalsHavingStep::describeActions(FormatSettings & settings) const
settings.out << prefix << "Mode: " << totalsModeToString(totals_mode, auto_include_threshold) << '\n';
bool first = true;
auto expression = actions->buildExpressions();
auto expression = std::make_shared<ExpressionActions>(actions);
for (const auto & action : expression->getActions())
{
settings.out << prefix << (first ? "Actions: "

2
src/Storages/MergeTree/MergeTreeData.cpp
View File

@ -438,7 +438,7 @@ void MergeTreeData::checkPartitionKeyAndInitMinMax(const KeyDescription & new_pa
/// Add all columns used in the partition key to the min-max index.
const NamesAndTypesList & minmax_idx_columns_with_types = new_partition_key.expression->getRequiredColumnsWithTypes();
minmax_idx_expr = std::make_shared<ActionsDAG>(minmax_idx_columns_with_types)->buildExpressions();
minmax_idx_expr = std::make_shared<ExpressionActions>(std::make_shared<ActionsDAG>(minmax_idx_columns_with_types));
for (const NameAndTypePair & column : minmax_idx_columns_with_types)
{
minmax_idx_columns.emplace_back(column.name);

2
src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
View File

@ -814,7 +814,7 @@ Pipe MergeTreeDataSelectExecutor::readFromParts(
if (result_projection)
{
auto result_projection_actions = result_projection->buildExpressions();
auto result_projection_actions = std::make_shared<ExpressionActions>(result_projection);
res.addSimpleTransform([&result_projection_actions](const Block & header)
{
return std::make_shared<ExpressionTransform>(header, result_projection_actions);