import time import pytest from helpers.client import QueryRuntimeException from helpers.cluster import ClickHouseCluster from helpers.network import PartitionManager from helpers.test_tools import TSV cluster = ClickHouseCluster(__file__) instance_test_reconnect = cluster.add_instance( "instance_test_reconnect", main_configs=["configs/remote_servers.xml"] ) instance_test_inserts_batching = cluster.add_instance( "instance_test_inserts_batching", main_configs=["configs/remote_servers.xml"], user_configs=["configs/enable_distributed_inserts_batching.xml"], ) remote = cluster.add_instance( "remote", main_configs=["configs/forbid_background_merges.xml"] ) instance_test_inserts_local_cluster = cluster.add_instance( "instance_test_inserts_local_cluster", main_configs=["configs/remote_servers.xml"] ) node1 = cluster.add_instance( "node1", main_configs=["configs/remote_servers.xml"], with_zookeeper=True ) node2 = cluster.add_instance( "node2", main_configs=["configs/remote_servers.xml"], with_zookeeper=True ) shard1 = cluster.add_instance( "shard1", main_configs=["configs/remote_servers.xml"], with_zookeeper=True ) shard2 = cluster.add_instance( "shard2", main_configs=["configs/remote_servers.xml"], with_zookeeper=True ) @pytest.fixture(scope="module") def started_cluster(): try: cluster.start() remote.query("CREATE TABLE local1 (x UInt32) ENGINE = Log") instance_test_reconnect.query( """ CREATE TABLE distributed (x UInt32) ENGINE = Distributed('test_cluster', 'default', 'local1') """ ) remote.query( "CREATE TABLE local2 (d Date, x UInt32, s String) ENGINE = MergeTree PARTITION BY toYYYYMM(d) ORDER BY x" ) instance_test_inserts_batching.query( """ CREATE TABLE distributed (d Date, x UInt32) ENGINE = Distributed('test_cluster', 'default', 'local2') SETTINGS fsync_after_insert=1, fsync_directories=1 """ ) instance_test_inserts_local_cluster.query( "CREATE TABLE local (d Date, x UInt32) ENGINE = MergeTree PARTITION BY toYYYYMM(d) ORDER BY x" ) instance_test_inserts_local_cluster.query( """ CREATE TABLE distributed_on_local (d Date, x UInt32) ENGINE = Distributed('test_local_cluster', 'default', 'local') """ ) node1.query( """ CREATE TABLE replicated(date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/0/replicated', 'node1') PARTITION BY toYYYYMM(date) ORDER BY id """ ) node2.query( """ CREATE TABLE replicated(date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/0/replicated', 'node2') PARTITION BY toYYYYMM(date) ORDER BY id """ ) node1.query( """ CREATE TABLE distributed (date Date, id UInt32) ENGINE = Distributed('shard_with_local_replica', 'default', 'replicated') """ ) node2.query( """ CREATE TABLE distributed (date Date, id UInt32) ENGINE = Distributed('shard_with_local_replica', 'default', 'replicated') """ ) shard1.query( """ CREATE TABLE low_cardinality (d Date, x UInt32, s LowCardinality(String)) ENGINE = MergeTree PARTITION BY toYYYYMM(d) ORDER BY x""" ) shard2.query( """ CREATE TABLE low_cardinality (d Date, x UInt32, s LowCardinality(String)) ENGINE = MergeTree PARTITION BY toYYYYMM(d) ORDER BY x""" ) shard1.query( """ CREATE TABLE low_cardinality_all (d Date, x UInt32, s LowCardinality(String)) ENGINE = Distributed('shard_with_low_cardinality', 'default', 'low_cardinality', sipHash64(s))""" ) node1.query( """ CREATE TABLE table_function (n UInt8, s String) ENGINE = MergeTree() ORDER BY n""" ) node2.query( """ CREATE TABLE table_function (n UInt8, s String) ENGINE = MergeTree() ORDER BY n""" ) node1.query( """ CREATE TABLE distributed_one_replica_internal_replication (date Date, id UInt32) ENGINE = Distributed('shard_with_local_replica_internal_replication', 'default', 'single_replicated') """ ) node2.query( """ CREATE TABLE distributed_one_replica_internal_replication (date Date, id UInt32) ENGINE = Distributed('shard_with_local_replica_internal_replication', 'default', 'single_replicated') """ ) node1.query( """ CREATE TABLE distributed_one_replica_no_internal_replication (date Date, id UInt32) ENGINE = Distributed('shard_with_local_replica', 'default', 'single_replicated') """ ) node2.query( """ CREATE TABLE distributed_one_replica_no_internal_replication (date Date, id UInt32) ENGINE = Distributed('shard_with_local_replica', 'default', 'single_replicated') """ ) node2.query( """ CREATE TABLE single_replicated(date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/0/single_replicated', 'node2') PARTITION BY toYYYYMM(date) ORDER BY id """ ) yield cluster finally: cluster.shutdown() def test_reconnect(started_cluster): instance = instance_test_reconnect with PartitionManager() as pm: # Open a connection for insertion. instance.query("INSERT INTO distributed VALUES (1)") time.sleep(1) assert remote.query("SELECT count(*) FROM local1").strip() == "1" # Now break the connection. pm.partition_instances( instance, remote, action="REJECT --reject-with tcp-reset" ) instance.query("INSERT INTO distributed VALUES (2)") time.sleep(1) # Heal the partition and insert more data. # The connection must be reestablished and after some time all data must be inserted. pm.heal_all() time.sleep(1) instance.query("INSERT INTO distributed VALUES (3)") time.sleep(5) assert remote.query("SELECT count(*) FROM local1").strip() == "3" def test_inserts_batching(started_cluster): instance = instance_test_inserts_batching with PartitionManager() as pm: pm.partition_instances(instance, remote) instance.query("INSERT INTO distributed(d, x) VALUES ('2000-01-01', 1)") # Sleep a bit so that this INSERT forms a batch of its own. time.sleep(0.1) instance.query("INSERT INTO distributed(x, d) VALUES (2, '2000-01-01')") for i in range(3, 7): instance.query( "INSERT INTO distributed(d, x) VALUES ('2000-01-01', {})".format(i) ) for i in range(7, 9): instance.query( "INSERT INTO distributed(x, d) VALUES ({}, '2000-01-01')".format(i) ) instance.query("INSERT INTO distributed(d, x) VALUES ('2000-01-01', 9)") # After ALTER the structure of the saved blocks will be different instance.query("ALTER TABLE distributed ADD COLUMN s String") for i in range(10, 13): instance.query( "INSERT INTO distributed(d, x) VALUES ('2000-01-01', {})".format(i) ) instance.query("SYSTEM FLUSH DISTRIBUTED distributed") time.sleep(1.0) result = remote.query( "SELECT _part, groupArray(x) FROM local2 GROUP BY _part ORDER BY _part" ) # Explanation: as merges are turned off on remote instance, active parts in local2 table correspond 1-to-1 # to inserted blocks. # Batches of max 3 rows are formed as min_insert_block_size_rows = 3. # Blocks: # 1. Failed batch that is retried with the same contents. # 2. Full batch of inserts before ALTER. # 3. Full batch of inserts before ALTER. # 4. Full batch of inserts after ALTER (that have different block structure). # 5. What was left to insert with the column structure before ALTER. expected = """\ 200001_1_1_0\t[1] 200001_2_2_0\t[2,3,4] 200001_3_3_0\t[5,6,7] 200001_4_4_0\t[10,11,12] 200001_5_5_0\t[8,9] """ assert TSV(result) == TSV(expected) def test_inserts_local(started_cluster): instance = instance_test_inserts_local_cluster instance.query("INSERT INTO distributed_on_local VALUES ('2000-01-01', 1)") time.sleep(0.5) assert instance.query("SELECT count(*) FROM local").strip() == "1" def test_inserts_single_replica_local_internal_replication(started_cluster): with pytest.raises( QueryRuntimeException, match="Table default.single_replicated doesn't exist" ): node1.query( "INSERT INTO distributed_one_replica_internal_replication VALUES ('2000-01-01', 1)", settings={ "insert_distributed_sync": "1", "prefer_localhost_replica": "1", # to make the test more deterministic "load_balancing": "first_or_random", }, ) assert node2.query("SELECT count(*) FROM single_replicated").strip() == "0" def test_inserts_single_replica_internal_replication(started_cluster): try: node1.query( "INSERT INTO distributed_one_replica_internal_replication VALUES ('2000-01-01', 1)", settings={ "insert_distributed_sync": "1", "prefer_localhost_replica": "0", # to make the test more deterministic "load_balancing": "first_or_random", }, ) assert node2.query("SELECT count(*) FROM single_replicated").strip() == "1" finally: node2.query("TRUNCATE TABLE single_replicated") def test_inserts_single_replica_no_internal_replication(started_cluster): try: with pytest.raises( QueryRuntimeException, match="Table default.single_replicated doesn't exist" ): node1.query( "INSERT INTO distributed_one_replica_no_internal_replication VALUES ('2000-01-01', 1)", settings={ "insert_distributed_sync": "1", "prefer_localhost_replica": "0", }, ) assert node2.query("SELECT count(*) FROM single_replicated").strip() == "1" finally: node2.query("TRUNCATE TABLE single_replicated") def test_prefer_localhost_replica(started_cluster): test_query = "SELECT * FROM distributed ORDER BY id" node1.query("INSERT INTO distributed VALUES (toDate('2017-06-17'), 11)") node2.query("INSERT INTO distributed VALUES (toDate('2017-06-17'), 22)") time.sleep(1.0) expected_distributed = """\ 2017-06-17\t11 2017-06-17\t22 """ expected_from_node2 = """\ 2017-06-17\t11 2017-06-17\t22 2017-06-17\t44 """ expected_from_node1 = """\ 2017-06-17\t11 2017-06-17\t22 2017-06-17\t33 """ assert TSV(node1.query(test_query)) == TSV(expected_distributed) assert TSV(node2.query(test_query)) == TSV(expected_distributed) # Make replicas inconsistent by disabling merges and fetches # for possibility of determining to which replica the query was send node1.query("SYSTEM STOP MERGES") node1.query("SYSTEM STOP FETCHES") node2.query("SYSTEM STOP MERGES") node2.query("SYSTEM STOP FETCHES") node1.query("INSERT INTO replicated VALUES (toDate('2017-06-17'), 33)") node2.query("INSERT INTO replicated VALUES (toDate('2017-06-17'), 44)") time.sleep(1.0) # Query is sent to node2, as it local and prefer_localhost_replica=1 assert TSV(node2.query(test_query)) == TSV(expected_from_node2) # Now query is sent to node1, as it higher in order assert TSV( node2.query( test_query + " SETTINGS load_balancing='in_order', prefer_localhost_replica=0" ) ) == TSV(expected_from_node1) def test_inserts_low_cardinality(started_cluster): instance = shard1 instance.query( "INSERT INTO low_cardinality_all (d,x,s) VALUES ('2018-11-12',1,'123')" ) time.sleep(0.5) assert instance.query("SELECT count(*) FROM low_cardinality_all").strip() == "1" def test_table_function(started_cluster): node1.query( "insert into table function cluster('shard_with_local_replica', 'default', 'table_function') select number, concat('str_', toString(number)) from numbers(100000)" ) assert ( node1.query( "select count() from cluster('shard_with_local_replica', 'default', 'table_function')" ).rstrip() == "100000" )