Improve CHECK TABLE system query

Resubmit PR #52745

docs/en/sql-reference/statements/check-table.md

@@ -5,19 +5,38 @@ sidebar_label: CHECK TABLE
 title: "CHECK TABLE Statement"
 ---
 
-Checks if the data in the table is corrupted.
+The `CHECK TABLE` query in ClickHouse is used to perform a validation check on a specific table or its partitions. It ensures the integrity of the data by verifying the checksums and other internal data structures.
 
-``` sql
-CHECK TABLE [db.]name [PARTITION partition_expr]
+Particularly it compares actual file sizes with the expected values which are stored on the server. If the file sizes do not match the stored values, it means the data is corrupted. This can be caused, for example, by a system crash during query execution.
+
+:::note
+The `CHECK TABLE`` query may read all the data in the table and hold some resources, making it resource-intensive.
+Consider the potential impact on performance and resource utilization before executing this query.
+:::
+
+## Syntax
+
+The basic syntax of the query is as follows:
+
+```sql
+CHECK TABLE table_name [PARTITION partition_expression] [FORMAT format] [SETTINGS check_query_single_value_result = (0|1) [, other_settings]  ]
 ```
 
-The `CHECK TABLE` query compares actual file sizes with the expected values which are stored on the server. If the file sizes do not match the stored values, it means the data is corrupted. This can be caused, for example, by a system crash during query execution.
+- `table_name`: Specifies the name of the table that you want to check.
+- `partition_expression`: (Optional) If you want to check a specific partition of the table, you can use this expression to specify the partition.
+- `FORMAT format`: (Optional) Allows you to specify the output format of the result.
+- `SETTINGS`: (Optional) Allows additional settings.
+	- **`check_query_single_value_result`**: (Optional) This setting allows you to toggle between a detailed result (`0`) or a summarized result (`1`).
+	- Other settings (e.g. `max_threads` can be applied as well).
 
-The query response contains the `result` column with a single row. The row has a value of
-[Boolean](../../sql-reference/data-types/boolean.md) type:
 
-- 0 - The data in the table is corrupted.
-- 1 - The data maintains integrity.
+The query response depends on the value of contains `check_query_single_value_result` setting.
+In case of `check_query_single_value_result = 1` only `result` column with a single row is returned. Value inside this row is `1` if the integrity check is passed and `0` if data is corrupted.
+
+With `check_query_single_value_result = 0` the query returns the following columns:
+    - `part_path`: Indicates the path to the data part or file name.
+    - `is_passed`: Returns 1 if the check for this part is successful, 0 otherwise.
+    - `message`: Any additional messages related to the check, such as errors or success messages.
 
 The `CHECK TABLE` query supports the following table engines:
 
@@ -26,30 +45,15 @@ The `CHECK TABLE` query supports the following table engines:
 - [StripeLog](../../engines/table-engines/log-family/stripelog.md)
 - [MergeTree family](../../engines/table-engines/mergetree-family/mergetree.md)
 
-Performed over the tables with another table engines causes an exception.
+Performed over the tables with another table engines causes an `NOT_IMPLEMETED` exception.
 
 Engines from the `*Log` family do not provide automatic data recovery on failure. Use the `CHECK TABLE` query to track data loss in a timely manner.
 
-## Checking the MergeTree Family Tables
+## Examples
 
