Repository: ClickHouse/ClickHouse
File: tests/integration/test_storage_s3_queue/test_migration.py (Python)

Code before:


@pytest.mark.not_repeatable
@pytest.mark.parametrize("setting_prefix", ["", "s3queue_"])
@pytest.mark.parametrize("buckets_num", [3, 1])
def test_migration(started_cluster, setting_prefix, buckets_num):
    node1 = started_cluster.instances["instance_24.5"]
    node2 = started_cluster.instances["instance2_24.5"]

    for node in [node1, node2]:
        if "24.5" not in node.query("select version()").strip():
            node.restart_with_original_version()

    table_name = f"test_replicated_{uuid.uuid4().hex[:8]}"
    dst_table_name = f"{table_name}_dst"
    mv_name = f"{table_name}_mv"
    keeper_path = f"/clickhouse/test_{table_name}_{buckets_num}"
    files_path = f"{table_name}_data"

    for node in [node1, node2]:
        node.query("DROP DATABASE IF EXISTS r")

    node1.query(
        "CREATE DATABASE r ENGINE=Replicated('/clickhouse/databases/replicateddb3', 'shard1', 'node1')"
    )
    node2.query(
        "CREATE DATABASE r ENGINE=Replicated('/clickhouse/databases/replicateddb3', 'shard1', 'node2')"
    )

    create_table(
        started_cluster,
        node1,
        table_name,
        "ordered",
        files_path,
        version="24.5",
        additional_settings={
            "keeper_path": keeper_path,
            "s3queue_polling_min_timeout_ms": 100,
            "s3queue_polling_max_timeout_ms": 1000,
            "s3queue_polling_backoff_ms": 100,
        },
        database_name="r",
    )

    for node in [node1, node2]:
        create_mv(node, f"r.{table_name}", dst_table_name, mv_name=mv_name)

    start_ind = [0]
    expected_rows = [0]
    last_processed_path = [""]

Review comment (suggestion):
Likely this could be fixed by adding a flag `clear_data_dir=True` in `node.restart_with_original_version()` above.

Code after:


@pytest.mark.not_repeatable
@pytest.mark.parametrize("setting_prefix", ["", "s3queue_"])
@pytest.mark.parametrize("buckets_num", [3, 1])
def test_migration(started_cluster, setting_prefix, buckets_num):
    node1 = started_cluster.instances["instance_24.5"]
    node2 = started_cluster.instances["instance2_24.5"]

    for node in [node1, node2]:
        if "24.5" not in node.query("select version()").strip():
            node.restart_with_original_version()

    table_name = f"test_replicated_{uuid.uuid4().hex[:8]}"
    dst_table_name = f"{table_name}_dst"
    mv_name = f"{table_name}_mv"
    keeper_path = f"/clickhouse/test_{table_name}_{buckets_num}"
    files_path = f"{table_name}_data"

    for node in [node1, node2]:
        node.query("DROP DATABASE IF EXISTS r SYNC")

    # Clean up the ZK path in case DROP DATABASE on an older binary didn't fully remove it
    zk_client = started_cluster.get_kazoo_client("zoo1")
    if zk_client.exists("/clickhouse/databases/replicateddb3"):
        zk_client.delete("/clickhouse/databases/replicateddb3", recursive=True)

    node1.query(
        "CREATE DATABASE r ENGINE=Replicated('/clickhouse/databases/replicateddb3', 'shard1', 'node1')"
    )
    node2.query(
        "CREATE DATABASE r ENGINE=Replicated('/clickhouse/databases/replicateddb3', 'shard1', 'node2')"
    )

    create_table(
        started_cluster,
        node1,
        table_name,
        "ordered",
        files_path,
        version="24.5",
        additional_settings={
            "keeper_path": keeper_path,
            "s3queue_polling_min_timeout_ms": 100,
            "s3queue_polling_max_timeout_ms": 1000,
            "s3queue_polling_backoff_ms": 100,
        },
        database_name="r",
    )

    for node in [node1, node2]: