Add result: selection method

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 @@ class PreviousState:
             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

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):
@@ -504,6 +505,23 @@ class StateSelectorMethod(SelectorMethod):
                 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,
@@ -516,6 +534,7 @@ class MethodManager:
         MethodName.TestType: TestTypeSelectorMethod,
         MethodName.State: StateSelectorMethod,
         MethodName.Exposure: ExposureSelectorMethod,
+        MethodName.Result: ResultSelectorMethod,
     }
 
     def __init__(