-For `MergeTree` family engines, if [check_query_single_value_result](../../operations/settings/settings.md#check_query_single_value_result) = 0, the `CHECK TABLE` query shows a check status for every individual data part of a table on the local server.
+By default `CHECK TABLE` query shows the general table check status:
 
 ```sql
-SET check_query_single_value_result = 0;
-CHECK TABLE test_table;
-```
-
-```text
-┌─part_path─┬─is_passed─┬─message─┐
-│ all_1_4_1 │         1 │         │
-│ all_1_4_2 │         1 │         │
-└───────────┴───────────┴─────────┘
-```
-
-If `check_query_single_value_result` = 1, the `CHECK TABLE` query shows the general table check status.
-
-```sql
-SET check_query_single_value_result = 1;
 CHECK TABLE test_table;
 ```
 
@@ -59,11 +63,60 @@ CHECK TABLE test_table;
 └────────┘
 ```
 
+If you want to see the check status for every individual data part you may use `check_query_single_value_result` setting.
+
+Also, to check a specific partition of the table, you can use the `PARTITION` keyword.
+
+```sql
+CHECK TABLE t0 PARTITION ID '201003'
+FORMAT PrettyCompactMonoBlock
+SETTINGS check_query_single_value_result = 0
+```
+
+Output:
+
+```text
+┌─part_path────┬─is_passed─┬─message─┐
+│ 201003_7_7_0 │         1 │         │
+│ 201003_3_3_0 │         1 │         │
+└──────────────┴───────────┴─────────┘
+```
+
+### Receiving a 'Corrupted' Result
+
+:::warning
+Disclaimer: The procedure described here, including the manual manipulating or removing files directly from the data directory, is for experimental or development environments only. Do **not** attempt this on a production server, as it may lead to data loss or other unintended consequences.
+:::
+
+Remove the existing checksum file:
+
+```bash
+rm /var/lib/clickhouse-server/data/default/t0/201003_3_3_0/checksums.txt
+```
+
+```sql
+CHECK TABLE t0 PARTITION ID '201003'
+FORMAT PrettyCompactMonoBlock
+SETTINGS check_query_single_value_result = 0
+
+
+Output:
+
+```text
+┌─part_path────┬─is_passed─┬─message──────────────────────────────────┐
+│ 201003_7_7_0 │         1 │                                          │
+│ 201003_3_3_0 │         1 │ Checksums recounted and written to disk. │
+└──────────────┴───────────┴──────────────────────────────────────────┘
+```
+
+If the checksums.txt file is missing, it can be restored. It will be recalculated and rewritten during the execution of the CHECK TABLE command for the specific partition, and the status will still be reported as 'success.'"
+
+
 ## If the Data Is Corrupted
 
 If the table is corrupted, you can copy the non-corrupted data to another table. To do this:
 
 1.  Create a new table with the same structure as damaged table. To do this execute the query `CREATE TABLE <new_table_name> AS <damaged_table_name>`.
-2.  Set the [max_threads](../../operations/settings/settings.md#settings-max_threads) value to 1 to process the next query in a single thread. To do this run the query `SET max_threads = 1`.
+2.  Set the `max_threads` value to 1 to process the next query in a single thread. To do this run the query `SET max_threads = 1`.
 3.  Execute the query `INSERT INTO <new_table_name> SELECT * FROM <damaged_table_name>`. This request copies the non-corrupted data from the damaged table to another table. Only the data before the corrupted part will be copied.
 4.  Restart the `clickhouse-client` to reset the `max_threads` value.

src/Common/FileChecker.cpp

@@ -82,33 +82,35 @@ size_t FileChecker::getTotalSize() const
 }
 
 
-CheckResults FileChecker::check() const
+FileChecker::DataValidationTasksPtr FileChecker::getDataValidationTasks()
 {
-    if (map.empty())
-        return {};
+    return std::make_unique<DataValidationTasks>(map);
+}
 
-    CheckResults results;
-
-    for (const auto & name_size : map)
+CheckResult FileChecker::checkNextEntry(DataValidationTasksPtr & check_data_tasks, bool & has_nothing_to_do) const
+{
+    String name;
+    size_t expected_size;
+    bool is_finished = check_data_tasks->next(name, expected_size);
+    if (is_finished)
     {
-        const String & name = name_size.first;
-        String path = parentPath(files_info_path) + name;
-        bool exists = fileReallyExists(path);
-        auto real_size = exists ? getRealFileSize(path) : 0;  /// No race condition assuming no one else is working with these files.
-
-        if (real_size != name_size.second)
-        {
-            String failure_message = exists
-                ? ("Size of " + path + " is wrong. Size is " + toString(real_size) + " but should be " + toString(name_size.second))
-                : ("File " + path + " doesn't exist");
-            results.emplace_back(name, false, failure_message);
-            break;
-        }
-
-        results.emplace_back(name, true, "");
+        has_nothing_to_do = true;
+        return {};
     }
 
-    return results;
+    String path = parentPath(files_info_path) + name;
+    bool exists = fileReallyExists(path);
+    auto real_size = exists ? getRealFileSize(path) : 0;  /// No race condition assuming no one else is working with these files.
+
+    if (real_size != expected_size)
+    {
+        String failure_message = exists
+            ? ("Size of " + path + " is wrong. Size is " + toString(real_size) + " but should be " + toString(expected_size))
+            : ("File " + path + " doesn't exist");
+        return CheckResult(name, false, failure_message);
+    }
+
+    return CheckResult(name, true, "");
 }
 
 void FileChecker::repair()

src/Common/FileChecker.h

@@ -3,6 +3,7 @@
 #include <Storages/CheckResults.h>
 #include <map>
 #include <base/types.h>
+#include <mutex>
 
 namespace Poco { class Logger; }
 
@@ -28,7 +29,11 @@ public:
     bool empty() const { return map.empty(); }
 
     /// Check the files whose parameters are specified in sizes.json
-    CheckResults check() const;
+    /// See comment in IStorage::checkDataNext
+    struct DataValidationTasks;
+    using DataValidationTasksPtr = std::unique_ptr<DataValidationTasks>;
+    DataValidationTasksPtr getDataValidationTasks();
+    CheckResult checkNextEntry(DataValidationTasksPtr & check_data_tasks, bool & has_nothing_to_do) const;
 
     /// Truncate files that have excessive size to the expected size.
     /// Throw exception if the file size is less than expected.
@@ -41,6 +46,36 @@ public:
     /// Returns total size of all files.
     size_t getTotalSize() const;
 
+    struct DataValidationTasks
+    {
+        DataValidationTasks(const std::map<String, size_t> & map_)
+            : map(map_), it(map.begin())
+        {}
+
+        bool next(String & out_name, size_t & out_size)
+        {
+            std::lock_guard lock(mutex);
+            if (it == map.end())
+                return true;
+            out_name = it->first;
+            out_size = it->second;
+            ++it;
+            return false;
+        }
+
+        size_t size() const
+        {
+            std::lock_guard lock(mutex);
+            return std::distance(it, map.end());
+        }
+
+        const std::map<String, size_t> & map;
+
+        mutable std::mutex mutex;
+        using Iterator = std::map<String, size_t>::const_iterator;
+        Iterator it;
+    };
+
 private:
     void load();
 

src/Interpreters/InterpreterCheckQuery.cpp

@@ -8,72 +8,201 @@
 #include <DataTypes/DataTypeString.h>
 #include <Columns/ColumnsNumber.h>
 #include <Common/typeid_cast.h>
+#include <Interpreters/ProcessList.h>
 #include <algorithm>
 
+#include <Columns/IColumn.h>
+
+#include <Processors/ResizeProcessor.h>
+#include <Processors/IAccumulatingTransform.h>
+
 
 namespace DB
 {
 
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
 namespace
 {
 
-NamesAndTypes getBlockStructure()
+Block getSingleValueBlock(UInt8 value)
 {
-    return {
+    return Block{{ColumnUInt8::create(1, value), std::make_shared<DataTypeUInt8>(), "result"}};
+}
+
+Block getHeaderForCheckResult()
+{
+    auto names_and_types = NamesAndTypes{
         {"part_path", std::make_shared<DataTypeString>()},
         {"is_passed", std::make_shared<DataTypeUInt8>()},
         {"message", std::make_shared<DataTypeString>()},
     };
+
+    return Block({
+        {names_and_types[0].type->createColumn(), names_and_types[0].type, names_and_types[0].name},
+        {names_and_types[1].type->createColumn(), names_and_types[1].type, names_and_types[1].name},
+        {names_and_types[2].type->createColumn(), names_and_types[2].type, names_and_types[2].name},
+    });
 }
 
+Chunk getChunkFromCheckResult(const CheckResult & check_result)
+{
+    MutableColumns columns = getHeaderForCheckResult().cloneEmptyColumns();
+    columns[0]->insert(check_result.fs_path);
+    columns[1]->insert(static_cast<UInt8>(check_result.success));
+    columns[2]->insert(check_result.failure_message);
+    return Chunk(std::move(columns), 1);
+}
+
+class TableCheckWorkerProcessor : public ISource
+{
+
+public:
+    TableCheckWorkerProcessor(IStorage::DataValidationTasksPtr check_data_tasks_, StoragePtr table_)
+        : ISource(getHeaderForCheckResult())
+        , table(table_)
+        , check_data_tasks(check_data_tasks_)
+    {
+    }
+
+    String getName() const override { return "TableCheckWorkerProcessor"; }
+
+protected:
+
+    std::optional<Chunk> tryGenerate() override
+    {
+        bool has_nothing_to_do = false;
+        auto check_result = table->checkDataNext(check_data_tasks, has_nothing_to_do);
+        if (has_nothing_to_do)
+            return {};
+
+        /// We can omit manual `progess` call, ISource will may count it automatically by returned chunk
+        /// However, we want to report only rows in progress
+        progress(1, 0);
+
+        if (!check_result.success)
+        {
+            LOG_WARNING(&Poco::Logger::get("InterpreterCheckQuery"),
+                "Check query for table {} failed, path {}, reason: {}",
+                table->getStorageID().getNameForLogs(),
+                check_result.fs_path,
+                check_result.failure_message);
+        }
+
+        return getChunkFromCheckResult(check_result);
+    }
+
+private:
+    StoragePtr table;
+    IStorage::DataValidationTasksPtr check_data_tasks;
+};
+
+class TableCheckResultEmitter : public IAccumulatingTransform
+{
+public:
+    TableCheckResultEmitter() : IAccumulatingTransform(getHeaderForCheckResult(), getSingleValueBlock(1).cloneEmpty()) {}
+
+    String getName() const override { return "TableCheckResultEmitter"; }
+
+    void consume(Chunk chunk) override
+    {
+        if (result_value == 0)
+            return;
+
+        auto columns = chunk.getColumns();
+        if (columns.size() != 3)
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong number of columns: {}", columns.size());
+
+        const auto * col = checkAndGetColumn<ColumnUInt8>(columns[1].get());
+        for (size_t i = 0; i < col->size(); ++i)
+        {
+            if (col->getElement(i) == 0)
+            {
+                result_value = 0;
+                return;
+            }
+        }
+    }
+
+    Chunk generate() override
+    {
+        if (is_value_emitted.exchange(true))
+            return {};
+        auto block = getSingleValueBlock(result_value);
+        return Chunk(block.getColumns(), block.rows());
+    }
+
+private:
+    std::atomic<UInt8> result_value{1};
+    std::atomic_bool is_value_emitted{false};
+};
+
 }
 
-
-InterpreterCheckQuery::InterpreterCheckQuery(const ASTPtr & query_ptr_, ContextPtr context_) : WithContext(context_), query_ptr(query_ptr_)
+InterpreterCheckQuery::InterpreterCheckQuery(const ASTPtr & query_ptr_, ContextPtr context_)
+    : WithContext(context_)
+    , query_ptr(query_ptr_)
 {
 }
 
-
 BlockIO InterpreterCheckQuery::execute()
 {
     const auto & check = query_ptr->as<ASTCheckQuery &>();
-    auto table_id = getContext()->resolveStorageID(check, Context::ResolveOrdinary);
+    const auto & context = getContext();
+    auto table_id = context->resolveStorageID(check, Context::ResolveOrdinary);
 
-    getContext()->checkAccess(AccessType::SHOW_TABLES, table_id);
-    StoragePtr table = DatabaseCatalog::instance().getTable(table_id, getContext());
-    auto check_results = table->checkData(query_ptr, getContext());
+    context->checkAccess(AccessType::SHOW_TABLES, table_id);
+    StoragePtr table = DatabaseCatalog::instance().getTable(table_id, context);
 
-    Block block;
-    if (getContext()->getSettingsRef().check_query_single_value_result)
-    {
-        bool result = std::all_of(check_results.begin(), check_results.end(), [] (const CheckResult & res) { return res.success; });
-        auto column = ColumnUInt8::create();
-        column->insertValue(static_cast<UInt64>(result));
-        block = Block{{std::move(column), std::make_shared<DataTypeUInt8>(), "result"}};
-    }
-    else
-    {
-        auto block_structure = getBlockStructure();
-        auto path_column = block_structure[0].type->createColumn();
-        auto is_passed_column = block_structure[1].type->createColumn();
-        auto message_column = block_structure[2].type->createColumn();
+    auto check_data_tasks = table->getCheckTaskList(query_ptr, context);
 
-        for (const auto & check_result : check_results)
-        {
-            path_column->insert(check_result.fs_path);
-            is_passed_column->insert(static_cast<UInt8>(check_result.success));
-            message_column->insert(check_result.failure_message);
-        }
-
-        block = Block({
-            {std::move(path_column), block_structure[0].type, block_structure[0].name},
-            {std::move(is_passed_column), block_structure[1].type, block_structure[1].name},
-            {std::move(message_column), block_structure[2].type, block_structure[2].name}});
-    }
+    const auto & settings = context->getSettingsRef();
 
     BlockIO res;
-    res.pipeline = QueryPipeline(std::make_shared<SourceFromSingleChunk>(std::move(block)));
+    {
+        auto processors = std::make_shared<Processors>();
 
+        std::vector<OutputPort *> worker_ports;
+
+        size_t num_streams = std::max<size_t>(1, settings.max_threads);
+
+        for (size_t i = 0; i < num_streams; ++i)
+        {
+            auto worker_processor = std::make_shared<TableCheckWorkerProcessor>(check_data_tasks, table);
+            if (i == 0)
+                worker_processor->addTotalRowsApprox(check_data_tasks->size());
+            worker_ports.emplace_back(&worker_processor->getPort());
+            processors->emplace_back(worker_processor);
+        }
+
+        OutputPort * resize_outport;
+        {
+            auto resize_processor = std::make_shared<ResizeProcessor>(getHeaderForCheckResult(), worker_ports.size(), 1);
+
+            auto & resize_inputs = resize_processor->getInputs();
+            auto resize_inport_it = resize_inputs.begin();
+            for (size_t i = 0; i < worker_ports.size(); ++i, ++resize_inport_it)
+                connect(*worker_ports[i], *resize_inport_it);
+
+            resize_outport = &resize_processor->getOutputs().front();
+            processors->emplace_back(resize_processor);
+        }
+
+        if (settings.check_query_single_value_result)
+        {
+            auto emitter_processor = std::make_shared<TableCheckResultEmitter>();
+            auto * input_port = &emitter_processor->getInputPort();
+            processors->emplace_back(emitter_processor);
+
+            connect(*resize_outport, *input_port);
+        }
+
+        res.pipeline = QueryPipeline(Pipe(std::move(processors)));
+        res.pipeline.setNumThreads(num_streams);
+    }
     return res;
 }
 

src/Storages/CheckResults.h

@@ -22,6 +22,4 @@ struct CheckResult
     {}
 };
 
-using CheckResults = std::vector<CheckResult>;
-
 }

src/Storages/IStorage.cpp

@@ -276,6 +276,17 @@ bool IStorage::isStaticStorage() const
     return false;
 }
 
