Repository: dbt-labs/dbt-core
Pull request: Smart Reruns from Failures/Errors
resolves #3891

<!---
  Include the number of the issue addressed by this PR above if applicable.
  PRs for code changes without an associated issue *will not be merged*.
  See CONTRIBUTING.md for more information.

  Example:
    resolves #1234
-->

### Description

<!--- Describe the Pull Request here -->
Instead of having to manually override dbt commands with the models in scope that need to rerun, dbt can now parse the `run_results.json` to do so automatically.

**Example Usage**
First dbt job model has an error.
![image](https://user-images.githubusercontent.com/19176976/136442293-b12d3a00-4d00-4387-8b21-d4b60e9c8c78.png)

Rerun dbt job errors and downstream nodes.
![image](https://user-images.githubusercontent.com/19176976/136442306-77fc8c00-6ae2-4288-b082-f48ab23803b0.png)


**Result Selectors in Scope**
Note: This pull request does NOT limit which selectors the user can call on from the below list. There are use cases to analyze the nodes for any of these result statuses(ex: `dbt ls --select result:skipped`) 
- `result:fail`
- `result:error`
- `result:warn`
- `result:success`
- `result:skipped`
- `result:pass`

**Testing Approach**
Tested result selectors relevant to the commands in scope. For example, you'll see a test for `dbt run --select result:error`, but you won't see a test for `dbt run --select result:fail`. This is because a `result:fail` selector will never exist for `dbt run` in the `run_results.json`. 

For concurrent selector tests, we tested the most common use cases below. 
- `dbt run --select state:modified+ result:error+ -—defer -—state ./target`
    - Rerun all my erroneous models AND run changes I made concurrently that may relate to the erroneous models for downstream use
- `dbt build --select state:modified+ result:error+ -—defer -—state ./target`
    - Rerun and retest all my erroneous models AND run changes I made concurrently that may relate to the erroneous models for downstream use
- `dbt build --select state:modified+ result:error+ result:fail+ --defer --state ./target`
    - Rerun all my erroneous models AND all my failed tests
    - Rerun all my erroneous models AND run changes I made concurrently that may relate to the erroneous models for downstream use
    - There's a failed test that's unrelated to modified or error nodes(think: source test that needs to refresh a data load in order to pass)
- `dbt test --select result:fail --exclude <example test> -—defer -—state ./target`
    - Rerun all my failed tests and exclude tests that I know will still fail
    - This can apply to updates in source data during the "EL" process that need to be rerun after they are refreshed

### 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:
Checkpointing in dbt
### Describe the feature
A checkpointing mechanism that can be used to start from a failure point when running a `dbt run` statement.

### Describe alternatives you've considered
As I'm running `dbt run` via cli, I'm using some shell/python scripts to create a checkpoint file (a txt file with the name of last executed model) that if the execution fails, I can re-run `dbt run` that it will start from the last execution model.

### Additional context
Using the `dbt run` command if any problem occurs, we need or (a) re-start the execution manually from the model that failed or (b) re-run the full execution again. 

### Who will this benefit?
- Teams that have a larger quantity of models (in my case I have more than 308) and suffers from the network and/or database issues that can break an execution
- Teams that want to reduce their recovery time if some outage happens during the dbt execution that instead to re-run `dbt run` again or `dbt run model_1, dbt run model_n` manually; start from the last point before the interruption 

### Are you interested in contributing this feature?
Yes.

Hints from issue discussion:
@fclesio Agree big-time!

The way I've been thinking about this: dbt should be able to read the artifact from a prior run (namely `run_results.json`), and select all nodes with a certain [status](https://schemas.getdbt.com/dbt/run-results/v2/index.html#tab-pane_results_items_status_oneOf_i0). That could look like `dbt run --select result:error result:skipped result:failed`, or it could be simple as `--select result:error+`: since skipped nodes are always the downstream children of errored/failed nodes, and a failing (as opposed to errored-out) test likely requires additional work that a simple retry won't solve.

In order to make this work, we'd need to teach dbt about remote results, in the same way we taught dbt about remote manifests for the sake of `state:` comparison. There's thinking about this over in https://github.com/dbt-labs/dbt/issues/2465, though that's now a fairly old issue. I think I'm going to close that issue, and open narrower issues targeting the outstanding use cases: selecting based on (a) past node results and (b) source freshness comparisons.

Does that artifact-based approach make sense to you? Is there anything you're thinking of that I'm missing?
@jtcohen6 thanks for the prompt response. 

The main issue that I see with the `dbt run --select result:error result:skipped result:failed` statement is that it assumes that the retry will be placed manually, and for the perspective of E(T)L/EL(T) perspective the checkpoint will be used in some situation of failure with some kind of automatic retry. 

For instance: Let's say after execute `dbt run` for my 308 models my job fails during the model materialization in the 97. At the time of an automatic retry, the command should be `dbt run` because the job itself does not know what failed or not. 

From the API perspective, I see something like `dbt run --checkpoint-name run_results.json --checkpoint-utilization:true` because in this way we can assure that:
1) Who needs the checkpoint will explicitly put that in the statement;

2.1) When the checkpoint is not chosen, dbt does not need to care about the remote states;

