Integration test for Distributed over Distributed (from #8640)

tests/integration/test_distributed_over_distributed/configs/remote_servers.xml

@@ -0,0 +1,18 @@
+<yandex>
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+            <shard>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_cluster>
+    </remote_servers>
+</yandex>

tests/integration/test_distributed_over_distributed/configs/set_distributed_defaults.xml

@@ -0,0 +1,35 @@
+<yandex>
+    <profiles>
+        <default>
+            <connections_with_failover_max_tries>3</connections_with_failover_max_tries>
+            <connect_timeout_with_failover_ms>1000</connect_timeout_with_failover_ms>
+            <min_insert_block_size_rows>1</min_insert_block_size_rows>
+        </default>
+        <delays>
+            <connections_with_failover_max_tries>5</connections_with_failover_max_tries>
+            <connect_timeout_with_failover_ms>3000</connect_timeout_with_failover_ms>
+            <min_insert_block_size_rows>1</min_insert_block_size_rows>
+        </delays>
+    </profiles>
+
+    <users>
+        <default>
+            <password></password>
+            <networks incl="networks" replace="replace">
+                <ip>::/0</ip>
+            </networks>
+            <profile>default</profile>
+            <quota>default</quota>
+        </default>
+        <ready_to_wait>
+            <password></password>
+            <networks incl="networks" replace="replace">
+                <ip>::/0</ip>
+            </networks>
+            <profile>delays</profile>
+            <quota>default</quota>
+        </ready_to_wait>
+    </users>
+
+    <quotas><default></default></quotas>
+</yandex>

tests/integration/test_distributed_over_distributed/test.py

@@ -0,0 +1,98 @@
+# This test is a subset of the 01223_dist_on_dist.
+# (just in case, with real separate instances).
+
+from __future__ import print_function
+
+import itertools
+import timeit
+import logging
+
+import pytest
+
+from helpers.cluster import ClickHouseCluster
+from helpers.network import PartitionManager
+from helpers.test_tools import TSV
+
+
+cluster = ClickHouseCluster(__file__)
+
+NODES = {'node' + str(i): cluster.add_instance(
+    'node' + str(i),
+    main_configs=['configs/remote_servers.xml'],
+    user_configs=['configs/set_distributed_defaults.xml'],
+) for i in (1, 2)}
+
+CREATE_TABLES_SQL = '''
+CREATE TABLE
+    base_table(
+        node String,
+        key Int32,
+        value Int32
+    )
+ENGINE = Memory;
+
+CREATE TABLE
+    distributed_table
+AS base_table
+ENGINE = Distributed(test_cluster, default, base_table);
+
+CREATE TABLE
+    distributed_over_distributed_table
+AS distributed_table
+ENGINE = Distributed('test_cluster', default, distributed_table);
+'''
+
+INSERT_SQL_TEMPLATE = "INSERT INTO base_table VALUES ('{node_id}', {key}, {value})"
+
+@pytest.fixture(scope="session")
+def started_cluster():
+    try:
+        cluster.start()
+        for node_index, (node_name, node) in enumerate(NODES.items()):
+            node.query(CREATE_TABLES_SQL)
+            for i in range(0, 2):
+                node.query(INSERT_SQL_TEMPLATE.format(node_id=node_name, key=i, value=i + (node_index * 10)))
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+@pytest.mark.parametrize("node", NODES.values())
+@pytest.mark.parametrize("source", ["distributed_over_distributed_table", "cluster('test_cluster', default, distributed_table)"])
+class TestDistributedOverDistributedSuite:
+    def test_select_with_order_by_node(self, started_cluster, node, source):
+        assert node.query("SELECT * FROM {source} ORDER BY node, key".format(source=source)) \
+            == """node1	0	0
+node1	0	0
+node1	1	1
+node1	1	1
+node2	0	10
+node2	0	10
+node2	1	11
+node2	1	11
+"""
+
+    def test_select_with_order_by_key(self, started_cluster, node, source):
+        assert node.query("SELECT * FROM {source} ORDER BY key, node".format(source=source)) \
+            == """node1	0	0
+node1	0	0
+node2	0	10
+node2	0	10
+node1	1	1
+node1	1	1
+node2	1	11
+node2	1	11
+"""
+
+    def test_select_with_group_by_node(self, started_cluster, node, source):
+        assert node.query("SELECT node, SUM(value) FROM {source} GROUP BY node ORDER BY node".format(source=source)) \
+            == "node1	2\nnode2	42\n"
+
+    def test_select_with_group_by_key(self, started_cluster, node, source):
+        assert node.query("SELECT key, SUM(value) FROM {source} GROUP BY key ORDER BY key".format(source=source)) \
+            == "0	20\n1	24\n"
+
+    def test_select_sum(self, started_cluster, node, source):
+        assert node.query("SELECT SUM(value) FROM {source}".format(source=source)) \
+            == "44\n"