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

Merge pull request #72126 from ClickHouse/chesema-partition-sink

Browse Source 
fix cancelation for PartitionedSink
This commit is contained in:
Sema Checherinda 2024-11-20 13:51:07 +00:00 committed by GitHub
commit 999bff236c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 13 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/Storages/PartitionedSink.cpp
View File

@ -146,6 +146,12 @@ String PartitionedSink::replaceWildcards(const String & haystack, const String &
return boost::replace_all_copy(haystack, PartitionedSink::PARTITION_ID_WILDCARD, partition_id); return boost::replace_all_copy(haystack, PartitionedSink::PARTITION_ID_WILDCARD, partition_id);
} }
PartitionedSink::~PartitionedSink()
{
if (isCancelled())
for (auto & item : partition_id_to_sink)
item.second->cancel();
}
} }
// NOLINTEND(clang-analyzer-optin.core.EnumCastOutOfRange) // NOLINTEND(clang-analyzer-optin.core.EnumCastOutOfRange)

2
src/Storages/PartitionedSink.h
View File

@ -18,6 +18,8 @@ public:
PartitionedSink(const ASTPtr & partition_by, ContextPtr context_, const Block & sample_block_); PartitionedSink(const ASTPtr & partition_by, ContextPtr context_, const Block & sample_block_);
~PartitionedSink() override;
String getName() const override { return "PartitionedSink"; } String getName() const override { return "PartitionedSink"; }
void consume(Chunk & chunk) override; void consume(Chunk & chunk) override;

5
src/Storages/StorageJoin.cpp
View File

@ -230,6 +230,11 @@ void StorageJoin::mutate(const MutationCommands & commands, ContextPtr context)
disk->replaceFile(path + tmp_backup_file_name, path + std::to_string(increment) + ".bin"); disk->replaceFile(path + tmp_backup_file_name, path + std::to_string(increment) + ".bin");
} }
else
{
compressed_backup_buf.cancel();
backup_buf->cancel();
}
} }
HashJoinPtr StorageJoin::getJoinLocked(std::shared_ptr<TableJoin> analyzed_join, ContextPtr context, const Names & required_columns_names) const HashJoinPtr StorageJoin::getJoinLocked(std::shared_ptr<TableJoin> analyzed_join, ContextPtr context, const Names & required_columns_names) const