2.2) Who chose the checkpoint will explicitly know that it will have some toll in terms of execution time, due to the remote states;

3) For job scheduling a single statement (i.e, `dbt run --checkpoint-name run_results.json --checkpoint-utilization:true` ) will solve all the retries, instead to run a command that assumes that something already failed (like `dbt run --select result:error result:skipped result:failed`). In other words, instead to have a command that says "_If something fails run me in those explicit terms!_" we can have a statement that says "_store all states along the time if something fails and you decide to retry automatically I know from which point I need to start over again..._"

I do not know if it's clear or not, but I'm happy to clarify more if needed. 



I prefer the `dbt run --select result:error+` semantics because it's consistent with how dbt users leverage similar syntax for Slim CI. To change the syntax using checkpoints will require dbt users to adopt another mental model for what's being run vs. not. 

I can also imagine a combination like this: `dbt run --select result:error+ state:modified+` that will provide a more coherent mental model for more nuanced runs.
Sorry for the stubbornness sungchun12, but considering a scenario of an automatic retry to avoid dbt run everything from scratch, it should be implicitly having an option establishing the checkpoint?

The idea is for every execution I should run `dbt run --select result:error+` regardless of the state, _i.e._ if failed previously or not, right? And in case of some failure that an automatic retry needs to take place, nothing should be changed. Is that correct from my side?
@fclesio Thanks for the thoughtful responses! I think there's an important distinction here:
- Automatic retry of a failed _query_: When a dbt-<adapter> plugin receives an error from the database that it believes to be transient, and conducive to successful retry, there should be a configuration option such that it retries _the specific query_. This exists already in `dbt-bigquery` + `dbt-spark`, and there's a proposal to add it on Postgres/Redshift as well: #3303. The real limitation here is the level of clarity and documentation from the database and its python client.
- Manual retry of a failed _job_: A pipeline has failed for any of several reasons. A model has a syntax error, a test found null records where we didn't expect, a source dropped a column without warning, a permission was accidentally revoked, etc. In this case, automatically re-running all the same queries would achieve little.

I took the thrust of this issue to mean the latter more than the former. If a specific model fails because its query timed out, or ran into intermittent network and/or database issues, the right answer feels like automatic retry—based on logic coded into the database adapter that checks for the timeout/transient error code. If that's more the thing you're after, I think #3303 is the right place to continue that conversation.

I'd be happy to open a separate issue for "smart manual retry," which is the `result:error+` syntax described above. I do think orchestration processes that wrap dbt Core could offer UX that makes the restart process pretty seamless: a click of a button, "Retry this job," that picks up from the failed run step with `result:error+`. Or better yet, to Sung's point, if you've needed to push code changes to account for an unexpected model/test/etc failure: `result:error+ state:modified+`.

@matt-winkler @sungchun12 To that end: I'd be thrilled to work with both of you on making this happen. The `state:` selection set a lot of the necessary groundwork here; I was able to get an experimental version of `dbt run -s result:<status> --state <state-path>` up and running with only a handful of code: [`experiment/previous-result-selection`](https://github.com/dbt-labs/dbt/compare/experiment/previous-result-selection).

One question: Should `result:<status>` be the only result-based selection criteria, or should we anticipate others and name it `result.status:<status>` instead? I'm actually leaning toward the former. [A year ago](https://github.com/dbt-labs/dbt/issues/2465#issuecomment-689823535), I'd thought about _timing_ (`result.timing:>600`), but I don't think that's really useful here—it's valuable metadata, just not as proactive selection criteria for a new run. Same goes for specific failure counts for tests (`result.failures:>500`); I think users should just use severity + thresholds (`warn_if` + `error_if`) to summarize up those details into specific result statuses (`result:warn`, `result:error`). Happy to hear disagreement, though!
@fclesio I don't see your response as stubborn at all! I welcome it and you raise a valid point with retrying `dbt run` regardless of state-very much aligned with @jtcohen6's mention of #3303 for transient errors that benefit from automatic retries. Based on the latest comment, this can be a both/and solution, not either/or. 

I expect the best of both worlds, even with implicit logic for transient errors. Given this paradigm already exists for specific database adapters, we shouldn't need an explicit flag like: `dbt --retry-on-transient-errors run` or `dbt run --checkpoint-name`.

