Clone doesnt require --defer

core/dbt/task/clone.py

@@ -1,9 +1,10 @@
 import threading
-from typing import AbstractSet, Any, List, Iterable, Set
+from typing import AbstractSet, Any, List, Iterable, Set, Optional
 
 from dbt.adapters.base import BaseRelation
 from dbt.clients.jinja import MacroGenerator
 from dbt.context.providers import generate_runtime_model_context
+from dbt.contracts.graph.manifest import WritableManifest
 from dbt.contracts.results import RunStatus, RunResult
 from dbt.dataclass_schema import dbtClassMixin
 from dbt.exceptions import DbtInternalError, CompilationError
@@ -93,6 +94,11 @@ class CloneTask(GraphRunnableTask):
     def raise_on_first_error(self):
         return False
 
+    def _get_deferred_manifest(self) -> Optional[WritableManifest]:
+        # Unlike other commands, 'clone' always requires a state manifest
+        # Load previous state, regardless of whether --defer flag has been set
+        return self._get_previous_state()
+
     def get_model_schemas(self, adapter, selected_uids: Iterable[str]) -> Set[BaseRelation]:
         if self.manifest is None:
             raise DbtInternalError("manifest was None in get_model_schemas")

core/dbt/task/runnable.py

@@ -617,10 +617,7 @@ def get_result(self, results, elapsed_time, generated_at):
     def task_end_messages(self, results):
         print_run_end_messages(results)
 
-    def _get_deferred_manifest(self) -> Optional[WritableManifest]:
-        if not self.args.defer:
-            return None
-
+    def _get_previous_state(self) -> Optional[WritableManifest]:
         state = self.previous_defer_state or self.previous_state
         if not state:
             raise DbtRuntimeError(
@@ -630,3 +627,6 @@ def _get_deferred_manifest(self) -> Optional[WritableManifest]:
         if not state.manifest:
             raise DbtRuntimeError(f'Could not find manifest in --state path: "{state}"')
         return state.manifest
+
+    def _get_deferred_manifest(self) -> Optional[WritableManifest]:
+        return self._get_previous_state() if self.args.defer else None