thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-10 09:32:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4372 from yandex/fix-cancel-of-vertical-merges

Browse Source 
Faster cancelling of vertical merges
This commit is contained in:
alexey-milovidov 2019-02-13 02:58:00 +03:00 committed by GitHub
commit 024709db7d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
dbms/src/DataStreams/CollapsingSortedBlockInputStream.h
View File

@ -3,6 +3,7 @@
#include <common/logger_useful.h>
#include <DataStreams/MergingSortedBlockInputStream.h>
#include <DataStreams/ColumnGathererStream.h>
namespace DB

1
dbms/src/DataStreams/MergingSortedBlockInputStream.cpp
View File

@ -3,6 +3,7 @@
#include <sstream>
#include <DataStreams/MergingSortedBlockInputStream.h>
#include <DataStreams/ColumnGathererStream.h>
namespace DB

1
dbms/src/DataStreams/MergingSortedBlockInputStream.h
View File

@ -13,7 +13,6 @@
#include <IO/WriteHelpers.h>
#include <DataStreams/IBlockInputStream.h>
#include <DataStreams/ColumnGathererStream.h>
namespace DB

1
dbms/src/DataStreams/ReplacingSortedBlockInputStream.h
View File

@ -3,6 +3,7 @@
#include <common/logger_useful.h>
#include <DataStreams/MergingSortedBlockInputStream.h>
#include <DataStreams/ColumnGathererStream.h>
namespace DB

1
dbms/src/DataStreams/VersionedCollapsingSortedBlockInputStream.h
View File

@ -3,6 +3,7 @@
#include <common/logger_useful.h>
#include <DataStreams/MergingSortedBlockInputStream.h>
#include <DataStreams/ColumnGathererStream.h>
#include <deque>

9
dbms/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
View File

@ -762,11 +762,15 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
size_t column_elems_written = 0;
column_to.writePrefix();
while ((block = column_gathered_stream.read()))
while (!actions_blocker.isCancelled() && (block = column_gathered_stream.read()))
{
column_elems_written += block.rows();
column_to.write(block);
}
if (actions_blocker.isCancelled())
throw Exception("Cancelled merging parts", ErrorCodes::ABORTED);
column_gathered_stream.readSuffix();
checksums_gathered_columns.add(column_to.writeSuffixAndGetChecksums());
@ -781,9 +785,6 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
merge_entry->columns_written += 1;
merge_entry->bytes_written_uncompressed += column_gathered_stream.getProfileInfo().bytes;
merge_entry->progress.store(progress_before + column_sizes.columnWeight(column_name), std::memory_order_relaxed);
if (actions_blocker.isCancelled())
throw Exception("Cancelled merging parts", ErrorCodes::ABORTED);
}
Poco::File(rows_sources_file_path).remove();