For result-based selection criteria, I recommend limiting it to  (`result:warn, result:error`). I have not heard of users in conversations or across the public slack showing strong desire for rerunning based on timing. Today, users can do that through [tags](https://docs.getdbt.com/reference/resource-properties/tags/) if needed. 

@jtcohen6 I would love to help out and branch off your experiment branch. @fclesio @matt-winkler, I would love to work with you too to make this a reality. @fclesio, I understand if you want to focus on the transient error checkpoint logic for your specific database instead. What database are you using today for dbt?




>What database are you using today for dbt?

@sungchun12 I'm using Redshift. 
@fclesio, cool! when we have a working pull request, we'd love for you to test drive it!

Commit message: Update changelog

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index bfa5948e5f6..fd562f339ff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,8 +1,15 @@
 ## dbt-core 1.0.0 (Release TBD)
 
-## dbt-core 1.0.0b1 (October 11, 2021)
+### Features
+
+- Add `result:<status>` selectors to automatically rerun failed tests and erroneous models. This makes it easier to rerun failed dbt jobs with a simple selector flag instead of restarting from the beginning or manually running the dbt models in scope. ([#3859](https://github.com/dbt-labs/dbt/issues/3891), [#4017](https://github.com/dbt-labs/dbt/pull/4017))
+
+### Contributors
 
-## dbt 1.0.0b1 (October 11, 2021)
+- [@sungchun12](https://github.com/sungchun12) ([#4017](https://github.com/dbt-labs/dbt/pull/4017))
+- [@matt-winkler](https://github.com/matt-winkler) ([#4017](https://github.com/dbt-labs/dbt/pull/4017))
+
+## dbt-core 1.0.0b1 (October 11, 2021)
 
 ### Breaking changes
 
@@ -38,7 +45,7 @@
 - Define the data type of `sources` as an array of arrays of string in the manifest artifacts. ([#3966](https://github.com/dbt-labs/dbt/issues/3966), [#3967](https://github.com/dbt-labs/dbt/pull/3967))
 - Marked `source-paths` and `data-paths` as deprecated keys in `dbt_project.yml` in favor of `model-paths` and `seed-paths` respectively.([#1607](https://github.com/dbt-labs/dbt/issues/1607))
 
-Contributors:
+### Contributors
 
 - [@dave-connors-3](https://github.com/dave-connors-3) ([#3920](https://github.com/dbt-labs/dbt/pull/3922))
 - [@kadero](https://github.com/kadero) ([#3952](https://github.com/dbt-labs/dbt/pull/3953))
diff --git a/core/dbt/contracts/state.py b/core/dbt/contracts/state.py
index 43999bba29c..1f269287ffb 100644
--- a/core/dbt/contracts/state.py
+++ b/core/dbt/contracts/state.py
@@ -1,5 +1,6 @@
 from pathlib import Path
 from .graph.manifest import WritableManifest
+from .results import RunResultsArtifact
 from typing import Optional
 from dbt.exceptions import IncompatibleSchemaException
 
@@ -8,6 +9,7 @@ class PreviousState:
     def __init__(self, path: Path):
         self.path: Path = path
         self.manifest: Optional[WritableManifest] = None
+        self.results: Optional[RunResultsArtifact] = None
 
         manifest_path = self.path / 'manifest.json'
         if manifest_path.exists() and manifest_path.is_file():
@@ -16,3 +18,11 @@ def __init__(self, path: Path):
             except IncompatibleSchemaException as exc:
                 exc.add_filename(str(manifest_path))
                 raise
+
+        results_path = self.path / 'run_results.json'
+        if results_path.exists() and results_path.is_file():
+            try:
+                self.results = RunResultsArtifact.read(str(results_path))
+            except IncompatibleSchemaException as exc:
+                exc.add_filename(str(results_path))
+                raise
diff --git a/core/dbt/graph/selector_methods.py b/core/dbt/graph/selector_methods.py
index d0c9ad7d589..7ac33136212 100644
--- a/core/dbt/graph/selector_methods.py
+++ b/core/dbt/graph/selector_methods.py
@@ -45,6 +45,7 @@ class MethodName(StrEnum):
     ResourceType = 'resource_type'
     State = 'state'
     Exposure = 'exposure'
+    Result = 'result'
 
 
 def is_selected_node(fqn: List[str], node_selector: str):
@@ -516,6 +517,23 @@ def search(
                 yield node
 
 
+class ResultSelectorMethod(SelectorMethod):
+    def search(
+        self, included_nodes: Set[UniqueId], selector: str
+    ) -> Iterator[UniqueId]:
+        if self.previous_state is None or self.previous_state.results is None:
+            raise InternalException(
+                'No comparison run_results'
+            )
+        matches = set(
+            result.unique_id for result in self.previous_state.results
+            if result.status == selector
+        )
+        for node, real_node in self.all_nodes(included_nodes):
+            if node in matches:
+                yield node
+
+
 class MethodManager:
     SELECTOR_METHODS: Dict[MethodName, Type[SelectorMethod]] = {
         MethodName.FQN: QualifiedNameSelectorMethod,
@@ -528,6 +546,7 @@ class MethodManager:
         MethodName.TestType: TestTypeSelectorMethod,
         MethodName.State: StateSelectorMethod,
         MethodName.Exposure: ExposureSelectorMethod,
+        MethodName.Result: ResultSelectorMethod,
     }
 
     def __init__(
diff --git a/test/integration/062_defer_state_test/models/schema.yml b/test/integration/062_defer_state_test/models/schema.yml
index 1ec506d3d19..342335148bf 100644
--- a/test/integration/062_defer_state_test/models/schema.yml
+++ b/test/integration/062_defer_state_test/models/schema.yml
@@ -4,6 +4,7 @@ models:
     columns:
       - name: id
         tests:
-          - unique
+          - unique:
+              severity: error
           - not_null
       - name: name
diff --git a/test/integration/062_defer_state_test/test_run_results_state.py b/test/integration/062_defer_state_test/test_run_results_state.py
new file mode 100644
index 00000000000..e6fce772e77
--- /dev/null
+++ b/test/integration/062_defer_state_test/test_run_results_state.py
@@ -0,0 +1,436 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import os
+import random
+import shutil
+import string
+
+import pytest
+
+from dbt.exceptions import CompilationException
+
+
+class TestRunResultsState(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_results_state_062"
+
+    @property
+    def models(self):
+        return "models"
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['macros'],
+            'seeds': {
+                'test': {
+                    'quote_columns': True,
+                }
+            }
+        }
+
+    def _symlink_test_folders(self):
+        # dbt's normal symlink behavior breaks this test. Copy the files
+        # so we can freely modify them.
+        for entry in os.listdir(self.test_original_source_path):
+            src = os.path.join(self.test_original_source_path, entry)
+            tst = os.path.join(self.test_root_dir, entry)
+            if entry in {'models', 'data', 'macros'}:
+                shutil.copytree(src, tst)
+            elif os.path.isdir(entry) or entry.endswith('.sql'):
+                os.symlink(src, tst)
+
+    def copy_state(self):
+        assert not os.path.exists('state')
+        os.makedirs('state')
+        shutil.copyfile('target/manifest.json', 'state/manifest.json')
+        shutil.copyfile('target/run_results.json', 'state/run_results.json')
+
+    def setUp(self):
+        super().setUp()
+        self.run_dbt(['build'])
+        self.copy_state()
+    
+    def rebuild_run_dbt(self, expect_pass=True):
+        shutil.rmtree('./state')
+        self.run_dbt(['build'], expect_pass=expect_pass)
+        self.copy_state()
+
+    @use_profile('postgres')
+    def test_postgres_seed_run_results_state(self):
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'])
+        self.copy_state()
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:success', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+        with open('data/seed.csv') as fp:
+            fp.readline()
+            newline = fp.newlines
+        with open('data/seed.csv', 'a') as fp:
+            fp.write(f'\"\'\'3,carl{newline}')
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'], expect_pass=False)
+        self.copy_state()
+
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:error', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+
+        with open('data/seed.csv') as fp:
+            fp.readline()
+            newline = fp.newlines
+        with open('data/seed.csv', 'a') as fp:
+            # assume each line is ~2 bytes + len(name)
+            target_size = 1*1024*1024
+            line_size = 64
+
+            num_lines = target_size // line_size
+
+            maxlines = num_lines + 4
+
+            for idx in range(4, maxlines):
+                value = ''.join(random.choices(string.ascii_letters, k=62))
+                fp.write(f'{idx},{value}{newline}')
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'], expect_pass=False)
+        self.copy_state()
+
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:error', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+    @use_profile('postgres')
+    def test_postgres_build_run_results_state(self):
+        results = self.run_dbt(['build', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 0
+
+        with open('models/view_model.sql') as fp:
+            fp.readline()
+            newline = fp.newlines
+
+        with open('models/view_model.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_error")
+            fp.write(newline)
+        
+        self.rebuild_run_dbt(expect_pass=False)
+
+        results = self.run_dbt(['build', '--select', 'result:error', '--state', './state'], expect_pass=False)
+        assert len(results) == 3
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'view_model', 'not_null_view_model_id','unique_view_model_id'}
+
+        results = self.run_dbt(['ls', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 3
+        assert set(results) == {'test.view_model', 'test.not_null_view_model_id', 'test.unique_view_model_id'}
+
+        results = self.run_dbt(['build', '--select', 'result:error+', '--state', './state'], expect_pass=False)
+        assert len(results) == 4
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'table_model','view_model', 'not_null_view_model_id','unique_view_model_id'}
+
+        results = self.run_dbt(['ls', '--select', 'result:error+', '--state', './state'])
+        assert len(results) == 6 # includes exposure
+        assert set(results) == {'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+        # test failure on build tests
+        # fail the unique test
+        with open('models/view_model.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select 1 as id union all select 1 as id")
+            fp.write(newline)
+        
+        self.rebuild_run_dbt(expect_pass=False)
+
+        results = self.run_dbt(['build', '--select', 'result:fail', '--state', './state'], expect_pass=False)
+        assert len(results) == 1
+        assert results[0].node.name == 'unique_view_model_id'
+
+        results = self.run_dbt(['ls', '--select', 'result:fail', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.unique_view_model_id'
+
+        results = self.run_dbt(['build', '--select', 'result:fail+', '--state', './state'], expect_pass=False)
+        assert len(results) == 2
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'table_model', 'unique_view_model_id'}
+
+        results = self.run_dbt(['ls', '--select', 'result:fail+', '--state', './state'])
+        assert len(results) == 2
+        assert set(results) == {'test.table_model', 'test.unique_view_model_id'}
+
+        # change the unique test severity from error to warn and reuse the same view_model.sql changes above
+        f = open('models/schema.yml', 'r')
+        filedata = f.read()
+        f.close()
+        newdata = filedata.replace('error','warn')
+        f = open('models/schema.yml', 'w')
+        f.write(newdata)
+        f.close()
+
+        self.rebuild_run_dbt(expect_pass=True)
+
+        results = self.run_dbt(['build', '--select', 'result:warn', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0].node.name == 'unique_view_model_id'
+
+        results = self.run_dbt(['ls', '--select', 'result:warn', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.unique_view_model_id'
+
+        results = self.run_dbt(['build', '--select', 'result:warn+', '--state', './state'], expect_pass=True)
+        assert len(results) == 2 # includes table_model to be run
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'table_model', 'unique_view_model_id'}
+
+        results = self.run_dbt(['ls', '--select', 'result:warn+', '--state', './state'])
+        assert len(results) == 2
+        assert set(results) == {'test.table_model', 'test.unique_view_model_id'}
+
+    @use_profile('postgres')
+    def test_postgres_run_run_results_state(self):
+        results = self.run_dbt(['run', '--select', 'result:success', '--state', './state'], expect_pass=True)
+        assert len(results) == 2
+        assert results[0].node.name == 'view_model'
+        assert results[1].node.name == 'table_model'
+        
+        # clear state and rerun upstream view model to test + operator
+        shutil.rmtree('./state')
+        self.run_dbt(['run', '--select', 'view_model'], expect_pass=True)
+        self.copy_state()
+        results = self.run_dbt(['run', '--select', 'result:success+', '--state', './state'], expect_pass=True)
+        assert len(results) == 2
+        assert results[0].node.name == 'view_model'
+        assert results[1].node.name == 'table_model'
+
+        # check we are starting from a place with 0 errors
+        results = self.run_dbt(['run', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 0
+        
+        # force an error in the view model to test error and skipped states
+        with open('models/view_model.sql') as fp:
+            fp.readline()
+            newline = fp.newlines
+
+        with open('models/view_model.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_error")
+            fp.write(newline)
+        
+        shutil.rmtree('./state')
+        self.run_dbt(['run'], expect_pass=False)
+        self.copy_state()
+
+        # test single result selector on error
+        results = self.run_dbt(['run', '--select', 'result:error', '--state', './state'], expect_pass=False)
+        assert len(results) == 1
+        assert results[0].node.name == 'view_model'
+        
+        # test + operator selection on error
+        results = self.run_dbt(['run', '--select', 'result:error+', '--state', './state'], expect_pass=False)
+        assert len(results) == 2
+        assert results[0].node.name == 'view_model'
+        assert results[1].node.name == 'table_model'
+
+        # single result selector on skipped. Expect this to pass becase underlying view already defined above
+        results = self.run_dbt(['run', '--select', 'result:skipped', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0].node.name == 'table_model'
+
+        # add a downstream model that depends on table_model for skipped+ selector
+        with open('models/table_model_downstream.sql', 'w') as fp:
+            fp.write("select * from {{ref('table_model')}}")
+        
+        shutil.rmtree('./state')
+        self.run_dbt(['run'], expect_pass=False)
+        self.copy_state()
+
+        results = self.run_dbt(['run', '--select', 'result:skipped+', '--state', './state'], expect_pass=True)
+        assert len(results) == 2
+        assert results[0].node.name == 'table_model'
+        assert results[1].node.name == 'table_model_downstream'
+    
+    
+    @use_profile('postgres')
+    def test_postgres_test_run_results_state(self):
+        # run passed nodes
+        results = self.run_dbt(['test', '--select', 'result:pass', '--state', './state'], expect_pass=True)
+        assert len(results) == 2
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'unique_view_model_id', 'not_null_view_model_id'}
+        
+        # run passed nodes with + operator
+        results = self.run_dbt(['test', '--select', 'result:pass+', '--state', './state'], expect_pass=True)
+        assert len(results) == 2
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'unique_view_model_id', 'not_null_view_model_id'}
+
+        # update view model to generate a failure case
+        os.remove('./models/view_model.sql')
+        with open('models/view_model.sql', 'w') as fp:
+            fp.write("select 1 as id union all select 1 as id")
+        
+        self.rebuild_run_dbt(expect_pass=False)
+
+        # test with failure selector
+        results = self.run_dbt(['test', '--select', 'result:fail', '--state', './state'], expect_pass=False)
+        assert len(results) == 1
+        assert results[0].node.name == 'unique_view_model_id'
+
+        # test with failure selector and + operator
+        results = self.run_dbt(['test', '--select', 'result:fail+', '--state', './state'], expect_pass=False)
+        assert len(results) == 1
+        assert results[0].node.name == 'unique_view_model_id'
+
+        # change the unique test severity from error to warn and reuse the same view_model.sql changes above
+        with open('models/schema.yml', 'r+') as f:
+            filedata = f.read()
+            newdata = filedata.replace('error','warn')
+            f.seek(0)
+            f.write(newdata)
+            f.truncate()
+        
+        # rebuild - expect_pass = True because we changed the error to a warning this time around
+        self.rebuild_run_dbt(expect_pass=True)
+
+        # test with warn selector
+        results = self.run_dbt(['test', '--select', 'result:warn', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0].node.name == 'unique_view_model_id'
+
+        # test with warn selector and + operator
+        results = self.run_dbt(['test', '--select', 'result:warn+', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0].node.name == 'unique_view_model_id'
+
+
+    @use_profile('postgres')
+    def test_postgres_concurrent_selectors_run_run_results_state(self):
+        results = self.run_dbt(['run', '--select', 'state:modified+', 'result:error+', '--state', './state'])
+        assert len(results) == 0
+
+        # force an error on a dbt model
+        with open('models/view_model.sql') as fp:
+            fp.readline()
+            newline = fp.newlines
+
+        with open('models/view_model.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_error")
+            fp.write(newline)
+        
+        shutil.rmtree('./state')
+        self.run_dbt(['run'], expect_pass=False)
+        self.copy_state()
+
+        # modify another dbt model
+        with open('models/table_model_modified_example.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_error")
+            fp.write(newline)
+        
+        results = self.run_dbt(['run', '--select', 'state:modified+', 'result:error+', '--state', './state'], expect_pass=False)
+        assert len(results) == 3
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'view_model', 'table_model_modified_example', 'table_model'}
+    
+
+    @use_profile('postgres')
+    def test_postgres_concurrent_selectors_test_run_results_state(self):
+        # create failure test case for result:fail selector
+        os.remove('./models/view_model.sql')
+        with open('./models/view_model.sql', 'w') as f:
+            f.write('select 1 as id union all select 1 as id union all select null as id')
+
+        # run dbt build again to trigger test errors
+        self.rebuild_run_dbt(expect_pass=False)
+        
+        # get the failures from 
+        results = self.run_dbt(['test', '--select', 'result:fail', '--exclude', 'not_null_view_model_id', '--state', './state'], expect_pass=False)
+        assert len(results) == 1
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'unique_view_model_id'}
+        
+        
+    @use_profile('postgres')
+    def test_postgres_concurrent_selectors_build_run_results_state(self):
+        results = self.run_dbt(['build', '--select', 'state:modified+', 'result:error+', '--state', './state'])
+        assert len(results) == 0
+
+        # force an error on a dbt model
+        with open('models/view_model.sql') as fp:
+            fp.readline()
+            newline = fp.newlines
+
+        with open('models/view_model.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_error")
+            fp.write(newline)
+        
+        self.rebuild_run_dbt(expect_pass=False)
+
+        # modify another dbt model
+        with open('models/table_model_modified_example.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_error")
+            fp.write(newline)
+        
+        results = self.run_dbt(['build', '--select', 'state:modified+', 'result:error+', '--state', './state'], expect_pass=False)
+        assert len(results) == 5
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'table_model_modified_example', 'view_model', 'table_model', 'not_null_view_model_id', 'unique_view_model_id'}
+        
+        # create failure test case for result:fail selector
+        os.remove('./models/view_model.sql')
+        with open('./models/view_model.sql', 'w') as f:
+            f.write('select 1 as id union all select 1 as id')
+
+        # create error model case for result:error selector
+        with open('./models/error_model.sql', 'w') as f:
+            f.write('select 1 as id from not_exists')
+        
+        # create something downstream from the error model to rerun
+        with open('./models/downstream_of_error_model.sql', 'w') as f:
+            f.write('select * from {{ ref("error_model") }} )')
+        
+        # regenerate build state
+        self.rebuild_run_dbt(expect_pass=False)
+
+        # modify model again to trigger the state:modified selector 
+        with open('models/table_model_modified_example.sql', 'w') as fp:
+            fp.write(newline)
+            fp.write("select * from forced_another_error")
+            fp.write(newline)
+        
+        results = self.run_dbt(['build', '--select', 'state:modified+', 'result:error+', 'result:fail+', '--state', './state'], expect_pass=False)
+        assert len(results) == 5
+        nodes = set([elem.node.name for elem in results])
+        assert nodes == {'error_model', 'downstream_of_error_model', 'table_model_modified_example', 'table_model', 'unique_view_model_id'}


Code review comments:
diff_hunk: @@ -0,0 +1,436 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import os
+import random
+import shutil
+import string
+
+import pytest
+
+from dbt.exceptions import CompilationException
+
+
+class TestRunResultsState(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_results_state_062"
+
+    @property
+    def models(self):
+        return "models"
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['macros'],
+            'seeds': {
+                'test': {
+                    'quote_columns': True,
+                }
+            }
+        }
+
+    def _symlink_test_folders(self):
+        # dbt's normal symlink behavior breaks this test. Copy the files
+        # so we can freely modify them.
+        for entry in os.listdir(self.test_original_source_path):
+            src = os.path.join(self.test_original_source_path, entry)
+            tst = os.path.join(self.test_root_dir, entry)
+            if entry in {'models', 'data', 'macros'}:
path: test/integration/062_defer_state_test/test_run_results_state.py
text: @user1:
Just need a quick fix here to get the integration tests passing:
```suggestion
            if entry in {'models', 'seeds', 'macros'}:
```
This is a side-effect of the renaming in rename (`data/` &rarr; `seeds/`) in https://github.com/dbt-labs/dbt/pull/4008. It just requires some quick renaming here and below.

Alternatively, you can set `seeds-path: data` up above in `project_config`.

diff_hunk: @@ -0,0 +1,436 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import os
+import random
+import shutil
+import string
+
+import pytest
+
+from dbt.exceptions import CompilationException
+
+
+class TestRunResultsState(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_results_state_062"
+
+    @property
+    def models(self):
+        return "models"
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['macros'],
+            'seeds': {
+                'test': {
+                    'quote_columns': True,
+                }
+            }
+        }
+
+    def _symlink_test_folders(self):
+        # dbt's normal symlink behavior breaks this test. Copy the files
+        # so we can freely modify them.
+        for entry in os.listdir(self.test_original_source_path):
+            src = os.path.join(self.test_original_source_path, entry)
+            tst = os.path.join(self.test_root_dir, entry)
+            if entry in {'models', 'data', 'macros'}:
+                shutil.copytree(src, tst)
+            elif os.path.isdir(entry) or entry.endswith('.sql'):
+                os.symlink(src, tst)
+
+    def copy_state(self):
+        assert not os.path.exists('state')
+        os.makedirs('state')
+        shutil.copyfile('target/manifest.json', 'state/manifest.json')
+        shutil.copyfile('target/run_results.json', 'state/run_results.json')
+
+    def setUp(self):
+        super().setUp()
+        self.run_dbt(['build'])
+        self.copy_state()
+    
+    def rebuild_run_dbt(self, expect_pass=True):
+        shutil.rmtree('./state')
+        self.run_dbt(['build'], expect_pass=expect_pass)
+        self.copy_state()
+
+    @use_profile('postgres')
+    def test_postgres_seed_run_results_state(self):
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'])
+        self.copy_state()
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:success', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+        with open('data/seed.csv') as fp:
+            fp.readline()
+            newline = fp.newlines
+        with open('data/seed.csv', 'a') as fp:
path: test/integration/062_defer_state_test/test_run_results_state.py
text: @user1:
```suggestion
        with open('seeds/seed.csv', 'a') as fp:
```

diff_hunk: @@ -0,0 +1,436 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import os
+import random
+import shutil
+import string
+
+import pytest
+
+from dbt.exceptions import CompilationException
+
+
+class TestRunResultsState(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_results_state_062"
+
+    @property
+    def models(self):
+        return "models"
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['macros'],
+            'seeds': {
+                'test': {
+                    'quote_columns': True,
+                }
+            }
+        }
+
+    def _symlink_test_folders(self):
+        # dbt's normal symlink behavior breaks this test. Copy the files
+        # so we can freely modify them.
+        for entry in os.listdir(self.test_original_source_path):
+            src = os.path.join(self.test_original_source_path, entry)
+            tst = os.path.join(self.test_root_dir, entry)
+            if entry in {'models', 'data', 'macros'}:
+                shutil.copytree(src, tst)
+            elif os.path.isdir(entry) or entry.endswith('.sql'):
+                os.symlink(src, tst)
+
+    def copy_state(self):
+        assert not os.path.exists('state')
+        os.makedirs('state')
+        shutil.copyfile('target/manifest.json', 'state/manifest.json')
+        shutil.copyfile('target/run_results.json', 'state/run_results.json')
+
+    def setUp(self):
+        super().setUp()
+        self.run_dbt(['build'])
+        self.copy_state()
+    
+    def rebuild_run_dbt(self, expect_pass=True):
+        shutil.rmtree('./state')
+        self.run_dbt(['build'], expect_pass=expect_pass)
+        self.copy_state()
+
+    @use_profile('postgres')
+    def test_postgres_seed_run_results_state(self):
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'])
+        self.copy_state()
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:success', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+        with open('data/seed.csv') as fp:
+            fp.readline()
+            newline = fp.newlines
+        with open('data/seed.csv', 'a') as fp:
+            fp.write(f'\"\'\'3,carl{newline}')
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'], expect_pass=False)
+        self.copy_state()
+
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:error', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+
+        with open('data/seed.csv') as fp:
path: test/integration/062_defer_state_test/test_run_results_state.py
text: @user1:
```suggestion
        with open('seeds/seed.csv') as fp:
```

diff_hunk: @@ -0,0 +1,436 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import os
+import random
+import shutil
+import string
+
+import pytest
+
+from dbt.exceptions import CompilationException
+
+
+class TestRunResultsState(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_results_state_062"
+
+    @property
+    def models(self):
+        return "models"
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['macros'],
+            'seeds': {
+                'test': {
+                    'quote_columns': True,
+                }
+            }
+        }
+
+    def _symlink_test_folders(self):
+        # dbt's normal symlink behavior breaks this test. Copy the files
+        # so we can freely modify them.
+        for entry in os.listdir(self.test_original_source_path):
+            src = os.path.join(self.test_original_source_path, entry)
+            tst = os.path.join(self.test_root_dir, entry)
+            if entry in {'models', 'data', 'macros'}:
+                shutil.copytree(src, tst)
+            elif os.path.isdir(entry) or entry.endswith('.sql'):
+                os.symlink(src, tst)
+
+    def copy_state(self):
+        assert not os.path.exists('state')
+        os.makedirs('state')
+        shutil.copyfile('target/manifest.json', 'state/manifest.json')
+        shutil.copyfile('target/run_results.json', 'state/run_results.json')
+
+    def setUp(self):
+        super().setUp()
+        self.run_dbt(['build'])
+        self.copy_state()
+    
+    def rebuild_run_dbt(self, expect_pass=True):
+        shutil.rmtree('./state')
+        self.run_dbt(['build'], expect_pass=expect_pass)
+        self.copy_state()
+
+    @use_profile('postgres')
+    def test_postgres_seed_run_results_state(self):
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'])
+        self.copy_state()
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:success', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+        with open('data/seed.csv') as fp:
+            fp.readline()
+            newline = fp.newlines
+        with open('data/seed.csv', 'a') as fp:
+            fp.write(f'\"\'\'3,carl{newline}')
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'], expect_pass=False)
+        self.copy_state()
+
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:error', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:error+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+
+        with open('data/seed.csv') as fp:
+            fp.readline()
+            newline = fp.newlines
+        with open('data/seed.csv', 'a') as fp:
path: test/integration/062_defer_state_test/test_run_results_state.py
text: @user1:
```suggestion
        with open('seeds/seed.csv', 'a') as fp:
```

diff_hunk: @@ -0,0 +1,436 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import os
+import random
+import shutil
+import string
+
+import pytest
+
+from dbt.exceptions import CompilationException
+
+
+class TestRunResultsState(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_results_state_062"
+
+    @property
+    def models(self):
+        return "models"
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['macros'],
+            'seeds': {
+                'test': {
+                    'quote_columns': True,
+                }
+            }
+        }
+
+    def _symlink_test_folders(self):
+        # dbt's normal symlink behavior breaks this test. Copy the files
+        # so we can freely modify them.
+        for entry in os.listdir(self.test_original_source_path):
+            src = os.path.join(self.test_original_source_path, entry)
+            tst = os.path.join(self.test_root_dir, entry)
+            if entry in {'models', 'data', 'macros'}:
+                shutil.copytree(src, tst)
+            elif os.path.isdir(entry) or entry.endswith('.sql'):
+                os.symlink(src, tst)
+
+    def copy_state(self):
+        assert not os.path.exists('state')
+        os.makedirs('state')
+        shutil.copyfile('target/manifest.json', 'state/manifest.json')
+        shutil.copyfile('target/run_results.json', 'state/run_results.json')
+
+    def setUp(self):
+        super().setUp()
+        self.run_dbt(['build'])
+        self.copy_state()
+    
+    def rebuild_run_dbt(self, expect_pass=True):
+        shutil.rmtree('./state')
+        self.run_dbt(['build'], expect_pass=expect_pass)
+        self.copy_state()
+
+    @use_profile('postgres')
+    def test_postgres_seed_run_results_state(self):
+        shutil.rmtree('./state')
+        self.run_dbt(['seed'])
+        self.copy_state()
+        results = self.run_dbt(['ls', '--resource-type', 'seed', '--select', 'result:success', '--state', './state'], expect_pass=True)
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success', '--state', './state'])
+        assert len(results) == 1
+        assert results[0] == 'test.seed'
+
+        results = self.run_dbt(['ls', '--select', 'result:success+', '--state', './state'])
+        assert len(results) == 7
+        assert set(results) == {'test.seed', 'test.table_model', 'test.view_model', 'test.ephemeral_model', 'test.not_null_view_model_id', 'test.unique_view_model_id', 'exposure:test.my_exposure'}
+
+        with open('data/seed.csv') as fp:
path: test/integration/062_defer_state_test/test_run_results_state.py
text: @user1:
```suggestion
        with open('seeds/seed.csv') as fp:
```