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

Better

Browse Source
This commit is contained in:
Antonio Andelic 2024-12-06 13:43:30 +01:00
parent d1a35d5b28
commit 6c6e1118c5
2 changed files with 8 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp
View File

@ -169,7 +169,7 @@ void DistributedAsyncInsertDirectoryQueue::flushAllData(const SettingsChanges &
std::lock_guard lock{mutex};
if (!hasPendingFiles())
return;
processFiles(settings_changes, /*force=*/true);
processFiles(/*force=*/true, settings_changes);
}
void DistributedAsyncInsertDirectoryQueue::shutdownAndDropAllData()
@ -212,7 +212,7 @@ void DistributedAsyncInsertDirectoryQueue::run()
{
try
{
processFiles();
processFiles(/*force=*/false);
/// No errors while processing existing files.
/// Let's see maybe there are more files to process.
do_sleep = false;
@ -380,11 +380,11 @@ void DistributedAsyncInsertDirectoryQueue::initializeFilesFromDisk()
status.broken_bytes_count = broken_bytes_count;
}
}
void DistributedAsyncInsertDirectoryQueue::processFiles(const SettingsChanges & settings_changes, bool force)
void DistributedAsyncInsertDirectoryQueue::processFiles(bool force, const SettingsChanges & settings_changes)
try
{
if (should_batch_inserts)
processFilesWithBatching(settings_changes, force);
processFilesWithBatching(force, settings_changes);
else
{
/// Process unprocessed file.
@ -539,7 +539,7 @@ DistributedAsyncInsertDirectoryQueue::Status DistributedAsyncInsertDirectoryQueu
return current_status;
}
void DistributedAsyncInsertDirectoryQueue::processFilesWithBatching(const SettingsChanges & settings_changes, bool force)
void DistributedAsyncInsertDirectoryQueue::processFilesWithBatching(bool force, const SettingsChanges & settings_changes)
{
/// Possibly, we failed to send a batch on the previous iteration. Try to send exactly the same batch.
if (fs::exists(current_batch_file_path))

5
src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.h
View File

@ -100,9 +100,10 @@ private:
void addFile(const std::string & file_path);
void initializeFilesFromDisk();
void processFiles(const SettingsChanges & settings_changes = {}, bool force = false);
/// Set `force = true` if processing of files must be finished fully despite cancellation flag being set
void processFiles(bool force, const SettingsChanges & settings_changes = {});
void processFile(std::string & file_path, const SettingsChanges & settings_changes);
void processFilesWithBatching(const SettingsChanges & settings_changes, bool force);
void processFilesWithBatching(bool force, const SettingsChanges & settings_changes);
void markAsBroken(const std::string & file_path);
void markAsSend(const std::string & file_path);