fix #7300: Enable state for deferral to be separate from state for se…

…lectors (#7690)

.changes/unreleased/Features-20230523-225955.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Enable state for deferral to be separate from state for selectors
+time: 2023-05-23T22:59:55.920975-07:00
+custom:
+  Author: aranke
+  Issue: "7300"

core/dbt/cli/main.py

@@ -179,6 +179,7 @@ def cli(ctx, **kwargs):
 @p.selector
 @p.show
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.store_failures
 @p.target
@@ -250,6 +251,7 @@ def docs(ctx, **kwargs):
 @p.selector
 @p.empty_catalog
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -322,6 +324,7 @@ def docs_serve(ctx, **kwargs):
 @p.selector
 @p.inline
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -368,6 +371,7 @@ def compile(ctx, **kwargs):
 @p.selector
 @p.inline
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -476,6 +480,7 @@ def init(ctx, **kwargs):
 @p.raw_select
 @p.selector
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -545,6 +550,7 @@ def parse(ctx, **kwargs):
 @p.select
 @p.selector
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -612,6 +618,7 @@ def run_operation(ctx, **kwargs):
 @p.selector
 @p.show
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -650,6 +657,7 @@ def seed(ctx, **kwargs):
 @p.select
 @p.selector
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -692,6 +700,7 @@ def source(ctx, **kwargs):
 @p.select
 @p.selector
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.target
 @p.target_path
@@ -738,6 +747,7 @@ def freshness(ctx, **kwargs):
 @p.select
 @p.selector
 @p.state
+@p.defer_state
 @p.deprecated_state
 @p.store_failures
 @p.target

core/dbt/cli/params.py

@@ -426,7 +426,20 @@
 state = click.option(
     "--state",
     envvar="DBT_STATE",
-    help="If set, use the given directory as the source for JSON files to compare with this project.",
+    help="Unless overridden, use this state directory for both state comparison and deferral.",
+    type=click.Path(
+        dir_okay=True,
+        file_okay=False,
+        readable=True,
+        resolve_path=True,
+        path_type=Path,
+    ),
+)
+
+defer_state = click.option(
+    "--defer-state",
+    envvar="DBT_DEFER_STATE",
+    help="Override the state directory for deferral only.",
     type=click.Path(
         dir_okay=True,
         file_okay=False,

core/dbt/task/compile.py

@@ -100,14 +100,14 @@ def _get_deferred_manifest(self) -> Optional[WritableManifest]:
         if not self.args.defer:
             return None
 
-        state = self.previous_state
-        if state is None:
+        state = self.previous_defer_state or self.previous_state
+        if not state:
             raise DbtRuntimeError(
                 "Received a --defer argument, but no value was provided to --state"
             )
 
-        if state.manifest is None:
-            raise DbtRuntimeError(f'Could not find manifest in --state path: "{self.args.state}"')
+        if not state.manifest:
+            raise DbtRuntimeError(f'Could not find manifest in --state path: "{state}"')
         return state.manifest
 
     def defer_to_manifest(self, adapter, selected_uids: AbstractSet[str]):

core/dbt/task/runnable.py

@@ -60,7 +60,6 @@
 
 
 class GraphRunnableTask(ConfiguredTask):
-
     MARK_DEPENDENT_ERRORS_STATUSES = [NodeStatus.Error]
 
     def __init__(self, args, config, manifest):
@@ -72,17 +71,20 @@ def __init__(self, args, config, manifest):
         self.node_results = []
         self.num_nodes: int = 0
         self.previous_state: Optional[PreviousState] = None
+        self.previous_defer_state: Optional[PreviousState] = None
         self.run_count: int = 0
         self.started_at: float = 0
 
-        self.set_previous_state()
-
-    def set_previous_state(self):
-        if self.args.state is not None:
+        if self.args.state:
             self.previous_state = PreviousState(
                 path=self.args.state, current_path=Path(self.config.target_path)
             )
 
+        if self.args.defer_state:
+            self.previous_defer_state = PreviousState(
+                path=self.args.defer_state, current_path=Path(self.config.target_path)
+            )
+
     def index_offset(self, value: int) -> int:
         return value
 

tests/functional/defer_state/test_defer_state.py

@@ -6,6 +6,7 @@
 import pytest
 
 from dbt.cli.exceptions import DbtUsageException
+from dbt.contracts.results import RunStatus
 from dbt.tests.util import run_dbt, write_file, rm_file
 
 from dbt.exceptions import DbtRuntimeError
@@ -23,6 +24,7 @@
     macros_sql,
     infinite_macros_sql,
     snapshot_sql,
+    view_model_now_table_sql,
 )
 
 
@@ -272,3 +274,68 @@ def test_run_defer_deleted_upstream(self, project, unique_schema, other_schema):
         )
         results = run_dbt(["test", "--state", "state", "--defer", "--favor-state"])
         assert other_schema not in results[0].node.compiled_code
+
+
+class TestDeferStateFlag(BaseDeferState):
+    def test_defer_state_flag(self, project, unique_schema, other_schema):
+        project.create_test_schema(other_schema)
+
+        # test that state deferral works correctly
+        run_dbt(["compile", "--target-path", "target_compile"])
+        write_file(view_model_now_table_sql, "models", "table_model.sql")
+
+        results = run_dbt(["ls", "--select", "state:modified", "--state", "target_compile"])
+        assert results == ["test.table_model"]
+
+        run_dbt(["seed", "--target", "otherschema", "--target-path", "target_otherschema"])
+
+        # this will fail because we haven't loaded the seed in the default schema
+        run_dbt(
+            [
+                "run",
+                "--select",
+                "state:modified",
+                "--defer",
+                "--state",
+                "target_compile",
+                "--favor-state",
+            ],
+            expect_pass=False,
+        )
+
+        # this will fail because we haven't passed in --state
+        with pytest.raises(
+            DbtRuntimeError, match="Got a state selector method, but no comparison manifest"
+        ):
+            run_dbt(
+                [
+                    "run",
+                    "--select",
+                    "state:modified",
+                    "--defer",
+                    "--defer-state",
+                    "target_otherschema",
+                    "--favor-state",
+                ],
+                expect_pass=False,
+            )
+
+        # this will succeed because we've loaded the seed in other schema and are successfully deferring to it instead
+        results = run_dbt(
+            [
+                "run",
+                "--select",
+                "state:modified",
+                "--defer",
+                "--state",
+                "target_compile",
+                "--defer-state",
+                "target_otherschema",
+                "--favor-state",
+            ]
+        )
+
+        assert len(results.results) == 1
+        assert results.results[0].status == RunStatus.Success
+        assert results.results[0].node.name == "table_model"
+        assert results.results[0].adapter_response["rows_affected"] == 2