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

Add log merge for replicas

Browse Source
This commit is contained in:
Bogdan Voronin 2017-03-15 22:57:30 +03:00 committed by alexey-milovidov
parent 98a866ee5b
commit a3ff193fe3
1 changed files with 49 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
dbms/src/Storages/StorageReplicatedMergeTree.cpp
View File

@ -1155,9 +1155,34 @@ bool StorageReplicatedMergeTree::executeLogEntry(const LogEntry & entry)
MergeTreeData::Transaction transaction;
size_t aio_threshold = context.getSettings().min_bytes_to_use_direct_io;
/// Logging
PartLogElement elem;
Stopwatch stopwatch;
elem.event_time = time(0);
elem.merged_from.reserve(parts.size());
for (const auto & part : parts)
elem.merged_from.push_back(part->name);
auto part = merger.mergePartsToTemporaryPart(
parts, entry.new_part_name, *merge_entry, aio_threshold, entry.create_time, reserved_space.get());
std::shared_ptr<PartLog> part_log = context.getPartLog();
if (part_log)
{
elem.event_type = PartLogElement::MERGE_PARTS;
elem.size_in_bytes = part->size_in_bytes;
elem.database_name = part->storage.getDatabaseName();
elem.table_name = part->storage.getTableName();
elem.part_name = part->name;
elem.duration_ms = stopwatch.elapsed() / 1000000;
part_log->add(elem);
}
part_log.reset();
zkutil::Ops ops;
try
@ -2331,10 +2356,34 @@ bool StorageReplicatedMergeTree::optimize(const String & partition, bool final,
{
MergeList::EntryPtr merge_entry = context.getMergeList().insert(database_name, table_name, merged_name, parts);
/// Logging
PartLogElement elem;
Stopwatch stopwatch;
elem.event_time = time(0);
elem.merged_from.reserve(parts.size());
for (const auto & part : parts)
elem.merged_from.push_back(part->name);
auto new_part = unreplicated_merger->mergePartsToTemporaryPart(
parts, merged_name, *merge_entry, settings.min_bytes_to_use_direct_io, time(0));
unreplicated_merger->renameMergedTemporaryPart(parts, new_part, merged_name, nullptr);
std::shared_ptr<PartLog> part_log = context.getPartLog();
if (part_log)
{
elem.event_type = PartLogElement::MERGE_PARTS;
elem.size_in_bytes = new_part->size_in_bytes;
elem.database_name = new_part->storage.getDatabaseName();
elem.table_name = new_part->storage.getTableName();
elem.part_name = new_part->name;
elem.duration_ms = stopwatch.elapsed() / 1000000;
part_log->add(elem);
}
return true;
}
}