Merge pull request ClickHouse#61936 from hanfei1991/hanfei/refreshabl…

…e_view_race fix a race in refreshable view
bigo-sg · Mar 27, 2024 · b824e94 · b824e94
2 parents 795026a + f3616f6
commit b824e94
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 4 deletions.
diff --git a/src/Storages/MaterializedView/RefreshSet.cpp b/src/Storages/MaterializedView/RefreshSet.cpp
@@ -73,15 +73,15 @@ void RefreshSet::Handle::reset()
 
 RefreshSet::RefreshSet() = default;
 
-RefreshSet::Handle RefreshSet::emplace(StorageID id, const std::vector<StorageID> & dependencies, RefreshTaskHolder task)
+void RefreshSet::emplace(StorageID id, const std::vector<StorageID> & dependencies, RefreshTaskHolder task)
 {
     std::lock_guard guard(mutex);
     auto [it, is_inserted] = tasks.emplace(id, task);
     if (!is_inserted)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Refresh set entry already exists for table {}", id.getFullTableName());
     addDependenciesLocked(id, dependencies);
 
-    return Handle(this, id, dependencies);
+    task->setRefreshSetHandleUnlock(Handle(this, id, dependencies));
 }
 
 void RefreshSet::addDependenciesLocked(const StorageID & id, const std::vector<StorageID> & dependencies)

diff --git a/src/Storages/MaterializedView/RefreshSet.h b/src/Storages/MaterializedView/RefreshSet.h
@@ -82,7 +82,7 @@ class RefreshSet
 
     RefreshSet();
 
-    Handle emplace(StorageID id, const std::vector<StorageID> & dependencies, RefreshTaskHolder task);
+    void emplace(StorageID id, const std::vector<StorageID> & dependencies, RefreshTaskHolder task);
 
     RefreshTaskHolder getTask(const StorageID & id) const;
 

diff --git a/src/Storages/MaterializedView/RefreshTask.cpp b/src/Storages/MaterializedView/RefreshTask.cpp
@@ -50,7 +50,7 @@ RefreshTaskHolder RefreshTask::create(
         for (auto && dependency : strategy.dependencies->children)
             deps.emplace_back(dependency->as<const ASTTableIdentifier &>());
 
-    task->set_handle = context->getRefreshSet().emplace(view.getStorageID(), deps, task);
+    context->getRefreshSet().emplace(view.getStorageID(), deps, task);
 
     return task;
 }
@@ -509,4 +509,9 @@ std::chrono::system_clock::time_point RefreshTask::currentTime() const
         return std::chrono::system_clock::time_point(std::chrono::seconds(fake));
 }
 
+void RefreshTask::setRefreshSetHandleUnlock(RefreshSet::Handle && set_handle_)
+{
+    set_handle = std::move(set_handle_);
+}
+
 }
diff --git a/src/Storages/MaterializedView/RefreshTask.h b/src/Storages/MaterializedView/RefreshTask.h
@@ -61,6 +61,9 @@ class RefreshTask : public std::enable_shared_from_this<RefreshTask>
     /// For tests
     void setFakeTime(std::optional<Int64> t);
 
+    /// RefreshSet will set handle for refresh tasks, to avoid race condition.
+    void setRefreshSetHandleUnlock(RefreshSet::Handle && set_handle_);
+
 private:
     LoggerPtr log = nullptr;
     std::weak_ptr<IStorage> view_to_refresh;