Skip to content

Change export partition log levels #2004

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
arthurpassos wants to merge 3 commits into
base: antalya-26.3
Choose a base branch
from
Open

Change export partition log levels #2004

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b62f3f0
change export partition log levels
arthurpassos Jul 3, 2026
93e07b1
adjustment
arthurpassos Jul 3, 2026
0c5ddc2
one more
arthurpassos Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 20 additions & 20 deletions src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ namespace
ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGetChildren);
if (Coordination::Error::ZOK != zk->tryGetChildren(container_path, children))
{
LOG_INFO(log, "ExportPartition Manifest Updating Task: failed to list last_exception leaves for {}, leaving in-memory copy untouched", log_key);
LOG_WARNING(log, "ExportPartition Manifest Updating Task: failed to list last_exception leaves for {}, leaving in-memory copy untouched", log_key);
return out;
}

Expand Down Expand Up @@ -140,7 +140,7 @@ namespace
fs::path(entry_path) / "commit_lock", *zk, storage.getReplicaName());
if (!commit_lock)
{
LOG_INFO(log, "ExportPartition Manifest Updating Task: commit in progress for {}, skipping timeout kill", entry_path);
LOG_DEBUG(log, "ExportPartition Manifest Updating Task: commit in progress for {}, skipping timeout kill", entry_path);
return;
}

Expand All @@ -153,14 +153,14 @@ namespace
ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGet);
if (!zk->tryGet(status_path, status_string, &status_stat))
{
LOG_INFO(log, "ExportPartition Manifest Updating Task: Failed to read status for {} while enforcing task timeout, skipping", entry_path);
LOG_WARNING(log, "ExportPartition Manifest Updating Task: Failed to read status for {} while enforcing task timeout, skipping", entry_path);
return;
}

const auto current_status = magic_enum::enum_cast<ExportReplicatedMergeTreePartitionTaskEntry::Status>(status_string);
if (!current_status || *current_status != ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING)
{
LOG_INFO(log, "ExportPartition Manifest Updating Task: Task {} is not PENDING, can't set to KILLED, skipping", entry_path);
LOG_DEBUG(log, "ExportPartition Manifest Updating Task: Task {} is not PENDING, can't set to KILLED, skipping", entry_path);
return;
}

Expand Down Expand Up @@ -195,7 +195,7 @@ namespace
/// ZBADVERSION (status changed), ZNODEEXISTS (lazy-create race with the scheduler),
/// counter race, or ZNONODE (entry concurrently removed). In all cases the batch
/// was rolled back atomically and the task will be re-evaluated on the next poll.
LOG_INFO(log,
LOG_DEBUG(log,
"ExportPartition Manifest Updating Task: atomic kill for {} failed (rc={}); "
"status was concurrently updated or a ZK op conflicted, will retry on next poll",
entry_path, rc);
Expand All @@ -215,19 +215,19 @@ namespace
if (Coordination::Error::ZOK != zk->tryGetChildren(fs::path(entry_path) / "processing", parts_in_processing_or_pending))
{

LOG_INFO(log, "ExportPartition Manifest Updating Task: Failed to get parts in processing or pending, skipping");
LOG_WARNING(log, "ExportPartition Manifest Updating Task: Failed to get parts in processing or pending, skipping");
return;
}

if (parts_in_processing_or_pending.empty())
{
LOG_INFO(log, "ExportPartition Manifest Updating Task: Cleanup found PENDING for {} with all parts exported, deferring commit recovery to post-lock phase", entry_path);
LOG_DEBUG(log, "ExportPartition Manifest Updating Task: Cleanup found PENDING for {} with all parts exported, deferring commit recovery to post-lock phase", entry_path);

const auto destination_storage_id = StorageID(QualifiedTableName {metadata.destination_database, metadata.destination_table});
const auto destination_storage = DatabaseCatalog::instance().tryGetTable(destination_storage_id, context);
if (!destination_storage)
{
LOG_INFO(log, "ExportPartition Manifest Updating Task: Failed to reconstruct destination storage: {}, skipping", destination_storage_id.getNameForLogs());
LOG_WARNING(log, "ExportPartition Manifest Updating Task: Failed to reconstruct destination storage: {}, skipping", destination_storage_id.getNameForLogs());
return;
}

Expand Down Expand Up @@ -312,7 +312,7 @@ void ExportPartitionManifestUpdatingTask::poll()
auto cleanup_lock = zkutil::EphemeralNodeHolder::tryCreate(cleanup_lock_path, *zk, storage.replica_name);
if (cleanup_lock)
{
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Cleanup lock acquired, will remove stale entries");
LOG_DEBUG(storage.log, "ExportPartition Manifest Updating Task: Cleanup lock acquired, will remove stale entries");
}

{
Expand All @@ -329,7 +329,7 @@ void ExportPartitionManifestUpdatingTask::poll()

auto & entries_by_key = working_model->get<ExportPartitionTaskEntryTagByCompositeKey>();

LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Polling for new entries for table {}. Current number of entries: {}", storage.getStorageID().getNameForLogs(), entries_by_key.size());
LOG_DEBUG(storage.log, "ExportPartition Manifest Updating Task: Polling for new entries for table {}. Current number of entries: {}", storage.getStorageID().getNameForLogs(), entries_by_key.size());

ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests);
ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGetChildrenWatch);
Expand All @@ -352,7 +352,7 @@ void ExportPartitionManifestUpdatingTask::poll()
std::string metadata_json;
if (!zk->tryGet(fs::path(entry_path) / "metadata.json", metadata_json))
{
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: missing metadata.json", key);
LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: missing metadata.json", key);
continue;
}

Expand Down Expand Up @@ -402,14 +402,14 @@ void ExportPartitionManifestUpdatingTask::poll()

if (status_string.empty())
{
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: missing status", key);
LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: missing status", key);
continue;
}

const auto status = magic_enum::enum_cast<ExportReplicatedMergeTreePartitionTaskEntry::Status>(status_string);
if (!status)
{
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", status_string, key);
LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", status_string, key);
continue;
}

Expand Down Expand Up @@ -458,7 +458,7 @@ void ExportPartitionManifestUpdatingTask::poll()
}
}
}
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: already exists", key);
LOG_DEBUG(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: already exists", key);

}

