Merge pull request ClickHouse#77893 from ClickHouse/bk3

Pause refreshable materialized views when restoring backup

Author: al13n321

docs/en/sql-reference/statements/system.md

@@ -580,21 +580,25 @@ SYSTEM REFRESH VIEW [db.]name
 
 Wait for the currently running refresh to complete. If the refresh fails, throws an exception. If no refresh is running, completes immediately, throwing an exception if previous refresh failed.
 
-### STOP VIEW, STOP VIEWS {#stop-view-stop-views}
+### STOP [REPLICATED] VIEW, STOP VIEWS {#stop-view-stop-views}
 
 Disable periodic refreshing of the given view or all refreshable views. If a refresh is in progress, cancel it too.
 
+If the view is in a Replicated or Shared database, `STOP VIEW` only affects the current replica, while `STOP REPLICATED VIEW` affects all replicas.
+
 ```sql
 SYSTEM STOP VIEW [db.]name
 ```
 ```sql
 SYSTEM STOP VIEWS
 ```
 
-### START VIEW, START VIEWS {#start-view-start-views}
+### START [REPLICATED] VIEW, START VIEWS {#start-view-start-views}
 
 Enable periodic refreshing for the given view or all refreshable views. No immediate refresh is triggered.
 
+If the view is in a Replicated or Shared database, `START VIEW` undoes the effect of `STOP VIEW`, and `START REPLICATED VIEW` undoes the effect of `STOP REPLICATED VIEW`.
+
 ```sql
 SYSTEM START VIEW [db.]name
 ```
@@ -604,7 +608,7 @@ SYSTEM START VIEWS
 
 ### CANCEL VIEW {#cancel-view}
 
-If there's a refresh in progress for the given view, interrupt and cancel it. Otherwise do nothing.
+If there's a refresh in progress for the given view on the current replica, interrupt and cancel it. Otherwise do nothing.
 
 ```sql
 SYSTEM CANCEL VIEW [db.]name
@@ -616,6 +620,8 @@ Waits for the running refresh to complete. If no refresh is running, returns imm
 
 Can be used right after creating a new refreshable materialized view (without EMPTY keyword) to wait for the initial refresh to complete.
 
+If the view is in a Replicated or Shared database, and refresh is running on another replica, waits for that refresh to complete.
+
 ```sql
 SYSTEM WAIT VIEW [db.]name
 ```

src/Backups/BackupCoordinationStage.h

@@ -40,6 +40,9 @@ namespace BackupCoordinationStage
     /// Inserting restored data to tables.
     constexpr const char * INSERTING_DATA_TO_TABLES = "inserting data to tables";
 
+    /// Unpausing refreshable materialized views.
+    constexpr const char * FINALIZING_TABLES = "finalizing tables";
+
     /// Coordination stage meaning that a host finished its work.
     constexpr const char * COMPLETED = "completed";
 }

src/Backups/BackupCoordinationStageSync.cpp

@@ -891,8 +891,19 @@ bool BackupCoordinationStageSync::checkIfHostsReachStage(const Strings & hosts,
         }
 
         if (host_info.finished)
+        {
+            if (stage_to_wait == BackupCoordinationStage::FINALIZING_TABLES)
+            {
+                /// This is a newly added stage. For compatibility with older server versions,
+                /// allow other replicas to skip this stage. This doesn't break anything: this stage
+                /// unpauses refreshable materialized views, but older server versions don't pause
+                /// them in the first place.
+                results[i] = "";
+                continue;
+            }
             throw Exception(ErrorCodes::FAILED_TO_SYNC_BACKUP_OR_RESTORE,
                             "{} finished without coming to stage {}", getHostDesc(host), stage_to_wait);
+        }
 
         if (should_stop_watching_thread)
             throw Exception(ErrorCodes::LOGICAL_ERROR, "waitHostsReachStage() can't wait for stage {} after the watching thread stopped", stage_to_wait);

src/Backups/RestorerFromBackup.cpp

@@ -176,6 +176,9 @@ void RestorerFromBackup::run(Mode mode_)
     insertDataToTables();
     runDataRestoreTasks();
 
+    setStage(Stage::FINALIZING_TABLES);
+    finalizeTables();
+
     /// Restored successfully!
     setStage(Stage::COMPLETED);
 }
@@ -1174,6 +1177,20 @@ void RestorerFromBackup::runDataRestoreTasks()
     }
 }
 
+void RestorerFromBackup::finalizeTables()
+{
+    std::vector<StoragePtr> tables;
+    {
+        std::lock_guard lock{mutex};
+        tables.reserve(table_infos.size());
+        for (const auto & [_, info] : table_infos)
+            tables.push_back(info.storage);
+    }
+
+    for (const auto & storage : tables)
+        storage->finalizeRestoreFromBackup();
+}
+
 void RestorerFromBackup::throwTableIsNotEmpty(const StorageID & storage_id)
 {
     throw Exception(

src/Backups/RestorerFromBackup.h

@@ -133,6 +133,8 @@ class RestorerFromBackup : private boost::noncopyable
 
     void runDataRestoreTasks();
 
+    void finalizeTables();
+
     void setStage(const String & new_stage, const String & message = "");
 
     /// Schedule a task from the thread pool and start executing it.

src/Interpreters/InterpreterCreateQuery.cpp

@@ -1778,7 +1778,8 @@ bool InterpreterCreateQuery::doCreateTable(ASTCreateQuery & create,
                 getContext()->getGlobalContext(),
                 properties.columns,
                 properties.constraints,
-                mode);
+                mode,
+                is_restore_from_backup);
         };
         auto temporary_table = TemporaryTableHolder(getContext(), creator, query_ptr);
 
@@ -1955,7 +1956,8 @@ bool InterpreterCreateQuery::doCreateTable(ASTCreateQuery & create,
             getContext()->getGlobalContext(),
             properties.columns,
             properties.constraints,
-            mode);
+            mode,
+            is_restore_from_backup);
 
         /// If schema was inferred while storage creation, add columns description to create query.
         auto & create_query = query_ptr->as<ASTCreateQuery &>();

src/Interpreters/InterpreterSystemQuery.cpp

@@ -718,6 +718,14 @@ BlockIO InterpreterSystemQuery::execute()
         case Type::STOP_VIEWS:
             startStopAction(ActionLocks::ViewRefresh, false);
             break;
+        case Type::START_REPLICATED_VIEW:
+            for (const auto & task : getRefreshTasks())
+                task->startReplicated();
+            break;
+        case Type::STOP_REPLICATED_VIEW:
+            for (const auto & task : getRefreshTasks())
+                task->stopReplicated("SYSTEM STOP REPLICATED VIEW");
+            break;
         case Type::REFRESH_VIEW:
             for (const auto & task : getRefreshTasks())
                 task->run();
@@ -1625,8 +1633,10 @@ AccessRightsElements InterpreterSystemQuery::getRequiredAccessForDDLOnCluster()
         case Type::WAIT_VIEW:
         case Type::START_VIEW:
         case Type::START_VIEWS:
+        case Type::START_REPLICATED_VIEW:
         case Type::STOP_VIEW:
         case Type::STOP_VIEWS:
+        case Type::STOP_REPLICATED_VIEW:
         case Type::CANCEL_VIEW:
         case Type::TEST_VIEW:
         {

src/Parsers/ASTSystemQuery.cpp

@@ -400,7 +400,9 @@ void ASTSystemQuery::formatImpl(WriteBuffer & ostr, const FormatSettings & setti
         }
         case Type::REFRESH_VIEW:
         case Type::START_VIEW:
+        case Type::START_REPLICATED_VIEW:
         case Type::STOP_VIEW:
+        case Type::STOP_REPLICATED_VIEW:
         case Type::CANCEL_VIEW:
         case Type::WAIT_VIEW:
         {

src/Parsers/ASTSystemQuery.h

@@ -107,8 +107,10 @@ class ASTSystemQuery : public IAST, public ASTQueryWithOnCluster
         WAIT_VIEW,
         START_VIEW,
         START_VIEWS,
+        START_REPLICATED_VIEW,
         STOP_VIEW,
         STOP_VIEWS,
+        STOP_REPLICATED_VIEW,
         CANCEL_VIEW,
         TEST_VIEW,
         LOAD_PRIMARY_KEY,

src/Parsers/ParserSystemQuery.cpp

@@ -441,7 +441,9 @@ bool ParserSystemQuery::parseImpl(IParser::Pos & pos, ASTPtr & node, Expected &
         case Type::REFRESH_VIEW:
         case Type::WAIT_VIEW:
         case Type::START_VIEW:
+        case Type::START_REPLICATED_VIEW:
         case Type::STOP_VIEW:
+        case Type::STOP_REPLICATED_VIEW:
         case Type::CANCEL_VIEW:
             if (!parseDatabaseAndTableAsAST(pos, expected, res->database, res->table))
                 return false;