+IStorage::DataValidationTasksPtr IStorage::getCheckTaskList(const ASTPtr & /* query */, ContextPtr /* context */)
+{
+    throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Check query is not supported for {} storage", getName());
+}
+
+CheckResult IStorage::checkDataNext(DataValidationTasksPtr & /* check_task_list */, bool & has_nothing_to_do)
+{
+    has_nothing_to_do = true;
+    return {};
+}
+
 void IStorage::adjustCreateQueryForBackup(ASTPtr &) const
 {
 }

src/Storages/IStorage.h

@@ -600,8 +600,45 @@ public:
     /// Provides a hint that the storage engine may evaluate the IN-condition by using an index.
     virtual bool mayBenefitFromIndexForIn(const ASTPtr & /* left_in_operand */, ContextPtr /* query_context */, const StorageMetadataPtr & /* metadata_snapshot */) const { return false; }
 
-    /// Checks validity of the data
-    virtual CheckResults checkData(const ASTPtr & /* query */, ContextPtr /* context */) { throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Check query is not supported for {} storage", getName()); }
+
+    /** A list of tasks to check a validity of data.
+      * Each IStorage implementation may interpret this task in its own way.
+      * E.g. for some storages it to check data it need to check a list of files in filesystem, for others it can be a list of parts.
+      * Also it may hold resources (e.g. locks) required during check.
+      */
+    struct DataValidationTasksBase
+    {
+        /// Number of entries left to check.
+        /// It decreases after each call to checkDataNext().
+        virtual size_t size() const = 0;
+        virtual ~DataValidationTasksBase() = default;
+    };
+
+    using DataValidationTasksPtr = std::shared_ptr<DataValidationTasksBase>;
+
+    virtual DataValidationTasksPtr getCheckTaskList(const ASTPtr & /* query */, ContextPtr /* context */);
+
+    /** Executes one task from the list.
+      * If no tasks left, sets has_nothing_to_do to true.
+      * Note: Function `checkDataNext` is accessing `check_task_list` thread-safely,
+      *   and can be called simultaneously for the same `getCheckTaskList` result
+      *   to process different tasks in parallel.
+      * Usage:
+      *
+      * auto check_task_list = storage.getCheckTaskList(query, context);
+      * size_t total_tasks = check_task_list->size();
+      * while (true)
+      * {
+      *     size_t tasks_left = check_task_list->size();
+      *     std::cout << "Checking data: " << (total_tasks - tasks_left) << " / " << total_tasks << " tasks done." << std::endl;
+      *     bool has_nothing_to_do = false;
+      *     auto result = storage.checkDataNext(check_task_list, has_nothing_to_do);
+      *     if (has_nothing_to_do)
+      *         break;
+      *     doSomething(result);
+      * }
+      */
+    virtual CheckResult checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do);
 
     /// Checks that table could be dropped right now
     /// Otherwise - throws an exception with detailed information.

