thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-21 17:20:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

dbms: fixed error [#METR-19700].

Browse Source
This commit is contained in:
Alexey Milovidov 2015-12-06 18:29:16 +03:00
parent 0ef797d9f3
commit 071b1e367c
3 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dbms/include/DB/DataStreams/ParallelAggregatingBlockInputStream.h
View File

@ -209,10 +209,14 @@ private:
void onFinishThread(size_t thread_num)
{
if (parent.aggregator.hasTemporaryFiles())
if (!parent.isCancelled() && parent.aggregator.hasTemporaryFiles())
{
/// Сбросим имеющиеся в оперативке данные тоже на диск. Так проще их потом объединять.
auto & data = *parent.many_data[thread_num];
if (data.isConvertibleToTwoLevel())
data.convertToTwoLevel();
size_t rows = data.sizeWithoutOverflowRow();
if (rows)
parent.aggregator.writeToTemporaryFile(data, rows);
@ -221,12 +225,15 @@ private:
void onFinish()
{
if (parent.aggregator.hasTemporaryFiles())
if (!parent.isCancelled() && parent.aggregator.hasTemporaryFiles())
{
/// Может так получиться, что какие-то данные ещё не сброшены на диск,
/// потому что во время вызова onFinishThread ещё никакие данные не были сброшены на диск, а потом какие-то - были.
for (auto & data : parent.many_data)
{
if (data->isConvertibleToTwoLevel())
data->convertToTwoLevel();
size_t rows = data->sizeWithoutOverflowRow();
if (rows)
parent.aggregator.writeToTemporaryFile(*data, rows);

2
dbms/include/DB/DataStreams/ParallelInputsProcessor.h
View File

@ -211,7 +211,7 @@ private:
}
}
handler.onFinish();
handler.onFinish(); /// TODO Если в onFinish или onFinishThread эксепшен, то вызывается std::terminate.
}
}

11
dbms/src/DataStreams/AggregatingBlockInputStream.cpp
View File

@ -32,10 +32,13 @@ Block AggregatingBlockInputStream::readImpl()
ProfileEvents::increment(ProfileEvents::ExternalAggregationMerge);
/// Сбросим имеющиеся в оперативке данные тоже на диск. Так проще.
size_t rows = data_variants.sizeWithoutOverflowRow();
if (rows)
aggregator.writeToTemporaryFile(data_variants, rows);
if (!isCancelled())
{
/// Сбросим имеющиеся в оперативке данные тоже на диск. Так проще.
size_t rows = data_variants.sizeWithoutOverflowRow();
if (rows)
aggregator.writeToTemporaryFile(data_variants, rows);
}
const auto & files = aggregator.getTemporaryFiles();
BlockInputStreams input_streams;