Merge pull request #10859 from ClickHouse/fix_watch_livelock_with_database_atomic

Fix livelock with WATCH queries and DatabaseAtomic

src/Storages/LiveView/LiveViewBlockInputStream.h

@@ -46,7 +46,7 @@ public:
 
     void cancel(bool kill) override
     {
-        if (isCancelled() || storage->is_dropped)
+        if (isCancelled() || storage->shutdown_called)
             return;
         IBlockInputStream::cancel(kill);
         std::lock_guard lock(storage->mutex);
@@ -115,7 +115,7 @@ protected:
             end = blocks->end();
         }
 
-        if (isCancelled() || storage->is_dropped)
+        if (isCancelled() || storage->shutdown_called)
         {
             return { Block(), true };
         }
@@ -155,7 +155,7 @@ protected:
                         bool signaled = std::cv_status::no_timeout == storage->condition.wait_for(lock,
                             std::chrono::microseconds(std::max(UInt64(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec))));
 
-                        if (isCancelled() || storage->is_dropped)
+                        if (isCancelled() || storage->shutdown_called)
                         {
                             return { Block(), true };
                         }

src/Storages/LiveView/LiveViewEventsBlockInputStream.h

@@ -65,7 +65,7 @@ public:
 
     void cancel(bool kill) override
     {
-        if (isCancelled() || storage->is_dropped)
+        if (isCancelled() || storage->shutdown_called)
             return;
         IBlockInputStream::cancel(kill);
         std::lock_guard lock(storage->mutex);
@@ -149,7 +149,7 @@ protected:
             end = blocks->end();
         }
 
-        if (isCancelled() || storage->is_dropped)
+        if (isCancelled() || storage->shutdown_called)
         {
             return { Block(), true };
         }
@@ -190,7 +190,7 @@ protected:
                         bool signaled = std::cv_status::no_timeout == storage->condition.wait_for(lock,
                             std::chrono::microseconds(std::max(UInt64(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec))));
 
-                        if (isCancelled() || storage->is_dropped)
+                        if (isCancelled() || storage->shutdown_called)
                         {
                             return { Block(), true };
                         }

src/Storages/LiveView/StorageLiveView.cpp

@@ -468,6 +468,10 @@ void StorageLiveView::shutdown()
     if (!shutdown_called.compare_exchange_strong(expected, true))
         return;
 
+    /// WATCH queries should be stopped after setting shutdown_called to true.
+    /// Otherwise livelock is possible for LiveView table in Atomic database:
+    /// WATCH query will wait for table to be dropped and DatabaseCatalog will wait for queries to finish
+
     {
         std::lock_guard no_users_thread_lock(no_users_thread_mutex);
         if (no_users_thread.joinable())