Repository: dbt-labs/dbt-core
Pull request: Defer iff unselected reference does not exist in current env
resolves #2909

### Description

Today, `--defer` is a blanket override: build the resources included by my selection criteria, and use the `--state` manifest's namespaces for any references to resources _not_ included by my selection criteria.

Instead, this PR would make it a more flexible fallback: If there's a resource I want to build, and it references a resource I haven't selected, use the other manifest's namespace _if and only if_ the upstream resource doesn't exist in my current environment namespace. In this way, `--defer` is more like a "fail-over" mechanism, using the deferred environment to fill in the gaps of whatever's missing in the current environment.

Pros:
- Fixes current unavoidable behavior whereby `dbt run --defer` would _always_ defer non-model resources (seeds and snapshots). If you've just seeded or snapshotted into your scratch/CI schema, you probably want to use those objects to build your models in the following step.
- Fixes annoying behavior whereby subsequent run steps would defer resources run by previous run steps. This is a common use case for testing changed incremental models in CI by rerunning only them (`dbt run -m state:modified ...` followed by `dbt run -m state:modified,config.materialized:incremental ...`).
- If we wanted to, we could reopen test deferral (#2701). I cherry-picked 280803c (from #2706), and it _just worked_. This is after months after convincing myself (and other people, too) that test deferral was A Bad Thing, Actually. Really, I just didn't see a way to do it right. Of course, many of the caveats still apply, e.g. that `relationships` tests don't make a lot of sense in CI environments.

Cons:
- If you have a lot of cruft hanging around in your scratch schema, and you want to use deferral as a "clean" override—use prod references for _everything_ I'm not currently building—you'll need to drop the cruft first. You could see this as a minor regression from current behavior, though IMO it's _very_ minor. The intended use case for `--defer` is dev, CI, and other scratch schemas that can be created, dropped, and recreated with aplomb.
- If you instantiate your dev/CI/scratch schema by first zero-copy cloning on Snowflake, you wouldn't be able to defer any of your references. But you also wouldn't need to; cloning and deferral accomplish the same purpose.

If you buy the premise, the solution is _very_ straightforward—a pleasant surprise. We made deferral a beta feature in v0.18 for exactly this reason: we knew there would be wrinkles to iron out. All the same, it's better to make this change sooner rather than later.

### Checklist
 - [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
 - [x] I have run this code in development and it appears to resolve the stated issue
 - [x] This PR includes tests, or tests are not required/relevant for this PR
 - [x] I have updated the `CHANGELOG.md` and added information about my change to the "dbt next" section.


Problem statement:
Defer refers to outdated seed 
### Describe the bug
I'm running the following on CI:

```bash
aws s3 cp $(MANIFEST_REMOTE_PATH) target/prod/manifest.json
dbt seed --select "state:modified+" --full-refresh --state target/prod/ --target dev-master
dbt run --models "state:modified+" --defer --state target/prod/ --target dev-master
...
```

Say we have a seed `countries.csv`:
```csv
name
Portugal
Ireland
Switzerland
```

And a model `dim_countries.sql`:

```sql
select to_upper(name) as name
from {{ ref("countries") }}
```

Now in one PR we change the seed to
```csv
id,name
1,Portugal
2,Ireland
3,Switzerland
```

And we change the model to:
```sql
select
  id, 
  to_upper(name) as name
from {{ ref("countries") }}
```


dbt correctly identifies that `countries.csv` changed, and that `dim_countries.sql` changed. However when the model runs, it fails with "column id does not exist" because the model tries to read the seed from the "main run" (it defers the seed) instead of identifying that it re-ran.

Note that I tried copying the manifest produced by `dbt seed` into `target/prod/` but what happens then is that dbt does not identify the model as "modified" 

### dbt version

`0.18.1`

Hints from issue discussion:
Thanks for the writeup @dmateusp. The issue here is that `--defer` simply switches the reference of any resource  not included in the selection criteria. As part of node selection, dbt excludes all resources from the selection criteria that are not relevant to the invocation type, e.g. a seed from `dbt run`. So today there's no _real_ way to _not_ defer a seed.

I think this fix is going to be tricky: we'd need dbt to defer on the basis of an intermediate set of selected nodes, rather than the final set. This is part of what made test deferral so cumbersome (#2701), and ultimately not worthwhile. In this case, though, I think we need to figure out an answer.

A much longer-term resolution here would be a generalized command that can operate on both seeds and models (#2743).

In the meantime, the only workarounds I can think of are quite hacky. E.g. if you have a staging model (`models/staging/seeds/*.sql`) that selects from the seed, and `ref()` that staging model instead of the seed downstream, you could try to run that view once (without deferral), before running the rest of the DAG _with_ deferral...

```bash
dbt seed -s state:modified --state .state
dbt run -m state:modified+1,staging.seeds --state .state # no deferral
dbt run -m state:modified+ --exclude state:modified+1,staging.seeds --defer --state .state
```
Thank you for the details @jtcohen6, the staging model is an interesting solution but not something practical to enforce in our project right now unfortunately.

For a short term solution would it be possible to extend dbt to allow deferring only models ? That way I could change my CI to always re-create all the seeds (which is not a super big deal since seeds should be small)

`dbt run -m state:modified+ --defer models <...>` ?
@dmateusp That's a good thought, and if it proves much more straightforward, that may be the move. In either case, it will require tweaking some of the logic of deferral—which isn't something we can squeeze in for the next minor version (v0.19), but possibly for the one after.

In the meantime, I'll plan to document this as a known caveat to state comparison + deferral.
One approach that's occurred to me, though I haven't thought through all the implications:

Today, we defer _all_ models that are not included in the node selection criteria. That means we defer:
- seeds (always)
- modified models that are included, then excluded, from shifting criteria, e.g. `dbt run -m state:modified && dbt run -m state:modified,config.materialized:incremental` (see [slack thread](https://getdbt.slack.com/archives/CMZ2V0X8V/p1607008955284500))

Perhaps we shouldn't use selection criteria as the basis for deferral. Instead, during compilation, we could check to see if a referent's "new" representation (`ci_schema.identifier`) exists in the database (via cache lookup). If it doesn't exist,  we "fall back" to the comparison manifest's representation (`prod_schema.identifier`) of a node with the same `unique_id`.

If we took this more-naive approach to deferral:
- Would it also help us avoid tricky issues around manifest construction when previous source or ephemeral model parents have gone missing (#2875)?
- Could we even revisit the question of how deferral might work for tests (#2701)?
I like that a lot!

Commit message: Check if relation exists before deferring

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9e596be3ba8..3175f028278 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,7 @@
 - dbt list command always return 0 as exit code ([#2886](https://github.com/fishtown-analytics/dbt/issues/2886), [#2892](https://github.com/fishtown-analytics/dbt/issues/2892))
 - Set default `materialized` for test node configs to `test` ([#2806](https://github.com/fishtown-analytics/dbt/issues/2806), [#2902](https://github.com/fishtown-analytics/dbt/pull/2902))
 - Use original file path instead of absolute path as checksum for big seeds ([#2927](https://github.com/fishtown-analytics/dbt/issues/2927), [#2939](https://github.com/fishtown-analytics/dbt/pull/2939))
+- Defer if and only if upstream reference does not exist in current environment namespace ([#2909](https://github.com/fishtown-analytics/dbt/issues/2909), [#2946](https://github.com/fishtown-analytics/dbt/pull/2946))
 
 ### Under the hood
 - Bump hologram version to 0.0.11. Add scripts/dtr.py ([#2888](https://github.com/fishtown-analytics/dbt/issues/2840),[#2889](https://github.com/fishtown-analytics/dbt/pull/2889))
diff --git a/core/dbt/contracts/graph/manifest.py b/core/dbt/contracts/graph/manifest.py
index 6f8bb26d8ce..48f68bba150 100644
--- a/core/dbt/contracts/graph/manifest.py
+++ b/core/dbt/contracts/graph/manifest.py
@@ -887,6 +887,7 @@ def resolve_doc(
 
     def merge_from_artifact(
         self,
+        adapter,
         other: 'WritableManifest',
         selected: AbstractSet[UniqueID],
     ) -> None:
@@ -898,10 +899,14 @@ def merge_from_artifact(
         refables = set(NodeType.refable())
         merged = set()
         for unique_id, node in other.nodes.items():
+            current = self.nodes[unique_id]
             if (
                 node.resource_type in refables and
                 not node.is_ephemeral and
-                unique_id not in selected
+                unique_id not in selected and
+                not adapter.get_relation(
+                    current.database, current.schema, current.identifier
+                )
             ):
                 merged.add(unique_id)
                 self.nodes[unique_id] = node.replace(deferred=True)
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index 086d423477e..495cd5523bb 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -372,7 +372,7 @@ def _get_deferred_manifest(self) -> Optional[WritableManifest]:
             )
         return state.manifest
 
-    def defer_to_manifest(self, selected_uids: AbstractSet[str]):
+    def defer_to_manifest(self, adapter, selected_uids: AbstractSet[str]):
         deferred_manifest = self._get_deferred_manifest()
         if deferred_manifest is None:
             return
@@ -382,6 +382,7 @@ def defer_to_manifest(self, selected_uids: AbstractSet[str]):
                 'manifest to defer from!'
             )
         self.manifest.merge_from_artifact(
+            adapter=adapter,
             other=deferred_manifest,
             selected=selected_uids,
         )
@@ -389,10 +390,10 @@ def defer_to_manifest(self, selected_uids: AbstractSet[str]):
         self.write_manifest()
 
     def before_run(self, adapter, selected_uids: AbstractSet[str]):
-        self.defer_to_manifest(selected_uids)
         with adapter.connection_named('master'):
             self.create_schemas(adapter, selected_uids)
             self.populate_adapter_cache(adapter)
+            self.defer_to_manifest(adapter, selected_uids)
             self.safe_run_hooks(adapter, RunHookType.Start, {})
 
     def after_run(self, adapter, results):
diff --git a/test/integration/062_defer_state_test/test_defer_state.py b/test/integration/062_defer_state_test/test_defer_state.py
index 1d3a4fc7b8d..ebdd2421e0c 100644
--- a/test/integration/062_defer_state_test/test_defer_state.py
+++ b/test/integration/062_defer_state_test/test_defer_state.py
@@ -102,6 +102,24 @@ def run_switchdirs_defer(self):
             expect_pass=False,
         )
 
+    def run_defer_iff_not_exists(self):
+        results = self.run_dbt(['seed', '--target', 'otherschema'])
+        assert len(results) == 1
+        results = self.run_dbt(['run', '--target', 'otherschema'])
+        assert len(results) == 2
+
+        # copy files over from the happy times when we had a good target
+        self.copy_state()
+        
+        results = self.run_dbt(['seed'])
+        assert len(results) == 1
+        results = self.run_dbt(['run', '--state', 'state', '--defer'])
+        assert len(results) == 2
+        
+        # because the seed now exists in our schema, we shouldn't defer it
+        assert self.other_schema not in results[0].node.compiled_sql
+        assert self.unique_schema() in results[0].node.compiled_sql
+
     @use_profile('postgres')
     def test_postgres_state_changetarget(self):
         self.run_and_defer()
@@ -118,8 +136,12 @@ def test_postgres_state_changetarget(self):
             self.run_dbt(['snapshot', '--defer'])
 
     @use_profile('postgres')
-    def test_postgres_stat_changedir(self):
+    def test_postgres_state_changedir(self):
         self.run_switchdirs_defer()
+        
+    @use_profile('postgres')
+    def test_postgres_state_defer_iffnotexists(self):
+        self.run_defer_iff_not_exists()
 
     @use_profile('snowflake')
     def test_snowflake_state_changetarget(self):


Code review comments:
diff_hunk: @@ -898,10 +899,14 @@ def merge_from_artifact(
         refables = set(NodeType.refable())
         merged = set()
         for unique_id, node in other.nodes.items():
+            current = self.nodes[unique_id]
path: core/dbt/contracts/graph/manifest.py
text: @user1:
is there a scenario where `unique_id` is present in the _other_ manifest, but not in _this_ manifest? I'm thinking that could happen if a model was deleted, for instance. If I'm thinking about that right, then I think this could raise a KeyError, and we'd probably want to use `self.nodes.get(unique_id)` and do something smart if the result is `None`. 

I might be way off-base here, it just jumped out at me though and wanted to confirm

@author:
really good point, I was able to produce a `KeyError` here and changed the line to handle. this also gave me a clue to #2875 :)