diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/OrcFileWriterFactory.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/OrcFileWriterFactory.java
index abf57e148..670b09386 100644
--- a/presto-hive/src/main/java/io/prestosql/plugin/hive/OrcFileWriterFactory.java
+++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/OrcFileWriterFactory.java
@@ -15,6 +15,7 @@ package io.prestosql.plugin.hive;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import io.airlift.log.Logger;
 import io.prestosql.orc.OrcDataSink;
 import io.prestosql.orc.OrcDataSource;
 import io.prestosql.orc.OrcDataSourceId;
@@ -61,6 +62,8 @@ import static java.util.stream.Collectors.toList;
 public class OrcFileWriterFactory
         implements HiveFileWriterFactory
 {
+    private static final Logger log = Logger.get(OrcFileWriterFactory.class);
+
     private final HdfsEnvironment hdfsEnvironment;
     private final TypeManager typeManager;
     private final NodeVersion nodeVersion;
@@ -195,6 +198,7 @@ public class OrcFileWriterFactory
             }
 
             Callable<Void> rollbackAction = () -> {
+                log.debug("RollBack action to delete file %s", path);
                 fileSystem.delete(path, false);
                 return null;
             };
@@ -238,6 +242,7 @@ public class OrcFileWriterFactory
     protected OrcDataSink createOrcDataSink(ConnectorSession session, FileSystem fileSystem, Path path)
             throws IOException
     {
+        log.debug("Creation of OrcDataSink for file %s", path);
         return new OutputStreamOrcDataSink(fileSystem.create(path));
     }
 
diff --git a/presto-main/src/main/java/io/prestosql/execution/SqlQueryExecution.java b/presto-main/src/main/java/io/prestosql/execution/SqlQueryExecution.java
index 145dc7e9b..203a758b2 100644
--- a/presto-main/src/main/java/io/prestosql/execution/SqlQueryExecution.java
+++ b/presto-main/src/main/java/io/prestosql/execution/SqlQueryExecution.java
@@ -607,7 +607,8 @@ public class SqlQueryExecution
                 heuristicIndexerManager,
                 snapshotManager,
                 // Require same number of tasks to be scheduled, but do not require it if starting from beginning
-                snapshotId.isPresent() ? queryScheduler.get().getStageTaskCounts() : null);
+                snapshotId.isPresent() ? queryScheduler.get().getStageTaskCounts() : null,
+                true);
     }
 
     private void resetOutputData(PlanRoot plan, OptionalLong snapshotId)
@@ -868,7 +869,8 @@ public class SqlQueryExecution
                 dynamicFilterService,
                 heuristicIndexerManager,
                 snapshotManager,
-                null);
+                null,
+                false);
 
         queryScheduler.set(scheduler);
 
diff --git a/presto-main/src/main/java/io/prestosql/execution/scheduler/NodeScheduler.java b/presto-main/src/main/java/io/prestosql/execution/scheduler/NodeScheduler.java
index 69ea476b1..af6872ff4 100644
--- a/presto-main/src/main/java/io/prestosql/execution/scheduler/NodeScheduler.java
+++ b/presto-main/src/main/java/io/prestosql/execution/scheduler/NodeScheduler.java
@@ -363,4 +363,9 @@ public class NodeScheduler
                 .collect(toImmutableList());
         return whenAnyCompleteCancelOthers(stateChangeFutures);
     }
+
+    public void refreshNodeStates()
+    {
+        nodeManager.refreshWorkerStates();
+    }
 }
diff --git a/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java b/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java
index 90d1cf75d..0eda7a566 100644
--- a/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java
+++ b/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java
@@ -180,7 +180,8 @@ public class SqlQueryScheduler
             DynamicFilterService dynamicFilterService,
             HeuristicIndexerManager heuristicIndexerManager,
             QuerySnapshotManager snapshotManager,
