ClickHouse/tests/integration/test_part_moves_between_shards/test.py
2021-09-17 18:11:32 +01:00

489 lines
19 KiB
Python

import pytest
import random
import threading
import time
from helpers.client import QueryRuntimeException
from helpers.cluster import ClickHouseCluster
from helpers.test_tools import TSV
transient_ch_errors = [23, 32, 210]
cluster = ClickHouseCluster(__file__)
s0r0 = cluster.add_instance(
's0r0',
main_configs=['configs/remote_servers.xml', 'configs/merge_tree.xml'],
stay_alive=True,
with_zookeeper=True)
s0r1 = cluster.add_instance(
's0r1',
main_configs=['configs/remote_servers.xml', 'configs/merge_tree.xml'],
stay_alive=True,
with_zookeeper=True)
s1r0 = cluster.add_instance(
's1r0',
main_configs=['configs/remote_servers.xml', 'configs/merge_tree.xml'],
stay_alive=True,
with_zookeeper=True)
s1r1 = cluster.add_instance(
's1r1',
main_configs=['configs/remote_servers.xml', 'configs/merge_tree.xml'],
stay_alive=True,
with_zookeeper=True)
@pytest.fixture(scope="module")
def started_cluster():
try:
cluster.start()
yield cluster
finally:
cluster.shutdown()
def test_move(started_cluster):
for shard_ix, rs in enumerate([[s0r0, s0r1], [s1r0, s1r1]]):
for replica_ix, r in enumerate(rs):
r.query("""
DROP TABLE IF EXISTS test_move;
CREATE TABLE test_move(v UInt64)
ENGINE ReplicatedMergeTree('/clickhouse/shard_{}/tables/test_move', '{}')
ORDER BY tuple()
""".format(shard_ix, r.name))
s0r0.query("SYSTEM STOP MERGES test_move")
s0r1.query("SYSTEM STOP MERGES test_move")
s0r0.query("INSERT INTO test_move VALUES (1)")
s0r0.query("INSERT INTO test_move VALUES (2)")
assert "2" == s0r0.query("SELECT count() FROM test_move").strip()
assert "0" == s1r0.query("SELECT count() FROM test_move").strip()
s0r0.query("ALTER TABLE test_move MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_1/tables/test_move'")
print(s0r0.query("SELECT * FROM system.part_moves_between_shards"))
s0r0.query("SYSTEM START MERGES test_move")
s0r0.query("OPTIMIZE TABLE test_move FINAL")
wait_for_state("DONE", s0r0, "test_move")
for n in [s0r0, s0r1]:
assert "1" == n.query("SELECT count() FROM test_move").strip()
for n in [s1r0, s1r1]:
assert "1" == n.query("SELECT count() FROM test_move").strip()
# Move part back
s1r0.query("ALTER TABLE test_move MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_0/tables/test_move'")
wait_for_state("DONE", s1r0, "test_move")
for n in [s0r0, s0r1]:
assert "2" == n.query("SELECT count() FROM test_move").strip()
for n in [s1r0, s1r1]:
assert "0" == n.query("SELECT count() FROM test_move").strip()
def test_deduplication_while_move(started_cluster):
for shard_ix, rs in enumerate([[s0r0, s0r1], [s1r0, s1r1]]):
for replica_ix, r in enumerate(rs):
r.query("""
DROP TABLE IF EXISTS test_deduplication;
CREATE TABLE test_deduplication(v UInt64)
ENGINE ReplicatedMergeTree('/clickhouse/shard_{}/tables/test_deduplication', '{}')
ORDER BY tuple()
""".format(shard_ix, r.name))
r.query("""
DROP TABLE IF EXISTS test_deduplication_d;
CREATE TABLE test_deduplication_d AS test_deduplication
ENGINE Distributed('test_cluster', '', test_deduplication)
""")
s0r0.query("SYSTEM STOP MERGES test_deduplication")
s0r1.query("SYSTEM STOP MERGES test_deduplication")
s0r0.query("INSERT INTO test_deduplication VALUES (1)")
s0r0.query("INSERT INTO test_deduplication VALUES (2)")
s0r1.query("SYSTEM SYNC REPLICA test_deduplication", timeout=20)
assert "2" == s0r0.query("SELECT count() FROM test_deduplication").strip()
assert "0" == s1r0.query("SELECT count() FROM test_deduplication").strip()
s0r0.query(
"ALTER TABLE test_deduplication MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_1/tables/test_deduplication'")
s0r0.query("SYSTEM START MERGES test_deduplication")
expected = """
1
2
"""
def deduplication_invariant_test():
n = random.choice(list(started_cluster.instances.values()))
assert TSV(
n.query("SELECT * FROM test_deduplication_d ORDER BY v",
settings={"allow_experimental_query_deduplication": 1})
) == TSV(expected)
deduplication_invariant = ConcurrentInvariant(deduplication_invariant_test)
deduplication_invariant.start()
wait_for_state("DONE", s0r0, "test_deduplication")
deduplication_invariant.stop_and_assert_no_exception()
def test_part_move_step_by_step(started_cluster):
for shard_ix, rs in enumerate([[s0r0, s0r1], [s1r0, s1r1]]):
for replica_ix, r in enumerate(rs):
r.query("""
DROP TABLE IF EXISTS test_part_move_step_by_step;
CREATE TABLE test_part_move_step_by_step(v UInt64)
ENGINE ReplicatedMergeTree('/clickhouse/shard_{}/tables/test_part_move_step_by_step', '{}')
ORDER BY tuple()
""".format(shard_ix, r.name))
r.query("""
DROP TABLE IF EXISTS test_part_move_step_by_step_d;
CREATE TABLE test_part_move_step_by_step_d AS test_part_move_step_by_step
ENGINE Distributed('test_cluster', currentDatabase(), test_part_move_step_by_step)
""")
s0r0.query("SYSTEM STOP MERGES test_part_move_step_by_step")
s0r1.query("SYSTEM STOP MERGES test_part_move_step_by_step")
s0r0.query("INSERT INTO test_part_move_step_by_step VALUES (1)")
s0r0.query("INSERT INTO test_part_move_step_by_step VALUES (2)")
s0r1.query("SYSTEM SYNC REPLICA test_part_move_step_by_step", timeout=20)
assert "2" == s0r0.query("SELECT count() FROM test_part_move_step_by_step").strip()
assert "0" == s1r0.query("SELECT count() FROM test_part_move_step_by_step").strip()
expected = """
1
2
"""
def deduplication_invariant_test():
n = random.choice(list(started_cluster.instances.values()))
try:
assert TSV(
n.query("SELECT * FROM test_part_move_step_by_step_d ORDER BY v",
settings={"allow_experimental_query_deduplication": 1})
) == TSV(expected)
except QueryRuntimeException as e:
# ignore transient errors that are caused by us restarting nodes
if e.returncode not in transient_ch_errors:
raise e
deduplication_invariant = ConcurrentInvariant(deduplication_invariant_test)
deduplication_invariant.start()
# Stop a source replica to prevent SYNC_SOURCE succeeding.
s0r1.stop_clickhouse()
s0r0.query(
"ALTER TABLE test_part_move_step_by_step MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_1/tables/test_part_move_step_by_step'")
# Should hang on SYNC_SOURCE until all source replicas acknowledge new pinned UUIDs.
wait_for_state("SYNC_SOURCE", s0r0, "test_part_move_step_by_step", "Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Start all replicas in source shard but stop a replica in destination shard
# to prevent SYNC_DESTINATION succeeding.
s1r1.stop_clickhouse()
s0r1.start_clickhouse()
# After SYNC_SOURCE step no merges will be assigned.
s0r0.query("SYSTEM START MERGES test_part_move_step_by_step; OPTIMIZE TABLE test_part_move_step_by_step;")
s0r1.query("SYSTEM START MERGES test_part_move_step_by_step; OPTIMIZE TABLE test_part_move_step_by_step;")
wait_for_state("SYNC_DESTINATION", s0r0, "test_part_move_step_by_step", "Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Start previously stopped replica in destination shard to let SYNC_DESTINATION
# succeed.
# Stop the other replica in destination shard to prevent DESTINATION_FETCH succeed.
s1r0.stop_clickhouse()
s1r1.start_clickhouse()
wait_for_state("DESTINATION_FETCH", s0r0, "test_part_move_step_by_step", "Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Start previously stopped replica in destination shard to let DESTINATION_FETCH
# succeed.
# Stop the other replica in destination shard to prevent DESTINATION_ATTACH succeed.
s1r1.stop_clickhouse()
s1r0.start_clickhouse()
wait_for_state("DESTINATION_ATTACH", s0r0, "test_part_move_step_by_step", "Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Start all replicas in destination shard to let DESTINATION_ATTACH succeed.
# Stop a source replica to prevent SOURCE_DROP succeeding.
s0r0.stop_clickhouse()
s1r1.start_clickhouse()
wait_for_state("SOURCE_DROP", s0r1, "test_part_move_step_by_step", "Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
s0r0.start_clickhouse()
wait_for_state("DONE", s0r1, "test_part_move_step_by_step")
deduplication_invariant.assert_no_exception()
# No hung tasks in replication queue. Would timeout otherwise.
for instance in started_cluster.instances.values():
instance.query("SYSTEM SYNC REPLICA test_part_move_step_by_step")
assert "1" == s0r0.query("SELECT count() FROM test_part_move_step_by_step").strip()
assert "1" == s1r0.query("SELECT count() FROM test_part_move_step_by_step").strip()
deduplication_invariant.stop_and_assert_no_exception()
def test_part_move_step_by_step_kill(started_cluster):
for shard_ix, rs in enumerate([[s0r0, s0r1], [s1r0, s1r1]]):
for replica_ix, r in enumerate(rs):
r.query("""
DROP TABLE IF EXISTS test_part_move_step_by_step_kill;
CREATE TABLE test_part_move_step_by_step_kill(v UInt64)
ENGINE ReplicatedMergeTree('/clickhouse/shard_{}/tables/test_part_move_step_by_step_kill', '{}')
ORDER BY tuple()
""".format(shard_ix, r.name))
r.query("""
DROP TABLE IF EXISTS test_part_move_step_by_step_kill_d;
CREATE TABLE test_part_move_step_by_step_kill_d AS test_part_move_step_by_step_kill
ENGINE Distributed('test_cluster', currentDatabase(), test_part_move_step_by_step_kill)
""")
s0r0.query("SYSTEM STOP MERGES test_part_move_step_by_step_kill")
s0r1.query("SYSTEM STOP MERGES test_part_move_step_by_step_kill")
s0r0.query("INSERT INTO test_part_move_step_by_step_kill VALUES (1)")
s0r0.query("INSERT INTO test_part_move_step_by_step_kill VALUES (2)")
s0r1.query("SYSTEM SYNC REPLICA test_part_move_step_by_step_kill", timeout=20)
assert "2" == s0r0.query("SELECT count() FROM test_part_move_step_by_step_kill").strip()
assert "0" == s1r0.query("SELECT count() FROM test_part_move_step_by_step_kill").strip()
expected = """
1
2
"""
def deduplication_invariant_test():
n = random.choice(list(started_cluster.instances.values()))
try:
assert TSV(
n.query("SELECT * FROM test_part_move_step_by_step_kill_d ORDER BY v",
settings={
"allow_experimental_query_deduplication": 1
})
) == TSV(expected)
except QueryRuntimeException as e:
# ignore transient errors that are caused by us restarting nodes
if e.returncode not in transient_ch_errors:
raise e
deduplication_invariant = ConcurrentInvariant(deduplication_invariant_test)
deduplication_invariant.start()
# Stop a source replica to prevent SYNC_SOURCE succeeding.
s0r1.stop_clickhouse()
s0r0.query(
"ALTER TABLE test_part_move_step_by_step_kill MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_1/tables/test_part_move_step_by_step_kill'")
# Should hang on SYNC_SOURCE until all source replicas acknowledge new pinned UUIDs.
wait_for_state("SYNC_SOURCE", s0r0, "test_part_move_step_by_step_kill", "Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Start all replicas in source shard but stop a replica in destination shard
# to prevent SYNC_DESTINATION succeeding.
s1r1.stop_clickhouse()
s0r1.start_clickhouse()
# After SYNC_SOURCE step no merges will be assigned.
s0r0.query("SYSTEM START MERGES test_part_move_step_by_step_kill; OPTIMIZE TABLE test_part_move_step_by_step_kill;")
s0r1.query("SYSTEM START MERGES test_part_move_step_by_step_kill; OPTIMIZE TABLE test_part_move_step_by_step_kill;")
wait_for_state("SYNC_DESTINATION", s0r0, "test_part_move_step_by_step_kill",
"Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Start previously stopped replica in destination shard to let SYNC_DESTINATION
# succeed.
# Stop the other replica in destination shard to prevent DESTINATION_FETCH succeed.
s1r0.stop_clickhouse()
s1r1.start_clickhouse()
wait_for_state("DESTINATION_FETCH", s0r0, "test_part_move_step_by_step_kill",
"Some replicas haven\\'t processed event")
# Start previously stopped replica in destination shard to let DESTINATION_FETCH
# succeed.
# Stop the other replica in destination shard to prevent DESTINATION_ATTACH succeed.
s1r1.stop_clickhouse()
s1r0.start_clickhouse()
wait_for_state("DESTINATION_ATTACH", s0r0, "test_part_move_step_by_step_kill",
"Some replicas haven\\'t processed event")
deduplication_invariant.assert_no_exception()
# Rollback here.
s0r0.query("""
KILL PART_MOVE_TO_SHARD
WHERE task_uuid = (SELECT task_uuid FROM system.part_moves_between_shards WHERE table = 'test_part_move_step_by_step_kill')
""")
wait_for_state("DESTINATION_ATTACH", s0r0, "test_part_move_step_by_step_kill",
assert_exception_msg="Some replicas haven\\'t processed event",
assert_rollback=True)
s1r1.start_clickhouse()
wait_for_state("CANCELLED", s0r0, "test_part_move_step_by_step_kill", assert_rollback=True)
deduplication_invariant.assert_no_exception()
# No hung tasks in replication queue. Would timeout otherwise.
for instance in started_cluster.instances.values():
instance.query("SYSTEM SYNC REPLICA test_part_move_step_by_step_kill")
assert "2" == s0r0.query("SELECT count() FROM test_part_move_step_by_step_kill").strip()
assert "0" == s1r0.query("SELECT count() FROM test_part_move_step_by_step_kill").strip()
deduplication_invariant.stop_and_assert_no_exception()
def test_move_not_permitted(started_cluster):
# Verify that invariants for part compatibility are checked.
# Tests are executed in order. Make sure cluster is up if previous test
# failed.
s0r0.start_clickhouse()
s1r0.start_clickhouse()
for ix, n in enumerate([s0r0, s1r0]):
n.query("""
DROP TABLE IF EXISTS not_permitted_columns;
CREATE TABLE not_permitted_columns(v_{ix} UInt64)
ENGINE ReplicatedMergeTree('/clickhouse/shard_{ix}/tables/not_permitted_columns', 'r')
ORDER BY tuple();
""".format(ix=ix))
partition = "date"
if ix > 0:
partition = "v"
n.query("""
DROP TABLE IF EXISTS not_permitted_partition;
CREATE TABLE not_permitted_partition(date Date, v UInt64)
ENGINE ReplicatedMergeTree('/clickhouse/shard_{ix}/tables/not_permitted_partition', 'r')
PARTITION BY ({partition})
ORDER BY tuple();
""".format(ix=ix, partition=partition))
s0r0.query("INSERT INTO not_permitted_columns VALUES (1)")
s0r0.query("INSERT INTO not_permitted_partition VALUES ('2021-09-03', 1)")
with pytest.raises(QueryRuntimeException, match="DB::Exception: Source and destination are the same"):
s0r0.query("ALTER TABLE not_permitted_columns MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_0/tables/not_permitted_columns'")
with pytest.raises(QueryRuntimeException, match="DB::Exception: Table columns structure in ZooKeeper is different from local table structure."):
s0r0.query("ALTER TABLE not_permitted_columns MOVE PART 'all_0_0_0' TO SHARD '/clickhouse/shard_1/tables/not_permitted_columns'")
with pytest.raises(QueryRuntimeException, match="DB::Exception: Existing table metadata in ZooKeeper differs in partition key expression."):
s0r0.query("ALTER TABLE not_permitted_partition MOVE PART '20210903_0_0_0' TO SHARD '/clickhouse/shard_1/tables/not_permitted_partition'")
def wait_for_state(desired_state, instance, test_table, assert_exception_msg=None, assert_rollback=False):
last_debug_print_time = time.time()
print("Waiting to reach state: {}".format(desired_state))
if assert_exception_msg:
print(" with exception contents: {}".format(assert_exception_msg))
if assert_rollback:
print(" and rollback: {}".format(assert_rollback))
while True:
tasks = TSV.toMat(instance.query(
"SELECT state, num_tries, last_exception, rollback FROM system.part_moves_between_shards WHERE table = '{}'".format(
test_table)))
assert len(tasks) == 1, "only one task expected in this test"
if time.time() - last_debug_print_time > 30:
last_debug_print_time = time.time()
print("Current state: ", tasks)
[state, num_tries, last_exception, rollback] = tasks[0]
if state == desired_state:
if assert_exception_msg and int(num_tries) < 3:
# Let the task be retried a few times when expecting an exception
# to make sure the exception is persistent and the code doesn't
# accidentally continue to run when we expect it not to.
continue
if assert_exception_msg:
assert assert_exception_msg in last_exception
if assert_rollback:
assert int(rollback) == 1, "rollback bit isn't set"
break
elif state in ["DONE", "CANCELLED"]:
raise Exception("Reached terminal state {}, but was waiting for {}".format(state, desired_state))
time.sleep(0.1)
class ConcurrentInvariant:
def __init__(self, invariant_test, loop_sleep=0.1):
self.invariant_test = invariant_test
self.loop_sleep = loop_sleep
self.started = False
self.exiting = False
self.exception = None
self.thread = threading.Thread(target=self._loop)
def start(self):
if self.started:
raise Exception('invariant thread already started')
self.started = True
self.thread.start()
def stop_and_assert_no_exception(self):
self._assert_started()
self.exiting = True
self.thread.join()
if self.exception:
raise self.exception
def assert_no_exception(self):
self._assert_started()
if self.exception:
raise self.exception
def _loop(self):
try:
while not self.exiting:
self.invariant_test()
time.sleep(self.loop_sleep)
except Exception as e:
self.exiting = True
self.exception = e
def _assert_started(self):
if not self.started:
raise Exception('invariant thread not started, forgot to call start?')