diff --git a/tests/integration/test_backup_restore_new/configs/cas_gc.xml b/tests/integration/test_backup_restore_new/configs/cas_gc.xml new file mode 100644 index 000000000000..f415af9d6ee6 --- /dev/null +++ b/tests/integration/test_backup_restore_new/configs/cas_gc.xml @@ -0,0 +1,10 @@ + + + + + 1 + 1 + + + + diff --git a/tests/integration/test_backup_restore_new/configs/cas_storage.xml b/tests/integration/test_backup_restore_new/configs/cas_storage.xml new file mode 100644 index 000000000000..349ce11ab215 --- /dev/null +++ b/tests/integration/test_backup_restore_new/configs/cas_storage.xml @@ -0,0 +1,24 @@ + + + + + object_storage + s3 + cas + itest-backup-restore-new + http://rustfs1:11121/test/backup_restore_new/ + clickhouse + clickhouse + + + + + + + cas + + + + + + diff --git a/tests/integration/test_backup_restore_new/test.py b/tests/integration/test_backup_restore_new/test.py index c300ca2b8a7a..6c8a5124afca 100644 --- a/tests/integration/test_backup_restore_new/test.py +++ b/tests/integration/test_backup_restore_new/test.py @@ -18,9 +18,10 @@ cluster = ClickHouseCluster(__file__) instance = cluster.add_instance( "instance", - main_configs=["configs/backups_disk.xml"], + main_configs=["configs/backups_disk.xml", "configs/cas_storage.xml"], user_configs=["configs/zookeeper_retries.xml"], external_dirs=["/backups/"], + with_rustfs=True, ) instance_with_short_timeout = cluster.add_instance( "instance_with_short_timeout", @@ -30,11 +31,14 @@ ) -def create_and_fill_table(engine="MergeTree", n=100): +def create_and_fill_table(engine="MergeTree", n=100, storage_policy=None): if engine == "MergeTree": engine = "MergeTree ORDER BY y PARTITION BY x%10" instance.query("CREATE DATABASE test") - instance.query(f"CREATE TABLE test.table(x UInt32, y String) ENGINE={engine}") + create_query = f"CREATE TABLE test.table(x UInt32, y String) ENGINE={engine}" + if storage_policy is not None: + create_query += f" SETTINGS storage_policy = '{storage_policy}'" + instance.query(create_query) instance.query( f"INSERT INTO test.table SELECT number, toString(number) FROM numbers({n})" ) @@ -224,6 +228,30 @@ def test_restore_table(engine): assert instance.query("SELECT count(), sum(x) FROM test.table") == "100\t4950\n" +def test_restore_table_on_cas_disk(): + backup_name = new_backup_name() + create_and_fill_table(storage_policy="cas_policy") + + assert instance.query("SELECT count(), sum(x) FROM test.table") == "100\t4950\n" + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table'") + == "cas_policy\n" + ) + instance.query(f"BACKUP TABLE test.table TO {backup_name}") + instance.query("DROP TABLE test.table") + instance.query(f"RESTORE TABLE test.table FROM {backup_name}") + + assert instance.query("SELECT count(), sum(x) FROM test.table") == "100\t4950\n" + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table'") + == "cas_policy\n" + ) + assert ( + instance.query("CHECK TABLE test.table SETTINGS check_query_single_value_result = 1") + == "1\n" + ) + + @pytest.mark.parametrize( "engine", ["MergeTree", "Log", "TinyLog", "StripeLog", "Memory"] ) @@ -1504,8 +1532,9 @@ def test_system_users_async(): ) -def test_projection(): - create_and_fill_table(n=3) +@pytest.mark.parametrize("storage_policy", [None, "cas_policy"]) +def test_projection(storage_policy): + create_and_fill_table(n=3, storage_policy=storage_policy) instance.query("ALTER TABLE test.table ADD PROJECTION prjmax (SELECT MAX(x))") instance.query("INSERT INTO test.table VALUES (100, 'a'), (101, 'b')") @@ -1555,6 +1584,18 @@ def test_projection(): == "2\n" ) + if storage_policy is not None: + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table'") + == f"{storage_policy}\n" + ) + assert ( + instance.query( + "CHECK TABLE test.table SETTINGS check_query_single_value_result = 1" + ) + == "1\n" + ) + def test_restore_table_not_evaluate_table_defaults(): instance.query("CREATE DATABASE test") @@ -1950,8 +1991,10 @@ def verify_restore_info(): assert info.bytes_read == 0 -def test_mutation(): - create_and_fill_table(engine="MergeTree ORDER BY tuple()", n=5) +def check_mutation(storage_policy=None): + create_and_fill_table( + engine="MergeTree ORDER BY tuple()", n=5, storage_policy=storage_policy + ) instance.query( "INSERT INTO test.table SELECT number, toString(number) FROM numbers(5, 5)" @@ -1973,6 +2016,12 @@ def test_mutation(): instance.query("ALTER TABLE test.table UPDATE x=x+1 WHERE 1") instance.query("ALTER TABLE test.table UPDATE x=x+1 WHERE 1") + if storage_policy is not None: + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table'") + == f"{storage_policy}\n" + ) + assert instance.query("SELECT count() FROM test.table") == "15\n" backup_name = new_backup_name() instance.query(f"BACKUP TABLE test.table TO {backup_name}") @@ -1986,6 +2035,20 @@ def test_mutation(): instance.query("DROP TABLE test.table") instance.query(f"RESTORE TABLE test.table FROM {backup_name}") + if storage_policy is not None: + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table'") + == f"{storage_policy}\n" + ) + assert instance.query("SELECT count() FROM test.table") == "15\n" + + +def test_mutation(): + check_mutation() + + +def test_mutation_on_cas_disk(): + check_mutation(storage_policy="cas_policy") def test_tables_dependency(): @@ -2215,12 +2278,17 @@ def test_restore_table_with_checksum_data_file_name(engine): assert instance.query("SELECT count(), sum(x) FROM test.table") == "100\t4950\n" -def test_incremental_backup_with_checksum_data_file_name(): +def check_incremental_backup_with_checksum_data_file_name(storage_policy=None): backup_name = new_backup_name() incremental_backup_name = new_backup_name() - create_and_fill_table() + create_and_fill_table(storage_policy=storage_policy) assert instance.query("SELECT count(), sum(x) FROM test.table") == "100\t4950\n" + if storage_policy is not None: + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table'") + == f"{storage_policy}\n" + ) instance.query( f"BACKUP TABLE test.table TO {backup_name} SETTINGS data_file_name_generator='checksum'" ) @@ -2236,6 +2304,19 @@ def test_incremental_backup_with_checksum_data_file_name(): f"RESTORE TABLE test.table AS test.table2 FROM {incremental_backup_name}" ) assert instance.query("SELECT count(), sum(x) FROM test.table2") == "102\t5081\n" + if storage_policy is not None: + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name = 'table2'") + == f"{storage_policy}\n" + ) + + +def test_incremental_backup_with_checksum_data_file_name(): + check_incremental_backup_with_checksum_data_file_name() + + +def test_incremental_backup_with_checksum_data_file_name_on_cas_disk(): + check_incremental_backup_with_checksum_data_file_name(storage_policy="cas_policy") def test_async_backup_restore_with_max_execution_time_zero(): diff --git a/tests/integration/test_backup_restore_new/test_cancel_backup.py b/tests/integration/test_backup_restore_new/test_cancel_backup.py index c6e378ca4bf6..0bcf43ebee78 100644 --- a/tests/integration/test_backup_restore_new/test_cancel_backup.py +++ b/tests/integration/test_backup_restore_new/test_cancel_backup.py @@ -1,4 +1,5 @@ import re +import time import uuid import pytest @@ -17,6 +18,8 @@ "configs/backups_disk.xml", "configs/slow_backups.xml", "configs/shutdown_cancel_backups.xml", + "configs/cas_storage.xml", + "configs/cas_gc.xml", ] node = cluster.add_instance( @@ -24,8 +27,11 @@ main_configs=main_configs, external_dirs=["/backups/"], stay_alive=True, + with_rustfs=True, ) +CAS_POOL_PREFIXES = ["backup_restore_new/blobs/"] + @pytest.fixture(scope="module", autouse=True) def start_cluster(): @@ -179,11 +185,57 @@ def cancel_restore(restore_id): assert kill_duration_ms < kill_duration_ms_threshold +def table_settings(storage_policy): + if storage_policy is None: + return "" + return f" SETTINGS storage_policy = '{storage_policy}'" + + +def count_cas_pool_objects(): + return sum( + len( + list( + cluster.rustfs_client.list_objects( + cluster.rustfs_bucket, prefix, recursive=True + ) + ) + ) + for prefix in CAS_POOL_PREFIXES + ) + + +def wait_cas_pool_settled(): + previous = count_cas_pool_objects() + for _ in range(40): + time.sleep(3) + current = count_cas_pool_objects() + if current == previous: + return current + previous = current + return previous + + +def wait_cas_pool_reclaimed(baseline): + current = count_cas_pool_objects() + for _ in range(120): + if current <= baseline: + break + time.sleep(1) + current = count_cas_pool_objects() + assert ( + current <= baseline + ), f"CAS pool objects were not reclaimed: baseline={baseline}, current={current}" + + # Test that BACKUP and RESTORE operations can be cancelled with KILL QUERY. -def test_cancel_backup(): +@pytest.mark.parametrize("storage_policy", [None, "cas_policy"]) +def test_cancel_backup(storage_policy): + cas_pool_baseline = wait_cas_pool_settled() if storage_policy == "cas_policy" else None + # We use partitioning so backups would contain more files. node.query( "CREATE TABLE tbl (x UInt64) ENGINE=MergeTree() ORDER BY tuple() PARTITION BY x%20" + + table_settings(storage_policy) ) node.query("INSERT INTO tbl SELECT number FROM numbers(500)") @@ -209,11 +261,25 @@ def test_cancel_backup(): start_restore(restore_id, backup_id) wait_restore(restore_id) + assert node.query("SELECT count(), sum(x) FROM tbl") == "500\t124750\n" + assert ( + node.query("CHECK TABLE tbl SETTINGS check_query_single_value_result = 1") + == "1\n" + ) + + if cas_pool_baseline is not None: + node.query("DROP TABLE tbl SYNC") + wait_cas_pool_reclaimed(cas_pool_baseline) + # Test that shutdown cancels a running backup and doesn't wait until it finishes. -def test_shutdown_cancel_backup(): +@pytest.mark.parametrize("storage_policy", [None, "cas_policy"]) +def test_shutdown_cancel_backup(storage_policy): + cas_pool_baseline = wait_cas_pool_settled() if storage_policy == "cas_policy" else None + node.query( "CREATE TABLE tbl (x UInt64) ENGINE=MergeTree() ORDER BY tuple() PARTITION BY x%5" + + table_settings(storage_policy) ) node.query("INSERT INTO tbl SELECT number FROM numbers(500)") @@ -237,3 +303,6 @@ def test_shutdown_cancel_backup(): f"RESTORE TABLE tbl FROM {get_backup_name(backup_id)}" ), ) + + if cas_pool_baseline is not None: + wait_cas_pool_reclaimed(cas_pool_baseline) diff --git a/tests/integration/test_backup_restore_on_cluster/concurrency_helper.py b/tests/integration/test_backup_restore_on_cluster/concurrency_helper.py index f75600d11296..724796de06c6 100644 --- a/tests/integration/test_backup_restore_on_cluster/concurrency_helper.py +++ b/tests/integration/test_backup_restore_on_cluster/concurrency_helper.py @@ -38,17 +38,50 @@ def generate_cluster_def(file: str, num_nodes: int) -> str: return str(path.absolute()) +def generate_cas_server_root_def(file: str, node_index: int) -> str: + path = ( + Path(__file__).parent + / f"_gen/cas_server_root_{Path(file).stem}_node{node_index}.xml" + ) + config = f""" + + + + itest-{Path(file).stem}-node{node_index} + + + +""" + if path.is_file() and path.read_text(encoding="utf-8") == config: + return str(path.absolute()) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(encoding="utf-8", data=config) + return str(path.absolute()) + + def add_nodes_to_cluster( cluster: ClickHouseCluster, num_nodes: int, main_configs: List[str], user_configs: List[str], + cas_file: str = None, **kwargs ) -> List[ClickHouseInstance]: + def node_main_configs(i): + if cas_file is None: + return main_configs + return main_configs + [ + "configs/cas_storage.xml", + generate_cas_server_root_def(cas_file, i), + ] + + if cas_file is not None: + kwargs["with_rustfs"] = True + nodes = [ cluster.add_instance( f"node{i}", - main_configs=main_configs, + main_configs=node_main_configs(i), user_configs=user_configs, external_dirs=["/backups/"], macros={"replica": f"node{i}", "shard": "shard1"}, @@ -60,9 +93,11 @@ def add_nodes_to_cluster( return nodes -def create_test_table(node: ClickHouseInstance) -> None: +def create_test_table(node: ClickHouseInstance, storage_policy: str = None) -> None: + settings = f" SETTINGS storage_policy = '{storage_policy}'" if storage_policy else "" node.query( """CREATE TABLE tbl ON CLUSTER 'cluster' ( x UInt64 ) ENGINE=ReplicatedMergeTree('/clickhouse/tables/tbl/', '{replica}') ORDER BY tuple()""" + + settings ) diff --git a/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node1.xml b/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node1.xml new file mode 100644 index 000000000000..e240cf0308ae --- /dev/null +++ b/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node1.xml @@ -0,0 +1,9 @@ + + + + + itest-backup-restore-on-cluster-node1 + + + + diff --git a/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node2.xml b/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node2.xml new file mode 100644 index 000000000000..fca0f5d5a472 --- /dev/null +++ b/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node2.xml @@ -0,0 +1,9 @@ + + + + + itest-backup-restore-on-cluster-node2 + + + + diff --git a/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node3.xml b/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node3.xml new file mode 100644 index 000000000000..49b8f8620c78 --- /dev/null +++ b/tests/integration/test_backup_restore_on_cluster/configs/cas_server_root_node3.xml @@ -0,0 +1,9 @@ + + + + + itest-backup-restore-on-cluster-node3 + + + + diff --git a/tests/integration/test_backup_restore_on_cluster/configs/cas_storage.xml b/tests/integration/test_backup_restore_on_cluster/configs/cas_storage.xml new file mode 100644 index 000000000000..5884c99c7f15 --- /dev/null +++ b/tests/integration/test_backup_restore_on_cluster/configs/cas_storage.xml @@ -0,0 +1,23 @@ + + + + + object_storage + s3 + cas + http://rustfs1:11121/test/backup_restore_on_cluster/ + clickhouse + clickhouse + + + + + + + cas + + + + + + diff --git a/tests/integration/test_backup_restore_on_cluster/test.py b/tests/integration/test_backup_restore_on_cluster/test.py index 6e1e0de0ec60..33ae40071a03 100644 --- a/tests/integration/test_backup_restore_on_cluster/test.py +++ b/tests/integration/test_backup_restore_on_cluster/test.py @@ -19,6 +19,8 @@ "configs/lesser_timeouts.xml", # Default timeouts are quite big (a few minutes), the tests don't need them to be that big. ] +cas_configs = ["configs/cas_storage.xml"] + user_configs = [ "configs/allow_database_types.xml", "configs/zookeeper_retries.xml", @@ -26,30 +28,33 @@ node1 = cluster.add_instance( "node1", - main_configs=main_configs, + main_configs=main_configs + cas_configs + ["configs/cas_server_root_node1.xml"], user_configs=user_configs, external_dirs=["/backups/"], macros={"replica": "node1", "shard": "shard1"}, with_zookeeper=True, + with_rustfs=True, ) node2 = cluster.add_instance( "node2", - main_configs=main_configs, + main_configs=main_configs + cas_configs + ["configs/cas_server_root_node2.xml"], user_configs=user_configs, external_dirs=["/backups/"], macros={"replica": "node2", "shard": "shard1"}, with_zookeeper=True, + with_rustfs=True, stay_alive=True, # Necessary for the "test_stop_other_host_while_backup" test ) node3 = cluster.add_instance( "node3", - main_configs=main_configs, + main_configs=main_configs + cas_configs + ["configs/cas_server_root_node3.xml"], user_configs=user_configs, external_dirs=["/backups/"], macros={"replica": "node3", "shard": "shard1"}, with_zookeeper=True, + with_rustfs=True, ) @@ -125,6 +130,46 @@ def test_replicated_table(): ) +def test_replicated_table_on_cas_disk(): + expected = TSV([[1, "Don\\'t"], [2, "count"], [3, "your"], [4, "chickens"]]) + node1.query( + "CREATE TABLE tbl ON CLUSTER 'cluster' (" + "x UInt8, y String" + ") ENGINE=ReplicatedMergeTree('/clickhouse/tables/tbl/', '{replica}')" + "ORDER BY x SETTINGS storage_policy = 'cas_policy'" + ) + + node1.query("INSERT INTO tbl VALUES (1, 'Don''t')") + node2.query("INSERT INTO tbl VALUES (2, 'count')") + node1.query("INSERT INTO tbl SETTINGS async_insert=true VALUES (3, 'your')") + node2.query("INSERT INTO tbl SETTINGS async_insert=true VALUES (4, 'chickens')") + node1.query("SYSTEM SYNC REPLICA ON CLUSTER 'cluster' tbl") + + assert node1.query("SELECT * FROM tbl ORDER BY x") == expected + assert node2.query("SELECT * FROM tbl ORDER BY x") == expected + for node in [node1, node2]: + assert ( + node.query("SELECT storage_policy FROM system.tables WHERE name = 'tbl'") + == "cas_policy\n" + ) + + backup_name = new_backup_name() + node1.query( + f"BACKUP TABLE tbl ON CLUSTER 'cluster' TO {backup_name} SETTINGS replica_num=1" + ) + node1.query("DROP TABLE tbl ON CLUSTER 'cluster' SYNC") + node2.query(f"RESTORE TABLE tbl ON CLUSTER 'cluster' FROM {backup_name}") + node1.query("SYSTEM SYNC REPLICA ON CLUSTER 'cluster' tbl") + + assert node1.query("SELECT * FROM tbl ORDER BY x") == expected + assert node2.query("SELECT * FROM tbl ORDER BY x") == expected + for node in [node1, node2]: + assert ( + node.query("SELECT storage_policy FROM system.tables WHERE name = 'tbl'") + == "cas_policy\n" + ) + + def test_empty_replicated_table(): node1.query( "CREATE TABLE tbl ON CLUSTER 'cluster' (" diff --git a/tests/integration/test_backup_restore_on_cluster/test_concurrency.py b/tests/integration/test_backup_restore_on_cluster/test_concurrency.py index 02431752f64b..cc9b1ebe52f2 100644 --- a/tests/integration/test_backup_restore_on_cluster/test_concurrency.py +++ b/tests/integration/test_backup_restore_on_cluster/test_concurrency.py @@ -25,7 +25,9 @@ # No [Zoo]Keeper retries for tests with concurrency user_configs = ["configs/allow_database_types.xml"] -nodes = add_nodes_to_cluster(cluster, num_nodes, main_configs, user_configs, cpu_limit=12) +nodes = add_nodes_to_cluster( + cluster, num_nodes, main_configs, user_configs, cas_file=__file__, cpu_limit=12 +) node0 = nodes[0] @@ -51,8 +53,8 @@ def drop_after_test(): backup_id_counter = 0 -def create_and_fill_table() -> None: - create_test_table(node0) +def create_and_fill_table(storage_policy=None) -> None: + create_test_table(node0, storage_policy) for i, node in enumerate(nodes): node.query(f"INSERT INTO tbl VALUES ({i})") @@ -66,8 +68,23 @@ def new_backup_name(): expected_sum = num_nodes * (num_nodes - 1) // 2 -def test_replicated_table(): - create_and_fill_table() +def check_restored_table(storage_policy): + for node in nodes: + assert node.query("SELECT sum(x) FROM tbl") == TSV([expected_sum]) + if storage_policy is not None: + assert ( + node.query("SELECT storage_policy FROM system.tables WHERE name = 'tbl'") + == f"{storage_policy}\n" + ) + assert ( + node.query("CHECK TABLE tbl SETTINGS check_query_single_value_result = 1") + == "1\n" + ) + + +@pytest.mark.parametrize("storage_policy", [None, "cas_policy"]) +def test_replicated_table(storage_policy): + create_and_fill_table(storage_policy) backup_name = new_backup_name() node0.query(f"BACKUP TABLE tbl ON CLUSTER 'cluster' TO {backup_name}") @@ -76,15 +93,15 @@ def test_replicated_table(): node0.query(f"RESTORE TABLE tbl ON CLUSTER 'cluster' FROM {backup_name}") node0.query("SYSTEM SYNC REPLICA ON CLUSTER 'cluster' tbl") - for i in range(num_nodes): - assert nodes[i].query("SELECT sum(x) FROM tbl") == TSV([expected_sum]) + check_restored_table(storage_policy) num_concurrent_backups = 4 -def test_concurrent_backups_on_same_node(): - create_and_fill_table() +@pytest.mark.parametrize("storage_policy", [None, "cas_policy"]) +def test_concurrent_backups_on_same_node(storage_policy): + create_and_fill_table(storage_policy) backup_names = [new_backup_name() for _ in range(num_concurrent_backups)] @@ -113,12 +130,12 @@ def test_concurrent_backups_on_same_node(): node0.query("DROP TABLE tbl ON CLUSTER 'cluster' SYNC") node0.query(f"RESTORE TABLE tbl ON CLUSTER 'cluster' FROM {backup_name}") node0.query("SYSTEM SYNC REPLICA ON CLUSTER 'cluster' tbl") - for i in range(num_nodes): - assert nodes[i].query("SELECT sum(x) FROM tbl") == TSV([expected_sum]) + check_restored_table(storage_policy) -def test_concurrent_backups_on_different_nodes(): - create_and_fill_table() +@pytest.mark.parametrize("storage_policy", [None, "cas_policy"]) +def test_concurrent_backups_on_different_nodes(storage_policy): + create_and_fill_table(storage_policy) assert num_concurrent_backups <= num_nodes backup_names = [new_backup_name() for _ in range(num_concurrent_backups)] @@ -150,8 +167,7 @@ def test_concurrent_backups_on_different_nodes(): nodes[i].query("DROP TABLE tbl ON CLUSTER 'cluster' SYNC") nodes[i].query(f"RESTORE TABLE tbl ON CLUSTER 'cluster' FROM {backup_names[i]}") nodes[i].query("SYSTEM SYNC REPLICA ON CLUSTER 'cluster' tbl") - for j in range(num_nodes): - assert nodes[j].query("SELECT sum(x) FROM tbl") == TSV([expected_sum]) + check_restored_table(storage_policy) @pytest.mark.parametrize( diff --git a/tests/integration/test_backup_restore_s3/configs/disk_cas.xml b/tests/integration/test_backup_restore_s3/configs/disk_cas.xml new file mode 100644 index 000000000000..83392e27c7e4 --- /dev/null +++ b/tests/integration/test_backup_restore_s3/configs/disk_cas.xml @@ -0,0 +1,41 @@ + + + + + + object_storage + s3 + cas + 30 + 10000 + itest-backup-restore-s3 + http://rustfs1:11121/test/backup_restore_s3_cas/ + clickhouse + clickhouse + + + cache + disk_cas + /tmp/cas_cache/ + 1000000000 + 1 + + + + + +
+ disk_cas +
+ + + + +
+ disk_cas_cache +
+ + + + + diff --git a/tests/integration/test_backup_restore_s3/test.py b/tests/integration/test_backup_restore_s3/test.py index 77e4e1c0dcbb..564c62607fd5 100644 --- a/tests/integration/test_backup_restore_s3/test.py +++ b/tests/integration/test_backup_restore_s3/test.py @@ -143,6 +143,7 @@ def setup_cluster(request): "node", main_configs=[ "configs/disk_s3.xml", + "configs/disk_cas.xml", "configs/named_collection_s3_backups.xml", "configs/s3_settings.xml", "configs/blob_log.xml", @@ -154,6 +155,7 @@ def setup_cluster(request): "configs/zookeeper_retries.xml", ], with_minio=True, + with_rustfs=True, # The test compares some S3 events. We disable the remote DB disk, so it doesn't affect the comparing events. with_remote_database_disk=False, with_zookeeper=True, @@ -177,6 +179,14 @@ def cluster(setup_cluster): backup_id_counter = 0 +CAS_POLICIES = ["policy_cas", "policy_cas_cache"] + + +def s3_backup_destination(storage_policy, backup_name): + if storage_policy in CAS_POLICIES: + return f"S3('http://rustfs1:11121/test/backups/{backup_name}', 'clickhouse', 'clickhouse')" + return f"S3('http://minio1:9001/root/data/backups/{backup_name}', 'minio', '{minio_secret_key}')" + def new_backup_name(): global backup_id_counter @@ -293,6 +303,8 @@ def check_system_tables(cluster, backup_query_id=None): ("disk_s3_plain", "ObjectStorage", "S3", "Plain"), ("disk_s3_plain_rewritable", "ObjectStorage", "S3", "PlainRewritable"), ("disk_s3_restricted_user", "ObjectStorage", "S3", "Local"), + ("disk_cas", "ObjectStorage", "S3", "CAS"), + ("disk_cas_cache", "ObjectStorage", "S3", "CAS"), ) assert len(expected_disks) == len(disks) for expected_disk in expected_disks: @@ -336,6 +348,16 @@ def check_system_tables(cluster, backup_query_id=None): "disk_s3_plain", id="from_local_to_s3_plain", ), + pytest.param( + "policy_cas", + "default", + id="from_cas_to_local", + ), + pytest.param( + "policy_cas", + "disk_s3", + id="from_cas_to_s3", + ), ], ) def test_backup_to_disk(cluster, storage_policy, to_disk): @@ -531,14 +553,29 @@ def test_backup_to_s3_multipart(cluster): "policy_s3", "policy_s3_other_bucket", "policy_s3_plain_rewritable", + "policy_cas", ], ) def test_backup_to_s3_native_copy(cluster, storage_policy): backup_name = new_backup_name() - backup_destination = f"S3('http://minio1:9001/root/data/backups/{backup_name}', 'minio', '{minio_secret_key}')" + backup_destination = s3_backup_destination(storage_policy, backup_name) (backup_events, restore_events) = check_backup_and_restore( cluster, storage_policy, backup_destination ) + + if storage_policy in CAS_POLICIES: + assert ( + backup_events.get("S3CopyObject", 0) + + backup_events.get("S3UploadPartCopy", 0) + > 0 + ) + assert ( + restore_events.get("S3CopyObject", 0) + + restore_events.get("S3UploadPartCopy", 0) + == 0 + ) + return + # single part upload assert backup_events["S3CopyObject"] > 0 assert restore_events["S3CopyObject"] > 0 @@ -556,6 +593,20 @@ def test_backup_to_s3_native_copy(cluster, storage_policy): ) +def test_backup_from_cas_to_s3_different_endpoint(cluster): + backup_name = new_backup_name() + backup_destination = f"S3('http://minio1:9001/root/data/backups/{backup_name}', 'minio', '{minio_secret_key}')" + backup_events, restore_events = check_backup_and_restore( + cluster, "policy_cas", backup_destination + ) + + for events in (backup_events, restore_events): + assert events.get("S3UploadPartCopy", 0) == 0 + assert events.get("S3CopyObject", 0) == 0 + assert events.get("S3GetObject", 0) > 0 + assert events.get("S3PutObject", 0) + events.get("S3UploadPart", 0) > 0 + + @pytest.mark.parametrize( "storage_policy", [ @@ -707,9 +758,8 @@ def test_backup_with_fs_cache( cluster, in_cache_initially, allow_backup_read_cache, allow_s3_native_copy ): storage_policy = "policy_s3_cache" - backup_name = new_backup_name() - backup_destination = f"S3('http://minio1:9001/root/data/backups/{backup_name}', 'minio', '{minio_secret_key}')" + backup_destination = s3_backup_destination(storage_policy, backup_name) insert_settings = { "enable_filesystem_cache_on_write_operations": int(in_cache_initially) diff --git a/tests/integration/test_backup_restore_storage_policy/configs/storage_config.xml b/tests/integration/test_backup_restore_storage_policy/configs/storage_config.xml index de9f68cbe870..6c841d4612e5 100644 --- a/tests/integration/test_backup_restore_storage_policy/configs/storage_config.xml +++ b/tests/integration/test_backup_restore_storage_policy/configs/storage_config.xml @@ -9,6 +9,15 @@ local /var/lib/disks/two/ + + object_storage + s3 + cas + itest-backup-restore-storage-policy + http://rustfs1:11121/test/backup_restore_storage_policy/ + clickhouse + clickhouse + @@ -25,6 +34,13 @@ + + + + cas + + + diff --git a/tests/integration/test_backup_restore_storage_policy/test.py b/tests/integration/test_backup_restore_storage_policy/test.py index b2e907dcaaa7..e45402408c6a 100644 --- a/tests/integration/test_backup_restore_storage_policy/test.py +++ b/tests/integration/test_backup_restore_storage_policy/test.py @@ -9,6 +9,7 @@ instance = cluster.add_instance( "instance", main_configs=["configs/storage_config.xml"], + with_rustfs=True, ) @@ -42,6 +43,12 @@ def create_table_backup(backup_name, storage_policy=None): create_query += f" SETTINGS storage_policy = '{storage_policy}'" instance.query(create_query) instance.query("INSERT INTO test.table SELECT number FROM numbers(10)") + assert instance.query("SELECT count(), sum(x) FROM test.table") == "10\t45\n" + if storage_policy is not None: + assert ( + instance.query("SELECT storage_policy FROM system.tables WHERE name='table'") + == f"{storage_policy}\n" + ) instance.query(f"BACKUP TABLE test.table TO {backup_name}") instance.query("DROP TABLE test.table SYNC") @@ -63,6 +70,11 @@ def restore_table(backup_name, storage_policy=None): ("policy1", "policy2", "policy2"), ("policy1", "", "default"), ("policy1", None, "policy1"), + ("cas_policy", None, "cas_policy"), + ("cas_policy", "policy1", "policy1"), + (None, "cas_policy", "cas_policy"), + ("policy1", "cas_policy", "cas_policy"), + ("cas_policy", "cas_policy", "cas_policy"), ], ) def test_storage_policies(origin_policy, restore_policy, expected_policy): @@ -74,3 +86,14 @@ def test_storage_policies(origin_policy, restore_policy, expected_policy): instance.query("SELECT storage_policy FROM system.tables WHERE name='table'") == f"{expected_policy}\n" ) + assert instance.query("SELECT count(), sum(x) FROM test.table") == "10\t45\n" + assert ( + instance.query( + "SELECT sum(rows) FROM system.parts WHERE database = 'test' AND table = 'table' AND active" + ) + == "10\n" + ) + assert ( + instance.query("CHECK TABLE test.table SETTINGS check_query_single_value_result = 1") + == "1\n" + ) diff --git a/tests/integration/test_database_backup/configs/backups.xml b/tests/integration/test_database_backup/configs/backups.xml index 32bc9dce7eec..d791dababff4 100644 --- a/tests/integration/test_database_backup/configs/backups.xml +++ b/tests/integration/test_database_backup/configs/backups.xml @@ -17,12 +17,37 @@ plain /local_plain/ + + s3_plain + http://rustfs1:11121/test/backup_disk_s3_plain_rustfs/ + clickhouse + clickhouse + + + object_storage + s3 + cas + itest-database-backup + http://rustfs1:11121/test/database_backup_cas/ + clickhouse + clickhouse + + + + + + cas + + + + backup_disk_local backup_disk_s3_plain backup_disk_object_storage_local_plain + backup_disk_s3_plain_rustfs /backups diff --git a/tests/integration/test_database_backup/test.py b/tests/integration/test_database_backup/test.py index 9dc7392d7797..9fcafffb4efb 100644 --- a/tests/integration/test_database_backup/test.py +++ b/tests/integration/test_database_backup/test.py @@ -10,6 +10,7 @@ main_configs=["configs/backups.xml"], stay_alive=True, with_minio=True, + with_rustfs=True, ) @@ -31,6 +32,12 @@ def cleanup_backup_files(instance): for s3_object in s3_objects: minio.remove_object(cluster.minio_bucket, s3_object) + rustfs = cluster.rustfs_client + for s3_object in rustfs.list_objects( + cluster.rustfs_bucket, prefix="backup_disk_s3_plain_rustfs/", recursive=True + ): + rustfs.remove_object(cluster.rustfs_bucket, s3_object.object_name) + @pytest.mark.parametrize( "backup_destination", @@ -39,9 +46,11 @@ def cleanup_backup_files(instance): "Disk('backup_disk_local', 'test_database_backup')", "Disk('backup_disk_s3_plain', 'test_database_backup')", "Disk('backup_disk_object_storage_local_plain', 'test_database_backup')", + "Disk('backup_disk_s3_plain_rustfs', 'test_database_backup')", ], ) -def test_database_backup_database(backup_destination): +@pytest.mark.parametrize("storage_policy", ["default", "cas_policy"]) +def test_database_backup_database(backup_destination, storage_policy): cleanup_backup_files(instance) instance.query( @@ -51,13 +60,13 @@ def test_database_backup_database(backup_destination): CREATE DATABASE test_database; - CREATE TABLE test_database.test_table_1 (id UInt64, value String) ENGINE=MergeTree ORDER BY id; + CREATE TABLE test_database.test_table_1 (id UInt64, value String) ENGINE=MergeTree ORDER BY id SETTINGS storage_policy = '{storage_policy}'; INSERT INTO test_database.test_table_1 VALUES (0, 'test_database.test_table_1'); - CREATE TABLE test_database.test_table_2 (id UInt64, value String) ENGINE=MergeTree ORDER BY id; + CREATE TABLE test_database.test_table_2 (id UInt64, value String) ENGINE=MergeTree ORDER BY id SETTINGS storage_policy = '{storage_policy}'; INSERT INTO test_database.test_table_2 VALUES (0, 'test_database.test_table_2'); - CREATE TABLE test_database.test_table_3 (id UInt64, value String) ENGINE=MergeTree ORDER BY id; + CREATE TABLE test_database.test_table_3 (id UInt64, value String) ENGINE=MergeTree ORDER BY id SETTINGS storage_policy = '{storage_policy}'; INSERT INTO test_database.test_table_3 VALUES (0, 'test_database.test_table_3'); BACKUP DATABASE test_database TO {backup_destination}; @@ -109,9 +118,11 @@ def test_database_backup_database(backup_destination): "Disk('backup_disk_local', 'test_table_backup')", "Disk('backup_disk_s3_plain', 'test_table_backup')", "Disk('backup_disk_object_storage_local_plain', 'test_table_backup')", + "Disk('backup_disk_s3_plain_rustfs', 'test_table_backup')", ], ) -def test_database_backup_table(backup_destination): +@pytest.mark.parametrize("storage_policy", ["default", "cas_policy"]) +def test_database_backup_table(backup_destination, storage_policy): cleanup_backup_files(instance) instance.query( @@ -121,7 +132,7 @@ def test_database_backup_table(backup_destination): CREATE DATABASE test_database; - CREATE TABLE test_database.test_table (id UInt64, value String) ENGINE=MergeTree ORDER BY id; + CREATE TABLE test_database.test_table (id UInt64, value String) ENGINE=MergeTree ORDER BY id SETTINGS storage_policy = '{storage_policy}'; INSERT INTO test_database.test_table VALUES (0, 'test_database.test_table'); BACKUP TABLE test_database.test_table TO {backup_destination};