-            Map<StageId, Integer> stageTaskCounts)
+            Map<StageId, Integer> stageTaskCounts,
+            boolean isResume)
     {
         SqlQueryScheduler sqlQueryScheduler = new SqlQueryScheduler(
                 queryStateMachine,
@@ -202,7 +203,8 @@ public class SqlQueryScheduler
                 dynamicFilterService,
                 heuristicIndexerManager,
                 snapshotManager,
-                stageTaskCounts);
+                stageTaskCounts,
+                isResume);
         sqlQueryScheduler.initialize();
         return sqlQueryScheduler;
     }
@@ -227,7 +229,8 @@ public class SqlQueryScheduler
             DynamicFilterService dynamicFilterService,
             HeuristicIndexerManager heuristicIndexerManager,
             QuerySnapshotManager snapshotManager,
-            Map<StageId, Integer> stageTaskCounts)
+            Map<StageId, Integer> stageTaskCounts,
+            boolean isResumeScheduler)
     {
         this.queryStateMachine = requireNonNull(queryStateMachine, "queryStateMachine is null");
         this.executionPolicy = requireNonNull(executionPolicy, "schedulerPolicyFactory is null");
@@ -270,7 +273,8 @@ public class SqlQueryScheduler
                 stageLinkageBuilder,
                 isSnapshotEnabled,
                 snapshotManager,
-                stageTaskCounts);
+                stageTaskCounts,
+                isResumeScheduler);
 
         SqlStageExecution rootStage = stageExecutions.get(0);
         rootStage.setOutputBuffers(rootOutputBuffers);
@@ -429,7 +433,8 @@ public class SqlQueryScheduler
             ImmutableMap.Builder<StageId, StageLinkage> stageLinkages,
             boolean isSnapshotEnabled,
             QuerySnapshotManager snapshotManager,
-            Map<StageId, Integer> stageTaskCounts)
+            Map<StageId, Integer> stageTaskCounts,
+            boolean isResumeScheduler)
     {
         ImmutableList.Builder<SqlStageExecution> localStages = ImmutableList.builder();
 
@@ -463,6 +468,9 @@ public class SqlQueryScheduler
             if (isInternalSystemConnector(catalogName)) {
                 catalogName = null;
             }
+            if (isResumeScheduler) {
+                nodeScheduler.refreshNodeStates();
+            }
             NodeSelector nodeSelector = nodeScheduler.createNodeSelector(catalogName, keepConsumerOnFeederNodes, feederScheduledNodes);
             if (isSnapshotEnabled) {
                 // When snapshot is enabled, then no task can be added after the query started running,
@@ -596,7 +604,8 @@ public class SqlQueryScheduler
                     stageLinkages,
                     isSnapshotEnabled,
                     snapshotManager,
-                    stageTaskCounts);
+                    stageTaskCounts,
+                    isResumeScheduler);
             localStages.addAll(subTree);
 
             SqlStageExecution childStage = subTree.get(0);
diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetector.java b/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetector.java
index 2b98f34f3..f9865fff9 100644
--- a/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetector.java
+++ b/presto-main/src/main/java/io/prestosql/failuredetector/FailureDetector.java
@@ -24,6 +24,11 @@ public interface FailureDetector
 
     State getState(HostAddress hostAddress);
 
