Antalya-26.6 - Backport flaky-fix commits from upstream (2026-08-10)

tests/config/config.d/storage_conf.xml

@@ -123,7 +123,13 @@
             <local_remote>
                 <volumes>
                     <local><disk>default</disk></local>
-                    <remote><disk>s3_disk</disk></remote>
+                    <!-- A part always starts on the 'local' volume, even when its move TTL
+                         expired while the INSERT was still running. The background mover
+                         ignores this setting, so TTL moves still happen. -->
+                    <remote>
+                        <disk>s3_disk</disk>
+                        <perform_ttl_move_on_insert>0</perform_ttl_move_on_insert>
+                    </remote>
                 </volumes>
             </local_remote>
             <s3_cache>

tests/integration/test_attach_table_from_s3_plain_readonly/test.py

@@ -77,17 +77,25 @@ def test_attach_table_from_s3_plain_readonly(started_cluster):
 
     assert int(node1.query("select num from local_db.test_table limit 1")) == 5
 
-    # Copy local MergeTree data into minio bucket
-    table_data_path = os.path.join(node1.path, "database/store")
+    table_uuid = node1.query(
+        "SELECT uuid FROM system.tables WHERE database='local_db' AND table='test_table'"
+    ).strip()
+
+    # Copy local MergeTree data into minio bucket. Scoped to this table: store/
+    # also holds the Atomic `system` database, whose parts a background flush can
+    # remove mid-walk.
+    table_data_path = os.path.join(
+        node1.path, "database/store", table_uuid[:3], table_uuid
+    )
     minio = cluster.minio_client
     upload_to_minio(
-        minio, cluster.minio_bucket, table_data_path, "data/disks/disk_s3_plain/store/"
+        minio,
+        cluster.minio_bucket,
+        table_data_path,
+        f"data/disks/disk_s3_plain/store/{table_uuid[:3]}/{table_uuid}/",
     )
 
     # Drop the non-replicated table, we don't need it anymore
-    table_uuid = node1.query(
-        "SELECT uuid FROM system.tables WHERE database='local_db' AND table='test_table'"
-    ).strip()
     node1.query("drop table local_db.test_table SYNC;")
 
     # Create a replicated database

tests/integration/test_dirty_pages_force_purge/configs/overrides.yaml

@@ -1,3 +1,7 @@
 ---
 max_server_memory_usage: 4Gi
-memory_worker_purge_dirty_pages_threshold_ratio: 0.2
+# The threshold must be reliably exceeded by a single iteration of the test query
+# (peak usage ~417 MiB). With a higher ratio, `pdirty` may plateau below the threshold,
+# because jemalloc reuses dirty pages within the same arenas across iterations,
+# and the number of touched arenas depends on the machine and the number of threads.
+memory_worker_purge_dirty_pages_threshold_ratio: 0.05

tests/integration/test_keeper_back_to_back/test.py

@@ -1,6 +1,7 @@
 import os
 import random
 import string
+import threading
 import time
 from multiprocessing.dummy import Pool
 
@@ -755,9 +756,19 @@ def test_concurrent_watches(started_cluster, request):
         all_paths_triggered = []
 
         existing_path = []
+        # A watch is one-shot per (path, session), so a mutation only notifies a registration
+        # that is live at that moment. Mutating a path this thread does not hold a token for can
+        # therefore consume another thread's registration, leaving it forever unnotified.
+        existing_path_lock = threading.Lock()
         all_paths_created = []
         watches_created = 0
 
+        def claim_path():
+            with existing_path_lock:
+                if not existing_path:
+                    return None
+                return existing_path.pop(random.randrange(len(existing_path)))
+
         def create_path_and_watch(i):
             nonlocal watches_created
             nonlocal all_paths_created
@@ -775,34 +786,28 @@ def dumb_watch(event):
             fake_zk.get(global_path + "/" + str(i), watch=dumb_watch)
             all_paths_created.append(global_path + "/" + str(i))
             watches_created += 1
-            existing_path.append(i)
+            with existing_path_lock:
+                existing_path.append(i)
 
         trigger_called = 0
 
         def trigger_watch(i):
             nonlocal trigger_called
             trigger_called += 1
             fake_zk.set(global_path + "/" + str(i), b"somevalue")
