thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-22 07:31:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge

Browse Source
This commit is contained in:
Michael Kolupaev 2014-04-04 16:47:57 +04:00
parent 847128152e
commit 1b5b9a1f1e
5 changed files with 49 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
dbms/include/DB/Storages/MergeTree/ReplicatedMergeTreePartsExchange.h
View File

@ -76,7 +76,6 @@ public:
const String & host,
int port)
{
LOG_TRACE(log, "Fetching part " << part_name);
ReadBufferFromHTTP::Params params = {
std::make_pair("endpoint", "ReplicatedMergeTree:" + replica_path),
std::make_pair("part", part_name),

20
dbms/include/DB/Storages/StorageReplicatedMergeTree.h
View File

@ -42,7 +42,7 @@ public:
return "Replicated" + data.getModePrefix() + "MergeTree";
}
std::string getTableName() const override { return name; }
std::string getTableName() const override { return table_name; }
std::string getSignColumnName() const { return data.getSignColumnName(); }
bool supportsSampling() const override { return data.supportsSampling(); }
bool supportsFinal() const override { return data.supportsFinal(); }
@ -152,17 +152,23 @@ private:
typedef std::list<LogEntry> LogEntries;
typedef std::set<String> StringSet;
typedef std::vector<std::thread> Threads;
Context & context;
zkutil::ZooKeeper & zookeeper;
/// Куски, для которых в очереди есть задание на слияние.
StringSet currently_merging;
Poco::FastMutex currently_merging_mutex;
/** "Очередь" того, что нужно сделать на этой реплике, чтобы всех догнать. Берется из ZooKeeper (/replicas/me/queue/).
* В ZK записи в хронологическом порядке. Здесь записи в том порядке, в котором их лучше выполнять.
*/
LogEntries queue;
Poco::FastMutex queue_mutex;
String path;
String name;
String table_name;
String full_path;
String zookeeper_path;
@ -186,8 +192,6 @@ private:
ReplicatedMergeTreePartsFetcher fetcher;
zkutil::LeaderElectionPtr leader_election;
typedef std::vector<std::thread> Threads;
/// Поток, следящий за обновлениями в логах всех реплик и загружающий их в очередь.
std::thread queue_updating_thread;
@ -197,12 +201,6 @@ private:
/// Поток, выбирающий куски для слияния.
std::thread merge_selecting_thread;
typedef std::set<String> StringSet;
/// Куски, для которых в очереди есть задание на слияние.
StringSet currently_merging;
Poco::FastMutex currently_merging_mutex;
Logger * log;
volatile bool shutdown_called;

6
dbms/src/Storages/MergeTree/MergeTreeDataMerger.cpp
View File

@ -38,8 +38,6 @@ static const double DISK_USAGE_COEFFICIENT_TO_RESERVE = 1.4;
bool MergeTreeDataMerger::selectPartsToMerge(MergeTreeData::DataPartsVector & parts, String & merged_name, size_t available_disk_space,
bool merge_anything_for_old_months, bool aggressive, bool only_small, const AllowedMergingPredicate & can_merge)
{
LOG_DEBUG(log, "Selecting parts to merge");
MergeTreeData::DataParts data_parts = data.getDataParts();
DateLUTSingleton & date_lut = DateLUTSingleton::instance();
@ -233,10 +231,6 @@ bool MergeTreeDataMerger::selectPartsToMerge(MergeTreeData::DataPartsVector & pa
LOG_DEBUG(log, "Selected " << parts.size() << " parts from " << parts.front()->name << " to " << parts.back()->name);
}
else
{
LOG_DEBUG(log, "No parts to merge");
}
return found;
}

5
dbms/src/Storages/StorageMergeTree.cpp
View File

@ -164,9 +164,14 @@ void StorageMergeTree::mergeThread(bool while_can, bool aggressive)
}
}
LOG_DEBUG(log, "Selecting parts to merge");
if (!merger.selectPartsToMerge(parts, merged_name, disk_space, false, aggressive, only_small, can_merge) &&
!merger.selectPartsToMerge(parts, merged_name, disk_space, true, aggressive, only_small, can_merge))
{
LOG_DEBUG(log, "No parts to merge");
break;
}
merging_tagger = new CurrentlyMergingPartsTagger(parts, merger.estimateDiskSpaceForMerge(parts), *this);
}

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