src/Storages/StorageLog.cpp

@@ -874,15 +874,18 @@ SinkToStoragePtr StorageLog::write(const ASTPtr & /*query*/, const StorageMetada
     return std::make_shared<LogSink>(*this, metadata_snapshot, std::move(lock));
 }
 
-CheckResults StorageLog::checkData(const ASTPtr & /* query */, ContextPtr local_context)
+IStorage::DataValidationTasksPtr StorageLog::getCheckTaskList(const ASTPtr & /* query */, ContextPtr local_context)
 {
     ReadLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception(ErrorCodes::TIMEOUT_EXCEEDED, "Lock timeout exceeded");
-
-    return file_checker.check();
+    return std::make_unique<DataValidationTasks>(file_checker.getDataValidationTasks(), std::move(lock));
 }
 
+CheckResult StorageLog::checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do)
+{
+    return file_checker.checkNextEntry(assert_cast<DataValidationTasks *>(check_task_list.get())->file_checker_tasks, has_nothing_to_do);
+}
 
 IStorage::ColumnSizeByName StorageLog::getColumnSizes() const
 {

src/Storages/StorageLog.h

@@ -59,7 +59,8 @@ public:
 
     void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;
 
-    CheckResults checkData(const ASTPtr & query, ContextPtr local_context) override;
+    DataValidationTasksPtr getCheckTaskList(const ASTPtr & query, ContextPtr context) override;
+    CheckResult checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do) override;
 
     void truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr, TableExclusiveLockHolder &) override;
 