-            try:
-                existing_path.remove(i)
-            except:
-                pass
 
         def call(total):
             for i in range(total):
                 create_path_and_watch(random.randint(0, 1000))
                 time.sleep(random.random() % 0.5)
-                try:
-                    rand_num = random.choice(existing_path)
-                    trigger_watch(rand_num)
-                except:
-                    pass
-            while existing_path:
-                try:
-                    rand_num = random.choice(existing_path)
+                rand_num = claim_path()
+                if rand_num is not None:
                     trigger_watch(rand_num)
-                except:
-                    pass
+            while True:
+                rand_num = claim_path()
+                if rand_num is None:
+                    break
+                trigger_watch(rand_num)
 
         p = Pool(10)
         arguments = [100] * 10

tests/integration/test_keeper_dynamic_log_level/test.py

@@ -60,21 +60,24 @@ def test_adjust_log_level(start_cluster):
             """,
         ]
     )
-    time.sleep(3)
-    node.query(
-        "SELECT * FROM system.zookeeper SETTINGS allow_unrestricted_reads_from_keeper = 'true'"
-    )
-    node.exec_in_container(
-        [
-            "bash",
-            "-c",
-            "sync",
-        ],
-        privileged=True,
-        user="root",
-    )
-    assert (
-        int(
+    # The config reloader applies the new logger settings asynchronously (it polls the config
+    # every couple of seconds), so poll until trace logging becomes active instead of relying
+    # on a fixed sleep, which is not enough on slow (e.g. sanitizer) runs.
+    trace_lines = 0
+    for _ in range(60):
+        node.query(
+            "SELECT * FROM system.zookeeper SETTINGS allow_unrestricted_reads_from_keeper = 'true'"
+        )
+        node.exec_in_container(
+            [
+                "bash",
+                "-c",
+                "sync",
+            ],
+            privileged=True,
+            user="root",
+        )
+        trace_lines = int(
             node.exec_in_container(
                 [
                     "bash",
@@ -85,5 +88,7 @@ def test_adjust_log_level(start_cluster):
                 user="root",
             )
         )
-        >= 1
-    )
+        if trace_lines >= 1:
+            break
+        time.sleep(1)
+    assert trace_lines >= 1

tests/integration/test_named_collections_encrypted2/test.py

@@ -94,6 +94,7 @@ def wait_zk_child_absent(zk, path, child, timeout=10):
 
 
 def check_encrypted(zk, collection):
+    zk.sync(ZK_PATH)
     content = zk.get(f"{ZK_PATH}/{collection}.sql")[0]
     assert content[:3] == b"ENC"
     return content
@@ -749,6 +750,7 @@ def test_new_replica_encrypted_data_integrity(stopped_node3):
         password='P@ssw0rd!Complex#123'
     """)
 
+    zk.sync(ZK_PATH)
     content = zk.get(f"{ZK_PATH}/encrypted_coll.sql")[0]
     assert content[:3] == b"ENC"
     assert b"super_secret_api_key_12345" not in content

tests/integration/test_replication_credentials/test.py

@@ -58,13 +58,13 @@ def test_same_credentials(same_credentials_cluster):
     node1.query("TRUNCATE TABLE test_table")
     node2.query("SYSTEM SYNC REPLICA test_table", timeout=10)
     node1.query("insert into test_table values ('2017-06-16', 111, 0)")
-    time.sleep(1)
+    node2.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node1.query("SELECT id FROM test_table order by id") == "111\n"
     assert node2.query("SELECT id FROM test_table order by id") == "111\n"
 
     node2.query("insert into test_table values ('2017-06-17', 222, 1)")
-    time.sleep(1)
+    node1.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node1.query("SELECT id FROM test_table order by id") == "111\n222\n"
     assert node2.query("SELECT id FROM test_table order by id") == "111\n222\n"
@@ -99,13 +99,13 @@ def test_no_credentials(no_credentials_cluster):
     node3.query("TRUNCATE TABLE test_table")
     node4.query("SYSTEM SYNC REPLICA test_table", timeout=10)
     node3.query("insert into test_table values ('2017-06-18', 111, 0)")