+    default void waitForServiceStateRefresh()
+    {
+        // do - nothing
+    }
+
     enum State
     {
         UNKNOWN,
diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java b/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java
index 1287c7d4e..567777907 100644
--- a/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java
+++ b/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java
@@ -61,6 +61,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -97,6 +98,7 @@ public class HeartbeatFailureDetector
     private final boolean httpsRequired;
 
     private final AtomicBoolean started = new AtomicBoolean();
+    private AtomicBoolean isTasksRemoved = new AtomicBoolean();
 
     @Inject
     public HeartbeatFailureDetector(
@@ -221,32 +223,57 @@ public class HeartbeatFailureDetector
         return builder.build();
     }
 
+    @Override
+    public void waitForServiceStateRefresh()
+    {
+        Map<ServiceDescriptor, Long> waitingTasks = getTasksTimestamp();
+        long currentTime = waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong();
+        updateMonitoredServices();
+        // remove expired tasks
+        synchronized (tasks) {
+            removeExpiredIds();
+            if (tasks.size() == 0) {
+                return;
+            }
+        }
+        Map<ServiceDescriptor, Long> finalCurrentTasks;
+        do {
+            Map<ServiceDescriptor, Long> currentTasks;
+            Set<ServiceDescriptor> failedTasks;
+            synchronized (tasks) {
+                currentTasks = getTasksTimestamp();
+                failedTasks = getFailed();
+            }
+            currentTasks.entrySet().stream().forEach(e -> {
+                if (e.getValue() > currentTime || failedTasks.contains(e.getKey())) {
+                    if (currentTasks.containsKey(e.getKey())) {
+                        currentTasks.remove(e.getKey());
+                    }
+                }
+            });
+            finalCurrentTasks = currentTasks;
+        } while (finalCurrentTasks.size() > 0);
+    }
+
+    private ConcurrentMap<ServiceDescriptor, Long> getTasksTimestamp()
+    {
+        return tasks.values().stream().collect(Collectors.toConcurrentMap(t -> t.getService(), t -> t.getLastCompleteTimestamp()));
+    }
+
     @VisibleForTesting
     void updateMonitoredServices()
     {
-        Set<ServiceDescriptor> online = selector.selectAllServices().stream()
-                .filter(descriptor -> !nodeInfo.getNodeId().equals(descriptor.getNodeId()))
-                .collect(toImmutableSet());
+        Set<ServiceDescriptor> online = getOnlineServiceDescriptors();
 
-        Set<UUID> onlineIds = online.stream()
-                .map(ServiceDescriptor::getId)
-                .collect(toImmutableSet());
+        Set<UUID> onlineIds = getOnlineIds(online);
 
         // make sure only one thread is updating the registrations
         synchronized (tasks) {
             // 1. remove expired tasks
-            List<UUID> expiredIds = tasks.values().stream()
-                    .filter(MonitoringTask::isExpired)
-                    .map(MonitoringTask::getService)
-                    .map(ServiceDescriptor::getId)
-                    .collect(toImmutableList());
-
-            tasks.keySet().removeAll(expiredIds);
+            removeExpiredIds();
 
             // 2. disable offline services
-            tasks.values().stream()
-                    .filter(task -> !onlineIds.contains(task.getService().getId()))
-                    .forEach(MonitoringTask::disable);
+            disableOfflineTasks(onlineIds);
 
             // 3. create tasks for new services
             Set<ServiceDescriptor> newServices = online.stream()
@@ -268,6 +295,38 @@ public class HeartbeatFailureDetector
         }
     }
 
+    private Set<ServiceDescriptor> getOnlineServiceDescriptors()
+    {
+        return selector.selectAllServices().stream()
+                .filter(descriptor -> !nodeInfo.getNodeId().equals(descriptor.getNodeId()))
+                .collect(toImmutableSet());
+    }
+
+    private Set<UUID> getOnlineIds(Set<ServiceDescriptor> online)
+    {
+        return online.stream()
+                .map(ServiceDescriptor::getId)
+                .collect(toImmutableSet());
+    }
+
+    private void disableOfflineTasks(Set<UUID> onlineIds)
+    {
+        tasks.values().stream()
+                .filter(task -> !onlineIds.contains(task.getService().getId()))
+                .forEach(MonitoringTask::disable);
+    }
+
+    private void removeExpiredIds()
+    {
+        List<UUID> expiredIds = tasks.values().stream()
+                .filter(MonitoringTask::isExpired)
+                .map(MonitoringTask::getService)
+                .map(ServiceDescriptor::getId)
+                .collect(toImmutableList());
+
+        tasks.keySet().removeAll(expiredIds);
+    }
+
     private URI getHttpUri(ServiceDescriptor descriptor)
     {
         String url = descriptor.getProperties().get(httpsRequired ? "https" : "http");
@@ -298,6 +357,12 @@ public class HeartbeatFailureDetector
         @GuardedBy("this")
         private Long successTransitionTimestamp;
 
+        @GuardedBy("this")
+        private long lastCompleteTimestamp;
+
+        @GuardedBy("this")
+        private double lastFailureCount;
+
         private MonitoringTask(ServiceDescriptor service, URI uri)
         {
             this.uri = uri;
@@ -358,6 +423,11 @@ public class HeartbeatFailureDetector
                     Duration.nanosSince(successTransitionTimestamp).compareTo(warmupInterval) < 0; // are we within the warmup period?
         }
 
+        public synchronized long getLastCompleteTimestamp()
+        {
+            return lastCompleteTimestamp;
+        }
+
         private void ping()
         {
             try {
@@ -396,6 +466,11 @@ public class HeartbeatFailureDetector
             }
             else if (successTransitionTimestamp == null) {
                 successTransitionTimestamp = System.nanoTime();
+                lastCompleteTimestamp = System.nanoTime();
+                lastFailureCount = stats.getRecentFailures();
+            }
+            else if (Duration.nanosSince(lastCompleteTimestamp).compareTo(new Duration(1, TimeUnit.SECONDS)) > 0 && stats.getRecentFailures() == lastFailureCount) {
+                lastCompleteTimestamp = System.nanoTime();
             }
         }
     }
diff --git a/presto-main/src/main/java/io/prestosql/metadata/DiscoveryNodeManager.java b/presto-main/src/main/java/io/prestosql/metadata/DiscoveryNodeManager.java
index cb90b2a17..4478dcfcb 100644
--- a/presto-main/src/main/java/io/prestosql/metadata/DiscoveryNodeManager.java
+++ b/presto-main/src/main/java/io/prestosql/metadata/DiscoveryNodeManager.java
@@ -160,6 +160,28 @@ public final class DiscoveryNodeManager
         pollWorkers();
     }
 
