Add allow_distributed_ddl_queries option to the cluster config

docs/en/engines/table-engines/special/distributed.md

@@ -141,6 +141,10 @@ Clusters are configured in the [server configuration file](../../../operations/c
              be used as current user for the query.
         -->
         <!-- <secret></secret> -->
+        
+        <!-- Optional. Whether distributed DDL queries (ON CLUSTER clause) are allowed for this cluster. Default: true (allowed). -->        
+        <!-- <allow_distributed_ddl_queries>true</allow_distributed_ddl_queries> -->
+        
         <shard>
             <!-- Optional. Shard weight when writing data. Default: 1. -->
             <weight>1</weight>

src/Interpreters/Cluster.cpp

@@ -396,6 +396,9 @@ Cluster::Cluster(const Poco::Util::AbstractConfiguration & config,
     secret = config.getString(config_prefix + "secret", "");
     boost::range::remove_erase(config_keys, "secret");
 
+    allow_distributed_ddl_queries = config.getBool(config_prefix + "allow_distributed_ddl_queries", true);
+    boost::range::remove_erase(config_keys, "allow_distributed_ddl_queries");
+
     if (config_keys.empty())
         throw Exception(ErrorCodes::SHARD_HAS_NO_CONNECTIONS, "No cluster elements (shard, node) specified in config at path {}", config_prefix);
 

src/Interpreters/Cluster.h

@@ -256,6 +256,9 @@ public:
     /// NOTE: true does not mean, that it's actually a cross-replication cluster.
     bool maybeCrossReplication() const;
 
+    /// Are distributed DDL Queries (ON CLUSTER Clause) allowed for this cluster
+    bool areDistributedDDLQueriesAllowed() const { return allow_distributed_ddl_queries; }
+
 private:
     SlotToShard slot_to_shard;
 
@@ -287,6 +290,8 @@ private:
     /// An array of shards. For each shard, an array of replica addresses (servers that are considered identical).
     AddressesWithFailover addresses_with_failover;
 
+    bool allow_distributed_ddl_queries = true;
+
     size_t remote_shard_count = 0;
     size_t local_shard_count = 0;
 

src/Interpreters/executeDDLQueryOnCluster.cpp

@@ -92,6 +92,9 @@ BlockIO executeDDLQueryOnCluster(const ASTPtr & query_ptr_, ContextPtr context,
 
     span.addAttribute("clickhouse.cluster", query->cluster);
 
+    if (!cluster->areDistributedDDLQueriesAllowed())
+        throw Exception(ErrorCodes::QUERY_IS_PROHIBITED, "Distributed DDL queries are prohibited for the cluster");
+
     /// TODO: support per-cluster grant
     context->checkAccess(AccessType::CLUSTER);
 

tests/integration/test_distributed_ddl/configs/config.d/clusters.xml

@@ -89,5 +89,16 @@
         </shard>
     </cluster_no_replicas>
 
+    <!-- Cluster with disabled distributed DDL queries -->
+    <cluster_disabled_ddl>
+        <allow_distributed_ddl_queries>false</allow_distributed_ddl_queries>
+        <shard>
+            <replica>
+                <host>ch1</host>
+                <port>9000</port>
+            </replica>
+        </shard>
+    </cluster_disabled_ddl>
+
 </remote_servers>
-</clickhouse>
+</clickhouse>

tests/integration/test_distributed_ddl/configs_secure/config.d/clusters.xml

@@ -102,5 +102,16 @@
         </shard>
     </cluster_no_replicas>
 
+    <!-- Cluster with disabled distributed DDL queries -->
+    <cluster_disabled_ddl>
+        <allow_distributed_ddl_queries>false</allow_distributed_ddl_queries>
+        <shard>
+            <replica>
+                <host>ch1</host>
+                <port>9000</port>
+            </replica>
+        </shard>
+    </cluster_disabled_ddl>
+
 </remote_servers>
 </clickhouse>

tests/integration/test_distributed_ddl/test.py

@@ -585,6 +585,17 @@ def test_replicated_without_arguments(test_cluster):
     test_cluster.pm_random_drops.push_rules(rules)
 
 
+def test_disabled_distributed_ddl(test_cluster):
+    instance = test_cluster.instances["ch1"]
+
+    assert (
+        "Distributed DDL queries are prohibited for the cluster"
+        in instance.query_and_get_error(
+            "CREATE DATABASE IF NOT EXISTS test ON CLUSTER 'cluster_disabled_ddl'"
+        )
+    )
+
+
 if __name__ == "__main__":
     with contextmanager(test_cluster)() as ctx_cluster:
         for name, instance in list(ctx_cluster.instances.items()):