Expand All @@ -470,7 +470,7 @@ void ExportPartitionManifestUpdatingTask::poll()
/// `entries_by_key` (a reference into it) must not be used afterwards.
storage.export_partition_manifests.set(std::move(working_model));

LOG_INFO(storage.log, "ExportPartition Manifest Updating task: finished polling for new entries. Number of entries: {}", entries_count);
LOG_DEBUG(storage.log, "ExportPartition Manifest Updating task: finished polling for new entries. Number of entries: {}", entries_count);
}

const auto log_ptr = storage.log.load();
Expand Down Expand Up @@ -603,7 +603,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges()

const bool had_changes = !local_status_changes.empty();

LOG_INFO(storage.log, "ExportPartition Manifest Updating task: handling status changes. Number of status changes: {}", local_status_changes.size());
LOG_DEBUG(storage.log, "ExportPartition Manifest Updating task: handling status changes. Number of status changes: {}", local_status_changes.size());

const auto current_model = storage.export_partition_manifests.get();
auto working_model = current_model
Expand Down Expand Up @@ -635,15 +635,15 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges()
std::string new_status_string;
if (!zk->tryGet(fs::path(storage.zookeeper_path) / "exports" / key / "status", new_status_string))
{
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Failed to get new status for task {}, skipping", key);
LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Failed to get new status for task {}, skipping", key);
local_status_changes.pop();
continue;
}

const auto new_status = magic_enum::enum_cast<ExportReplicatedMergeTreePartitionTaskEntry::Status>(new_status_string);
if (!new_status)
{
LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", new_status_string, key);
LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", new_status_string, key);
local_status_changes.pop();
continue;
}
Expand Down Expand Up @@ -691,7 +691,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges()
{
tryLogCurrentException(storage.log, __PRETTY_FUNCTION__);

LOG_INFO(storage.log, "ExportPartition Manifest Updating task: exception thrown while handling status changes; nothing was published, requeuing the whole batch. Batch size: {}", batch.size());
LOG_WARNING(storage.log, "ExportPartition Manifest Updating task: exception thrown while handling status changes; nothing was published, requeuing the whole batch. Batch size: {}", batch.size());

std::lock_guard lock(status_changes_mutex);

Expand All @@ -708,7 +708,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges()
std::swap(status_changes, requeued);
}

LOG_INFO(storage.log, "ExportPartition Manifest Updating task: pending status changes after requeue: {}", status_changes.size());
LOG_DEBUG(storage.log, "ExportPartition Manifest Updating task: pending status changes after requeue: {}", status_changes.size());

throw;
}
Expand Down
Loading
Loading