thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-12-04 05:22:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge

Browse Source
This commit is contained in:
Alexey Milovidov 2016-04-28 22:44:43 +03:00
parent f3ad486e14
commit 2b1550105c
1 changed files with 10 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
dbms/src/DataStreams/GraphiteRollupSortedBlockInputStream.cpp
View File

@ -122,6 +122,7 @@ void GraphiteRollupSortedBlockInputStream::merge(ColumnPlainPtrs & merged_column
auto prev_pattern = current_pattern; auto prev_pattern = current_pattern;
bool path_differs = is_first || next_path != current_path; bool path_differs = is_first || next_path != current_path;
is_first = false;
if (path_differs) if (path_differs)
current_pattern = selectPatternForPath(next_path); current_pattern = selectPatternForPath(next_path);
@ -135,23 +136,17 @@ void GraphiteRollupSortedBlockInputStream::merge(ColumnPlainPtrs & merged_column
bool is_new_key = path_differs || next_time != current_time; bool is_new_key = path_differs || next_time != current_time;
/// если накопилось достаточно строк и последняя посчитана полностью
if (is_new_key && merged_rows >= max_block_size)
{
finishCurrentRow(merged_columns);
return;
}
queue.pop();
bool was_first = is_first;
is_first = false;
if (is_new_key) if (is_new_key)
{ {
if (!was_first) if (merged_rows)
{
finishCurrentRow(merged_columns); finishCurrentRow(merged_columns);
/// если накопилось достаточно строк
if (merged_rows >= max_block_size)
return;
}
startNextRow(merged_columns, current); startNextRow(merged_columns, current);
owned_current_block = source_blocks[current.impl->order]; owned_current_block = source_blocks[current.impl->order];
@ -170,6 +165,8 @@ void GraphiteRollupSortedBlockInputStream::merge(ColumnPlainPtrs & merged_column
accumulateRow(current); accumulateRow(current);
current_max_version = std::max(current_max_version, current->all_columns[version_column_num]->get64(current->pos)); current_max_version = std::max(current_max_version, current->all_columns[version_column_num]->get64(current->pos));
queue.pop();
if (!current->isLast()) if (!current->isLast())
{ {
current->next(); current->next();