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

dbms: fixing error [#METR-19767].

Browse Source
This commit is contained in:
Alexey Milovidov 2016-01-26 00:40:13 +03:00
parent 525f961110
commit 869a521aed
4 changed files with 58 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dbms/include/DB/DataStreams/BlockStreamProfileInfo.h
View File

@ -42,10 +42,13 @@ struct BlockStreamProfileInfo
void update(Block & block);
/// Методы для бинарной [де]сериализации
/// Методы для бинарной [де]сериализации. Передаются не все поля.
void read(ReadBuffer & in);
void write(WriteBuffer & out) const;
/// Установить поля из другого объекта, но только те, которые передаются по сети (методами выше).
void setFrom(const BlockStreamProfileInfo & rhs);
private:
void calculateRowsBeforeLimit() const;

61
dbms/src/DataStreams/BlockStreamProfileInfo.cpp
View File

@ -31,6 +31,17 @@ void BlockStreamProfileInfo::write(WriteBuffer & out) const
}
void BlockStreamProfileInfo::setFrom(const BlockStreamProfileInfo & rhs)
{
rows = rhs.rows;
blocks = rhs.blocks;
bytes = rhs.bytes;
applied_limit = rhs.applied_limit;
rows_before_limit = rhs.rows_before_limit;
calculated_rows_before_limit = rhs.calculated_rows_before_limit;
}
size_t BlockStreamProfileInfo::getRowsBeforeLimit() const
{
if (!calculated_rows_before_limit)
@ -75,22 +86,48 @@ void BlockStreamProfileInfo::calculateRowsBeforeLimit() const
/// есть ли Limit?
BlockStreamProfileInfos limits;
collectInfosForStreamsWithName("Limit", limits);
if (limits.empty())
return;
applied_limit = true;
if (!limits.empty())
{
applied_limit = true;
/** Берём количество строчек, прочитанных ниже PartialSorting-а, если есть, или ниже Limit-а.
* Это нужно, потому что сортировка может вернуть только часть строк.
*/
BlockStreamProfileInfos partial_sortings;
collectInfosForStreamsWithName("PartialSorting", partial_sortings);
/** Берём количество строчек, прочитанных ниже PartialSorting-а, если есть, или ниже Limit-а.
* Это нужно, потому что сортировка может вернуть только часть строк.
*/
BlockStreamProfileInfos partial_sortings;
collectInfosForStreamsWithName("PartialSorting", partial_sortings);
BlockStreamProfileInfos & limits_or_sortings = partial_sortings.empty() ? limits : partial_sortings;
BlockStreamProfileInfos & limits_or_sortings = partial_sortings.empty() ? limits : partial_sortings;
for (const auto & info_limit_or_sort : limits_or_sortings)
for (const auto & nested_info : info_limit_or_sort->nested_infos)
rows_before_limit += nested_info->rows;
for (const auto & info_limit_or_sort : limits_or_sortings)
for (const auto & nested_info : info_limit_or_sort->nested_infos)
rows_before_limit += nested_info->rows;
}
else
{
/// Тогда данные о rows_before_limit могут быть в RemoteBlockInputStream-е (приехать с удалённого сервера).
std::cerr << "Has no Limit\n";
BlockStreamProfileInfos remotes;
collectInfosForStreamsWithName("Remote", remotes);
if (remotes.empty())
return;
std::cerr << "Found Remote\n";
for (const auto & info : remotes)
{
if (info->applied_limit)
{
std::cerr << "Has info->applied_limit\n";
applied_limit = true;
rows_before_limit += info->rows_before_limit;
}
}
}
}
}

3
dbms/src/DataStreams/RemoteBlockInputStream.cpp
View File

@ -175,7 +175,8 @@ Block RemoteBlockInputStream::readImpl()
break;
case Protocol::Server::ProfileInfo:
info = packet.profile_info;
info.setFrom(packet.profile_info);
std::cerr << "received, has_applied_limit: " << info.hasAppliedLimit() << ", rows_before_limit: " << info.getRowsBeforeLimit() << "\n";
break;
case Protocol::Server::Totals:

3
dbms/src/Server/TCPHandler.cpp
View File

@ -372,6 +372,9 @@ void TCPHandler::sendProfileInfo()
if (const IProfilingBlockInputStream * input = dynamic_cast<const IProfilingBlockInputStream *>(&*state.io.in))
{
writeVarUInt(Protocol::Server::ProfileInfo, *out);
std::cerr << "Sending profile info, rows_before_limit: " << input->getInfo().getRowsBeforeLimit() << "\n";
input->getInfo().write(*out);
out->next();
}