mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-17 21:24:28 +00:00
Less verbose logging
This commit is contained in:
parent
45ebf08925
commit
a6938cf5d9
@ -1756,60 +1756,35 @@ std::optional<std::pair<Int64, int>> ReplicatedMergeTreeMergePredicate::getDesir
|
|||||||
/// the part (checked by querying queue.virtual_parts), we can confidently assign a mutation to
|
/// the part (checked by querying queue.virtual_parts), we can confidently assign a mutation to
|
||||||
/// version X for this part.
|
/// version X for this part.
|
||||||
|
|
||||||
LOG_DEBUG(queue.log, "LOOKING for desired mutation version for part:" << part->name);
|
|
||||||
if (last_quorum_parts.find(part->name) != last_quorum_parts.end()
|
if (last_quorum_parts.find(part->name) != last_quorum_parts.end()
|
||||||
|| part->name == inprogress_quorum_part)
|
|| part->name == inprogress_quorum_part)
|
||||||
{
|
|
||||||
LOG_DEBUG(queue.log, "PART " << part->name <<" NAME NOT IN QUORUM");
|
|
||||||
return {};
|
return {};
|
||||||
}
|
|
||||||
|
|
||||||
std::lock_guard lock(queue.state_mutex);
|
std::lock_guard lock(queue.state_mutex);
|
||||||
|
|
||||||
if (queue.virtual_parts.getContainingPart(part->info) != part->name)
|
if (queue.virtual_parts.getContainingPart(part->info) != part->name)
|
||||||
{
|
|
||||||
LOG_DEBUG(queue.log, "VIRTUAL PARTS HAVE CONTAINING PART " << part->name);
|
|
||||||
return {};
|
return {};
|
||||||
}
|
|
||||||
|
|
||||||
auto in_partition = queue.mutations_by_partition.find(part->info.partition_id);
|
auto in_partition = queue.mutations_by_partition.find(part->info.partition_id);
|
||||||
if (in_partition == queue.mutations_by_partition.end())
|
if (in_partition == queue.mutations_by_partition.end())
|
||||||
{
|
|
||||||
LOG_DEBUG(queue.log, "NO PARTITION FOR MUTATION FOR PART " << part->name);
|
|
||||||
return {};
|
return {};
|
||||||
}
|
|
||||||
|
|
||||||
Int64 current_version = queue.getCurrentMutationVersionImpl(part->info.partition_id, part->info.getDataVersion(), lock);
|
Int64 current_version = queue.getCurrentMutationVersionImpl(part->info.partition_id, part->info.getDataVersion(), lock);
|
||||||
Int64 max_version = in_partition->second.rbegin()->first;
|
Int64 max_version = in_partition->second.rbegin()->first;
|
||||||
|
|
||||||
if (current_version >= max_version)
|
|
||||||
{
|
|
||||||
LOG_DEBUG(queue.log, "PART VERSION FOR " << part->name << " IS BIGGER THAN MAX");
|
|
||||||
//std::cerr << "But current version is:" << current_version << std::endl;
|
|
||||||
return {};
|
|
||||||
}
|
|
||||||
|
|
||||||
int alter_version = -1;
|
int alter_version = -1;
|
||||||
|
|
||||||
//std::cerr << "Looking for alter version for mutation\n";
|
|
||||||
String version;
|
|
||||||
for (auto [mutation_version, mutation_status] : in_partition->second)
|
for (auto [mutation_version, mutation_status] : in_partition->second)
|
||||||
{
|
{
|
||||||
max_version = mutation_version;
|
max_version = mutation_version;
|
||||||
if (mutation_version > current_version && mutation_status->entry->alter_version != -1)
|
if (mutation_version > current_version && mutation_status->entry->alter_version != -1)
|
||||||
{
|
{
|
||||||
alter_version = mutation_status->entry->alter_version;
|
alter_version = mutation_status->entry->alter_version;
|
||||||
version = mutation_status->entry->znode_name;
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
//std::cerr << "FOUND alter version:" << alter_version << " and mutation znode name:" << version << std::endl;
|
|
||||||
if (current_version >= max_version)
|
|
||||||
{
|
|
||||||
LOG_DEBUG(queue.log, "PART VERSION FOR " << part->name << " IS BIGGER THAN MAX AFTER ALTER");
|
|
||||||
return {};
|
|
||||||
}
|
|
||||||
|
|
||||||
|
if (current_version >= max_version)
|
||||||
|
return {};
|
||||||
|
|
||||||
return std::make_pair(max_version, alter_version);
|
return std::make_pair(max_version, alter_version);
|
||||||
}
|
}
|
||||||
|
@ -2251,11 +2251,7 @@ BackgroundProcessingPoolTaskResult StorageReplicatedMergeTree::movePartsTask()
|
|||||||
void StorageReplicatedMergeTree::mergeSelectingTask()
|
void StorageReplicatedMergeTree::mergeSelectingTask()
|
||||||
{
|
{
|
||||||
if (!is_leader)
|
if (!is_leader)
|
||||||
{
|
|
||||||
LOG_DEBUG(log, "I'm not leader, I don't want to assign anything");
|
|
||||||
return;
|
return;
|
||||||
}
|
|
||||||
LOG_DEBUG(log, "Merge selecting started");
|
|
||||||
|
|
||||||
const auto storage_settings_ptr = getSettings();
|
const auto storage_settings_ptr = getSettings();
|
||||||
const bool deduplicate = false; /// TODO: read deduplicate option from table config
|
const bool deduplicate = false; /// TODO: read deduplicate option from table config
|
||||||
@ -2295,7 +2291,6 @@ void StorageReplicatedMergeTree::mergeSelectingTask()
|
|||||||
if (max_source_parts_size_for_merge > 0 &&
|
if (max_source_parts_size_for_merge > 0 &&
|
||||||
merger_mutator.selectPartsToMerge(future_merged_part, false, max_source_parts_size_for_merge, merge_pred))
|
merger_mutator.selectPartsToMerge(future_merged_part, false, max_source_parts_size_for_merge, merge_pred))
|
||||||
{
|
{
|
||||||
LOG_DEBUG(log, "ASSIGNING MERGE");
|
|
||||||
success = createLogEntryToMergeParts(zookeeper, future_merged_part.parts,
|
success = createLogEntryToMergeParts(zookeeper, future_merged_part.parts,
|
||||||
future_merged_part.name, deduplicate, force_ttl);
|
future_merged_part.name, deduplicate, force_ttl);
|
||||||
}
|
}
|
||||||
@ -2307,20 +2302,12 @@ void StorageReplicatedMergeTree::mergeSelectingTask()
|
|||||||
DataPartsVector data_parts = getDataPartsVector();
|
DataPartsVector data_parts = getDataPartsVector();
|
||||||
for (const auto & part : data_parts)
|
for (const auto & part : data_parts)
|
||||||
{
|
{
|
||||||
LOG_DEBUG(log, "ASSIGNING MUTATIONS LOOKING AT PART " << part->name);
|
|
||||||
if (part->bytes_on_disk > max_source_part_size_for_mutation)
|
if (part->bytes_on_disk > max_source_part_size_for_mutation)
|
||||||
continue;
|
continue;
|
||||||
|
|
||||||
std::optional<std::pair<Int64, int>> desired_mutation_version = merge_pred.getDesiredMutationVersion(part);
|
std::optional<std::pair<Int64, int>> desired_mutation_version = merge_pred.getDesiredMutationVersion(part);
|
||||||
if (!desired_mutation_version)
|
if (!desired_mutation_version)
|
||||||
{
|
|
||||||
LOG_DEBUG(log, "NO Desired version found");
|
|
||||||
continue;
|
continue;
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
LOG_DEBUG(log, "Desired mutation version: " << desired_mutation_version->first << " alter version:" << desired_mutation_version->second);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (createLogEntryToMutatePart(*part, desired_mutation_version->first, desired_mutation_version->second))
|
if (createLogEntryToMutatePart(*part, desired_mutation_version->first, desired_mutation_version->second))
|
||||||
{
|
{
|
||||||
@ -2329,10 +2316,6 @@ void StorageReplicatedMergeTree::mergeSelectingTask()
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
else
|
|
||||||
{
|
|
||||||
LOG_DEBUG(log, "TOO MANY MUTATIONS IN QUEUE");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
catch (...)
|
catch (...)
|
||||||
|
Loading…
Reference in New Issue
Block a user