Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor DAGQueryBlockInterpreter #3639

Merged
merged 9 commits into from
Dec 14, 2021
Merged

refactor DAGQueryBlockInterpreter #3639

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a23bba5
refactor
SeaRise Dec 13, 2021
90a93a2
fix
SeaRise Dec 13, 2021
9840834
Merge branch 'master' into refactor_DAGQueryBlockInterpreter
SeaRise Dec 14, 2021
9cf0047
fix
SeaRise Dec 14, 2021
ed0a10e
update
SeaRise Dec 14, 2021
810235e
Update dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp
SeaRise Dec 14, 2021
742de59
address comments
SeaRise Dec 14, 2021
b0c54f8
Merge branch 'refactor_DAGQueryBlockInterpreter' of https://github.co…
SeaRise Dec 14, 2021
2dbe165
rename
SeaRise Dec 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
192 changes: 109 additions & 83 deletions dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -911,6 +911,105 @@ void DAGQueryBlockInterpreter::executeRemoteQueryImpl(
}
}

void DAGQueryBlockInterpreter::executeExchangeReceiver(DAGPipeline & pipeline)
{
auto it = exchange_receiver_map.find(query_block.source_name);
if (unlikely(it == exchange_receiver_map.end()))
throw Exception("Can not find exchange receiver for " + query_block.source_name, ErrorCodes::LOGICAL_ERROR);
// todo choose a more reasonable stream number
for (size_t i = 0; i < max_streams; i++)
SeaRise marked this conversation as resolved.
Show resolved Hide resolved
{
BlockInputStreamPtr stream = std::make_shared<ExchangeReceiverInputStream>(it->second, taskLogger());
dagContext().getRemoteInputStreams().push_back(stream);
stream = std::make_shared<SquashingBlockInputStream>(stream, 8192, 0, taskLogger());
pipeline.streams.push_back(stream);
}
std::vector<NameAndTypePair> source_columns;
Block block = pipeline.firstStream()->getHeader();
for (const auto & col : block.getColumnsWithTypeAndName())
{
source_columns.emplace_back(NameAndTypePair(col.name, col.type));
}
analyzer = std::make_unique<DAGExpressionAnalyzer>(std::move(source_columns), context);
}

void DAGQueryBlockInterpreter::executeProjection(const tipb::Projection & projection, DAGPipeline & pipeline)
{
std::vector<NameAndTypePair> input_columns;
pipeline.streams = input_streams_vec[0];
for (auto const & p : pipeline.firstStream()->getHeader().getNamesAndTypesList())
input_columns.emplace_back(p.name, p.type);
DAGExpressionAnalyzer dag_analyzer(std::move(input_columns), context);
ExpressionActionsChain chain;
dag_analyzer.initChain(chain, dag_analyzer.getCurrentInputColumns());
ExpressionActionsChain::Step & last_step = chain.steps.back();
std::vector<NameAndTypePair> output_columns;
NamesWithAliases project_cols;
UniqueNameGenerator unique_name_generator;
for (const auto & expr : projection.exprs())
{
auto expr_name = dag_analyzer.getActions(expr, last_step.actions);
last_step.required_output.emplace_back(expr_name);
const auto & col = last_step.actions->getSampleBlock().getByName(expr_name);
String alias = unique_name_generator.toUniqueName(col.name);
output_columns.emplace_back(alias, col.type);
project_cols.emplace_back(col.name, alias);
}
pipeline.transform([&](auto & stream) { stream = std::make_shared<ExpressionBlockInputStream>(stream, chain.getLastActions(), taskLogger()); });
executeProject(pipeline, project_cols);
analyzer = std::make_unique<DAGExpressionAnalyzer>(std::move(output_columns), context);
}