+    public void refreshWorkerStates()
+    {
+        failureDetector.waitForServiceStateRefresh();
+        pollWorkers();
+        AllNodes allNodesOverDiscovery = getAllNodes();
+        Set<InternalNode> aliveNodes = ImmutableSet.<InternalNode>builder()
+                .addAll(allNodesOverDiscovery.getActiveNodes())
+                .addAll(allNodesOverDiscovery.getIsolatingNodes())
+                .addAll(allNodesOverDiscovery.getIsolatedNodes())
+                .addAll(allNodesOverDiscovery.getShuttingDownNodes())
+                .build();
+
+        ImmutableSet<String> aliveNodeIds = aliveNodes.stream()
+                .map(InternalNode::getNodeIdentifier)
+                .collect(toImmutableSet());
+
+        // Remove nodes that don't exist anymore
+        // Make a copy to materialize the set difference
+        Set<String> deadNodes = difference(nodeStates.keySet(), aliveNodeIds).immutableCopy();
+        nodeStates.keySet().removeAll(deadNodes);
+    }
+
     private void pollWorkers()
     {
         AllNodes allNodesOverDiscovery = getAllNodes();
diff --git a/presto-main/src/main/java/io/prestosql/metadata/InMemoryNodeManager.java b/presto-main/src/main/java/io/prestosql/metadata/InMemoryNodeManager.java
index b535d1475..1190bc474 100644
--- a/presto-main/src/main/java/io/prestosql/metadata/InMemoryNodeManager.java
+++ b/presto-main/src/main/java/io/prestosql/metadata/InMemoryNodeManager.java
@@ -143,4 +143,10 @@ public class InMemoryNodeManager
     {
         listeners.remove(requireNonNull(listener, "listener is null"));
     }
+
+    @Override
+    public void refreshWorkerStates()
+    {
+        // no-op
+    }
 }
diff --git a/presto-main/src/main/java/io/prestosql/metadata/InternalNodeManager.java b/presto-main/src/main/java/io/prestosql/metadata/InternalNodeManager.java
index 638aa4f26..ef13cda0c 100644
--- a/presto-main/src/main/java/io/prestosql/metadata/InternalNodeManager.java
+++ b/presto-main/src/main/java/io/prestosql/metadata/InternalNodeManager.java
@@ -37,4 +37,6 @@ public interface InternalNodeManager
     void addNodeChangeListener(Consumer<AllNodes> listener);
 
     void removeNodeChangeListener(Consumer<AllNodes> listener);
+
+    void refreshWorkerStates();
 }