-    time.sleep(1)
+    node4.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node3.query("SELECT id FROM test_table order by id") == "111\n"
     assert node4.query("SELECT id FROM test_table order by id") == "111\n"
 
     node4.query("insert into test_table values ('2017-06-19', 222, 1)")
-    time.sleep(1)
+    node3.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node3.query("SELECT id FROM test_table order by id") == "111\n222\n"
     assert node4.query("SELECT id FROM test_table order by id") == "111\n222\n"

tests/integration/test_storage_delta/test.py

@@ -1701,12 +1701,20 @@ def test_replicated_database_and_unavailable_s3(started_cluster, use_delta_kerne
 
         node2.restart_clickhouse()
 
-        assert (
-            node2.query(
+        # `restart_clickhouse` only waits until the server answers a query, but the
+        # digest is rewritten by the background `startup Replicated database` job, so
+        # poll instead of reading it once.
+        digest = None
+        deadline = time.monotonic() + 60
+        while time.monotonic() < deadline:
+            digest = node2.query(
                 f"SELECT value FROM system.zookeeper WHERE path = '{replica_path}' AND name = 'digest'"
             ).strip()
-            != "123456"
-        )
+            if digest != "123456":
+                break
+            time.sleep(1)
+
+        assert digest != "123456"
 
 
 def test_session_token(started_cluster):

tests/integration/test_tcp_handler_connection_limits/test.py

@@ -1,6 +1,5 @@
 import pytest
 import subprocess
-import time
 from helpers.cluster import ClickHouseCluster
 
 cluster = ClickHouseCluster(__file__)
@@ -14,11 +13,6 @@ def started_cluster():
     finally:
         cluster.shutdown()
 
-@pytest.fixture(scope="module", autouse=True)
-def stabilize_container(started_cluster):
-    """Wait for container startup processes to complete before running tests"""
-    time.sleep(1)
-
 def execute_queries_persistent_connection(queries):
     """Execute multiple queries through a single persistent clickhouse-client connection"""
     proc = subprocess.Popen(
@@ -34,17 +28,19 @@ def execute_queries_persistent_connection(queries):
 
     return stdout, stderr
 
-def get_connection_done_count():
-    try:
-        log_result = node.exec_in_container(
-            ["grep", "-c", "Done processing connection", "/var/log/clickhouse-server/clickhouse-server.log"]
-        )
-        return int(log_result.strip())
-    except Exception:
-        return 0
+def get_limit_closed_count(reason):
+    """Count the connections that the server closed because a limit was reached.
+
+    Counting every closed connection instead would be racy: the readiness probe of
+    `cluster.start` connects to the port and closes it without sending any data, and the
+    server accepts that connection only once it starts serving, which can happen after the
+    test has already sampled the initial count. Connections closed for other reasons never
+    report a limit, so counting only those keeps the assertion exact.
+    """
+    return int(node.count_in_log(f"Closing connection due to limits: {reason}").strip())
 
 def test_query_count_limit(started_cluster):
-    initial_count = get_connection_done_count()
+    initial_count = get_limit_closed_count("queries=")
 
     queries = ["SELECT 1;", "SELECT 2;", "SELECT 3;", "SELECT 4;", "SELECT 5;"]
     stdout, stderr = execute_queries_persistent_connection(queries)
@@ -53,17 +49,17 @@ def test_query_count_limit(started_cluster):
     assert "4" not in stdout and "5" not in stdout
     assert "TCP_CONNECTION_LIMIT_REACHED" in stderr
 
-    final_count = get_connection_done_count()
+    final_count = get_limit_closed_count("queries=")
     assert final_count == initial_count + 1, f"Expected exactly 1 connection closure, got {final_count - initial_count}"
 
 def test_time_limit(started_cluster):
-    initial_count = get_connection_done_count()
+    initial_count = get_limit_closed_count("elapsed=")
 
     queries = ["SELECT sleep(3);", "SELECT 1;", "SELECT 2;"]
     stdout, stderr = execute_queries_persistent_connection(queries)
 
     assert "1" not in stdout and "2" not in stdout
     assert "TCP_CONNECTION_LIMIT_REACHED" in stderr
 
-    final_count = get_connection_done_count()
+    final_count = get_limit_closed_count("elapsed=")
     assert final_count == initial_count + 1, f"Expected exactly 1 connection closure, got {final_count - initial_count}"

tests/integration/test_trace_log_memory_context/test.py

@@ -58,13 +58,20 @@ def get_trace_events(memory_context, memory_blocked_context, trace_type, query_i
     #     server-wide allocations including background activity, not just
     #     this query's allocations.
     # Retrying bounds the test runtime while keeping it reliable.
-    for _ in range(0, 15):
+    for attempt in range(0, 15):
         # Generate some logs to generate entries with memory_blocked_context=Global and trace_type=JemallocSample
         for i in range(10):
             node.query("SELECT logTrace('foo')")
         query_id = uuid.uuid4().hex
         node.query("SELECT * FROM numbers(100000) ORDER BY number", query_id=query_id)
 
+        # `Memory`/`MemoryPeak` with `memory_context = 'Global'` are sent only when the
+        # server-wide memory usage grows past its previous peak by at least
+        # `total_memory_profiler_step` (4 MiB). The peak reached during server startup can
+        # be above anything the small queries here allocate, in which case retrying alone
+        # never helps. Force a new global peak by allocating more memory on every attempt.
+        node.query(f"SELECT groupArray(number) FROM numbers({(attempt + 1) * 12500000}) FORMAT Null")
+
         node.query("SYSTEM FLUSH LOGS system.trace_log")
         if (
             get_trace_events("Unknown", "Max", "MemorySample", query_id) > 0 and

tests/queries/0_stateless/01184_long_insert_values_huge_strings.sh

@@ -17,7 +17,8 @@ done;
 wait
 
 $CLICKHOUSE_CLIENT -q "select count() from huge_strings"
-$CLICKHOUSE_CLIENT -q "select sum(l = length(s)) from huge_strings"
-$CLICKHOUSE_CLIENT -q "select sum(h = cityHash64(s)) from huge_strings"
+# Pin `max_threads`: each read stream holds its own buffer for a ~9 MB row of `s`, so the randomized 32-thread draw exceeds `max_memory_usage`.
+$CLICKHOUSE_CLIENT -q "select sum(l = length(s)) from huge_strings SETTINGS max_threads = 3"
+$CLICKHOUSE_CLIENT -q "select sum(h = cityHash64(s)) from huge_strings SETTINGS max_threads = 3"
 
 $CLICKHOUSE_CLIENT -q "drop table huge_strings"

tests/queries/0_stateless/02040_clickhouse_benchmark_query_id_pass_through.sh

@@ -6,6 +6,11 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 
 query_id="${CLICKHOUSE_DATABASE}_$$"
 benchmark_args=(
+    # A loaded runner can take longer than the default 10 s handshake_timeout_ms
+    # to send Hello; the benchmark then exits without running a query and
+    # query_log has 0 rows instead of 3.
+    --connect_timeout 60
+    --handshake_timeout_ms 60000
     --iterations 1
     --log_queries 1
     --query_id "$query_id"

tests/queries/0_stateless/02559_add_parts.reference

@@ -1,4 +1,4 @@
 0	0	0
 1	1	2
 2	2	4
-2	1	2
+1

tests/queries/0_stateless/02559_add_parts.sql

@@ -16,5 +16,7 @@ SELECT parts, active_parts,total_marks FROM system.tables WHERE name = 'check_sy
 INSERT INTO check_system_tables VALUES (1, 2, 1);
 SELECT parts, active_parts,total_marks FROM system.tables WHERE name = 'check_system_tables' AND database = currentDatabase();
 ALTER TABLE check_system_tables DETACH PARTITION 1;
-SELECT parts, active_parts,total_marks FROM system.tables WHERE name = 'check_system_tables' AND database = currentDatabase();
+-- `parts` and `total_marks` count Outdated parts too, and reclamation after DETACH is best-effort,
+-- so only `active_parts` is well defined here.
+SELECT active_parts FROM system.tables WHERE name = 'check_system_tables' AND database = currentDatabase();
 DROP TABLE IF EXISTS check_system_tables;