Fix possible deadlock with allow_asynchronous_read_from_io_pool_for_merge_tree in case of exception from ThreadPool::schedule

src/Storages/MergeTree/MergeTreeSource.cpp

@@ -103,9 +103,18 @@ struct MergeTreeSource::AsyncReadingState
     }
 
     void schedule(ThreadPool::Job job)
+    {
+        try
         {
             callback_runner(std::move(job), 0);
         }
+        catch (...)
+        {
+            /// Roll back stage in case of exception from ThreadPool::schedule
+            control->stage = Stage::NotStarted;
+            throw;
+        }
+    }
 
     ChunkAndProgress getResult()
     {