diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index b2fcbc77c56..bc4f2cb43d2 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -63,6 +63,7 @@ static struct InitFiu REGULAR(keepermap_fail_drop_data) \ REGULAR(lazy_pipe_fds_fail_close) \ PAUSEABLE(infinite_sleep) \ + PAUSEABLE(stop_moving_part_before_swap_with_active) \ namespace FailPoints diff --git a/src/Storages/MergeTree/MergeTreePartsMover.cpp b/src/Storages/MergeTree/MergeTreePartsMover.cpp index 9223d6fd5b1..8daa48a2339 100644 --- a/src/Storages/MergeTree/MergeTreePartsMover.cpp +++ b/src/Storages/MergeTree/MergeTreePartsMover.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -15,6 +16,11 @@ namespace ErrorCodes extern const int DIRECTORY_ALREADY_EXISTS; } +namespace FailPoints +{ + extern const char stop_moving_part_before_swap_with_active[]; +} + namespace { @@ -272,7 +278,13 @@ MergeTreePartsMover::TemporaryClonedPart MergeTreePartsMover::clonePart(const Me cloned_part.part = std::move(builder).withPartFormatFromDisk().build(); LOG_TRACE(log, "Part {} was cloned to {}", part->name, cloned_part.part->getDataPartStorage().getFullPath()); - cloned_part.part->is_temp = data->allowRemoveStaleMovingParts(); + if (data->allowRemoveStaleMovingParts()) + { + cloned_part.part->is_temp = data->allowRemoveStaleMovingParts(); + /// Setting it in case connection to zookeeper is lost while moving + /// Otherwise part might be stuck in the moving directory due to the KEEPER_EXCEPTION in part's destructor + cloned_part.part->remove_tmp_policy = IMergeTreeDataPart::BlobsRemovalPolicyForTemporaryParts::REMOVE_BLOBS; + } cloned_part.part->loadColumnsChecksumsIndexes(true, true); cloned_part.part->loadVersionMetadata(); cloned_part.part->modification_time = cloned_part.part->getDataPartStorage().getLastModified().epochTime(); @@ -282,6 +294,8 @@ MergeTreePartsMover::TemporaryClonedPart MergeTreePartsMover::clonePart(const Me void MergeTreePartsMover::swapClonedPart(TemporaryClonedPart & cloned_part) const { + /// Used to get some stuck parts in the moving directory by stopping moves while pause is active + FailPointInjection::pauseFailPoint(FailPoints::stop_moving_part_before_swap_with_active); if (moves_blocker.isCancelled()) throw Exception(ErrorCodes::ABORTED, "Cancelled moving parts."); diff --git a/tests/integration/test_remove_stale_moving_parts/__init__.py b/tests/integration/test_remove_stale_moving_parts/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/integration/test_remove_stale_moving_parts/config.xml b/tests/integration/test_remove_stale_moving_parts/config.xml new file mode 100644 index 00000000000..968e07fae51 --- /dev/null +++ b/tests/integration/test_remove_stale_moving_parts/config.xml @@ -0,0 +1,46 @@ + + + + + + ch1 + 9000 + + + + + + 01 + + + + + s3 + http://minio1:9001/root/data/ + minio + minio123 + + + + + + + default + False + + + s3 + False + + + 0.0 + + + + + + true + s3 + + true + diff --git a/tests/integration/test_remove_stale_moving_parts/test.py b/tests/integration/test_remove_stale_moving_parts/test.py new file mode 100644 index 00000000000..dcbcf38d25a --- /dev/null +++ b/tests/integration/test_remove_stale_moving_parts/test.py @@ -0,0 +1,126 @@ +from pathlib import Path +import time +import pytest +from helpers.cluster import ClickHouseCluster + +cluster = ClickHouseCluster(__file__) +ch1 = cluster.add_instance( + "ch1", + main_configs=[ + "config.xml", + ], + macros={"replica": "node1"}, + with_zookeeper=True, + with_minio=True, +) + +DATABASE_NAME = "stale_moving_parts" + + +@pytest.fixture(scope="module") +def started_cluster(): + try: + cluster.start() + yield cluster + + finally: + cluster.shutdown() + + +def q(node, query): + return node.query(database=DATABASE_NAME, sql=query) + + +# .../disks/s3/store/ +def get_table_path(node, table): + return ( + node.query( + sql=f"SELECT data_paths FROM system.tables WHERE table = '{table}' and database = '{DATABASE_NAME}' LIMIT 1" + ) + .strip('"\n[]') + .split(",")[1] + .strip("'") + ) + + +def exec(node, cmd, path): + return node.exec_in_container( + [ + "bash", + "-c", + f"{cmd} {path}", + ] + ) + + +def stop_zookeeper(node): + node.exec_in_container(["bash", "-c", "/opt/zookeeper/bin/zkServer.sh stop"]) + timeout = time.time() + 60 + while node.get_process_pid("zookeeper") != None: + if time.time() > timeout: + raise Exception("Failed to stop ZooKeeper in 60 secs") + time.sleep(0.2) + + +def wait_part_is_stuck(node, table_moving_path, moving_part): + num_tries = 5 + while q(node, "SELECT part_name FROM system.moves").strip() != moving_part: + if num_tries == 0: + raise Exception("Part has not started to move") + num_tries -= 1 + time.sleep(1) + num_tries = 5 + while exec(node, "ls", table_moving_path).strip() != moving_part: + if num_tries == 0: + raise Exception("Part is not stuck in the moving directory") + num_tries -= 1 + time.sleep(1) + + +def wait_zookeeper_node_to_start(zk_nodes, timeout=60): + start = time.time() + while time.time() - start < timeout: + try: + for instance in zk_nodes: + conn = cluster.get_kazoo_client(instance) + conn.get_children("/") + print("All instances of ZooKeeper started") + return + except Exception as ex: + print(("Can't connect to ZooKeeper " + str(ex))) + time.sleep(0.5) + + +def test_remove_stale_moving_parts_without_zookeeper(started_cluster): + ch1.query(f"CREATE DATABASE IF NOT EXISTS {DATABASE_NAME}") + + q( + ch1, + "CREATE TABLE test_remove ON CLUSTER cluster ( id UInt32 ) ENGINE ReplicatedMergeTree() ORDER BY id;", + ) + + table_moving_path = Path(get_table_path(ch1, "test_remove")) / "moving" + + q(ch1, "SYSTEM ENABLE FAILPOINT stop_moving_part_before_swap_with_active") + q(ch1, "INSERT INTO test_remove SELECT number FROM numbers(100);") + moving_part = "all_0_0_0" + move_response = ch1.get_query_request( + sql=f"ALTER TABLE test_remove MOVE PART '{moving_part}' TO DISK 's3'", + database=DATABASE_NAME, + ) + + wait_part_is_stuck(ch1, table_moving_path, moving_part) + + cluster.stop_zookeeper_nodes(["zoo1", "zoo2", "zoo3"]) + # Stop moves in case table is not read-only yet + q(ch1, "SYSTEM STOP MOVES") + q(ch1, "SYSTEM DISABLE FAILPOINT stop_moving_part_before_swap_with_active") + + assert "Cancelled moving parts" in move_response.get_error() + assert exec(ch1, "ls", table_moving_path).strip() == "" + + cluster.start_zookeeper_nodes(["zoo1", "zoo2", "zoo3"]) + wait_zookeeper_node_to_start(["zoo1", "zoo2", "zoo3"]) + q(ch1, "SYSTEM START MOVES") + + q(ch1, f"DROP TABLE test_remove")