AmateurMergeTree: ci

korowa · korowa · commit c34bf42e0cc2 · 2026-07-16T06:42:30.000+03:00
diff --git a/.github/workflows/amateur_ci.yaml b/.github/workflows/amateur_ci.yaml
@@ -0,0 +1,54 @@
+---
+name: AmateurCI
+
+on:
+  push:
+    branches: ['amateur-merge-tree']
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    name: "Build"
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v6
+        with:
+          submodules: recursive
+
+      - name: Restore ccache
+        uses: actions/cache/restore@v4
+        with:
+          path: /tmp/ccache
+          key: ccache-${{ github.sha }}
+          restore-keys: |
+            ccache-
+
+      - name: Build
+        run: |
+          docker run --rm -v $PWD:/ClickHouse -v /tmp/ccache:/tmp/ccache clickhouse/fasttest:b63954f8a1b316d0686e \
+            /bin/bash -c "
+              set -e
+              apt-get update && apt-get install -y ccache
+              export CC=clang-21
+              export CXX=clang++-21
+              export CCACHE_DIR=/tmp/ccache
+              cd /ClickHouse/
+              cmake -DCOMPILER_CACHE=ccache -DCMAKE_BUILD_TYPE=Debug -DOMIT_HEAVY_DEBUG_SYMBOLS=1 -DENABLE_LIBRARIES=0 -DENABLE_UTILS=0 -DENABLE_TESTS=0 -DENABLE_RUST=0 -DENABLE_JEMALLOC=1 -DENABLE_YAML_CPP=1 -DENABLE_AWS_S3=1 -G Ninja -S . -B build
+              ninja -C build
+              ccache --show-stats
+            "
+
+      - name: Save ccache
+        if: github.ref_name == 'amateur-merge-tree'
+        uses: actions/cache/save@v4
+        with:
+          path: /tmp/ccache
+          key: ccache-${{ github.sha }}
+
+      - name: Integration tests
+        run: |
+          docker run --rm -v /var/run/docker.sock:/var/run/docker.sock -v $PWD:$PWD --network host clickhouse/integration-tests-runner:c909f2e80ec78b3a6532 \
+            /bin/bash -c "
+              CLICKHOUSE_TESTS_SERVER_BIN_PATH=$PWD/build/programs/clickhouse \
+              pytest $PWD/tests/integration/test_amateur_merge_tree/ -v --timeout=600
+            "
diff --git a/tests/integration/test_amateur_merge_tree/__init__.py b/tests/integration/test_amateur_merge_tree/__init__.py
diff --git a/tests/integration/test_amateur_merge_tree/configs/config.yaml b/tests/integration/test_amateur_merge_tree/configs/config.yaml
@@ -0,0 +1,17 @@
+---
+storage_configuration:
+  disks:
+    s3:
+      type: object_storage
+      object_storage_type: s3
+      metadata_type: amateur
+      endpoint: http://minio1:9001/root/data/
+      keeper_prefix: /clickhouse/disks/s3
+      access_key_id: minio
+      secret_access_key: ClickHouse_Minio_P@ssw0rd
+  policies:
+    s3:
+      volumes:
+      - main:
+          disks:
+          - s3
diff --git a/tests/integration/test_amateur_merge_tree/helpers.py b/tests/integration/test_amateur_merge_tree/helpers.py
@@ -0,0 +1,18 @@
+import time
+
+
+def wait_table_sync(zk_client, zk_table_path):
+    _, parts_stat = zk_client.get(f"{zk_table_path}/parts")
+    nodes = zk_client.get_children(f"{zk_table_path}/replicas")
+
+    for _ in range(60):
+        nodes_in_sync = 0
+        for node in nodes:
+            node_epoch, _ = zk_client.get(f"{zk_table_path}/replicas/{node}/epoch")
+            if int(node_epoch) >= parts_stat.pzxid:
+                nodes_in_sync += 1
+
+        if nodes_in_sync == len(nodes):
+            return
+
+        time.sleep(0.5)
diff --git a/tests/integration/test_amateur_merge_tree/test.py b/tests/integration/test_amateur_merge_tree/test.py
@@ -0,0 +1,286 @@
+#!/usr/bin/env python3
+
+import time
+import pytest
+
+from helpers.cluster import ClickHouseCluster
+from helpers.test_tools import TSV
+
+from test_amateur_merge_tree.helpers import wait_table_sync
+
+cluster = ClickHouseCluster(__file__)
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.add_instance(
+            "node1",
+            main_configs=["configs/config.yaml"],
+            with_minio=True,
+            with_zookeeper=True,
+            stay_alive=True,
+            use_keeper=False,
+            cpu_limit=3,
+        )
+        cluster.add_instance(
+            "node2",
+            main_configs=["configs/config.yaml"],
+            with_minio=True,
+            with_zookeeper=True,
+            stay_alive=True,
+            use_keeper=False,
+            cpu_limit=3
+        )
+        cluster.start()
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+
+def test_insert_select(started_cluster):
+    """
+    Basic scenario with inserts and replica synchronization
+    """
+
+    for node_name, node in cluster.instances.items():
+        data = node.query(f"""
+            CREATE TABLE test_basic (key UInt32, val String)
+            ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_basic', '{node_name}')
+            ORDER BY key
+            SETTINGS storage_policy = 's3'
+        """)
+
+    for i in range(10):
+        node = list(cluster.instances.values())[i % 2]
+        node.query(f"""
+            INSERT INTO test_basic SELECT number + {i} * 10 as key, 'val-' || key AS val FROM numbers(10)
+        """)
+
+    zk_client = cluster.get_kazoo_client("zoo1")
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_basic")
+
+    expected = [[100, 0, 'val-0', 99, 'val-99']]
+    for node_name, node in cluster.instances.items():
+        actual = node.query("""
+            SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key) FROM test_basic
+        """)
+
+        assert TSV(actual) == TSV(expected), f"Received an unexpected result from node {node_name}"
+
+
+def test_drop_partititon(started_cluster):
+    """
+    Insert into partitioned tables and drop partition
+    """
+
+    for node_name, node in cluster.instances.items():
+        data = node.query(f"""
+            CREATE TABLE test_partitioned (key UInt32, prt UInt32, val String)
+            ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_partitioned', '{node_name}')
+            PARTITION BY prt
+            ORDER BY key
+            SETTINGS storage_policy = 's3'
+        """)
+
+    for i in range(10):
+        node = list(cluster.instances.values())[i % 2]
+        node.query(f"""
+            INSERT INTO test_partitioned SELECT number + {i} * 10 as key, number % 5 AS prt, 'val-' || key FROM numbers(10)
+        """)
+
+    cluster.instances["node1"].query("ALTER TABLE test_partitioned DROP PARTITION 1")
+    cluster.instances["node2"].query("ALTER TABLE test_partitioned DROP PARTITION 3")
+
+    zk_client = cluster.get_kazoo_client("zoo1")
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_partitioned")
+
+    expected = [[60, 0, 'val-0', 99, 'val-99', 0, 0]]
+    for node_name, node in cluster.instances.items():
+        actual = node.query("""
+            SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key),
+                   countIf(prt = 1), countIf(prt = 3)
+            FROM test_partitioned
+        """)
+
+        assert TSV(actual) == TSV(expected), f"Received an unexpected result from node {node_name}"
+
+
+def test_optimize(started_cluster):
+    """
+    Run optimize on both replicas
+    """
+
+    for node_name, node in cluster.instances.items():
+        data = node.query(f"""
+            CREATE TABLE test_optimize(key UInt32, val String)
+            ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_optimize', '{node_name}')
+            ORDER BY key
+            SETTINGS storage_policy = 's3'
+        """)
+
+    for i in range(10):
+        node = list(cluster.instances.values())[i % 2]
+        node.query(f"""
+            INSERT INTO test_optimize SELECT number + {i} * 10 as key, 'val-' || key FROM numbers(10)
+        """)
+
+    zk_client = cluster.get_kazoo_client("zoo1")
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_optimize")
+
+    for node_name, node in cluster.instances.items():
+        actual = node.query("OPTIMIZE TABLE test_optimize FINAL")
+
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_optimize")
+
+    expected = [[60, 0, 'val-0', 99, 'val-99', 0, 0, 'all_0_9']]
+    for node_name, node in cluster.instances.items():
+        actual = node.query("""
+            SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key),
+                   any(substr(_part, 1, 7))
+            FROM test_optimize
+        """)
+
+
+def test_add_remove_replica(started_cluster):
+    """
+    Add table replica while the data already exists
+    """
+
+    cluster.instances["node1"].query("""
+        CREATE TABLE test_add_remove_replica (key UInt32, val String)
+        ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_add_remove_replica', 'node1')
+        ORDER BY key
+        SETTINGS storage_policy = 's3'
+    """)
+
+    for i in range(10):
+        cluster.instances["node1"].query(f"""
+            INSERT INTO test_add_remove_replica SELECT number + {i} * 10 as key, 'val-' || key AS val FROM numbers(10)
+        """)
+
+    expected = [[100, 0, 'val-0', 99, 'val-99']]
+
+    actual_first = cluster.instances["node1"].query("""
+        SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key) FROM test_add_remove_replica
+    """)
+    assert TSV(actual_first) == TSV(expected), "Received an unexpected result from node1"
+
+    cluster.instances["node2"].query("""
+        CREATE TABLE test_add_remove_replica (key UInt32, val String)
+        ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_add_remove_replica', 'node2')
+        ORDER BY key
+        SETTINGS storage_policy = 's3'
+    """)
+
+    zk_client = cluster.get_kazoo_client("zoo1")
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_add_remove_replica")
+
+    actual_replica = cluster.instances["node2"].query("""
+        SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key) FROM test_add_remove_replica
+    """)
+    assert TSV(actual_first) == TSV(expected), "Received an unexpected result from node2"
+
+    cluster.instances["node2"].query("DROP TABLE test_add_remove_replica SYNC")
+
+    actual_first = cluster.instances["node1"].query("""
+        SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key) FROM test_add_remove_replica
+    """)
+    assert TSV(actual_first) == TSV(expected), "Received an unexpected result from node1 after replica drop"
+
+
+def test_epoch_hold_by_snapshot(started_cluster):
+    """
+    Assert epoch (used for cleanup) not propagating when snapshot is hold by long running query
+    """
+
+    for node_name, node in cluster.instances.items():
+        data = node.query(f"""
+            CREATE TABLE test_epoch_hold_by_snapshot (key UInt32, val String)
+            ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_epoch_hold_by_snapshot', '{node_name}')
+            ORDER BY key
+            SETTINGS storage_policy = 's3', old_parts_lifetime = 1
+        """)
+
+    cluster.instances["node1"].query(f"""
+        INSERT INTO test_epoch_hold_by_snapshot SELECT number, 'val-' || number FROM numbers(10)
+    """)
+
+    zk_client = cluster.get_kazoo_client("zoo1")
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_epoch_hold_by_snapshot")
+
+    node1_initial_epoch = cluster.instances["node1"].query("""
+        SELECT value FROM system.zookeeper WHERE path = '/clickhouse/tables/default/test_epoch_hold_by_snapshot/replicas/node1'
+    """).strip()
+    node2_initial_epoch = cluster.instances["node1"].query("""
+        SELECT value FROM system.zookeeper WHERE path = '/clickhouse/tables/default/test_epoch_hold_by_snapshot/replicas/node2'
+    """).strip()
+
+    assert int(node1_initial_epoch) == int(node2_initial_epoch), "Expected initial epochs to be equal"
+
+    response = cluster.instances["node2"].get_query_request(
+        f"SELECT * FROM loop(test_epoch_hold_by_snapshot) settings max_threads = 1",
+        query_id="hold_snapshot"
+    )
+
+    for _ in range(3):
+        cluster.instances["node1"].query(f"""
+            INSERT INTO test_epoch_hold_by_snapshot SELECT number, 'val-' || number FROM numbers(10)
+        """)
+
+    cluster.instances["node1"].query("""
+        OPTIMIZE TABLE test_epoch_hold_by_snapshot FINAL
+        SETTINGS alter_sync = 1, optimize_throw_if_noop = 1
+    """)
+
+    time.sleep(2)
+
+    node1_after_opt_epoch = cluster.instances["node1"].query("""
+        SELECT value FROM system.zookeeper WHERE path = '/clickhouse/tables/default/test_epoch_hold_by_snapshot/replicas/node1'
+    """).strip()
+    node2_after_opt_epoch = cluster.instances["node1"].query("""
+        SELECT value FROM system.zookeeper WHERE path = '/clickhouse/tables/default/test_epoch_hold_by_snapshot/replicas/node2'
+    """).strip()
+
+    assert int(node1_after_opt_epoch) > int(node2_after_opt_epoch), "Expected node1 epoch to be greater than node2 epoch"
+    assert int(node2_initial_epoch) == int(node2_after_opt_epoch), "Expected node2 unchanged after inserts / optimize"
+
+    response = cluster.instances["node2"].query("KILL QUERY WHERE query_id = 'hold_snapshot' SYNC")
+
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_epoch_hold_by_snapshot")
+
+    node2_after_sync_epoch = cluster.instances["node1"].query("""
+        SELECT value FROM system.zookeeper WHERE path = '/clickhouse/tables/default/test_epoch_hold_by_snapshot/replicas/node2'
+    """).strip()
+
+    assert int(node1_after_opt_epoch) == int(node2_after_sync_epoch), "Expected epochs to be equal after final sync"
+
+
+def test_deduplicate(started_cluster):
+    """
+    Assert block deduplication
+    """
+
+    for node_name, node in cluster.instances.items():
+        data = node.query(f"""
+            CREATE TABLE test_deduplicate(key UInt32, val String)
+            ENGINE = AmateurMergeTree('/clickhouse/tables/default/test_deduplicate', '{node_name}')
+            ORDER BY key
+            SETTINGS storage_policy = 's3'
+        """)
+
+    for i in range(6):
+        node = list(cluster.instances.values())[i % 2]
+        node.query(f"""
+            INSERT INTO test_deduplicate VALUES (0, 'a'), (1, 'b'), (2, 'c'), (4, 'd'), (5, 'e')
+        """)
+
+    zk_client = cluster.get_kazoo_client("zoo1")
+    wait_table_sync(zk_client, "/clickhouse/tables/default/test_deduplicate")
+
+    expected = [[5, 0, 'a', 5, 'e']]
+    for node_name, node in cluster.instances.items():
+        actual = node.query("""
+            SELECT count(), min(key), argMin(val, key), max(key), argMax(val, key),
+            FROM test_deduplicate
+        """)
diff --git a/tests/integration/test_amateur_merge_tree/test_failpoints.py b/tests/integration/test_amateur_merge_tree/test_failpoints.py
