Prevent conversion to Replicated if zookeeper path already exists

src/Databases/DatabaseOrdinary.cpp

@@ -44,6 +44,7 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
     extern const int UNKNOWN_DATABASE_ENGINE;
     extern const int NOT_IMPLEMENTED;
+    extern const int UNEXPECTED_NODE_IN_ZOOKEEPER;
 }
 
 static constexpr size_t METADATA_FILE_BUFFER_SIZE = 32768;
@@ -76,6 +77,20 @@ static void setReplicatedEngine(ASTCreateQuery * create_query, ContextPtr contex
     String replica_path = server_settings.default_replica_path;
     String replica_name = server_settings.default_replica_name;
 
+    /// Check that replica path doesn't exist
+    Macros::MacroExpansionInfo info;
+    StorageID table_id = StorageID(create_query->getDatabase(), create_query->getTable(), create_query->uuid);
+    info.table_id = table_id;
+    info.expand_special_macros_only = false;
+
+    String zookeeper_path = context->getMacros()->expand(replica_path, info);
+    if (context->getZooKeeper()->exists(zookeeper_path))
+        throw Exception(
+            ErrorCodes::UNEXPECTED_NODE_IN_ZOOKEEPER,
+            "Found existing ZooKeeper path {} while trying to convert table {} to replicated. Table will not be converted.",
+            zookeeper_path, backQuote(table_id.getFullTableName())
+        );
+
     auto args = std::make_shared<ASTExpressionList>();
     args->children.push_back(std::make_shared<ASTLiteral>(replica_path));
     args->children.push_back(std::make_shared<ASTLiteral>(replica_name));

tests/integration/test_modify_engine_on_restart/test_zk_path_exists.py

@@ -0,0 +1,69 @@
+import pytest
+from test_modify_engine_on_restart.common import (
+    get_table_path,
+    set_convert_flags,
+)
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+ch1 = cluster.add_instance(
+    "ch1",
+    main_configs=[
+        "configs/config.d/clusters.xml",
+        "configs/config.d/distributed_ddl.xml",
+    ],
+    with_zookeeper=True,
+    macros={"replica": "node1"},
+    stay_alive=True,
+)
+
+database_name = "modify_engine_zk_path"
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def q(node, query):
+    return node.query(database=database_name, sql=query)
+
+
+def test_modify_engine_fails_if_zk_path_exists(started_cluster):
+    ch1.query("CREATE DATABASE " + database_name)
+
+    q(
+        ch1,
+        "CREATE TABLE already_exists_1 ( A Int64, D Date, S String ) ENGINE MergeTree() PARTITION BY toYYYYMM(D) ORDER BY A;",
+    )
+    uuid = q(
+        ch1,
+        f"SELECT uuid FROM system.tables WHERE table = 'already_exists_1' and database = '{database_name}'",
+    ).strip("'[]\n")
+
+    q(
+        ch1,
+        f"CREATE TABLE already_exists_2 ( A Int64, D Date, S String ) ENGINE ReplicatedMergeTree('/clickhouse/tables/{uuid}/{{shard}}', 'node2') PARTITION BY toYYYYMM(D) ORDER BY A;",
+    )
+
+    set_convert_flags(ch1, database_name, ["already_exists_1"])
+
+    table_data_path = get_table_path(ch1, "already_exists_1", database_name)
+
+    ch1.stop_clickhouse()
+    ch1.start_clickhouse(start_wait_sec=120, expected_to_fail=True)
+
+    # Check if we can cancel convertation
+    ch1.exec_in_container(
+        [
+            "bash",
+            "-c",
+            f"rm {table_data_path}convert_to_replicated",
+        ]
+    )
+    ch1.start_clickhouse()