PR feedback

- Actually set deferred=True
- add a test for that
- combine/unify environment vars vs cli vars behavior between defer/state better
  - make the state path start with DBT_

core/dbt/contracts/graph/manifest.py

@@ -1030,7 +1030,7 @@ def merge_from_artifact(
                 unique_id not in selected
             ):
                 merged.add(unique_id)
-                self.nodes[unique_id] = node
+                self.nodes[unique_id] = node.replace(deferred=True)
 
         # log up to 5 items
         sample = list(islice(merged, 5))

core/dbt/flags.py

@@ -1,5 +1,6 @@
 import os
 import multiprocessing
+from pathlib import Path
 from typing import Optional
 # initially all flags are set to None, the on-load call of reset() will set
 # them for their first time.
@@ -22,10 +23,19 @@ def env_set_truthy(key: str) -> Optional[str]:
     return value
 
 
+def env_set_path(key: str) -> Optional[Path]:
+    value = os.getenv(key)
+    if value is None:
+        return value
+    else:
+        return Path(value)
+
+
 SINGLE_THREADED_WEBSERVER = env_set_truthy('DBT_SINGLE_THREADED_WEBSERVER')
 SINGLE_THREADED_HANDLER = env_set_truthy('DBT_SINGLE_THREADED_HANDLER')
 MACRO_DEBUGGING = env_set_truthy('DBT_MACRO_DEBUGGING')
 DEFER_MODE = env_set_truthy('DBT_DEFER_TO_STATE')
+ARTIFACT_STATE_PATH = env_set_path('DBT_ARTIFACT_STATE_PATH')
 
 
 def _get_context():

core/dbt/main.py

@@ -461,6 +461,7 @@ def _build_run_subparser(subparsers, base_subparser):
         with this project.
         ''',
         type=Path,
+        default=flags.ARTIFACT_STATE_PATH,
     )
     run_sub.add_optional_argument_inverse(
         '--defer',

core/dbt/task/run.py

@@ -1,5 +1,4 @@
 import functools
-import os
 import time
 from pathlib import Path
 from typing import List, Dict, Any, Iterable, Set, Tuple, Optional
@@ -47,9 +46,6 @@
 from dbt.node_types import NodeType, RunHookType
 
 
-ARTIFACT_STATE_PATH: Optional[str] = os.getenv('ARTIFACT_STATE_PATH')
-
-
 class Timer:
     def __init__(self):
         self.start = None
@@ -256,8 +252,6 @@ def __init__(self, args, config):
     def _get_state_path(self) -> Path:
         if self.args.state is not None:
             return self.args.state
-        elif ARTIFACT_STATE_PATH is not None:
-            return Path(ARTIFACT_STATE_PATH)
         else:
             raise RuntimeException(
                 'Received a --defer argument, but no value was provided '

test/integration/062_defer_state_test/test_defer_state.py

@@ -1,10 +1,12 @@
 from test.integration.base import DBTIntegrationTest, use_profile
 import copy
+import json
 import os
 import shutil
 
 import pytest
 
+
 class TestDeferState(DBTIntegrationTest):
     @property
     def schema(self):
@@ -49,8 +51,10 @@ def copy_state(self):
     def run_and_defer(self):
         results = self.run_dbt(['seed'])
         assert len(results) == 1
+        assert not any(r.node.deferred for r in results)
         results = self.run_dbt(['run'])
         assert len(results) == 2
+        assert not any(r.node.deferred for r in results)
 
         # copy files over from the happy times when we had a good target
         self.copy_state()
@@ -62,6 +66,11 @@ def run_and_defer(self):
         results = self.run_dbt(['run', '-m', 'view_model', '--state', 'state', '--defer', '--target', 'otherschema'])
         assert self.other_schema not in results[0].node.injected_sql
         assert self.unique_schema() in results[0].node.injected_sql
+
+        with open('target/manifest.json') as fp:
+            data = json.load(fp)
+        assert data['nodes']['seed.test.seed']['deferred']
+
         assert len(results) == 1
 
     def run_switchdirs_defer(self):