void DAGQueryBlockInterpreter::executeExtraCastAndSelection(
DAGPipeline & pipeline,
const ExpressionActionsPtr & extra_cast,
const NamesWithAliases & project_after_ts_and_filter_for_remote_read,
const ExpressionActionsPtr & before_where,
const ExpressionActionsPtr & project_after_where,
const String & filter_column_name)
{
/// execute timezone cast and the selection
ExpressionActionsPtr project_for_cop_read;
for (auto & stream : pipeline.streams)
{
if (dynamic_cast<CoprocessorBlockInputStream *>(stream.get()) != nullptr)
{
/// for cop read, just execute the project is enough, because timezone cast and the selection are already done in remote TiFlash
if (!project_after_ts_and_filter_for_remote_read.empty())
{
if (project_for_cop_read == nullptr)
{
project_for_cop_read = generateProjectExpressionActions(stream, context, project_after_ts_and_filter_for_remote_read);
}
stream = std::make_shared<ExpressionBlockInputStream>(stream, project_for_cop_read, taskLogger());
}
}
else
{
/// execute timezone cast or duration cast if needed
if (extra_cast)
stream = std::make_shared<ExpressionBlockInputStream>(stream, extra_cast, taskLogger());
/// execute selection if needed
if (before_where)
{
stream = std::make_shared<FilterBlockInputStream>(stream, before_where, filter_column_name, taskLogger());
if (project_after_where)
stream = std::make_shared<ExpressionBlockInputStream>(stream, project_after_where, taskLogger());
}
}
}
for (auto & stream : pipeline.streams_with_non_joined_data)
{
/// execute selection if needed
if (before_where)
{
stream = std::make_shared<FilterBlockInputStream>(stream, before_where, filter_column_name, taskLogger());
if (project_after_where)
stream = std::make_shared<ExpressionBlockInputStream>(stream, project_after_where, taskLogger());
}
}
}

// To execute a query block, you have to:
// 1. generate the date stream and push it to pipeline.
// 2. assign the analyzer
Expand Down Expand Up @@ -940,51 +1039,12 @@ void DAGQueryBlockInterpreter::executeImpl(DAGPipeline & pipeline)
}
else if (query_block.source->tp() == tipb::ExecType::TypeExchangeReceiver)
{
auto it = exchange_receiver_map.find(query_block.source_name);
if (unlikely(it == exchange_receiver_map.end()))
throw Exception("Can not find exchange receiver for " + query_block.source_name, ErrorCodes::LOGICAL_ERROR);
// todo choose a more reasonable stream number
for (size_t i = 0; i < max_streams; i++)
{
BlockInputStreamPtr stream = std::make_shared<ExchangeReceiverInputStream>(it->second, taskLogger());
dagContext().getRemoteInputStreams().push_back(stream);
stream = std::make_shared<SquashingBlockInputStream>(stream, 8192, 0, taskLogger());
pipeline.streams.push_back(stream);
}
std::vector<NameAndTypePair> source_columns;
Block block = pipeline.firstStream()->getHeader();
for (const auto & col : block.getColumnsWithTypeAndName())
{
source_columns.emplace_back(NameAndTypePair(col.name, col.type));
}
analyzer = std::make_unique<DAGExpressionAnalyzer>(std::move(source_columns), context);
executeExchangeReceiver(pipeline);
recordProfileStreams(pipeline, query_block.source_name);
}
else if (query_block.source->tp() == tipb::ExecType::TypeProjection)
{
std::vector<NameAndTypePair> input_columns;
pipeline.streams = input_streams_vec[0];
for (auto const & p : pipeline.firstStream()->getHeader().getNamesAndTypesList())
input_columns.emplace_back(p.name, p.type);
DAGExpressionAnalyzer dag_analyzer(std::move(input_columns), context);
ExpressionActionsChain chain;
dag_analyzer.initChain(chain, dag_analyzer.getCurrentInputColumns());
ExpressionActionsChain::Step & last_step = chain.steps.back();
std::vector<NameAndTypePair> output_columns;
NamesWithAliases project_cols;
UniqueNameGenerator unique_name_generator;
for (const auto & expr : query_block.source->projection().exprs())
{
auto expr_name = dag_analyzer.getActions(expr, last_step.actions);
last_step.required_output.emplace_back(expr_name);
const auto & col = last_step.actions->getSampleBlock().getByName(expr_name);
String alias = unique_name_generator.toUniqueName(col.name);
output_columns.emplace_back(alias, col.type);
project_cols.emplace_back(col.name, alias);
}
pipeline.transform([&](auto & stream) { stream = std::make_shared<ExpressionBlockInputStream>(stream, chain.getLastActions(), taskLogger()); });
executeProject(pipeline, project_cols);
analyzer = std::make_unique<DAGExpressionAnalyzer>(std::move(output_columns), context);
executeProjection(query_block.source->projection(), pipeline);
recordProfileStreams(pipeline, query_block.source_name);
}
else
Expand All @@ -1005,55 +1065,21 @@ void DAGQueryBlockInterpreter::executeImpl(DAGPipeline & pipeline)

