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

show the correct number of threads of agg in explain analyze #5367

Closed
wants to merge 3 commits into from
Closed

show the correct number of threads of agg in explain analyze #5367

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
49cc00d
fix
SeaRise Jul 14, 2022
5a6d4df
fix
SeaRise Jul 14, 2022
6c62793
Update dbms/src/Flash/Statistics/ExecutorStatistics.h
SeaRise Jul 19, 2022
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
13 changes: 12 additions & 1 deletion dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -409,7 +409,7 @@ void DAGQueryBlockInterpreter::executeAggregation(
pipeline.firstStream() = std::move(stream);

// should record for agg before restore concurrency. See #3804.
recordProfileStreams(pipeline, query_block.aggregation_name);
recordProfileStream(pipeline.firstStream(), max_streams, query_block.aggregation_name);
restorePipelineConcurrency(pipeline);
}
else
Expand Down Expand Up @@ -451,6 +451,17 @@ void DAGQueryBlockInterpreter::recordProfileStreams(DAGPipeline & pipeline, cons
pipeline.transform([&profile_streams](auto & stream) { profile_streams.push_back(stream); });
}

void DAGQueryBlockInterpreter::recordProfileStream(const BlockInputStreamPtr & stream, size_t concurrency, const String & key)
{
assert(stream);
auto & profile_streams = dagContext().getProfileStreamsMap()[key];
profile_streams.push_back(stream);
const auto & header = stream->getHeader();
/// We consider the number of streams as the concurrency of the operator and pass it to tidb.
for (size_t i = 1; i < concurrency; ++i)
profile_streams.emplace_back(std::make_shared<NullBlockInputStream>(header));
Comment on lines +460 to +462
Copy link
Contributor

@JaySon-Huang JaySon-Huang Jul 14, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Do we fill the concurrency to the response in here?

tiflash/dbms/src/Flash/Coprocessor/DAGResponseWriter.cpp

Lines 94 to 103 in f97567a

/// part 1: local execution info
for (const auto & stream_ptr : streams)
{
if (auto * p_stream = dynamic_cast<IProfilingBlockInputStream *>(stream_ptr.get()))
{
current.time_processed_ns = std::max(current.time_processed_ns, p_stream->getProfileInfo().execution_time);
current.num_produced_rows += p_stream->getProfileInfo().rows;
current.num_iterations += p_stream->getProfileInfo().blocks;
}
current.concurrency++;

I don't get it that why using NullBlockInputStream here won't affect the number of actual rows, actual process time, etc.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

because NullBlockInputStream is not IProfilingBlockInputStream..

}

void DAGQueryBlockInterpreter::handleExchangeReceiver(DAGPipeline & pipeline)
{
auto exchange_receiver = dagContext().getMPPExchangeReceiver(query_block.source_name);
Expand Down
2 changes: 2 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 @@ -87,6 +87,8 @@ class DAGQueryBlockInterpreter

void recordProfileStreams(DAGPipeline & pipeline, const String & key);

void recordProfileStream(const BlockInputStreamPtr & stream, size_t concurrency, const String & key);

void recordJoinExecuteInfo(size_t build_side_index, const JoinPtr & join_ptr);

void restorePipelineConcurrency(DAGPipeline & pipeline);
Expand Down
9 changes: 5 additions & 4 deletions dbms/src/Flash/Statistics/ExecutorStatistics.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,10 +84,11 @@ class ExecutorStatistics : public ExecutorStatisticsBase
{
for (const auto & input_stream : it->second)
{
auto * p_stream = dynamic_cast<IProfilingBlockInputStream *>(input_stream.get());
assert(p_stream);
const auto & profile_info = p_stream->getProfileInfo();
base.append(profile_info);
if (auto * p_stream = dynamic_cast<IProfilingBlockInputStream *>(input_stream.get()); p_stream)
{
const auto & profile_info = p_stream->getProfileInfo();
base.append(profile_info);
}
}
}
if constexpr (ExecutorImpl::has_extra_info)
Expand Down