Dictionaries ddl on cluster

dbms/src/Interpreters/InterpreterCreateQuery.cpp

@@ -708,6 +708,9 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
 
 BlockIO InterpreterCreateQuery::createDictionary(ASTCreateQuery & create)
 {
+    if (!create.cluster.empty())
+        return executeDDLQueryOnCluster(query_ptr, context, {create.database});
+
     String dictionary_name = create.table;
 
     String database_name = !create.database.empty() ? create.database : context.getCurrentDatabase();

dbms/src/Parsers/ASTCreateQuery.cpp

@@ -234,14 +234,15 @@ void ASTCreateQuery::formatQueryImpl(const FormatSettings & settings, FormatStat
                 << (if_not_exists ? "IF NOT EXISTS " : "")
             << (settings.hilite ? hilite_none : "")
             << (!database.empty() ? backQuoteIfNeed(database) + "." : "") << backQuoteIfNeed(table);
-            formatOnCluster(settings);
+        formatOnCluster(settings);
     }
     else
     {
         /// Always DICTIONARY
         settings.ostr << (settings.hilite ? hilite_keyword : "") << (attach ? "ATTACH " : "CREATE ") << "DICTIONARY "
                       << (if_not_exists ? "IF NOT EXISTS " : "") << (settings.hilite ? hilite_none : "")
                       << (!database.empty() ? backQuoteIfNeed(database) + "." : "") << backQuoteIfNeed(table);
+        formatOnCluster(settings);
     }
 
     if (as_table_function)

dbms/src/Parsers/ParserCreateQuery.cpp

@@ -826,6 +826,7 @@ bool ParserCreateDictionaryQuery::parseImpl(IParser::Pos & pos, ASTPtr & node, E
     ParserKeyword s_attach("ATTACH");
     ParserKeyword s_dictionary("DICTIONARY");
     ParserKeyword s_if_not_exists("IF NOT EXISTS");
+    ParserKeyword s_on("ON");
     ParserIdentifier name_p;
     ParserToken s_left_paren(TokenType::OpeningRoundBracket);
     ParserToken s_right_paren(TokenType::ClosingRoundBracket);
@@ -840,6 +841,7 @@ bool ParserCreateDictionaryQuery::parseImpl(IParser::Pos & pos, ASTPtr & node, E
     ASTPtr name;
     ASTPtr attributes;
     ASTPtr dictionary;
+    String cluster_str;
 
     bool attach = false;
     if (!s_create.ignore(pos, expected))
@@ -866,6 +868,12 @@ bool ParserCreateDictionaryQuery::parseImpl(IParser::Pos & pos, ASTPtr & node, E
             return false;
     }
 
+    if (s_on.ignore(pos, expected))
+    {
+        if (!ASTQueryWithOnCluster::parse(pos, cluster_str, expected))
+            return false;
+    }
+
     if (!attach)
     {
         if (!s_left_paren.ignore(pos, expected))
@@ -894,6 +902,7 @@ bool ParserCreateDictionaryQuery::parseImpl(IParser::Pos & pos, ASTPtr & node, E
     query->if_not_exists = if_not_exists;
     query->set(query->dictionary_attributes_list, attributes);
     query->set(query->dictionary, dictionary);
+    query->cluster = cluster_str;
 
     return true;
 }

dbms/tests/integration/test_dictionary_ddl_on_cluster/configs/config.d/clusters.xml

@@ -0,0 +1,28 @@
+<yandex>
+<remote_servers>
+    <cluster>
+        <shard>
+            <internal_replication>true</internal_replication>
+            <replica>
+                <host>ch1</host>
+                <port>9000</port>
+            </replica>
+            <replica>
+                <host>ch2</host>
+                <port>9000</port>
+            </replica>
+        </shard>
+        <shard>
+            <internal_replication>true</internal_replication>
+            <replica>
+                <host>ch3</host>
+                <port>9000</port>
+            </replica>
+            <replica>
+                <host>ch4</host>
+                <port>9000</port>
+            </replica>
+        </shard>
+    </cluster>
+</remote_servers>
+</yandex>

dbms/tests/integration/test_dictionary_ddl_on_cluster/configs/config.d/ddl.xml

@@ -0,0 +1,8 @@
+<yandex>
+<distributed_ddl>
+    <path>/clickhouse/task_queue/ddl</path>
+    <max_tasks_in_queue>10</max_tasks_in_queue>
+    <task_max_lifetime>3600</task_max_lifetime>
+    <cleanup_delay_period>5</cleanup_delay_period>
+</distributed_ddl>
+</yandex>

dbms/tests/integration/test_dictionary_ddl_on_cluster/test.py

@@ -0,0 +1,62 @@
+import pytest
+from helpers.cluster import ClickHouseCluster
+from helpers.client import QueryRuntimeException
+
+cluster = ClickHouseCluster(__file__)
+ch1 = cluster.add_instance('ch1', config_dir="configs", with_zookeeper=True)
+ch2 = cluster.add_instance('ch2', config_dir="configs", with_zookeeper=True)
+ch3 = cluster.add_instance('ch3', config_dir="configs", with_zookeeper=True)
+ch4 = cluster.add_instance('ch4', config_dir="configs", with_zookeeper=True)
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+        ch1.query("CREATE TABLE sometbl ON CLUSTER 'cluster' (key UInt64, value String) ENGINE = MergeTree ORDER by key")
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def test_dictionary_ddl_on_cluster(started_cluster):
+    for node in [ch1, ch2, ch3, ch4]:
+        assert node.query("SELECT count() from sometbl") == "0\n"
+
+    for num, node in enumerate([ch1, ch2, ch3, ch4]):
+        node.query("insert into sometbl values ({}, '{}')".format(num, node.name))
+
+
+    ch1.query(
+        """
+        CREATE DICTIONARY somedict ON CLUSTER 'cluster' (
+            key UInt64,
+            value String
+        )
+        PRIMARY KEY key
+        LAYOUT(FLAT())
+        SOURCE(CLICKHOUSE(HOST 'localhost' PORT 9000 USER 'default' TABLE 'sometbl' DB 'default'))
+        LIFETIME(10)
+        """)
+
+    for num, node in enumerate([ch1, ch2, ch3, ch4]):
+        assert node.query("SELECT count() from sometbl") == "1\n"
+        assert node.query("SELECT dictGetString('default.somedict', 'value', toUInt64({}))".format(num)) == node.name + '\n'
+
+    ch1.query("DETACH DICTIONARY default.somedict ON CLUSTER 'cluster'")
+
+    for node in [ch1, ch2, ch3, ch4]:
+        with pytest.raises(QueryRuntimeException):
+            node.query("SELECT dictGetString('default.somedict', 'value', toUInt64(1))")
+
+    ch1.query("ATTACH DICTIONARY default.somedict ON CLUSTER 'cluster'")
+
+    for num, node in enumerate([ch1, ch2, ch3, ch4]):
+        assert node.query("SELECT count() from sometbl") == "1\n"
+        assert node.query("SELECT dictGetString('default.somedict', 'value', toUInt64({}))".format(num)) == node.name + '\n'
+
+    ch1.query("DROP DICTIONARY default.somedict ON CLUSTER 'cluster'")
+
+    for node in [ch1, ch2, ch3, ch4]:
+        with pytest.raises(QueryRuntimeException):
+            node.query("SELECT dictGetString('default.somedict', 'value', toUInt64(1))")