@@ -142,6 +143,19 @@ private:
     std::atomic<UInt64> total_rows = 0;
     std::atomic<UInt64> total_bytes = 0;
 
+    struct DataValidationTasks : public IStorage::DataValidationTasksBase
+    {
+        DataValidationTasks(FileChecker::DataValidationTasksPtr file_checker_tasks_, ReadLock && lock_)
+            : file_checker_tasks(std::move(file_checker_tasks_)), lock(std::move(lock_))
+        {}
+
+        size_t size() const override { return file_checker_tasks->size(); }
+
+        FileChecker::DataValidationTasksPtr file_checker_tasks;
+        /// Lock to prevent table modification while checking
+        ReadLock lock;
+    };
+
     FileChecker file_checker;
 
     const size_t max_compress_block_size;

src/Storages/StorageMergeTree.cpp

@@ -2200,9 +2200,8 @@ void StorageMergeTree::onActionLockRemove(StorageActionBlockType action_type)
         background_moves_assignee.trigger();
 }
 
-CheckResults StorageMergeTree::checkData(const ASTPtr & query, ContextPtr local_context)
+IStorage::DataValidationTasksPtr StorageMergeTree::getCheckTaskList(const ASTPtr & query, ContextPtr local_context)
 {
-    CheckResults results;
     DataPartsVector data_parts;
     if (const auto & check_query = query->as<ASTCheckQuery &>(); check_query.partition)
     {
@@ -2212,7 +2211,14 @@ CheckResults StorageMergeTree::checkData(const ASTPtr & query, ContextPtr local_
     else
         data_parts = getVisibleDataPartsVector(local_context);
 
-    for (auto & part : data_parts)
+    return std::make_unique<DataValidationTasks>(std::move(data_parts), local_context);
+}
+
+CheckResult StorageMergeTree::checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do)
+{
+    auto * data_validation_tasks = assert_cast<DataValidationTasks *>(check_task_list.get());
+    auto local_context = data_validation_tasks->context;
+    if (auto part = data_validation_tasks->next())
     {
         /// If the checksums file is not present, calculate the checksums and write them to disk.
         static constexpr auto checksums_path = "checksums.txt";
@@ -2226,7 +2232,7 @@ CheckResults StorageMergeTree::checkData(const ASTPtr & query, ContextPtr local_
                 auto & part_mutable = const_cast<IMergeTreeDataPart &>(*part);
                 part_mutable.writeChecksums(part->checksums, local_context->getWriteSettings());
 
-                results.emplace_back(part->name, true, "Checksums recounted and written to disk.");
+                return CheckResult(part->name, true, "Checksums recounted and written to disk.");
             }
             catch (...)
             {
@@ -2234,7 +2240,7 @@ CheckResults StorageMergeTree::checkData(const ASTPtr & query, ContextPtr local_
                     throw;
 
                 tryLogCurrentException(log, __PRETTY_FUNCTION__);
-                results.emplace_back(part->name, false, "Check of part finished with error: '" + getCurrentExceptionMessage(false) + "'");
+                return CheckResult(part->name, false, "Check of part finished with error: '" + getCurrentExceptionMessage(false) + "'");
             }
         }
         else
@@ -2242,18 +2248,22 @@ CheckResults StorageMergeTree::checkData(const ASTPtr & query, ContextPtr local_
             try
             {
                 checkDataPart(part, true);
-                results.emplace_back(part->name, true, "");
+                return CheckResult(part->name, true, "");
             }
             catch (...)
             {
                 if (isRetryableException(std::current_exception()))
                     throw;
 
-                results.emplace_back(part->name, false, getCurrentExceptionMessage(false));
+                return CheckResult(part->name, false, getCurrentExceptionMessage(false));
             }
         }
     }
-    return results;
+    else
+    {
+        has_nothing_to_do = true;
+        return {};
+    }
 }
 
 

src/Storages/StorageMergeTree.h

@@ -108,7 +108,8 @@ public:
 
     void onActionLockRemove(StorageActionBlockType action_type) override;
 
-    CheckResults checkData(const ASTPtr & query, ContextPtr context) override;
+    DataValidationTasksPtr getCheckTaskList(const ASTPtr & query, ContextPtr context) override;
+    CheckResult checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do) override;
 
     bool scheduleDataProcessingJob(BackgroundJobsAssignee & assignee) override;
 
