diff --git a/tests/config/config.d/storage_conf.xml b/tests/config/config.d/storage_conf.xml index a5216fc05249..1ef64249f2eb 100644 --- a/tests/config/config.d/storage_conf.xml +++ b/tests/config/config.d/storage_conf.xml @@ -123,7 +123,13 @@ default - s3_disk + + + s3_disk + 0 + diff --git a/tests/integration/test_attach_table_from_s3_plain_readonly/test.py b/tests/integration/test_attach_table_from_s3_plain_readonly/test.py index 2aae3563d955..140551a9fc29 100644 --- a/tests/integration/test_attach_table_from_s3_plain_readonly/test.py +++ b/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 diff --git a/tests/integration/test_dirty_pages_force_purge/configs/overrides.yaml b/tests/integration/test_dirty_pages_force_purge/configs/overrides.yaml index 195236e51dde..229d9175b265 100644 --- a/tests/integration/test_dirty_pages_force_purge/configs/overrides.yaml +++ b/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 diff --git a/tests/integration/test_keeper_back_to_back/test.py b/tests/integration/test_keeper_back_to_back/test.py index 5b73a79b65e8..01e7effb5595 100644 --- a/tests/integration/test_keeper_back_to_back/test.py +++ b/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,7 +786,8 @@ 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 @@ -783,26 +795,19 @@ 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 diff --git a/tests/integration/test_keeper_dynamic_log_level/test.py b/tests/integration/test_keeper_dynamic_log_level/test.py index 891e474f6e38..a4b418916618 100644 --- a/tests/integration/test_keeper_dynamic_log_level/test.py +++ b/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 diff --git a/tests/integration/test_named_collections_encrypted2/test.py b/tests/integration/test_named_collections_encrypted2/test.py index bf2df9767875..647447a9a6c7 100644 --- a/tests/integration/test_named_collections_encrypted2/test.py +++ b/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 diff --git a/tests/integration/test_replication_credentials/test.py b/tests/integration/test_replication_credentials/test.py index 7d46bef3b8f2..5e0fb1b088aa 100644 --- a/tests/integration/test_replication_credentials/test.py +++ b/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" diff --git a/tests/integration/test_storage_delta/test.py b/tests/integration/test_storage_delta/test.py index 5d9400a04584..39684966f801 100644 --- a/tests/integration/test_storage_delta/test.py +++ b/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): diff --git a/tests/integration/test_tcp_handler_connection_limits/test.py b/tests/integration/test_tcp_handler_connection_limits/test.py index ef9a35f40f2a..ac9ec0a4eddd 100644 --- a/tests/integration/test_tcp_handler_connection_limits/test.py +++ b/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,11 +49,11 @@ 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) @@ -65,5 +61,5 @@ def test_time_limit(started_cluster): 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}" diff --git a/tests/integration/test_trace_log_memory_context/test.py b/tests/integration/test_trace_log_memory_context/test.py index 47ad24fa12d3..a6f52c55b58d 100644 --- a/tests/integration/test_trace_log_memory_context/test.py +++ b/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 diff --git a/tests/queries/0_stateless/01184_long_insert_values_huge_strings.sh b/tests/queries/0_stateless/01184_long_insert_values_huge_strings.sh index 8d41c32467d2..0973bf5d17d2 100755 --- a/tests/queries/0_stateless/01184_long_insert_values_huge_strings.sh +++ b/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" diff --git a/tests/queries/0_stateless/02040_clickhouse_benchmark_query_id_pass_through.sh b/tests/queries/0_stateless/02040_clickhouse_benchmark_query_id_pass_through.sh index 59538534fa71..6d084cfd9858 100755 --- a/tests/queries/0_stateless/02040_clickhouse_benchmark_query_id_pass_through.sh +++ b/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" diff --git a/tests/queries/0_stateless/02559_add_parts.reference b/tests/queries/0_stateless/02559_add_parts.reference index 50bd3725056e..845bf13dd8d9 100644 --- a/tests/queries/0_stateless/02559_add_parts.reference +++ b/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 diff --git a/tests/queries/0_stateless/02559_add_parts.sql b/tests/queries/0_stateless/02559_add_parts.sql index 9f4e85a32589..b8f427a90537 100644 --- a/tests/queries/0_stateless/02559_add_parts.sql +++ b/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; diff --git a/tests/queries/0_stateless/02703_max_local_read_bandwidth.sh b/tests/queries/0_stateless/02703_max_local_read_bandwidth.sh index fb7f47613c1d..9a049fd34d7f 100755 --- a/tests/queries/0_stateless/02703_max_local_read_bandwidth.sh +++ b/tests/queries/0_stateless/02703_max_local_read_bandwidth.sh @@ -11,8 +11,10 @@ $CLICKHOUSE_CLIENT -m -q " create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9; " -# reading 1e6*8 bytes with 1M bandwith it should take (8-1)/1=7 seconds -$CLICKHOUSE_CLIENT -q "insert into data select * from numbers(1e6)" +# Reading 2e5*8 bytes at 160000 B/s takes 1.6e6/160000-1 = 9 seconds (-1 is the 1s token burst). +# The throttler only sleeps while the arrival rate exceeds the cap, so the cap must stay far +# below the natural read rate or the sleep assertion flaps on loaded runners. +$CLICKHOUSE_CLIENT -q "insert into data select * from numbers(2e5)" read_methods=( read @@ -25,14 +27,14 @@ read_methods=( ) for read_method in "${read_methods[@]}"; do query_id=$(random_str 10) - $CLICKHOUSE_CLIENT --query_id "$query_id" -q "select * from data format Null settings max_local_read_bandwidth='1M', local_filesystem_read_method='$read_method'" + $CLICKHOUSE_CLIENT --query_id "$query_id" -q "select * from data format Null settings max_local_read_bandwidth=160000, local_filesystem_read_method='$read_method'" $CLICKHOUSE_CLIENT -m -q " SYSTEM FLUSH LOGS query_log; SELECT '$read_method', query_duration_ms >= 7e3, - ProfileEvents['ReadBufferFromFileDescriptorReadBytes'] > 8e6, - ProfileEvents['QueryLocalReadThrottlerBytes'] > 8e6, + ProfileEvents['ReadBufferFromFileDescriptorReadBytes'] > 1.5e6, + ProfileEvents['QueryLocalReadThrottlerBytes'] > 1.5e6, ProfileEvents['QueryLocalReadThrottlerSleepMicroseconds'] > 7e6*0.5 FROM system.query_log WHERE event_date >= yesterday() AND event_time >= now() - 600 AND current_database = '$CLICKHOUSE_DATABASE' AND query_id = '$query_id' AND type != 'QueryStart' diff --git a/tests/queries/0_stateless/02932_refreshable_materialized_views_2.reference b/tests/queries/0_stateless/02932_refreshable_materialized_views_2.reference index 7861f52373f8..235d5ef0ef5c 100644 --- a/tests/queries/0_stateless/02932_refreshable_materialized_views_2.reference +++ b/tests/queries/0_stateless/02932_refreshable_materialized_views_2.reference @@ -5,7 +5,6 @@ <23: simple refresh> 1 <24: rename during refresh> 1 <25: rename during refresh> rmv_f Running -<27: cancelled> rmv_f Scheduled 1 <28: drop during refresh> 0 0 CREATE MATERIALIZED VIEW default.rmv_g\nREFRESH EVERY 1 WEEK OFFSET 3 DAY 4 HOUR RANDOMIZE FOR 4 DAY 1 HOUR\n(\n `x` Int64\n)\nENGINE = Memory\nDEFINER = default SQL SECURITY DEFINER\nAS SELECT 42 AS x diff --git a/tests/queries/0_stateless/02932_refreshable_materialized_views_2.sh b/tests/queries/0_stateless/02932_refreshable_materialized_views_2.sh index 6553895b6356..741318f89719 100755 --- a/tests/queries/0_stateless/02932_refreshable_materialized_views_2.sh +++ b/tests/queries/0_stateless/02932_refreshable_materialized_views_2.sh @@ -82,16 +82,6 @@ $CLICKHOUSE_CLIENT -q " select '<24: rename during refresh>', * from rmv_f;" query_no_scheduling "select '<25: rename during refresh>', view, status from refreshes where view = 'rmv_f'" $CLICKHOUSE_CLIENT -q "alter table rmv_f modify refresh after 10 year settings refresh_retries = 0;" -sleep 1 # make it likely that at least one row was processed -# Cancel. -$CLICKHOUSE_CLIENT -q " - system cancel view rmv_f;" -while [ "`$CLICKHOUSE_CLIENT -q "select status from refreshes -- $LINENO" | xargs`" != 'Scheduled' ] -do - sleep 0.5 -done -# Check that another refresh doesn't immediately start after the cancelled one. -query_no_scheduling "select '<27: cancelled>', view, status, exception != '' from refreshes where view = 'rmv_f'" $CLICKHOUSE_CLIENT -q "system refresh view rmv_f;" while [ "`$CLICKHOUSE_CLIENT -q "select status from refreshes where view = 'rmv_f' -- $LINENO" | xargs`" != 'Running' ] do diff --git a/tests/queries/0_stateless/03469_json_read_subcolumns_combined_2_compact_merge_tree.sql b/tests/queries/0_stateless/03469_json_read_subcolumns_combined_2_compact_merge_tree.sql index 773e2c24f142..a34894479dcf 100644 --- a/tests/queries/0_stateless/03469_json_read_subcolumns_combined_2_compact_merge_tree.sql +++ b/tests/queries/0_stateless/03469_json_read_subcolumns_combined_2_compact_merge_tree.sql @@ -1,5 +1,7 @@ -- Tags: no-fasttest, long --- Random settings limits: index_granularity=(100, None); index_granularity_bytes=(100000, None); max_threads=(4, 32) +-- Random settings limits: index_granularity=(8192, None); index_granularity_bytes=(100000, None); max_threads=(4, 32) +-- index_granularity is floored at the engine default on purpose: tiny granules multiply the +-- mark count over these 71 SELECTs and timed the test out on slow builds. SET enable_json_type = 1; set allow_experimental_variant_type = 1; diff --git a/tests/queries/0_stateless/03717_async_deduplication_with_mv.sql b/tests/queries/0_stateless/03717_async_deduplication_with_mv.sql index 79f095bd35bb..a46083c95379 100644 --- a/tests/queries/0_stateless/03717_async_deduplication_with_mv.sql +++ b/tests/queries/0_stateless/03717_async_deduplication_with_mv.sql @@ -62,7 +62,9 @@ SELECT count() as value FROM 03717_table; SET async_insert = 1, insert_deduplicate = 1, async_insert_deduplicate = 1, wait_for_async_insert = 0, deduplicate_blocks_in_dependent_materialized_views=1; -set async_insert_use_adaptive_busy_timeout=0, async_insert_busy_timeout_min_ms=1000, async_insert_busy_timeout_max_ms=5000; +-- The busy timeout must outlast this test: the table-scoped flush below waits only for the jobs it +-- schedules itself, so a batch the deadline timer already drained is not waited for at all. +set async_insert_use_adaptive_busy_timeout=0, async_insert_busy_timeout_min_ms=1000, async_insert_busy_timeout_max_ms=600000; SET max_block_size=1; SET max_insert_block_size=1; diff --git a/tests/queries/0_stateless/04105_system_pause_view.reference b/tests/queries/0_stateless/04105_system_pause_view.reference index e97f918273c3..33cbc01c2312 100644 --- a/tests/queries/0_stateless/04105_system_pause_view.reference +++ b/tests/queries/0_stateless/04105_system_pause_view.reference @@ -6,3 +6,4 @@ <5: start views resumes all> 2 2 <6: granted pause works> <6: denied pause errors as expected> +<7: cancel during refresh records an exception> 1 1 1 diff --git a/tests/queries/0_stateless/04105_system_pause_view.sh b/tests/queries/0_stateless/04105_system_pause_view.sh index f5ece885ae33..3d3306fc695f 100755 --- a/tests/queries/0_stateless/04105_system_pause_view.sh +++ b/tests/queries/0_stateless/04105_system_pause_view.sh @@ -216,3 +216,56 @@ $CLICKHOUSE_CLIENT -q " drop table denied; drop table src; drop user $test_user;" + +# --------------------------------------------------------------------------- +# Test 5: SYSTEM CANCEL VIEW records the cancellation as an exception. +# +# The cancel must reach a live `PipelineExecutor`, so the refresh is parked at the +# `infinite_sleep` failpoint (hit from `sleepEachRow` inside `executor.execute()`) +# instead of being timed. `infinite_sleep` is server-global and fires on every +# `sleep`/`sleepEachRow` call, so this block must stay LAST in the file: the views +# above use `sleepEachRow(1)` and would park too. +# --------------------------------------------------------------------------- + +# `SYSTEM DISABLE FAILPOINT` is also the resume mechanism below; this trap only covers an +# early exit between the enable and that disable, which would otherwise leave the global +# failpoint active and park every later `sleep`/`sleepEachRow` call in the run. Disabling an +# already-disabled failpoint is a no-op. +trap ' + $CLICKHOUSE_CLIENT -q "SYSTEM DISABLE FAILPOINT infinite_sleep" 2>/dev/null || true +' EXIT + +$CLICKHOUSE_CLIENT -q " + create table src (x Int64) engine Memory; + insert into src values (1); + create materialized view c refresh every 1 year settings refresh_retries = 0 (x Int64) engine Memory empty as + select x + sleepEachRow(0) as x from src settings max_block_size = 1, max_threads = 1; + system enable failpoint infinite_sleep; + system refresh view c;" + +if ! timeout 60 $CLICKHOUSE_CLIENT -q "SYSTEM WAIT FAILPOINT infinite_sleep PAUSE" +then + echo "FAIL: refresh did not reach the infinite_sleep failpoint" +fi + +# The refresh is parked inside `executor.execute()`, so the cancel cannot be outrun. Disabling +# the failpoint resumes it. +$CLICKHOUSE_CLIENT -q " + system cancel view c; + system disable failpoint infinite_sleep;" + +wait_status c Scheduled + +# `Cancelling refresh in ...` is logged only when the interrupt finds a non-null +# `execution.executor`, so it proves the cancel hit the running pipeline rather than an +# already-finished attempt. Matching 'cancelled' distinguishes a cancellation from any +# other refresh failure. +$CLICKHOUSE_CLIENT -q " + system flush logs text_log; + select '<7: cancel during refresh records an exception>', + (select exception != '' from refreshes where view = 'c'), + (select position(exception, 'cancelled') > 0 from refreshes where view = 'c'), + (select count() > 0 from system.text_log + where message = 'Cancelling refresh in ' || currentDatabase() || '.c'); + drop table c; + drop table src;"