ClickHouse/tests/integration/test_attach_partition_with_large_destination/test.py

import pytest

from helpers.cluster import ClickHouseCluster

cluster = ClickHouseCluster(__file__)
node = cluster.add_instance(
    "node", main_configs=["configs/config.xml"], with_zookeeper=True
)


@pytest.fixture(scope="module")
def started_cluster():
    try:
        cluster.start()
        yield cluster
    finally:
        cluster.shutdown()


def create_force_drop_flag(node):
    force_drop_flag_path = "/var/lib/clickhouse/flags/force_drop_table"
    node.exec_in_container(
        [
            "bash",
            "-c",
            "touch {} && chmod a=rw {}".format(
                force_drop_flag_path, force_drop_flag_path
            ),
        ],
        user="root",
    )


@pytest.mark.parametrize("engine", ["Ordinary", "Atomic"])
def test_attach_partition_with_large_destination(started_cluster, engine):
    # Initialize
    node.query("CREATE DATABASE db ENGINE={}".format(engine))
    node.query(
        "CREATE TABLE db.destination (n UInt64) ENGINE=ReplicatedMergeTree('/test/destination', 'r1') ORDER BY n PARTITION BY n % 2"
    )
    node.query(
        "CREATE TABLE db.source_1 (n UInt64) ENGINE=ReplicatedMergeTree('/test/source_1', 'r1') ORDER BY n PARTITION BY n % 2"
    )
    node.query("INSERT INTO db.source_1 VALUES (1), (2), (3), (4)")
    node.query(
        "CREATE TABLE db.source_2 (n UInt64) ENGINE=ReplicatedMergeTree('/test/source_2', 'r1') ORDER BY n PARTITION BY n % 2"
    )
    node.query("INSERT INTO db.source_2 VALUES (5), (6), (7), (8)")

    # Attach partition when destination partition is empty
    node.query("ALTER TABLE db.destination ATTACH PARTITION 0 FROM db.source_1")
    assert node.query("SELECT n FROM db.destination ORDER BY n") == "2\n4\n"

    # REPLACE PARTITION should still respect max_partition_size_to_drop
    assert node.query_and_get_error(
        "ALTER TABLE db.destination REPLACE PARTITION 0 FROM db.source_2"
    )
    assert node.query("SELECT n FROM db.destination ORDER BY n") == "2\n4\n"

    # Attach partition when destination partition is larger than max_partition_size_to_drop
    node.query("ALTER TABLE db.destination ATTACH PARTITION 0 FROM db.source_2")
    assert node.query("SELECT n FROM db.destination ORDER BY n") == "2\n4\n6\n8\n"

    # Cleanup
    create_force_drop_flag(node)
    node.query("DROP TABLE db.source_1 SYNC")
    create_force_drop_flag(node)
    node.query("DROP TABLE db.source_2 SYNC")
    create_force_drop_flag(node)
    node.query("DROP TABLE db.destination SYNC")
    node.query("DROP DATABASE db")