@@ -278,6 +279,32 @@ private:
     friend class MergePlainMergeTreeTask;
     friend class MutatePlainMergeTreeTask;
 
+    struct DataValidationTasks : public IStorage::DataValidationTasksBase
+    {
+        DataValidationTasks(DataPartsVector && parts_, ContextPtr context_)
+            : parts(std::move(parts_)), it(parts.begin()), context(std::move(context_))
+        {}
+
+        DataPartPtr next()
+        {
+            std::lock_guard lock(mutex);
+            if (it == parts.end())
+                return nullptr;
+            return *(it++);
+        }
+
+        size_t size() const override
+        {
+            std::lock_guard lock(mutex);
+            return std::distance(it, parts.end());
+        }
+
+        mutable std::mutex mutex;
+        DataPartsVector parts;
+        DataPartsVector::const_iterator it;
+
+        ContextPtr context;
+    };
 
 protected:
     std::map<int64_t, MutationCommands> getAlterMutationCommandsForPart(const DataPartPtr & part) const override;

src/Storages/StorageProxy.h

@@ -149,8 +149,11 @@ public:
         return getNested()->mayBenefitFromIndexForIn(left_in_operand, query_context, metadata_snapshot);
     }
 
-    CheckResults checkData(const ASTPtr & query, ContextPtr context) override { return getNested()->checkData(query, context); }
+    DataValidationTasksPtr getCheckTaskList(const ASTPtr & query, ContextPtr context) override { return getNested()->getCheckTaskList(query, context); }
+    CheckResult checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do) override { return getNested()->checkDataNext(check_task_list, has_nothing_to_do); }
+
     void checkTableCanBeDropped([[ maybe_unused ]] ContextPtr query_context) const override { getNested()->checkTableCanBeDropped(query_context); }
+
     bool storesDataOnDisk() const override { return getNested()->storesDataOnDisk(); }
     Strings getDataPaths() const override { return getNested()->getDataPaths(); }
     StoragePolicyPtr getStoragePolicy() const override { return getNested()->getStoragePolicy(); }

src/Storages/StorageReplicatedMergeTree.cpp

@@ -8562,9 +8562,8 @@ void StorageReplicatedMergeTree::enqueuePartForCheck(const String & part_name, t
     part_check_thread.enqueuePart(part_name, delay_to_check_seconds);
 }
 
-CheckResults StorageReplicatedMergeTree::checkData(const ASTPtr & query, ContextPtr local_context)
+IStorage::DataValidationTasksPtr StorageReplicatedMergeTree::getCheckTaskList(const ASTPtr & query, ContextPtr local_context)
 {
-    CheckResults results;
     DataPartsVector data_parts;
     if (const auto & check_query = query->as<ASTCheckQuery &>(); check_query.partition)
     {
@@ -8574,24 +8573,30 @@ CheckResults StorageReplicatedMergeTree::checkData(const ASTPtr & query, Context
     else
         data_parts = getVisibleDataPartsVector(local_context);
 
-    {
-        auto part_check_lock = part_check_thread.pausePartsCheck();
+    auto part_check_lock = part_check_thread.pausePartsCheck();
+    return std::make_unique<DataValidationTasks>(std::move(data_parts), std::move(part_check_lock));
+}
 
-        for (auto & part : data_parts)
+CheckResult StorageReplicatedMergeTree::checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do)
+{
+
+    if (auto part = assert_cast<DataValidationTasks *>(check_task_list.get())->next())
+    {
+        try
         {
-            try
-            {
-                results.push_back(part_check_thread.checkPartAndFix(part->name));
-            }
-            catch (const Exception & ex)
-            {
-                tryLogCurrentException(log, __PRETTY_FUNCTION__);
-                results.emplace_back(part->name, false, "Check of part finished with error: '" + ex.message() + "'");
-            }
+            return CheckResult(part_check_thread.checkPartAndFix(part->name));
+        }
+        catch (const Exception & ex)
+        {
+            tryLogCurrentException(log, __PRETTY_FUNCTION__);
+            return CheckResult(part->name, false, "Check of part finished with error: '" + ex.message() + "'");
         }
     }
-
-    return results;
+    else
+    {
+        has_nothing_to_do = true;
+        return {};
+    }
 }
 
 

src/Storages/StorageReplicatedMergeTree.h

@@ -230,7 +230,8 @@ public:
     /// Add a part to the queue of parts whose data you want to check in the background thread.
     void enqueuePartForCheck(const String & part_name, time_t delay_to_check_seconds = 0);
 
-    CheckResults checkData(const ASTPtr & query, ContextPtr context) override;
+    DataValidationTasksPtr getCheckTaskList(const ASTPtr & query, ContextPtr context) override;
+    CheckResult checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do) override;
 
     /// Checks ability to use granularity
     bool canUseAdaptiveGranularity() const override;
@@ -995,6 +996,34 @@ private:
     bool waitZeroCopyLockToDisappear(const ZeroCopyLock & lock, size_t milliseconds_to_wait) override;
 
     void startupImpl(bool from_attach_thread);
+
+    struct DataValidationTasks : public IStorage::DataValidationTasksBase
+    {
+        explicit DataValidationTasks(DataPartsVector && parts_, std::unique_lock<std::mutex> && parts_check_lock_)
+            : parts_check_lock(std::move(parts_check_lock_)), parts(std::move(parts_)), it(parts.begin())
+        {}
+
+        DataPartPtr next()
+        {
+            std::lock_guard lock(mutex);
+            if (it == parts.end())
+                return nullptr;
+            return *(it++);
+        }
+
+        size_t size() const override
+        {
+            std::lock_guard lock(mutex);
+            return std::distance(it, parts.end());
+        }
+
+        std::unique_lock<std::mutex> parts_check_lock;
+
+        mutable std::mutex mutex;
+        DataPartsVector parts;
+        DataPartsVector::const_iterator it;
+    };
+
 };
 
 String getPartNamePossiblyFake(MergeTreeDataFormatVersion format_version, const MergeTreePartInfo & part_info);

