Repository: ClickHouse/ClickHouse
File: src/Databases/DatabaseReplicated.cpp (C++)

Code before:
{
    chassert(!ddl_worker);
    chassert(!is_probably_dropped.load());
    chassert(!ddl_worker_initialized.load());

    ddl_worker = std::make_unique<DatabaseReplicatedDDLWorker>(this, getContext());
    ddl_worker->startup();
    ddl_worker_initialized = true;
}

void DatabaseReplicated::restoreDatabaseNodesInKeeper(const ZooKeeperPtr & zookeeper)
{
    auto local_context = getContext();

    zookeeper->createAncestors(zookeeper_path);

    Coordination::Requests ops;
    auto add_ops = [&ops](Coordination::Requests && others)
    { ops.insert(ops.end(), std::make_move_iterator(others.begin()), std::make_move_iterator(others.end())); };

    add_ops(buildDatabaseNodesInZooKeeper());

    for (auto existing_tables_it = getTablesIterator(local_context, {}, /*skip_not_loaded=*/false); existing_tables_it->isValid();
         existing_tables_it->next())
    {
        const String table_name = existing_tables_it->name();
        LOG_TEST(log, "Restoring metadata in Keeper of table {}", table_name);

        const String statement = getObjectDefinitionFromCreateQuery(getCreateTableQuery(table_name, local_context));
        const String table_metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_name);
        add_ops({zkutil::makeCreateRequest(table_metadata_zk_path, statement, zkutil::CreateMode::Persistent)});
    }

    /// Compute the digest atomically from the current tables map, using the same method
    /// as checkDigestValid (readMetadataFile). Computing it outside the lock from the
    /// iterator above would race with concurrent DDL operations modifying the tables map.
    {
        std::lock_guard lock{metadata_mutex};
        UInt64 digest = 0;
        {
            std::lock_guard tables_lock{mutex};
            for (const auto & table : tables)
                digest += getMetadataHash(table.first);
        }
        tables_metadata_digest = digest;
    }
    Coordination::Responses responses;
    auto code = zookeeper->tryMulti(ops, responses);

    if (code == Coordination::Error::ZOK)
        return;

Review comment (suggestion):
We should move the `table_lock{mutex}` before `for (auto existing_tables_it = getTablesIterator(local_context, {}, /*skip_not_loaded=*/false); existing_tables_it->isValid();`, and calculate the digest inside the for-loop. So that the digest and keeper path nodes are consistent.

Code after:
{
    chassert(!ddl_worker);
    chassert(!is_probably_dropped.load());
    chassert(!ddl_worker_initialized.load());

    ddl_worker = std::make_unique<DatabaseReplicatedDDLWorker>(this, getContext());
    ddl_worker->startup();
    ddl_worker_initialized = true;
}

void DatabaseReplicated::restoreDatabaseNodesInKeeper(const ZooKeeperPtr & zookeeper)
{
    auto local_context = getContext();

    zookeeper->createAncestors(zookeeper_path);

    Coordination::Requests ops;
    auto add_ops = [&ops](Coordination::Requests && others)
    { ops.insert(ops.end(), std::make_move_iterator(others.begin()), std::make_move_iterator(others.end())); };

    add_ops(buildDatabaseNodesInZooKeeper());

    /// Hold metadata_mutex so that the digest and keeper path nodes are consistent:
    /// both are computed from the same set of tables under the same lock.
    {
        std::lock_guard lock{metadata_mutex};
        UInt64 digest = 0;

        for (auto existing_tables_it = getTablesIterator(local_context, {}, /*skip_not_loaded=*/false); existing_tables_it->isValid();
             existing_tables_it->next())
        {
            const String table_name = existing_tables_it->name();
            LOG_TEST(log, "Restoring metadata in Keeper of table {}", table_name);

            const String statement = readMetadataFile(table_name);
            const String table_metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_name);
            add_ops({zkutil::makeCreateRequest(table_metadata_zk_path, statement, zkutil::CreateMode::Persistent)});

            digest += DB::getMetadataHash(table_name, statement);
        }

        tables_metadata_digest = digest;
    }
    Coordination::Responses responses;
    auto code = zookeeper->tryMulti(ops, responses);

    if (code == Coordination::Error::ZOK)
        return;

    zkutil::KeeperMultiException::check(code, ops, responses);
    UNREACHABLE();