Fix possible deadlock with allow_asynchronous_read_from_io_pool_for_m…

…erge_tree in case of exception from ThreadPool::schedule

src/Storages/MergeTree/MergeTreeSource.cpp

@@ -104,7 +104,16 @@ struct MergeTreeSource::AsyncReadingState
 
     void schedule(ThreadPool::Job job)
     {
-        callback_runner(std::move(job), 0);
+        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()