src/Storages/StorageStripeLog.cpp

@@ -403,16 +403,18 @@ SinkToStoragePtr StorageStripeLog::write(const ASTPtr & /*query*/, const Storage
     return std::make_shared<StripeLogSink>(*this, metadata_snapshot, std::move(lock));
 }
 
-
-CheckResults StorageStripeLog::checkData(const ASTPtr & /* query */, ContextPtr local_context)
+IStorage::DataValidationTasksPtr StorageStripeLog::getCheckTaskList(const ASTPtr & /* query */, ContextPtr local_context)
 {
     ReadLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception(ErrorCodes::TIMEOUT_EXCEEDED, "Lock timeout exceeded");
-
-    return file_checker.check();
+    return std::make_unique<DataValidationTasks>(file_checker.getDataValidationTasks(), std::move(lock));
 }
 
+CheckResult StorageStripeLog::checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do)
+{
+    return file_checker.checkNextEntry(assert_cast<DataValidationTasks *>(check_task_list.get())->file_checker_tasks, has_nothing_to_do);
+}
 
 void StorageStripeLog::truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr, TableExclusiveLockHolder &)
 {

src/Storages/StorageStripeLog.h

@@ -53,7 +53,8 @@ public:
 
     void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;
 
-    CheckResults checkData(const ASTPtr & query, ContextPtr ocal_context) override;
+    DataValidationTasksPtr getCheckTaskList(const ASTPtr & query, ContextPtr context) override;
+    CheckResult checkDataNext(DataValidationTasksPtr & check_task_list, bool & has_nothing_to_do) override;
 
     bool storesDataOnDisk() const override { return true; }
     Strings getDataPaths() const override { return {DB::fullPath(disk, table_path)}; }
@@ -93,6 +94,20 @@ private:
     const DiskPtr disk;
     String table_path;
 
+    struct DataValidationTasks : public IStorage::DataValidationTasksBase
+    {
+        DataValidationTasks(FileChecker::DataValidationTasksPtr file_checker_tasks_, ReadLock && lock_)
+            : file_checker_tasks(std::move(file_checker_tasks_)), lock(std::move(lock_))
+        {}
+
+        size_t size() const override { return file_checker_tasks->size(); }
+
+        FileChecker::DataValidationTasksPtr file_checker_tasks;
+
+        /// Lock to prevent table modification while checking
+        ReadLock lock;
+    };
+
     String data_file_path;
     String index_file_path;
     FileChecker file_checker;

tests/queries/0_stateless/00063_check_query.reference

@@ -1,2 +1,3 @@
 1
 1
+1

tests/queries/0_stateless/00063_check_query.sql

@@ -8,6 +8,9 @@ INSERT INTO check_query_tiny_log VALUES (1, 'A'), (2, 'B'), (3, 'C');
 
 CHECK TABLE check_query_tiny_log;
 
+-- Settings and FORMAT are supported
+CHECK TABLE check_query_tiny_log SETTINGS max_threads = 16;
+CHECK TABLE check_query_tiny_log FORMAT Null SETTINGS max_threads = 8, check_query_single_value_result = 0;
 
 DROP TABLE IF EXISTS check_query_log;
 

tests/queries/0_stateless/00961_check_table.sql

@@ -3,29 +3,29 @@ DROP TABLE IF EXISTS mt_table;
 
 CREATE TABLE mt_table (d Date, key UInt64, data String) ENGINE = MergeTree() PARTITION BY toYYYYMM(d) ORDER BY key;
 
-CHECK TABLE mt_table;
+CHECK TABLE mt_table SETTINGS max_threads = 1;
 
 INSERT INTO mt_table VALUES (toDate('2019-01-02'), 1, 'Hello'), (toDate('2019-01-02'), 2, 'World');
 
-CHECK TABLE mt_table;
+CHECK TABLE mt_table SETTINGS max_threads = 1;
 
 INSERT INTO mt_table VALUES (toDate('2019-01-02'), 3, 'quick'), (toDate('2019-01-02'), 4, 'brown');
 
 SELECT '========';
 
-CHECK TABLE mt_table;
+CHECK TABLE mt_table SETTINGS max_threads = 1;
 
 OPTIMIZE TABLE mt_table FINAL;
 
 SELECT '========';
 
-CHECK TABLE mt_table;
+CHECK TABLE mt_table SETTINGS max_threads = 1;
 
 SELECT '========';
 
 INSERT INTO mt_table VALUES (toDate('2019-02-03'), 5, '!'), (toDate('2019-02-03'), 6, '?');
 
-CHECK TABLE mt_table;
+CHECK TABLE mt_table SETTINGS max_threads = 1;
 
 SELECT '========';
 
@@ -33,6 +33,6 @@ INSERT INTO mt_table VALUES (toDate('2019-02-03'), 7, 'jump'), (toDate('2019-02-
 
 OPTIMIZE TABLE mt_table FINAL;
 
-CHECK TABLE mt_table PARTITION 201902;
+CHECK TABLE mt_table PARTITION 201902 SETTINGS max_threads = 1;
 
 DROP TABLE IF EXISTS mt_table;

tests/queries/0_stateless/01042_check_query_and_last_granule_size.sql

@@ -7,11 +7,11 @@ CREATE TABLE check_query_test (SomeKey UInt64, SomeValue String) ENGINE = MergeT
 -- Rows in this table are short, so granularity will be 8192.
 INSERT INTO check_query_test SELECT number, toString(number) FROM system.numbers LIMIT 81920;
 
-CHECK TABLE check_query_test;
+CHECK TABLE check_query_test SETTINGS max_threads = 1;
 
 OPTIMIZE TABLE check_query_test;
 
-CHECK TABLE check_query_test;
+CHECK TABLE check_query_test SETTINGS max_threads = 1;
 
 DROP TABLE IF EXISTS check_query_test;
 
@@ -21,18 +21,18 @@ CREATE TABLE check_query_test_non_adaptive (SomeKey UInt64, SomeValue String) EN
 
 INSERT INTO check_query_test_non_adaptive SELECT number, toString(number) FROM system.numbers LIMIT 81920;
 
-CHECK TABLE check_query_test_non_adaptive;
+CHECK TABLE check_query_test_non_adaptive SETTINGS max_threads = 1;
 
 OPTIMIZE TABLE check_query_test_non_adaptive;
 
-CHECK TABLE check_query_test_non_adaptive;
+CHECK TABLE check_query_test_non_adaptive SETTINGS max_threads = 1;
 
 INSERT INTO check_query_test_non_adaptive SELECT number, toString(number) FROM system.numbers LIMIT 77;
 
-CHECK TABLE check_query_test_non_adaptive;
+CHECK TABLE check_query_test_non_adaptive SETTINGS max_threads = 1;
 
 OPTIMIZE TABLE check_query_test_non_adaptive;
 
-CHECK TABLE check_query_test_non_adaptive;
+CHECK TABLE check_query_test_non_adaptive SETTINGS max_threads = 1;
 
 DROP TABLE IF EXISTS check_query_test_non_adaptive;

tests/queries/0_stateless/02235_check_table_sparse_serialization.sql

@@ -12,7 +12,6 @@ SELECT name, column, serialization_kind FROM system.parts_columns
 WHERE database = currentDatabase() AND table = 't_sparse_02235'
 ORDER BY name, column;
 
-SET check_query_single_value_result = 0;
-CHECK TABLE t_sparse_02235;
+CHECK TABLE t_sparse_02235 SETTINGS check_query_single_value_result = 0, max_threads = 1;
 
 DROP TABLE t_sparse_02235;

tests/queries/0_stateless/02841_check_table_progress.reference

@@ -0,0 +1,2 @@
+Ok
+Ok

tests/queries/0_stateless/02841_check_table_progress.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS t0";
+${CLICKHOUSE_CLIENT} -q "CREATE TABLE t0 (x UInt64, val String) ENGINE = MergeTree ORDER BY x PARTITION BY x % 100";
+${CLICKHOUSE_CLIENT} -q "INSERT INTO t0 SELECT sipHash64(number), randomPrintableASCII(1000) FROM numbers(1000)";
+
+
+# Check that we have at least 3 different values for read_rows
+UNIQUE_VALUES=$(
+    ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&send_progress_in_http_headers=1&http_headers_progress_interval_ms=0" -d @- <<< "CHECK TABLE t0" -v |& {
+        grep -F -e X-ClickHouse-Progress: -e X-ClickHouse-Summary:  | grep -o '"read_rows"\s*:\s*"[0-9]*"'
+    } | uniq | wc -l
+)
+
+[ "$UNIQUE_VALUES" -ge "3" ] && echo "Ok" || echo "Fail: got $UNIQUE_VALUES"
+
+
+# Check that we have we have at least 100 total_rows_to_read (at least one check task per partition)
+MAX_TOTAL_VALUE=$(
+    ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&send_progress_in_http_headers=1&http_headers_progress_interval_ms=0" -d @- <<< "CHECK TABLE t0" -v |& {
+        grep -F -e X-ClickHouse-Progress: -e X-ClickHouse-Summary:  | grep -o '"total_rows_to_read"\s*:\s*"[0-9]*"' | grep -o '[0-9]*'
+    } | sort -n | tail -1
+)
+
+[ "$MAX_TOTAL_VALUE" -ge "100" ] && echo "Ok" || echo "Fail: got $MAX_TOTAL_VALUE"