@ -31,12 +31,12 @@ StorageReplicatedMergeTree::StorageReplicatedMergeTree(
const MergeTreeSettings & settings_)
:
context(context_), zookeeper(context.getZooKeeper()),
path(path_), name(name_), full_path(path + escapeForFileName(name) + '/'), zookeeper_path(zookeeper_path_),
table_name(name_), full_path(path_ + escapeForFileName(table_name) + '/'), zookeeper_path(zookeeper_path_),
replica_name(replica_name_), is_leader_node(false),
data( full_path, columns_, context_, primary_expr_ast_, date_column_name_, sampling_expression_,
index_granularity_,mode_, sign_column_, settings_),
reader(data), writer(data), merger(data), fetcher(data),
log(&Logger::get("StorageReplicatedMergeTree")),
log(&Logger::get("StorageReplicatedMergeTree: " + table_name)),
shutdown_called(false)
{
if (!zookeeper_path.empty() && *zookeeper_path.rbegin() == '/')
@ -337,10 +337,13 @@ void StorageReplicatedMergeTree::pullLogsToQueue()
priority_queue.push(iterator);
}
size_t count = 0;
while (!priority_queue.empty())
{
LogIterator iterator = priority_queue.top();
priority_queue.pop();
++count;
LogEntry entry = LogEntry::parse(iterator.entry_str);
@ -353,7 +356,7 @@ void StorageReplicatedMergeTree::pullLogsToQueue()
auto results = zookeeper.multi(ops);
String path_created = dynamic_cast<zkutil::OpResult::Create &>((*results)[0]).getPathCreated();
entry.znode_name = path.substr(path.find_last_of('/') + 1);
entry.znode_name = path_created.substr(path_created.find_last_of('/') + 1);
entry.tagPartsAsCurrentlyMerging(*this);
queue.push_back(entry);
@ -361,6 +364,9 @@ void StorageReplicatedMergeTree::pullLogsToQueue()
if (iterator.readEntry(zookeeper, zookeeper_path))
priority_queue.push(iterator);
}
if (count > 0)
LOG_DEBUG(log, "Pulled " << count << " entries to queue");
}
void StorageReplicatedMergeTree::optimizeQueue()
@ -377,14 +383,25 @@ void StorageReplicatedMergeTree::executeLogEntry(const LogEntry & entry)
/// Даже если кусок есть локально, его (в исключительных случаях) может не быть в zookeeper.
if (containing_part && zookeeper.exists(replica_path + "/parts/" + containing_part->name))
{
LOG_DEBUG(log, "Skipping action for part " + entry.new_part_name + " - part already exists");
return;
}
}
if (entry.type != LogEntry::GET_PART)
if (entry.type == LogEntry::GET_PART)
{
String replica = findActiveReplicaHavingPart(entry.new_part_name);
fetchPart(entry.new_part_name, replica);
}
else if (entry.type == LogEntry::MERGE_PARTS)
{
throw Exception("Merging is not implemented.", ErrorCodes::NOT_IMPLEMENTED);
String replica = findActiveReplicaHavingPart(entry.new_part_name);
fetchPart(entry.new_part_name, replica);
}
else
{
throw Exception("Unexpected log entry type: " + toString(static_cast<int>(entry.type)));
}
}
void StorageReplicatedMergeTree::queueUpdatingThread()
@ -514,6 +531,8 @@ void StorageReplicatedMergeTree::mergeSelectingThread()
String merged_name;
auto can_merge = std::bind(&StorageReplicatedMergeTree::canMergeParts, this, std::placeholders::_1, std::placeholders::_2);
LOG_TRACE(log, "Selecting parts to merge" << (has_big_merge ? " (only small)" : ""));
if (merger.selectPartsToMerge(parts, merged_name, 0, false, false, has_big_merge, can_merge) ||
merger.selectPartsToMerge(parts, merged_name, 0, true, false, has_big_merge, can_merge))
{
@ -543,7 +562,7 @@ void StorageReplicatedMergeTree::mergeSelectingThread()
break;
if (!success)
std::this_thread::sleep_for(QUEUE_AFTER_WORK_SLEEP);
std::this_thread::sleep_for(MERGE_SELECTING_SLEEP);
}
}
@ -561,7 +580,10 @@ bool StorageReplicatedMergeTree::canMergeParts(const MergeTreeData::DataPartPtr
String path = zookeeper_path + "/block_numbers/block-" + number_str;
if (AbandonableLockInZooKeeper::check(path, zookeeper) != AbandonableLockInZooKeeper::ABANDONED)
{
LOG_DEBUG(log, "Can't merge parts " << left->name << " and " << right->name << " because block " << path << " exists");
return false;
}
}
return true;
@ -569,6 +591,7 @@ bool StorageReplicatedMergeTree::canMergeParts(const MergeTreeData::DataPartPtr
void StorageReplicatedMergeTree::becomeLeader()
{
LOG_INFO(log, "Became leader");
is_leader_node = true;
merge_selecting_thread = std::thread(&StorageReplicatedMergeTree::mergeSelectingThread, this);
}
@ -592,6 +615,8 @@ String StorageReplicatedMergeTree::findActiveReplicaHavingPart(const String & pa
void StorageReplicatedMergeTree::fetchPart(const String & part_name, const String & replica_name)
{
LOG_DEBUG(log, "Fetching part " << part_name << " from " << replica_name);
auto table_lock = lockStructure(true);
String host;
@ -621,6 +646,8 @@ void StorageReplicatedMergeTree::fetchPart(const String & part_name, const Strin
zookeeper.getDefaultACL(),
zkutil::CreateMode::Persistent));
zookeeper.multi(ops);
LOG_DEBUG(log, "Fetched part");
}
void StorageReplicatedMergeTree::shutdown()