ClickHouse/tests/integration/test_backup_restore/test.py

209 lines
7.8 KiB
Python
Raw Normal View History

import os.path
import pytest
2021-06-02 14:33:47 +00:00
import logging
from helpers.cluster import ClickHouseCluster
from helpers.test_tools import TSV
cluster = ClickHouseCluster(__file__)
instance = cluster.add_instance("node")
path_to_data = "/var/lib/clickhouse/"
@pytest.fixture(scope="module")
def started_cluster():
try:
cluster.start()
instance.query(
"CREATE DATABASE test ENGINE = Ordinary"
) # Different path in shadow/ with Atomic
2021-06-02 14:33:47 +00:00
instance.query("DROP TABLE IF EXISTS test.tbl")
instance.query(
"CREATE TABLE test.tbl (p Date, k Int8) ENGINE = MergeTree PARTITION BY toYYYYMM(p) ORDER BY p"
)
2021-06-02 14:33:47 +00:00
for i in range(1, 4):
instance.query(
"INSERT INTO test.tbl (p, k) VALUES(toDate({}), {})".format(i, i)
)
2021-06-02 14:33:47 +00:00
for i in range(31, 34):
instance.query(
"INSERT INTO test.tbl (p, k) VALUES(toDate({}), {})".format(i, i)
)
2021-06-02 14:33:47 +00:00
expected = TSV(
"1970-01-02\t1\n1970-01-03\t2\n1970-01-04\t3\n1970-02-01\t31\n1970-02-02\t32\n1970-02-03\t33"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl FREEZE")
yield cluster
finally:
cluster.shutdown()
2021-10-01 19:05:36 +00:00
def get_last_backup_path(instance, database, table):
fp_increment = os.path.join(path_to_data, "shadow/increment.txt")
increment = instance.exec_in_container(["cat", fp_increment]).strip()
return os.path.join(path_to_data, "shadow", increment, "data", database, table)
2021-06-02 14:33:47 +00:00
def copy_backup_to_detached(instance, database, src_table, dst_table):
fp_backup = os.path.join(path_to_data, "shadow", "*", "data", database, src_table)
fp_detached = os.path.join(path_to_data, "data", database, dst_table, "detached")
logging.debug(f"copy from {fp_backup} to {fp_detached}")
instance.exec_in_container(["bash", "-c", f"cp -r {fp_backup} -T {fp_detached}"])
2021-06-02 14:33:47 +00:00
def test_restore(started_cluster):
instance.query("CREATE TABLE test.tbl1 AS test.tbl")
copy_backup_to_detached(started_cluster.instances["node"], "test", "tbl", "tbl1")
# The data_version of parts to be attached are larger than the newly created table's data_version.
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl1 ATTACH PARTITION 197001")
instance.query("ALTER TABLE test.tbl1 ATTACH PARTITION 197002")
instance.query("SELECT sleep(2)")
# Validate the attached parts are identical to the backup.
expected = TSV(
"1970-01-02\t1\n1970-01-03\t2\n1970-01-04\t3\n1970-02-01\t31\n1970-02-02\t32\n1970-02-03\t33"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl1 ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl1 UPDATE k=10 WHERE 1")
instance.query("SELECT sleep(2)")
# Validate mutation has been applied to all attached parts.
expected = TSV(
"1970-01-02\t10\n1970-01-03\t10\n1970-01-04\t10\n1970-02-01\t10\n1970-02-02\t10\n1970-02-03\t10"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl1 ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("DROP TABLE IF EXISTS test.tbl1")
2021-06-02 14:33:47 +00:00
def test_attach_partition(started_cluster):
instance.query("CREATE TABLE test.tbl2 AS test.tbl")
for i in range(3, 5):
instance.query(
"INSERT INTO test.tbl2(p, k) VALUES(toDate({}), {})".format(i, i)
)
for i in range(33, 35):
instance.query(
"INSERT INTO test.tbl2(p, k) VALUES(toDate({}), {})".format(i, i)
)
expected = TSV("1970-01-04\t3\n1970-01-05\t4\n1970-02-03\t33\n1970-02-04\t34")
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl2 ORDER BY p")
assert TSV(res) == expected
copy_backup_to_detached(started_cluster.instances["node"], "test", "tbl", "tbl2")
# The data_version of parts to be attached
# - may be less than, equal to or larger than the current table's data_version.
# - may intersect with the existing parts of a partition.
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl2 ATTACH PARTITION 197001")
instance.query("ALTER TABLE test.tbl2 ATTACH PARTITION 197002")
instance.query("SELECT sleep(2)")
expected = TSV(
"1970-01-02\t1\n1970-01-03\t2\n1970-01-04\t3\n1970-01-04\t3\n1970-01-05\t4\n1970-02-01\t31\n1970-02-02\t32\n1970-02-03\t33\n1970-02-03\t33\n1970-02-04\t34"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl2 ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl2 UPDATE k=10 WHERE 1")
instance.query("SELECT sleep(2)")
# Validate mutation has been applied to all attached parts.
expected = TSV(
"1970-01-02\t10\n1970-01-03\t10\n1970-01-04\t10\n1970-01-04\t10\n1970-01-05\t10\n1970-02-01\t10\n1970-02-02\t10\n1970-02-03\t10\n1970-02-03\t10\n1970-02-04\t10"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl2 ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("DROP TABLE IF EXISTS test.tbl2")
2021-06-02 14:33:47 +00:00
def test_replace_partition(started_cluster):
instance.query("CREATE TABLE test.tbl3 AS test.tbl")
for i in range(3, 5):
instance.query(
"INSERT INTO test.tbl3(p, k) VALUES(toDate({}), {})".format(i, i)
)
for i in range(33, 35):
instance.query(
"INSERT INTO test.tbl3(p, k) VALUES(toDate({}), {})".format(i, i)
)
expected = TSV("1970-01-04\t3\n1970-01-05\t4\n1970-02-03\t33\n1970-02-04\t34")
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl3 ORDER BY p")
assert TSV(res) == expected
copy_backup_to_detached(started_cluster.instances["node"], "test", "tbl", "tbl3")
# The data_version of parts to be copied
# - may be less than, equal to or larger than the current table data_version.
# - may intersect with the existing parts of a partition.
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl3 REPLACE PARTITION 197002 FROM test.tbl")
instance.query("SELECT sleep(2)")
expected = TSV(
"1970-01-04\t3\n1970-01-05\t4\n1970-02-01\t31\n1970-02-02\t32\n1970-02-03\t33"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl3 ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("ALTER TABLE test.tbl3 UPDATE k=10 WHERE 1")
instance.query("SELECT sleep(2)")
# Validate mutation has been applied to all copied parts.
expected = TSV(
"1970-01-04\t10\n1970-01-05\t10\n1970-02-01\t10\n1970-02-02\t10\n1970-02-03\t10"
)
2021-06-02 14:33:47 +00:00
res = instance.query("SELECT * FROM test.tbl3 ORDER BY p")
assert TSV(res) == expected
2021-06-02 14:33:47 +00:00
instance.query("DROP TABLE IF EXISTS test.tbl3")
2021-10-01 19:05:36 +00:00
2021-10-01 19:05:36 +00:00
def test_freeze_in_memory(started_cluster):
instance.query(
"CREATE TABLE test.t_in_memory(a UInt32, s String) ENGINE = MergeTree ORDER BY a SETTINGS min_rows_for_compact_part = 1000"
)
2021-10-01 19:05:36 +00:00
instance.query("INSERT INTO test.t_in_memory VALUES (1, 'a')")
instance.query("ALTER TABLE test.t_in_memory FREEZE")
fp_backup = get_last_backup_path(
started_cluster.instances["node"], "test", "t_in_memory"
)
part_path = fp_backup + "/all_1_1_0/"
assert TSV(
instance.query(
"SELECT part_type, is_frozen FROM system.parts WHERE database = 'test' AND table = 't_in_memory'"
)
) == TSV("InMemory\t1\n")
instance.exec_in_container(["test", "-f", part_path + "/data.bin"])
assert instance.exec_in_container(["cat", part_path + "/count.txt"]).strip() == "1"
instance.query(
"CREATE TABLE test.t_in_memory_2(a UInt32, s String) ENGINE = MergeTree ORDER BY a"
)
copy_backup_to_detached(
started_cluster.instances["node"], "test", "t_in_memory", "t_in_memory_2"
)
2021-10-01 19:05:36 +00:00
instance.query("ALTER TABLE test.t_in_memory_2 ATTACH PARTITION ID 'all'")
assert TSV(
instance.query(
"SELECT part_type FROM system.parts WHERE database = 'test' AND table = 't_in_memory_2'"
)
) == TSV("Compact\n")
2021-10-01 19:05:36 +00:00
assert TSV(instance.query("SELECT a, s FROM test.t_in_memory_2")) == TSV("1\ta\n")