if (res.extra_cast || res.before_where)
{
/// execute timezone cast and the selection
ExpressionActionsPtr project_for_cop_read;
for (auto & stream : pipeline.streams)
{
if (dynamic_cast<CoprocessorBlockInputStream *>(stream.get()) != nullptr)
{
/// for cop read, just execute the project is enough, because timezone cast and the selection are already done in remote TiFlash
if (!res.project_after_ts_and_filter_for_remote_read.empty())
{
if (project_for_cop_read == nullptr)
{
project_for_cop_read = generateProjectExpressionActions(stream, context, res.project_after_ts_and_filter_for_remote_read);
}
stream = std::make_shared<ExpressionBlockInputStream>(stream, project_for_cop_read, taskLogger());
}
}
else
{
/// execute timezone cast or duration cast if needed
if (res.extra_cast)
stream = std::make_shared<ExpressionBlockInputStream>(stream, res.extra_cast, taskLogger());
/// execute selection if needed
if (res.before_where)
{
stream = std::make_shared<FilterBlockInputStream>(stream, res.before_where, res.filter_column_name, taskLogger());
if (res.project_after_where)
stream = std::make_shared<ExpressionBlockInputStream>(stream, res.project_after_where, taskLogger());
}
}
}
for (auto & stream : pipeline.streams_with_non_joined_data)
{
/// execute selection if needed
if (res.before_where)
{
stream = std::make_shared<FilterBlockInputStream>(stream, res.before_where, res.filter_column_name, taskLogger());
if (res.project_after_where)
stream = std::make_shared<ExpressionBlockInputStream>(stream, res.project_after_where, taskLogger());
}
}
executeExtraCastAndSelection(
pipeline,
res.extra_cast,
res.project_after_ts_and_filter_for_remote_read,
res.before_where,
res.project_after_where,
res.filter_column_name);
}
if (res.before_where)
{
recordProfileStreams(pipeline, query_block.selection_name);
}

// this log measures the concurrent degree in this mpp task
LOG_INFO(log,
"execution stream size for query block(before aggregation) " << query_block.qb_column_prefix << " is " << pipeline.streams.size());
LOG_INFO(log, "execution stream size for query block(before aggregation) " << query_block.qb_column_prefix << " is " << pipeline.streams.size());

dagContext().final_concurrency = std::max(dagContext().final_concurrency, pipeline.streams.size());

Expand Down
9 changes: 9 additions & 0 deletions dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,15 @@ class DAGQueryBlockInterpreter
bool is_right_out_join,
const google::protobuf::RepeatedPtrField<tipb::Expr> & filters,
String & filter_column_name);
void executeExchangeReceiver(DAGPipeline & pipeline);
void executeProjection(const tipb::Projection & projection, DAGPipeline & pipeline);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

executeProjection and executeProject confused me.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, I'd like pipeline be the first arg to align with other functions.

void executeExtraCastAndSelection(
DAGPipeline & pipeline,
const ExpressionActionsPtr & extra_cast,
const NamesWithAliases & project_after_ts_and_filter_for_remote_read,
const ExpressionActionsPtr & before_where,
const ExpressionActionsPtr & project_after_where,
const String & filter_column_name);
ExpressionActionsPtr genJoinOtherConditionAction(
const tipb::Join & join,
std::vector<NameAndTypePair> & source_columns,
Expand Down