Merge pull request ClickHouse#89789 from ClickHouse/backport/25.8/89692

Backport ClickHouse#89692 to 25.8: Speed up test_refreshable_mv_skip_old_temp_table_ddls

Author: tuanpach

src/Common/FailPoint.cpp

@@ -117,7 +117,7 @@ static struct InitFiu
     REGULAR(sleep_in_logs_flush) \
     ONCE(smt_commit_exception_before_op) \
     ONCE(backup_add_empty_memory_table) \
-    REGULAR(refresh_task_delay_update_coordination_state_running)
+    REGULAR(refresh_task_stop_racing_for_running_refresh)
 
 
 namespace FailPoints

src/Storages/MaterializedView/RefreshTask.cpp

@@ -78,7 +78,7 @@ namespace ErrorCodes
 
 namespace FailPoints
 {
-extern const char refresh_task_delay_update_coordination_state_running[];
+extern const char refresh_task_stop_racing_for_running_refresh[];
 }
 
 RefreshTask::RefreshTask(
@@ -1000,10 +1000,11 @@ bool RefreshTask::updateCoordinationState(CoordinationZnode root, bool running,
         ops.emplace_back(zkutil::makeSetRequest(coordination.path, root.toString(), root.version));
         if (running)
         {
-            fiu_do_on(FailPoints::refresh_task_delay_update_coordination_state_running, {
-                std::chrono::milliseconds sleep_time{3000 + thread_local_rng() % 2000};
-                std::this_thread::sleep_for(sleep_time);
-            });
+            bool stop_racing_for_running_refresh = false;
+            fiu_do_on(FailPoints::refresh_task_stop_racing_for_running_refresh, { stop_racing_for_running_refresh = true; });
+            if (stop_racing_for_running_refresh)
+                return false;
+
             ops.emplace_back(
                 zkutil::makeCreateRequest(coordination.path + "/running", coordination.replica_name, zkutil::CreateMode::Ephemeral));
         }

tests/integration/test_refreshable_mv_skip_old_temp_table_ddls/test.py

@@ -80,9 +80,7 @@ def test_refreshable_mv_skip_old_temp_tables_ddls(
     node1.query(f"DROP DATABASE IF EXISTS {db_name} SYNC")
     node2.query(f"DROP DATABASE IF EXISTS {db_name}  SYNC")
     # Make sure that the MV is refreshed on node1
-    node2.query(
-        "SYSTEM ENABLE FAILPOINT refresh_task_delay_update_coordination_state_running"
-    )
+    node2.query("SYSTEM ENABLE FAILPOINT refresh_task_stop_racing_for_running_refresh")
     node2.query("SYSTEM ENABLE FAILPOINT database_replicated_delay_entry_execution")
 
     node1.query(
@@ -98,11 +96,11 @@ def test_refreshable_mv_skip_old_temp_tables_ddls(
             f"CREATE TABLE {db_name}.target (x DateTime) ENGINE ReplicatedMergeTree ORDER BY x"
         )
         node1.query(
-            f"CREATE MATERIALIZED VIEW {db_name}.mv REFRESH EVERY 1 SECOND {append_clause} TO {db_name}.target AS SELECT now() AS x"
+            f"CREATE MATERIALIZED VIEW {db_name}.mv REFRESH EVERY 1 HOUR {append_clause} TO {db_name}.target AS SELECT now() AS x"
         )
     else:
         node1.query(
-            f"CREATE MATERIALIZED VIEW {db_name}.mv REFRESH EVERY 1 SECOND {append_clause} (x DateTime) ENGINE ReplicatedMergeTree ORDER BY x AS SELECT now() AS x"
+            f"CREATE MATERIALIZED VIEW {db_name}.mv REFRESH EVERY 1 HOUR {append_clause} (x DateTime) ENGINE ReplicatedMergeTree ORDER BY x AS SELECT now() AS x"
         )
 
     node2.query(
@@ -122,10 +120,16 @@ def test_refreshable_mv_skip_old_temp_tables_ddls(
 
     last_log_ts = get_last_ddl_worker_log_ts(node2, db_name)
 
-    # Wait for node1 to refresh the view several times
-    time.sleep(5)
-
-    node1.query(f"ALTER TABLE {db_name}.mv MODIFY REFRESH EVERY 1 HOUR {append_clause}")
+    last_refresh_time = node1.query(
+        "SELECT last_refresh_time FROM system.view_refreshes WHERE view='mv'"
+    )
+    for i in range(2):
+        node1.query(f"SYSTEM REFRESH VIEW {db_name}.mv")
+    # Ensure that the mv is refresh
+    node1.query_with_retry(
+        "SELECT last_refresh_time FROM system.view_refreshes WHERE view='mv'",
+        check_callback=lambda x: x != last_refresh_time,
+    )
 
     # Make sure that the view is not refreshing, and it is scheduled to be refreshed in at least 10 minutes
     node1.query_with_retry(
@@ -135,9 +139,7 @@ def test_refreshable_mv_skip_old_temp_tables_ddls(
         > datetime.timedelta(minutes=10),
     )
 
-    node2.query(
-        "SYSTEM DISABLE FAILPOINT refresh_task_delay_update_coordination_state_running"
-    )
+    node2.query("SYSTEM DISABLE FAILPOINT refresh_task_stop_racing_for_running_refresh")
     node2.query("SYSTEM DISABLE FAILPOINT database_replicated_delay_entry_execution")
 
     table_info1 = node1.query(f"SELECT uuid, name FROM system.tables WHERE database='{db_name}'")