From 32c559838d692c572a6a33446c5d71b63b88d257 Mon Sep 17 00:00:00 2001
From: Jacob Beck <jake@fishtownanalytics.com>
Date: Tue, 23 Jun 2020 10:14:24 -0600
Subject: [PATCH] Try to make imports a little more sane, ordering-wise

consolidate dbt.ui, move non-rpc node_runners into their tasks
move parse_cli_vars into config.utils
get rid of logger/exceptions requirements in dbt.utils
---
 core/dbt/adapters/base/connections.py         |   8 +-
 core/dbt/adapters/base/impl.py                |  14 +-
 core/dbt/adapters/sql/connections.py          |   7 +-
 core/dbt/adapters/sql/impl.py                 |   1 -
 core/dbt/clients/jinja.py                     |   6 +-
 core/dbt/compilation.py                       |   6 +-
 core/dbt/config/runtime.py                    |   6 +-
 core/dbt/config/utils.py                      |  23 +
 core/dbt/contracts/graph/model_config.py      |   2 +-
 core/dbt/contracts/graph/parsed.py            |   6 +-
 core/dbt/contracts/project.py                 |   6 +-
 core/dbt/deprecations.py                      |   6 +-
 core/dbt/deps/git.py                          |   4 +-
 core/dbt/exceptions.py                        |   8 +-
 core/dbt/logger.py                            |  17 +-
 core/dbt/main.py                              |   6 +-
 core/dbt/node_runners.py                      | 662 ------------------
 core/dbt/parser/base.py                       |   6 +-
 core/dbt/parser/manifest.py                   |  65 +-
 core/dbt/parser/results.py                    |   4 +-
 core/dbt/parser/sources.py                    |   4 +-
 core/dbt/rpc/gc.py                            |   1 -
 core/dbt/rpc/node_runners.py                  |   2 +-
 core/dbt/rpc/task_handler.py                  |   6 +-
 core/dbt/rpc/task_handler_protocol.py         |   1 -
 core/dbt/rpc/task_manager.py                  |   7 +-
 core/dbt/task/base.py                         | 296 +++++++-
 core/dbt/task/compile.py                      |  26 +-
 core/dbt/task/debug.py                        |  18 +-
 core/dbt/task/freshness.py                    | 101 ++-
 core/dbt/task/generate.py                     |  15 +-
 core/dbt/{ui => task}/printer.py              | 144 +---
 core/dbt/task/rpc/cli.py                      |   2 +-
 core/dbt/task/run.py                          | 194 ++++-
 core/dbt/task/run_operation.py                |  13 +-
 core/dbt/task/runnable.py                     |  31 +-
 core/dbt/task/seed.py                         |  35 +-
 core/dbt/task/snapshot.py                     |  16 +-
 core/dbt/task/test.py                         |  91 ++-
 core/dbt/ui.py                                |  76 ++
 core/dbt/ui/__init__.py                       |   0
 core/dbt/ui/colors.py                         |  10 -
 core/dbt/utils.py                             |  75 --
 .../bigquery/dbt/adapters/bigquery/impl.py    |   9 +-
 test/unit/test_main.py                        |   2 +-
 test/unit/utils.py                            |   4 +-
 46 files changed, 1033 insertions(+), 1009 deletions(-)
 create mode 100644 core/dbt/config/utils.py
 delete mode 100644 core/dbt/node_runners.py
 rename core/dbt/{ui => task}/printer.py (71%)
 create mode 100644 core/dbt/ui.py
 delete mode 100644 core/dbt/ui/__init__.py
 delete mode 100644 core/dbt/ui/colors.py

diff --git a/core/dbt/adapters/base/connections.py b/core/dbt/adapters/base/connections.py
index 07088dc0083..e39d3681403 100644
--- a/core/dbt/adapters/base/connections.py
+++ b/core/dbt/adapters/base/connections.py
@@ -10,7 +10,6 @@
 import agate
 
 import dbt.exceptions
-import dbt.flags
 from dbt.contracts.connection import (
     Connection, Identifier, ConnectionState, AdapterRequiredConfig, LazyHandle
 )
@@ -19,6 +18,7 @@
     MacroQueryStringSetter,
 )
 from dbt.logger import GLOBAL_LOGGER as logger
+from dbt import flags
 
 
 class BaseConnectionManager(metaclass=abc.ABCMeta):
@@ -39,7 +39,7 @@ class BaseConnectionManager(metaclass=abc.ABCMeta):
     def __init__(self, profile: AdapterRequiredConfig):
         self.profile = profile
         self.thread_connections: Dict[Hashable, Connection] = {}
-        self.lock: RLock = dbt.flags.MP_CONTEXT.RLock()
+        self.lock: RLock = flags.MP_CONTEXT.RLock()
         self.query_header: Optional[MacroQueryStringSetter] = None
 
     def set_query_header(self, manifest: Manifest) -> None:
@@ -235,7 +235,7 @@ def _close_handle(cls, connection: Connection) -> None:
     @classmethod
     def _rollback(cls, connection: Connection) -> None:
         """Roll back the given connection."""
-        if dbt.flags.STRICT_MODE:
+        if flags.STRICT_MODE:
             if not isinstance(connection, Connection):
                 raise dbt.exceptions.CompilerException(
                     f'In _rollback, got {connection} - not a Connection!'
@@ -253,7 +253,7 @@ def _rollback(cls, connection: Connection) -> None:
 
     @classmethod
     def close(cls, connection: Connection) -> Connection:
-        if dbt.flags.STRICT_MODE:
+        if flags.STRICT_MODE:
             if not isinstance(connection, Connection):
                 raise dbt.exceptions.CompilerException(
                     f'In close, got {connection} - not a Connection!'
diff --git a/core/dbt/adapters/base/impl.py b/core/dbt/adapters/base/impl.py
index 6af38b5264b..ab6064bb90b 100644
--- a/core/dbt/adapters/base/impl.py
+++ b/core/dbt/adapters/base/impl.py
@@ -17,7 +17,7 @@
     get_relation_returned_multiple_results,
     InternalException, NotImplementedException, RuntimeException,
 )
-import dbt.flags
+from dbt import flags
 
 from dbt import deprecations
 from dbt.clients.agate_helper import empty_table, merge_tables, table_from_rows
@@ -267,7 +267,7 @@ def load_internal_manifest(self) -> Manifest:
     def _schema_is_cached(self, database: Optional[str], schema: str) -> bool:
         """Check if the schema is cached, and by default logs if it is not."""
 
-        if dbt.flags.USE_CACHE is False:
+        if flags.USE_CACHE is False:
             return False
         elif (database, schema) not in self.cache:
             logger.debug(
@@ -323,7 +323,7 @@ def _relations_cache_for_schemas(self, manifest: Manifest) -> None:
         """Populate the relations cache for the given schemas. Returns an
         iterable of the schemas populated, as strings.
         """
-        if not dbt.flags.USE_CACHE:
+        if not flags.USE_CACHE:
             return
 
         cache_schemas = self._get_cache_schemas(manifest)
@@ -352,7 +352,7 @@ def set_relations_cache(
         """Run a query that gets a populated cache of the relations in the
         database and set the cache on this adapter.
         """
-        if not dbt.flags.USE_CACHE:
+        if not flags.USE_CACHE:
             return
 
         with self.cache.lock:
@@ -368,7 +368,7 @@ def cache_added(self, relation: Optional[BaseRelation]) -> str:
             raise_compiler_error(
                 'Attempted to cache a null relation for {}'.format(name)
             )
-        if dbt.flags.USE_CACHE:
+        if flags.USE_CACHE:
             self.cache.add(relation)
         # so jinja doesn't render things
         return ''
@@ -383,7 +383,7 @@ def cache_dropped(self, relation: Optional[BaseRelation]) -> str:
             raise_compiler_error(
                 'Attempted to drop a null relation for {}'.format(name)
             )
-        if dbt.flags.USE_CACHE:
+        if flags.USE_CACHE:
             self.cache.drop(relation)
         return ''
 
@@ -405,7 +405,7 @@ def cache_renamed(
                 .format(src_name, dst_name, name)
             )
 
-        if dbt.flags.USE_CACHE:
+        if flags.USE_CACHE:
             self.cache.rename(from_relation, to_relation)
         return ''
 
diff --git a/core/dbt/adapters/sql/connections.py b/core/dbt/adapters/sql/connections.py
index 3fe1dad2ec7..91ebe4e0d85 100644
--- a/core/dbt/adapters/sql/connections.py
+++ b/core/dbt/adapters/sql/connections.py
@@ -6,9 +6,10 @@
 
 import dbt.clients.agate_helper
 import dbt.exceptions
-from dbt.contracts.connection import Connection
 from dbt.adapters.base import BaseConnectionManager
+from dbt.contracts.connection import Connection
 from dbt.logger import GLOBAL_LOGGER as logger
+from dbt import flags
 
 
 class SQLConnectionManager(BaseConnectionManager):
@@ -133,7 +134,7 @@ def add_commit_query(self):
     def begin(self):
         connection = self.get_thread_connection()
 
-        if dbt.flags.STRICT_MODE:
+        if flags.STRICT_MODE:
             if not isinstance(connection, Connection):
                 raise dbt.exceptions.CompilerException(
                     f'In begin, got {connection} - not a Connection!'
@@ -151,7 +152,7 @@ def begin(self):
 
     def commit(self):
         connection = self.get_thread_connection()
-        if dbt.flags.STRICT_MODE:
+        if flags.STRICT_MODE:
             if not isinstance(connection, Connection):
                 raise dbt.exceptions.CompilerException(
                     f'In commit, got {connection} - not a Connection!'
diff --git a/core/dbt/adapters/sql/impl.py b/core/dbt/adapters/sql/impl.py
index 84268b34d16..93b7aae0906 100644
--- a/core/dbt/adapters/sql/impl.py
+++ b/core/dbt/adapters/sql/impl.py
@@ -4,7 +4,6 @@
 import dbt.clients.agate_helper
 from dbt.contracts.connection import Connection
 import dbt.exceptions
-import dbt.flags
 from dbt.adapters.base import BaseAdapter, available
 from dbt.adapters.sql import SQLConnectionManager
 from dbt.logger import GLOBAL_LOGGER as logger
diff --git a/core/dbt/clients/jinja.py b/core/dbt/clients/jinja.py
index 6138e8902c8..07a344338b7 100644
--- a/core/dbt/clients/jinja.py
+++ b/core/dbt/clients/jinja.py
@@ -30,7 +30,7 @@
     InternalException, raise_compiler_error, CompilationException,
     invalid_materialization_argument, MacroReturn
 )
-from dbt.flags import MACRO_DEBUGGING
+from dbt import flags
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
 
 
@@ -93,8 +93,8 @@ def _compile(self, source, filename):
         If the value is 'write', also write the files to disk.
         WARNING: This can write a ton of data if you aren't careful.
         """
-        if filename == '<template>' and MACRO_DEBUGGING:
-            write = MACRO_DEBUGGING == 'write'
+        if filename == '<template>' and flags.MACRO_DEBUGGING:
+            write = flags.MACRO_DEBUGGING == 'write'
             filename = _linecache_inject(source, write)
 
         return super()._compile(source, filename)  # type: ignore
diff --git a/core/dbt/compilation.py b/core/dbt/compilation.py
index e02a21f2b18..3383f90299f 100644
--- a/core/dbt/compilation.py
+++ b/core/dbt/compilation.py
@@ -6,6 +6,7 @@
 import dbt.include
 import dbt.tracking
 
+from dbt import flags
 from dbt.node_types import NodeType
 from dbt.linker import Linker
 
@@ -13,7 +14,6 @@
 from dbt.contracts.graph.compiled import NonSourceNode
 from dbt.contracts.graph.manifest import Manifest
 import dbt.exceptions
-import dbt.flags
 import dbt.config
 from dbt.contracts.graph.compiled import (
     InjectedCTE,
@@ -104,7 +104,7 @@ def recursively_prepend_ctes(model, manifest):
     if model.extra_ctes_injected:
         return (model, model.extra_ctes, manifest)
 
-    if dbt.flags.STRICT_MODE:
+    if flags.STRICT_MODE:
         if not isinstance(model, tuple(COMPILED_TYPES.values())):
             raise dbt.exceptions.InternalException(
                 'Bad model type: {}'.format(type(model))
@@ -187,7 +187,7 @@ def compile_node(self, node, manifest, extra_context=None):
     def write_graph_file(self, linker: Linker, manifest: Manifest):
         filename = graph_file_name
         graph_path = os.path.join(self.config.target_path, filename)
-        if dbt.flags.WRITE_JSON:
+        if flags.WRITE_JSON:
             linker.write_graph(graph_path, manifest)
 
     def link_node(
diff --git a/core/dbt/config/runtime.py b/core/dbt/config/runtime.py
index bad39fa8b6c..342169e74ab 100644
--- a/core/dbt/config/runtime.py
+++ b/core/dbt/config/runtime.py
@@ -11,6 +11,7 @@
 from .profile import Profile
 from .project import Project
 from .renderer import DbtProjectYamlRenderer, ProfileRenderer
+from .utils import parse_cli_vars
 from dbt import tracking
 from dbt.adapters.factory import get_relation_class_by_name, get_include_paths
 from dbt.helper_types import FQNPath, PathSet
@@ -19,8 +20,7 @@
 from dbt.contracts.connection import AdapterRequiredConfig, Credentials
 from dbt.contracts.graph.manifest import ManifestMetadata
 from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.ui import printer
-from dbt.utils import parse_cli_vars
+from dbt.ui import warning_tag
 
 from dbt.contracts.project import Configuration, UserConfig
 from dbt.exceptions import (
@@ -317,7 +317,7 @@ def warn_for_unused_resource_config_paths(
             '\n'.join('- {}'.format('.'.join(u)) for u in unused)
         )
 
-        warn_or_error(msg, log_fmt=printer.warning_tag('{}'))
+        warn_or_error(msg, log_fmt=warning_tag('{}'))
 
     def load_dependencies(self) -> Mapping[str, 'RuntimeConfig']:
         if self.dependencies is None:
diff --git a/core/dbt/config/utils.py b/core/dbt/config/utils.py
new file mode 100644
index 00000000000..a0f5832a396
--- /dev/null
+++ b/core/dbt/config/utils.py
@@ -0,0 +1,23 @@
+from typing import Dict, Any
+
+from dbt.clients import yaml_helper
+from dbt.exceptions import raise_compiler_error, ValidationException
+from dbt.logger import GLOBAL_LOGGER as logger
+
+
+def parse_cli_vars(var_string: str) -> Dict[str, Any]:
+    try:
+        cli_vars = yaml_helper.load_yaml_text(var_string)
+        var_type = type(cli_vars)
+        if var_type is dict:
+            return cli_vars
+        else:
+            type_name = var_type.__name__
+            raise_compiler_error(
+                "The --vars argument must be a YAML dictionary, but was "
+                "of type '{}'".format(type_name))
+    except ValidationException:
+        logger.error(
+            "The YAML provided in the --vars argument is not valid.\n"
+        )
+        raise
diff --git a/core/dbt/contracts/graph/model_config.py b/core/dbt/contracts/graph/model_config.py
index 3375f2a02f5..dfd5555b37c 100644
--- a/core/dbt/contracts/graph/model_config.py
+++ b/core/dbt/contracts/graph/model_config.py
@@ -14,10 +14,10 @@
 from hologram import JsonSchemaMixin, ValidationError
 from hologram.helpers import StrEnum, register_pattern
 
-from dbt import hooks
 from dbt.contracts.graph.unparsed import AdditionalPropertiesAllowed
 from dbt.exceptions import CompilationException, InternalException
 from dbt.contracts.util import Replaceable, list_str
+from dbt import hooks
 from dbt.node_types import NodeType
 
 
diff --git a/core/dbt/contracts/graph/parsed.py b/core/dbt/contracts/graph/parsed.py
index d988da24ff9..a8387f93aa6 100644
--- a/core/dbt/contracts/graph/parsed.py
+++ b/core/dbt/contracts/graph/parsed.py
@@ -15,7 +15,6 @@
 from hologram import JsonSchemaMixin
 
 from dbt.clients.system import write_file
-import dbt.flags
 from dbt.contracts.graph.unparsed import (
     UnparsedNode, UnparsedDocumentation, Quoting, Docs,
     UnparsedBaseNode, FreshnessThreshold, ExternalTable,
@@ -24,6 +23,7 @@
 )
 from dbt.contracts.util import Replaceable
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt import flags
 from dbt.node_types import NodeType
 
 
@@ -117,7 +117,7 @@ def patch(self, patch: 'ParsedNodePatch'):
         self.columns = patch.columns
         self.meta = patch.meta
         self.docs = patch.docs
-        if dbt.flags.STRICT_MODE:
+        if flags.STRICT_MODE:
             assert isinstance(self, JsonSchemaMixin)
             self.to_dict(validate=True)
 
@@ -300,7 +300,7 @@ def patch(self, patch: ParsedMacroPatch):
         self.meta = patch.meta
         self.docs = patch.docs
         self.arguments = patch.arguments
-        if dbt.flags.STRICT_MODE:
+        if flags.STRICT_MODE:
             assert isinstance(self, JsonSchemaMixin)
             self.to_dict(validate=True)
 
diff --git a/core/dbt/contracts/project.py b/core/dbt/contracts/project.py
index 884d82d4577..b5a2a14f61e 100644
--- a/core/dbt/contracts/project.py
+++ b/core/dbt/contracts/project.py
@@ -3,7 +3,7 @@
 from dbt.helper_types import NoValue
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
 from dbt import tracking
-from dbt.ui import printer
+from dbt import ui
 
 from hologram import JsonSchemaMixin, ValidationError
 from hologram.helpers import HyphenatedJsonSchemaMixin, register_pattern, \
@@ -268,10 +268,10 @@ def set_values(self, cookie_dir):
             tracking.do_not_track()
 
         if self.use_colors:
-            printer.use_colors()
+            ui.use_colors()
 
         if self.printer_width:
-            printer.printer_width(self.printer_width)
+            ui.printer_width(self.printer_width)
 
 
 @dataclass
diff --git a/core/dbt/deprecations.py b/core/dbt/deprecations.py
index 64a46358a8f..63e4239b27b 100644
--- a/core/dbt/deprecations.py
+++ b/core/dbt/deprecations.py
@@ -1,9 +1,7 @@
 from typing import Optional, Set, List, Dict, ClassVar
 
-import dbt.links
 import dbt.exceptions
-import dbt.flags
-from dbt.ui import printer
+from dbt import ui
 
 
 class DBTDeprecation:
@@ -29,7 +27,7 @@ def description(self) -> str:
     def show(self, *args, **kwargs) -> None:
         if self.name not in active_deprecations:
             desc = self.description.format(**kwargs)
-            msg = printer.line_wrap_message(
+            msg = ui.line_wrap_message(
                 desc, prefix='* Deprecation Warning: '
             )
             dbt.exceptions.warn_or_error(msg)
diff --git a/core/dbt/deps/git.py b/core/dbt/deps/git.py
index 8d77775f654..b8018e1bb07 100644
--- a/core/dbt/deps/git.py
+++ b/core/dbt/deps/git.py
@@ -13,7 +13,7 @@
     ExecutableError, warn_or_error, raise_dependency_error
 )
 from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.ui import printer
+from dbt import ui
 
 PIN_PACKAGE_URL = 'https://docs.getdbt.com/docs/package-management#section-specifying-package-versions' # noqa
 
@@ -77,7 +77,7 @@ def _fetch_metadata(self, project, renderer) -> ProjectPackageMetadata:
                 'The git package "{}" is not pinned.\n\tThis can introduce '
                 'breaking changes into your project without warning!\n\nSee {}'
                 .format(self.git, PIN_PACKAGE_URL),
-                log_fmt=printer.yellow('WARNING: {}')
+                log_fmt=ui.yellow('WARNING: {}')
             )
         loaded = Project.from_project_root(path, renderer)
         return ProjectPackageMetadata.from_project(loaded)
diff --git a/core/dbt/exceptions.py b/core/dbt/exceptions.py
index f5d539c8f00..729c29e0c28 100644
--- a/core/dbt/exceptions.py
+++ b/core/dbt/exceptions.py
@@ -4,7 +4,8 @@
 
 from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.node_types import NodeType
-import dbt.flags
+from dbt import flags
+from dbt.ui import line_wrap_message
 
 import hologram
 
@@ -867,7 +868,6 @@ def raise_unrecognized_credentials_type(typename, supported_types):
 def raise_invalid_patch(
     node, patch_section: str, patch_path: str,
 ) -> NoReturn:
-    from dbt.ui.printer import line_wrap_message
     msg = line_wrap_message(
         f'''\
         '{node.name}' is a {node.resource_type} node, but it is
@@ -904,7 +904,7 @@ def raise_duplicate_alias(
 
 
 def warn_or_error(msg, node=None, log_fmt=None):
-    if dbt.flags.WARN_ERROR:
+    if flags.WARN_ERROR:
         raise_compiler_error(msg, node)
     else:
         if log_fmt is not None:
@@ -913,7 +913,7 @@ def warn_or_error(msg, node=None, log_fmt=None):
 
 
 def warn_or_raise(exc, log_fmt=None):
-    if dbt.flags.WARN_ERROR:
+    if flags.WARN_ERROR:
         raise exc
     else:
         msg = str(exc)
diff --git a/core/dbt/logger.py b/core/dbt/logger.py
index a209dbad31d..090f672485c 100644
--- a/core/dbt/logger.py
+++ b/core/dbt/logger.py
@@ -1,10 +1,10 @@
-import dbt.flags
-import dbt.ui.colors
+import dbt.ui
 
 import json
 import logging
 import os
 import sys
+import time
 import warnings
 from dataclasses import dataclass
 from datetime import datetime
@@ -440,7 +440,7 @@ def _replay_buffered(self):
     def format(self, record: logbook.LogRecord) -> str:
         msg = super().format(record)
         subbed = str(msg)
-        for escape_sequence in dbt.ui.colors.COLORS.values():
+        for escape_sequence in dbt.ui.COLORS.values():
             subbed = subbed.replace(escape_sequence, '')
         return subbed
 
@@ -615,3 +615,14 @@ def list_handler(
     """Return a context manager that temporarly attaches a list to the logger.
     """
     return ListLogHandler(lst=lst, level=level, bubble=True)
+
+
+def get_timestamp():
+    return time.strftime("%H:%M:%S")
+
+
+def print_timestamped_line(msg: str, use_color: Optional[str] = None):
+    if use_color is not None:
+        msg = dbt.ui.color(msg, use_color)
+
+    GLOBAL_LOGGER.info("{} | {}".format(get_timestamp(), msg))
diff --git a/core/dbt/main.py b/core/dbt/main.py
index 3c6d99883f6..cde94c77c71 100644
--- a/core/dbt/main.py
+++ b/core/dbt/main.py
@@ -22,14 +22,12 @@
 import dbt.task.serve as serve_task
 import dbt.task.freshness as freshness_task
 import dbt.task.run_operation as run_operation_task
+from dbt.profiler import profiler
 from dbt.task.list import ListTask
 from dbt.task.rpc.server import RPCServerTask
 from dbt.adapters.factory import reset_adapters, cleanup_connections
 
 import dbt.tracking
-import dbt.ui.printer
-import dbt.deprecations
-import dbt.profiler
 
 from dbt.utils import ExitCodes
 from dbt.config import PROFILES_DIR, read_user_config
@@ -147,7 +145,7 @@ def handle_and_check(args):
         if parsed.record_timing_info:
             profiler_enabled = True
 
-        with dbt.profiler.profiler(
+        with profiler(
             enable=profiler_enabled,
             outfile=parsed.record_timing_info
         ):
diff --git a/core/dbt/node_runners.py b/core/dbt/node_runners.py
deleted file mode 100644
index e1f84836bd3..00000000000
--- a/core/dbt/node_runners.py
+++ /dev/null
@@ -1,662 +0,0 @@
-import abc
-import threading
-import time
-import traceback
-from typing import List, Dict, Any, Optional
-
-from dbt import deprecations
-from dbt.adapters.base import BaseRelation
-from dbt.clients.jinja import MacroGenerator
-from dbt.compilation import compile_node
-from dbt.context.providers import generate_runtime_model
-from dbt.contracts.graph.compiled import (
-    CompiledDataTestNode,
-    CompiledSchemaTestNode,
-    CompiledTestNode,
-)
-from dbt.contracts.graph.manifest import Manifest
-from dbt.contracts.results import (
-    RunModelResult, collect_timing_info, SourceFreshnessResult, PartialResult,
-)
-from dbt.exceptions import (
-    NotImplementedException, CompilationException, RuntimeException,
-    InternalException, missing_materialization, raise_compiler_error
-)
-from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.node_types import NodeType
-import dbt.tracking
-import dbt.ui.printer
-import dbt.flags
-import dbt.utils
-
-
-INTERNAL_ERROR_STRING = """This is an error in dbt. Please try again. If \
-the error persists, open an issue at https://github.com/fishtown-analytics/dbt
-""".strip()
-
-
-def track_model_run(index, num_nodes, run_model_result):
-    if dbt.tracking.active_user is None:
-        raise InternalException('cannot track model run with no active user')
-    invocation_id = dbt.tracking.active_user.invocation_id
-    dbt.tracking.track_model_run({
-        "invocation_id": invocation_id,
-        "index": index,
-        "total": num_nodes,
-        "execution_time": run_model_result.execution_time,
-        "run_status": run_model_result.status,
-        "run_skipped": run_model_result.skip,
-        "run_error": None,
-        "model_materialization": run_model_result.node.get_materialization(),
-        "model_id": dbt.utils.get_hash(run_model_result.node),
-        "hashed_contents": dbt.utils.get_hashed_contents(
-            run_model_result.node
-        ),
-        "timing": [t.to_dict() for t in run_model_result.timing],
-    })
-
-
-class ExecutionContext:
-    """During execution and error handling, dbt makes use of mutable state:
-    timing information and the newest (compiled vs executed) form of the node.
-    """
-    def __init__(self, node):
-        self.timing = []
-        self.node = node
-
-
-class BaseRunner(metaclass=abc.ABCMeta):
-    def __init__(self, config, adapter, node, node_index, num_nodes):
-        self.config = config
-        self.adapter = adapter
-        self.node = node
-        self.node_index = node_index
-        self.num_nodes = num_nodes
-
-        self.skip = False
-        self.skip_cause: Optional[RunModelResult] = None
-
-    @abc.abstractmethod
-    def compile(self, manifest: Manifest) -> Any:
-        pass
-
-    def get_result_status(self, result) -> Dict[str, str]:
-        if result.error:
-            return {'node_status': 'error', 'node_error': str(result.error)}
-        elif result.skip:
-            return {'node_status': 'skipped'}
-        elif result.fail:
-            return {'node_status': 'failed'}
-        elif result.warn:
-            return {'node_status': 'warn'}
-        else:
-            return {'node_status': 'passed'}
-
-    def run_with_hooks(self, manifest):
-        if self.skip:
-            return self.on_skip()
-
-        # no before/after printing for ephemeral mdoels
-        if not self.node.is_ephemeral_model:
-            self.before_execute()
-
-        result = self.safe_run(manifest)
-
-        if not self.node.is_ephemeral_model:
-            self.after_execute(result)
-
-        return result
-
-    def _build_run_result(self, node, start_time, error, status, timing_info,
-                          skip=False, fail=None, warn=None, agate_table=None):
-        execution_time = time.time() - start_time
-        thread_id = threading.current_thread().name
-        return RunModelResult(
-            node=node,
-            error=error,
-            skip=skip,
-            status=status,
-            fail=fail,
-            warn=warn,
-            execution_time=execution_time,
-            thread_id=thread_id,
-            timing=timing_info,
-            agate_table=agate_table,
-        )
-
-    def error_result(self, node, error, start_time, timing_info):
-        return self._build_run_result(
-            node=node,
-            start_time=start_time,
-            error=error,
-            status='ERROR',
-            timing_info=timing_info
-        )
-
-    def ephemeral_result(self, node, start_time, timing_info):
-        return self._build_run_result(
-            node=node,
-            start_time=start_time,
-            error=None,
-            status=None,
-            timing_info=timing_info
-        )
-
-    def from_run_result(self, result, start_time, timing_info):
-        return self._build_run_result(
-            node=result.node,
-            start_time=start_time,
-            error=result.error,
-            skip=result.skip,
-            status=result.status,
-            fail=result.fail,
-            warn=result.warn,
-            timing_info=timing_info,
-            agate_table=result.agate_table,
-        )
-
-    def compile_and_execute(self, manifest, ctx):
-        result = None
-        with self.adapter.connection_for(self.node):
-            with collect_timing_info('compile') as timing_info:
-                # if we fail here, we still have a compiled node to return
-                # this has the benefit of showing a build path for the errant
-                # model
-                ctx.node = self.compile(manifest)
-            ctx.timing.append(timing_info)
-
-            # for ephemeral nodes, we only want to compile, not run
-            if not ctx.node.is_ephemeral_model:
-                with collect_timing_info('execute') as timing_info:
-                    result = self.run(ctx.node, manifest)
-                    ctx.node = result.node
-
-                ctx.timing.append(timing_info)
-
-        return result
-
-    def _handle_catchable_exception(self, e, ctx):
-        if e.node is None:
-            e.add_node(ctx.node)
-
-        logger.debug(str(e), exc_info=True)
-        return str(e)
-
-    def _handle_internal_exception(self, e, ctx):
-        build_path = self.node.build_path
-        prefix = 'Internal error executing {}'.format(build_path)
-
-        error = "{prefix}\n{error}\n\n{note}".format(
-            prefix=dbt.ui.printer.red(prefix),
-            error=str(e).strip(),
-            note=INTERNAL_ERROR_STRING
-        )
-        logger.debug(error, exc_info=True)
-        return str(e)
-
-    def _handle_generic_exception(self, e, ctx):
-        node_description = self.node.build_path
-        if node_description is None:
-            node_description = self.node.unique_id
-        prefix = "Unhandled error while executing {}".format(node_description)
-        error = "{prefix}\n{error}".format(
-            prefix=dbt.ui.printer.red(prefix),
-            error=str(e).strip()
-        )
-
-        logger.error(error)
-        logger.debug('', exc_info=True)
-        return str(e)
-
-    def handle_exception(self, e, ctx):
-        catchable_errors = (CompilationException, RuntimeException)
-        if isinstance(e, catchable_errors):
-            error = self._handle_catchable_exception(e, ctx)
-        elif isinstance(e, InternalException):
-            error = self._handle_internal_exception(e, ctx)
-        else:
-            error = self._handle_generic_exception(e, ctx)
-        return error
-
-    def safe_run(self, manifest):
-        started = time.time()
-        ctx = ExecutionContext(self.node)
-        error = None
-        result = None
-
-        try:
-            result = self.compile_and_execute(manifest, ctx)
-        except Exception as e:
-            error = self.handle_exception(e, ctx)
-        finally:
-            exc_str = self._safe_release_connection()
-
-            # if releasing failed and the result doesn't have an error yet, set
-            # an error
-            if (
-                exc_str is not None and result is not None and
-                result.error is None and error is None
-            ):
-                error = exc_str
-
-        if error is not None:
-            # we could include compile time for runtime errors here
-            result = self.error_result(ctx.node, error, started, [])
-        elif result is not None:
-            result = self.from_run_result(result, started, ctx.timing)
-        else:
-            result = self.ephemeral_result(ctx.node, started, ctx.timing)
-        return result
-
-    def _safe_release_connection(self):
-        """Try to release a connection. If an exception is hit, log and return
-        the error string.
-        """
-        try:
-            self.adapter.release_connection()
-        except Exception as exc:
-            logger.debug(
-                'Error releasing connection for node {}: {!s}\n{}'
-                .format(self.node.name, exc, traceback.format_exc())
-            )
-            return str(exc)
-
-        return None
-
-    def before_execute(self):
-        raise NotImplementedException()
-
-    def execute(self, compiled_node, manifest):
-        raise NotImplementedException()
-
-    def run(self, compiled_node, manifest):
-        return self.execute(compiled_node, manifest)
-
-    def after_execute(self, result):
-        raise NotImplementedException()
-
-    def _skip_caused_by_ephemeral_failure(self):
-        if self.skip_cause is None or self.skip_cause.node is None:
-            return False
-        return self.skip_cause.node.is_ephemeral_model
-
-    def on_skip(self):
-        schema_name = self.node.schema
-        node_name = self.node.name
-
-        error = None
-        if not self.node.is_ephemeral_model:
-            # if this model was skipped due to an upstream ephemeral model
-            # failure, print a special 'error skip' message.
-            if self._skip_caused_by_ephemeral_failure():
-                dbt.ui.printer.print_skip_caused_by_error(
-                    self.node,
-                    schema_name,
-                    node_name,
-                    self.node_index,
-                    self.num_nodes,
-                    self.skip_cause
-                )
-                if self.skip_cause is None:  # mypy appeasement
-                    raise InternalException(
-                        'Skip cause not set but skip was somehow caused by '
-                        'an ephemeral failure'
-                    )
-                # set an error so dbt will exit with an error code
-                error = (
-                    'Compilation Error in {}, caused by compilation error '
-                    'in referenced ephemeral model {}'
-                    .format(self.node.unique_id,
-                            self.skip_cause.node.unique_id)
-                )
-            else:
-                dbt.ui.printer.print_skip_line(
-                    self.node,
-                    schema_name,
-                    node_name,
-                    self.node_index,
-                    self.num_nodes
-                )
-
-        node_result = RunModelResult(self.node, skip=True, error=error)
-        return node_result
-
-    def do_skip(self, cause=None):
-        self.skip = True
-        self.skip_cause = cause
-
-
-class CompileRunner(BaseRunner):
-    def before_execute(self):
-        pass
-
-    def after_execute(self, result):
-        pass
-
-    def execute(self, compiled_node, manifest):
-        return RunModelResult(compiled_node)
-
-    def compile(self, manifest):
-        return compile_node(self.adapter, self.config, self.node, manifest, {})
-
-
-# make sure that we got an ok result back from a materialization
-def _validate_materialization_relations_dict(
-    inp: Dict[Any, Any], model
-) -> List[BaseRelation]:
-    try:
-        relations_value = inp['relations']
-    except KeyError:
-        msg = (
-            'Invalid return value from materialization, "relations" '
-            'not found, got keys: {}'.format(list(inp))
-        )
-        raise CompilationException(msg, node=model) from None
-
-    if not isinstance(relations_value, list):
-        msg = (
-            'Invalid return value from materialization, "relations" '
-            'not a list, got: {}'.format(relations_value)
-        )
-        raise CompilationException(msg, node=model) from None
-
-    relations: List[BaseRelation] = []
-    for relation in relations_value:
-        if not isinstance(relation, BaseRelation):
-            msg = (
-                'Invalid return value from materialization, '
-                '"relations" contains non-Relation: {}'
-                .format(relation)
-            )
-            raise CompilationException(msg, node=model)
-
-        assert isinstance(relation, BaseRelation)
-        relations.append(relation)
-    return relations
-
-
-class ModelRunner(CompileRunner):
-    def get_node_representation(self):
-        display_quote_policy = {
-            'database': False, 'schema': False, 'identifier': False
-        }
-        relation = self.adapter.Relation.create_from(
-            self.config, self.node, quote_policy=display_quote_policy
-        )
-        # exclude the database from output if it's the default
-        if self.node.database == self.config.credentials.database:
-            relation = relation.include(database=False)
-        return str(relation)
-
-    def describe_node(self):
-        return "{} model {}".format(self.node.get_materialization(),
-                                    self.get_node_representation())
-
-    def print_start_line(self):
-        description = self.describe_node()
-        dbt.ui.printer.print_start_line(description, self.node_index,
-                                        self.num_nodes)
-
-    def print_result_line(self, result):
-        description = self.describe_node()
-        dbt.ui.printer.print_model_result_line(result,
-                                               description,
-                                               self.node_index,
-                                               self.num_nodes)
-
-    def before_execute(self):
-        self.print_start_line()
-
-    def after_execute(self, result):
-        track_model_run(self.node_index, self.num_nodes, result)
-        self.print_result_line(result)
-
-    def _build_run_model_result(self, model, context):
-        result = context['load_result']('main')
-        return RunModelResult(model, status=result.status)
-
-    def _materialization_relations(
-        self, result: Any, model
-    ) -> List[BaseRelation]:
-        if isinstance(result, str):
-            deprecations.warn('materialization-return',
-                              materialization=model.get_materialization())
-            return [
-                self.adapter.Relation.create_from(self.config, model)
-            ]
-
-        if isinstance(result, dict):
-            return _validate_materialization_relations_dict(result, model)
-
-        msg = (
-            'Invalid return value from materialization, expected a dict '
-            'with key "relations", got: {}'.format(str(result))
-        )
-        raise CompilationException(msg, node=model)
-
-    def execute(self, model, manifest):
-        context = generate_runtime_model(
-            model, self.config, manifest
-        )
-
-        materialization_macro = manifest.find_materialization_macro_by_name(
-            self.config.project_name,
-            model.get_materialization(),
-            self.adapter.type())
-
-        if materialization_macro is None:
-            missing_materialization(model, self.adapter.type())
-
-        if 'config' not in context:
-            raise InternalException(
-                'Invalid materialization context generated, missing config: {}'
-                .format(context)
-            )
-        context_config = context['config']
-
-        hook_ctx = self.adapter.pre_model_hook(context_config)
-        try:
-            result = MacroGenerator(materialization_macro, context)()
-        finally:
-            self.adapter.post_model_hook(context_config, hook_ctx)
-
-        for relation in self._materialization_relations(result, model):
-            self.adapter.cache_added(relation.incorporate(dbt_created=True))
-
-        return self._build_run_model_result(model, context)
-
-
-class FreshnessRunner(BaseRunner):
-    def on_skip(self):
-        raise RuntimeException(
-            'Freshness: nodes cannot be skipped!'
-        )
-
-    def get_result_status(self, result) -> Dict[str, str]:
-        if result.error:
-            return {'node_status': 'error', 'node_error': str(result.error)}
-        else:
-            return {'node_status': str(result.status)}
-
-    def before_execute(self):
-        description = 'freshness of {0.source_name}.{0.name}'.format(self.node)
-        dbt.ui.printer.print_start_line(description, self.node_index,
-                                        self.num_nodes)
-
-    def after_execute(self, result):
-        dbt.ui.printer.print_freshness_result_line(result,
-                                                   self.node_index,
-                                                   self.num_nodes)
-
-    def _build_run_result(self, node, start_time, error, status, timing_info,
-                          skip=False, failed=None):
-        execution_time = time.time() - start_time
-        thread_id = threading.current_thread().name
-        status = dbt.utils.lowercase(status)
-        return PartialResult(
-            node=node,
-            status=status,
-            error=error,
-            execution_time=execution_time,
-            thread_id=thread_id,
-            timing=timing_info,
-        )
-
-    def from_run_result(self, result, start_time, timing_info):
-        result.execution_time = (time.time() - start_time)
-        result.timing.extend(timing_info)
-        return result
-
-    def execute(self, compiled_node, manifest):
-        # we should only be here if we compiled_node.has_freshness, and
-        # therefore loaded_at_field should be a str. If this invariant is
-        # broken, raise!
-        if compiled_node.loaded_at_field is None:
-            raise InternalException(
-                'Got to execute for source freshness of a source that has no '
-                'loaded_at_field!'
-            )
-
-        relation = self.adapter.Relation.create_from_source(compiled_node)
-        # given a Source, calculate its fresnhess.
-        with self.adapter.connection_for(compiled_node):
-            self.adapter.clear_transaction()
-            freshness = self.adapter.calculate_freshness(
-                relation,
-                compiled_node.loaded_at_field,
-                compiled_node.freshness.filter,
-                manifest=manifest
-            )
-
-        status = compiled_node.freshness.status(freshness['age'])
-
-        return SourceFreshnessResult(
-            node=compiled_node,
-            status=status,
-            thread_id=threading.current_thread().name,
-            **freshness
-        )
-
-    def compile(self, manifest):
-        if self.node.resource_type != NodeType.Source:
-            # should be unreachable...
-            raise RuntimeException('fresnhess runner: got a non-Source')
-        # we don't do anything interesting when we compile a source node
-        return self.node
-
-
-class TestRunner(CompileRunner):
-    def describe_node(self):
-        node_name = self.node.name
-        return "test {}".format(node_name)
-
-    def print_result_line(self, result):
-        schema_name = self.node.schema
-        dbt.ui.printer.print_test_result_line(result,
-                                              schema_name,
-                                              self.node_index,
-                                              self.num_nodes)
-
-    def print_start_line(self):
-        description = self.describe_node()
-        dbt.ui.printer.print_start_line(description, self.node_index,
-                                        self.num_nodes)
-
-    def execute_data_test(self, test: CompiledDataTestNode):
-        sql = (
-            f'select count(*) as errors from (\n{test.injected_sql}\n) sbq'
-        )
-        res, table = self.adapter.execute(sql, auto_begin=True, fetch=True)
-
-        num_rows = len(table.rows)
-        if num_rows != 1:
-            num_cols = len(table.columns)
-            # since we just wrapped our query in `select count(*)`, we are in
-            # big trouble!
-            raise InternalException(
-                f"dbt internally failed to execute {test.unique_id}: "
-                f"Returned {num_rows} rows and {num_cols} cols, but expected "
-                f"1 row and 1 column"
-            )
-        return table[0][0]
-
-    def execute_schema_test(self, test: CompiledSchemaTestNode):
-        res, table = self.adapter.execute(
-            test.injected_sql,
-            auto_begin=True,
-            fetch=True,
-        )
-
-        num_rows = len(table.rows)
-        if num_rows != 1:
-            num_cols = len(table.columns)
-            raise_compiler_error(
-                f"Bad test {test.test_metadata.name}: "
-                f"Returned {num_rows} rows and {num_cols} cols, but expected "
-                f"1 row and 1 column"
-            )
-        return table[0][0]
-
-    def before_execute(self):
-        self.print_start_line()
-
-    def execute(self, test: CompiledTestNode, manifest: Manifest):
-        if isinstance(test, CompiledDataTestNode):
-            failed_rows = self.execute_data_test(test)
-        elif isinstance(test, CompiledSchemaTestNode):
-            failed_rows = self.execute_schema_test(test)
-        else:
-
-            raise InternalException(
-                f'Expected compiled schema test or compiled data test, got '
-                f'{type(test)}'
-            )
-        severity = test.config.severity.upper()
-
-        if failed_rows == 0:
-            return RunModelResult(test, status=failed_rows)
-        elif severity == 'ERROR' or dbt.flags.WARN_ERROR:
-            return RunModelResult(test, status=failed_rows, fail=True)
-        else:
-            return RunModelResult(test, status=failed_rows, warn=True)
-
-    def after_execute(self, result):
-        self.print_result_line(result)
-
-
-class SnapshotRunner(ModelRunner):
-    def describe_node(self):
-        return "snapshot {}".format(self.get_node_representation())
-
-    def print_result_line(self, result):
-        dbt.ui.printer.print_snapshot_result_line(
-            result,
-            self.get_node_representation(),
-            self.node_index,
-            self.num_nodes)
-
-
-class SeedRunner(ModelRunner):
-    def describe_node(self):
-        return "seed file {}".format(self.get_node_representation())
-
-    def before_execute(self):
-        description = self.describe_node()
-        dbt.ui.printer.print_start_line(description, self.node_index,
-                                        self.num_nodes)
-
-    def _build_run_model_result(self, model, context):
-        result = super()._build_run_model_result(model, context)
-        agate_result = context['load_result']('agate_table')
-        result.agate_table = agate_result.table
-        return result
-
-    def compile(self, manifest):
-        return self.node
-
-    def print_result_line(self, result):
-        schema_name = self.node.schema
-        dbt.ui.printer.print_seed_result_line(result,
-                                              schema_name,
-                                              self.node_index,
-                                              self.num_nodes)
diff --git a/core/dbt/parser/base.py b/core/dbt/parser/base.py
index ff3fa0a0e9a..3397fe3024a 100644
--- a/core/dbt/parser/base.py
+++ b/core/dbt/parser/base.py
@@ -7,14 +7,13 @@
 
 from hologram import ValidationError
 
+from dbt import utils
 from dbt.clients.jinja import MacroGenerator
 from dbt.clients.system import load_file_contents
 from dbt.context.providers import (
     generate_parser_model,
     generate_generate_component_name_macro,
 )
-import dbt.flags
-from dbt import hooks
 from dbt.adapters.factory import get_adapter
 from dbt.clients.jinja import get_rendered
 from dbt.config import Project, RuntimeConfig
@@ -29,6 +28,7 @@
 from dbt.exceptions import (
     CompilationException, validator_error_message, InternalException
 )
+from dbt import hooks
 from dbt.node_types import NodeType
 from dbt.parser.results import ParseResult, ManifestNodes
 from dbt.parser.search import FileBlock
@@ -172,7 +172,7 @@ def default_database(self):
     def get_fqn_prefix(self, path: str) -> List[str]:
         no_ext = os.path.splitext(path)[0]
         fqn = [self.project.project_name]
-        fqn.extend(dbt.utils.split_path(no_ext)[:-1])
+        fqn.extend(utils.split_path(no_ext)[:-1])
         return fqn
 
     def get_fqn(self, path: str, name: str) -> List[str]:
diff --git a/core/dbt/parser/manifest.py b/core/dbt/parser/manifest.py
index 341e9de8a08..1d20021f6dd 100644
--- a/core/dbt/parser/manifest.py
+++ b/core/dbt/parser/manifest.py
@@ -6,7 +6,7 @@
 )
 
 import dbt.exceptions
-import dbt.flags
+import dbt.flags as flags
 
 from dbt import deprecations
 from dbt.adapters.factory import (
@@ -25,6 +25,14 @@
 from dbt.contracts.graph.parsed import (
     ParsedSourceDefinition, ParsedNode, ParsedMacro, ColumnInfo,
 )
+from dbt.exceptions import (
+    ref_target_not_found,
+    get_target_not_found_or_disabled_msg,
+    source_target_not_found,
+    get_source_not_found_or_disabled_msg,
+    warn_or_error,
+)
+from dbt.include.global_project import PACKAGES
 from dbt.parser.base import BaseParser, Parser
 from dbt.parser.analysis import AnalysisParser
 from dbt.parser.data_test import DataTestParser
@@ -38,6 +46,7 @@
 from dbt.parser.seeds import SeedParser
 from dbt.parser.snapshots import SnapshotParser
 from dbt.parser.sources import patch_sources
+from dbt.ui import warning_tag
 from dbt.version import __version__
 
 
@@ -270,8 +279,8 @@ def matching_parse_results(self, result: ParseResult) -> bool:
 
     def _partial_parse_enabled(self):
         # if the CLI is set, follow that
-        if dbt.flags.PARTIAL_PARSE is not None:
-            return dbt.flags.PARTIAL_PARSE
+        if flags.PARTIAL_PARSE is not None:
+            return flags.PARTIAL_PARSE
         # if the config is set, follow that
         elif self.root_project.config.partial_parse is not None:
             return self.root_project.config.partial_parse
@@ -371,6 +380,52 @@ def load_internal(cls, root_config: RuntimeConfig) -> Manifest:
             return loader.load_only_macros()
 
 
+def invalid_ref_fail_unless_test(node, target_model_name,
+                                 target_model_package, disabled):
+
+    if node.resource_type == NodeType.Test:
+        msg = get_target_not_found_or_disabled_msg(
+            node, target_model_name, target_model_package, disabled
+        )
+        if disabled:
+            logger.debug(warning_tag(msg))
+        else:
+            warn_or_error(
+                msg,
+                log_fmt=warning_tag('{}')
+            )
+    else:
+        ref_target_not_found(
+            node,
+            target_model_name,
+            target_model_package,
+            disabled=disabled,
+        )
+
+
+def invalid_source_fail_unless_test(
+    node, target_name, target_table_name, disabled
+):
+    if node.resource_type == NodeType.Test:
+        msg = get_source_not_found_or_disabled_msg(
+            node, target_name, target_table_name, disabled
+        )
+        if disabled:
+            logger.debug(warning_tag(msg))
+        else:
+            warn_or_error(
+                msg,
+                log_fmt=warning_tag('{}')
+            )
+    else:
+        source_target_not_found(
+            node,
+            target_name,
+            target_table_name,
+            disabled=disabled
+        )
+
+
 def _check_resource_uniqueness(
     manifest: Manifest,
     config: RuntimeConfig,
@@ -542,7 +597,7 @@ def _process_refs_for_node(
             # This may raise. Even if it doesn't, we don't want to add
             # this node to the graph b/c there is no destination node
             node.config.enabled = False
-            dbt.utils.invalid_ref_fail_unless_test(
+            invalid_ref_fail_unless_test(
                 node, target_model_name, target_model_package,
                 disabled=(isinstance(target_model, Disabled))
             )
@@ -579,7 +634,7 @@ def _process_sources_for_node(
         if target_source is None or isinstance(target_source, Disabled):
             # this folows the same pattern as refs
             node.config.enabled = False
-            dbt.utils.invalid_source_fail_unless_test(
+            invalid_source_fail_unless_test(
                 node,
                 source_name,
                 table_name,
diff --git a/core/dbt/parser/results.py b/core/dbt/parser/results.py
index aa7ffca06aa..6e55c601b14 100644
--- a/core/dbt/parser/results.py
+++ b/core/dbt/parser/results.py
@@ -31,7 +31,7 @@
     raise_compiler_error, raise_duplicate_source_patch_name
 )
 from dbt.node_types import NodeType
-from dbt.ui import printer
+from dbt.ui import line_wrap_message
 from dbt.version import __version__
 
 
@@ -120,7 +120,7 @@ def add_macro(self, source_file: SourceFile, macro: ParsedMacro):
             # subtract 2 for the "Compilation Error" indent
             # note that the line wrap eats newlines, so if you want newlines,
             # this is the result :(
-            msg = printer.line_wrap_message(
+            msg = line_wrap_message(
                 f'''\
                 dbt found two macros named "{macro.name}" in the project
                 "{macro.package_name}".
diff --git a/core/dbt/parser/sources.py b/core/dbt/parser/sources.py
index 4a3a6bc55fa..6dd9ab739f8 100644
--- a/core/dbt/parser/sources.py
+++ b/core/dbt/parser/sources.py
@@ -22,7 +22,7 @@
 
 from dbt.parser.schemas import SchemaParser, ParserRef
 from dbt.parser.results import ParseResult
-from dbt.ui import printer
+from dbt import ui
 
 
 class SourcePatcher:
@@ -157,7 +157,7 @@ def warn_unused(self) -> None:
 
         if unused_tables:
             msg = self.get_unused_msg(unused_tables)
-            warn_or_error(msg, log_fmt=printer.yellow('WARNING: {}'))
+            warn_or_error(msg, log_fmt=ui.warning_tag('{}'))
 
     def get_unused_msg(
         self,
diff --git a/core/dbt/rpc/gc.py b/core/dbt/rpc/gc.py
index 492eca13da3..ef0462d0497 100644
--- a/core/dbt/rpc/gc.py
+++ b/core/dbt/rpc/gc.py
@@ -3,7 +3,6 @@
 from typing import Optional, List, Iterable, Tuple
 
 import dbt.exceptions
-import dbt.flags
 from dbt.contracts.rpc import (
     GCSettings,
     GCResultState,
diff --git a/core/dbt/rpc/node_runners.py b/core/dbt/rpc/node_runners.py
index 21bf3c0a1ee..58130b17415 100644
--- a/core/dbt/rpc/node_runners.py
+++ b/core/dbt/rpc/node_runners.py
@@ -7,7 +7,7 @@
     RemoteCompileResult, RemoteRunResult, ResultTable,
 )
 from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.node_runners import CompileRunner
+from dbt.task.compile import CompileRunner
 from dbt.rpc.error import dbt_error, RPCException, server_error
 
 
diff --git a/core/dbt/rpc/task_handler.py b/core/dbt/rpc/task_handler.py
index 60b3619953c..a195a1d491c 100644
--- a/core/dbt/rpc/task_handler.py
+++ b/core/dbt/rpc/task_handler.py
@@ -38,7 +38,6 @@
     QueueTimeoutMessage,
 )
 from dbt.rpc.method import RemoteMethod
-from dbt.flags import SINGLE_THREADED_HANDLER
 
 
 # we use this in typing only...
@@ -311,7 +310,10 @@ def method(self) -> str:
 
     @property
     def _single_threaded(self):
-        return bool(self.task.args.single_threaded or SINGLE_THREADED_HANDLER)
+        return bool(
+            self.task.args.single_threaded or
+            dbt.flags.SINGLE_THREADED_HANDLER
+        )
 
     @property
     def timeout(self) -> Optional[float]:
diff --git a/core/dbt/rpc/task_handler_protocol.py b/core/dbt/rpc/task_handler_protocol.py
index 7b4c3d2d0e6..1479d9030e2 100644
--- a/core/dbt/rpc/task_handler_protocol.py
+++ b/core/dbt/rpc/task_handler_protocol.py
@@ -4,7 +4,6 @@
 from typing_extensions import Protocol
 
 import dbt.exceptions
-import dbt.flags
 from dbt.contracts.rpc import (
     TaskHandlerState,
     TaskID,
diff --git a/core/dbt/rpc/task_manager.py b/core/dbt/rpc/task_manager.py
index 2becd7cb838..83ab7c0eabc 100644
--- a/core/dbt/rpc/task_manager.py
+++ b/core/dbt/rpc/task_manager.py
@@ -7,7 +7,7 @@
 
 
 import dbt.exceptions
-import dbt.flags
+import dbt.flags as flags
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.rpc import (
     LastParse,
@@ -32,7 +32,6 @@
 
 # import this to make sure our timedelta encoder is registered
 from dbt import helper_types  # noqa
-from dbt.flags import SINGLE_THREADED_WEBSERVER
 
 
 WrappedHandler = Callable[..., Dict[str, Any]]
@@ -86,12 +85,12 @@ def __init__(self, args, config, task_types: TaskTypes) -> None:
         self.active_tasks: TaskHandlerMap = {}
         self.gc = GarbageCollector(active_tasks=self.active_tasks)
         self.last_parse: LastParse = LastParse(state=ManifestStatus.Init)
-        self._lock: dbt.flags.MP_CONTEXT.Lock = dbt.flags.MP_CONTEXT.Lock()
+        self._lock: flags.MP_CONTEXT.Lock = flags.MP_CONTEXT.Lock()
         self._reloader: Optional[ManifestReloader] = None
         self.reload_manifest()
 
     def single_threaded(self):
-        return SINGLE_THREADED_WEBSERVER or self.args.single_threaded
+        return flags.SINGLE_THREADED_WEBSERVER or self.args.single_threaded
 
     def _reload_task_manager_thread(self, reloader: ManifestReloader):
         """This function can only be running once at a time, as it runs in the
diff --git a/core/dbt/task/base.py b/core/dbt/task/base.py
index a872115dc4a..d421cdc6592 100644
--- a/core/dbt/task/base.py
+++ b/core/dbt/task/base.py
@@ -1,12 +1,27 @@
 import os
+import threading
+import time
+import traceback
 from abc import ABCMeta, abstractmethod
-from typing import Type, Union
+from typing import Type, Union, Dict, Any, Optional
+
+from dbt import tracking
+from dbt import ui
+from dbt.contracts.graph.manifest import Manifest
+from dbt.contracts.results import (
+    RunModelResult, collect_timing_info
+)
+from dbt.exceptions import (
+    NotImplementedException, CompilationException, RuntimeException,
+    InternalException
+)
+from dbt.logger import GLOBAL_LOGGER as logger, log_manager
+from .printer import print_skip_caused_by_error, print_skip_line
+
 
 from dbt.adapters.factory import register_adapter
 from dbt.config import RuntimeConfig, Project
 from dbt.config.profile import read_profile, PROFILES_DIR
-from dbt import tracking
-from dbt.logger import GLOBAL_LOGGER as logger, log_manager
 import dbt.exceptions
 
 
@@ -139,3 +154,278 @@ def __init__(self, args, config):
     def from_args(cls, args):
         move_to_nearest_project_dir(args)
         return super().from_args(args)
+
+
+INTERNAL_ERROR_STRING = """This is an error in dbt. Please try again. If \
+the error persists, open an issue at https://github.com/fishtown-analytics/dbt
+""".strip()
+
+
+class ExecutionContext:
+    """During execution and error handling, dbt makes use of mutable state:
+    timing information and the newest (compiled vs executed) form of the node.
+    """
+    def __init__(self, node):
+        self.timing = []
+        self.node = node
+
+
+class BaseRunner(metaclass=ABCMeta):
+    def __init__(self, config, adapter, node, node_index, num_nodes):
+        self.config = config
+        self.adapter = adapter
+        self.node = node
+        self.node_index = node_index
+        self.num_nodes = num_nodes
+
+        self.skip = False
+        self.skip_cause: Optional[RunModelResult] = None
+
+    @abstractmethod
+    def compile(self, manifest: Manifest) -> Any:
+        pass
+
+    def get_result_status(self, result) -> Dict[str, str]:
+        if result.error:
+            return {'node_status': 'error', 'node_error': str(result.error)}
+        elif result.skip:
+            return {'node_status': 'skipped'}
+        elif result.fail:
+            return {'node_status': 'failed'}
+        elif result.warn:
+            return {'node_status': 'warn'}
+        else:
+            return {'node_status': 'passed'}
+
+    def run_with_hooks(self, manifest):
+        if self.skip:
+            return self.on_skip()
+
+        # no before/after printing for ephemeral mdoels
+        if not self.node.is_ephemeral_model:
+            self.before_execute()
+
+        result = self.safe_run(manifest)
+
+        if not self.node.is_ephemeral_model:
+            self.after_execute(result)
+
+        return result
+
+    def _build_run_result(self, node, start_time, error, status, timing_info,
+                          skip=False, fail=None, warn=None, agate_table=None):
+        execution_time = time.time() - start_time
+        thread_id = threading.current_thread().name
+        return RunModelResult(
+            node=node,
+            error=error,
+            skip=skip,
+            status=status,
+            fail=fail,
+            warn=warn,
+            execution_time=execution_time,
+            thread_id=thread_id,
+            timing=timing_info,
+            agate_table=agate_table,
+        )
+
+    def error_result(self, node, error, start_time, timing_info):
+        return self._build_run_result(
+            node=node,
+            start_time=start_time,
+            error=error,
+            status='ERROR',
+            timing_info=timing_info
+        )
+
+    def ephemeral_result(self, node, start_time, timing_info):
+        return self._build_run_result(
+            node=node,
+            start_time=start_time,
+            error=None,
+            status=None,
+            timing_info=timing_info
+        )
+
+    def from_run_result(self, result, start_time, timing_info):
+        return self._build_run_result(
+            node=result.node,
+            start_time=start_time,
+            error=result.error,
+            skip=result.skip,
+            status=result.status,
+            fail=result.fail,
+            warn=result.warn,
+            timing_info=timing_info,
+            agate_table=result.agate_table,
+        )
+
+    def compile_and_execute(self, manifest, ctx):
+        result = None
+        with self.adapter.connection_for(self.node):
+            with collect_timing_info('compile') as timing_info:
+                # if we fail here, we still have a compiled node to return
+                # this has the benefit of showing a build path for the errant
+                # model
+                ctx.node = self.compile(manifest)
+            ctx.timing.append(timing_info)
+
+            # for ephemeral nodes, we only want to compile, not run
+            if not ctx.node.is_ephemeral_model:
+                with collect_timing_info('execute') as timing_info:
+                    result = self.run(ctx.node, manifest)
+                    ctx.node = result.node
+
+                ctx.timing.append(timing_info)
+
+        return result
+
+    def _handle_catchable_exception(self, e, ctx):
+        if e.node is None:
+            e.add_node(ctx.node)
+
+        logger.debug(str(e), exc_info=True)
+        return str(e)
+
+    def _handle_internal_exception(self, e, ctx):
+        build_path = self.node.build_path
+        prefix = 'Internal error executing {}'.format(build_path)
+
+        error = "{prefix}\n{error}\n\n{note}".format(
+            prefix=ui.red(prefix),
+            error=str(e).strip(),
+            note=INTERNAL_ERROR_STRING
+        )
+        logger.debug(error, exc_info=True)
+        return str(e)
+
+    def _handle_generic_exception(self, e, ctx):
+        node_description = self.node.build_path
+        if node_description is None:
+            node_description = self.node.unique_id
+        prefix = "Unhandled error while executing {}".format(node_description)
+        error = "{prefix}\n{error}".format(
+            prefix=ui.red(prefix),
+            error=str(e).strip()
+        )
+
+        logger.error(error)
+        logger.debug('', exc_info=True)
+        return str(e)
+
+    def handle_exception(self, e, ctx):
+        catchable_errors = (CompilationException, RuntimeException)
+        if isinstance(e, catchable_errors):
+            error = self._handle_catchable_exception(e, ctx)
+        elif isinstance(e, InternalException):
+            error = self._handle_internal_exception(e, ctx)
+        else:
+            error = self._handle_generic_exception(e, ctx)
+        return error
+
+    def safe_run(self, manifest):
+        started = time.time()
+        ctx = ExecutionContext(self.node)
+        error = None
+        result = None
+
+        try:
+            result = self.compile_and_execute(manifest, ctx)
+        except Exception as e:
+            error = self.handle_exception(e, ctx)
+        finally:
+            exc_str = self._safe_release_connection()
+
+            # if releasing failed and the result doesn't have an error yet, set
+            # an error
+            if (
+                exc_str is not None and result is not None and
+                result.error is None and error is None
+            ):
+                error = exc_str
+
+        if error is not None:
+            # we could include compile time for runtime errors here
+            result = self.error_result(ctx.node, error, started, [])
+        elif result is not None:
+            result = self.from_run_result(result, started, ctx.timing)
+        else:
+            result = self.ephemeral_result(ctx.node, started, ctx.timing)
+        return result
+
+    def _safe_release_connection(self):
+        """Try to release a connection. If an exception is hit, log and return
+        the error string.
+        """
+        try:
+            self.adapter.release_connection()
+        except Exception as exc:
+            logger.debug(
+                'Error releasing connection for node {}: {!s}\n{}'
+                .format(self.node.name, exc, traceback.format_exc())
+            )
+            return str(exc)
+
+        return None
+
+    def before_execute(self):
+        raise NotImplementedException()
+
+    def execute(self, compiled_node, manifest):
+        raise NotImplementedException()
+
+    def run(self, compiled_node, manifest):
+        return self.execute(compiled_node, manifest)
+
+    def after_execute(self, result):
+        raise NotImplementedException()
+
+    def _skip_caused_by_ephemeral_failure(self):
+        if self.skip_cause is None or self.skip_cause.node is None:
+            return False
+        return self.skip_cause.node.is_ephemeral_model
+
+    def on_skip(self):
+        schema_name = self.node.schema
+        node_name = self.node.name
+
+        error = None
+        if not self.node.is_ephemeral_model:
+            # if this model was skipped due to an upstream ephemeral model
+            # failure, print a special 'error skip' message.
+            if self._skip_caused_by_ephemeral_failure():
+                print_skip_caused_by_error(
+                    self.node,
+                    schema_name,
+                    node_name,
+                    self.node_index,
+                    self.num_nodes,
+                    self.skip_cause
+                )
+                if self.skip_cause is None:  # mypy appeasement
+                    raise InternalException(
+                        'Skip cause not set but skip was somehow caused by '
+                        'an ephemeral failure'
+                    )
+                # set an error so dbt will exit with an error code
+                error = (
+                    'Compilation Error in {}, caused by compilation error '
+                    'in referenced ephemeral model {}'
+                    .format(self.node.unique_id,
+                            self.skip_cause.node.unique_id)
+                )
+            else:
+                print_skip_line(
+                    self.node,
+                    schema_name,
+                    node_name,
+                    self.node_index,
+                    self.num_nodes
+                )
+
+        node_result = RunModelResult(self.node, skip=True, error=error)
+        return node_result
+
+    def do_skip(self, cause=None):
+        self.skip = True
+        self.skip_cause = cause
diff --git a/core/dbt/task/compile.py b/core/dbt/task/compile.py
index b25015dfe86..320d60d0651 100644
--- a/core/dbt/task/compile.py
+++ b/core/dbt/task/compile.py
@@ -1,8 +1,26 @@
-from dbt.node_runners import CompileRunner
+
+
+from .runnable import GraphRunnableTask
+from .base import BaseRunner
+
+from dbt.compilation import compile_node
+from dbt.contracts.results import RunModelResult
+from dbt.logger import print_timestamped_line
 from dbt.node_types import NodeType
-import dbt.ui.printer
 
-from dbt.task.runnable import GraphRunnableTask
+
+class CompileRunner(BaseRunner):
+    def before_execute(self):
+        pass
+
+    def after_execute(self, result):
+        pass
+
+    def execute(self, compiled_node, manifest):
+        return RunModelResult(compiled_node)
+
+    def compile(self, manifest):
+        return compile_node(self.adapter, self.config, self.node, manifest, {})
 
 
 class CompileTask(GraphRunnableTask):
@@ -21,4 +39,4 @@ def get_runner_type(self):
         return CompileRunner
 
     def task_end_messages(self, results):
-        dbt.ui.printer.print_timestamped_line('Done.')
+        print_timestamped_line('Done.')
diff --git a/core/dbt/task/debug.py b/core/dbt/task/debug.py
index 14b374e13a6..84fdd420713 100644
--- a/core/dbt/task/debug.py
+++ b/core/dbt/task/debug.py
@@ -6,18 +6,17 @@
 
 from dbt.logger import GLOBAL_LOGGER as logger
 import dbt.clients.system
-import dbt.config
-import dbt.utils
 import dbt.exceptions
-from dbt.links import ProfileConfigDocs
 from dbt.adapters.factory import get_adapter, register_adapter
-from dbt.version import get_installed_version
-from dbt.config import Project, Profile
+from dbt.config import Project, Profile, PROFILES_DIR
 from dbt.config.renderer import DbtProjectYamlRenderer, ProfileRenderer
+from dbt.config.utils import parse_cli_vars
 from dbt.context.base import generate_base_context
 from dbt.context.target import generate_target_context
 from dbt.clients.yaml_helper import load_yaml_text
-from dbt.ui.printer import green, red
+from dbt.links import ProfileConfigDocs
+from dbt.ui import green, red
+from dbt.version import get_installed_version
 
 from dbt.task.base import BaseTask, get_nearest_project_dir
 
@@ -70,8 +69,7 @@ def __init__(self, *args, **kwargs):
 class DebugTask(BaseTask):
     def __init__(self, args, config):
         super().__init__(args, config)
-        self.profiles_dir = getattr(self.args, 'profiles_dir',
-                                    dbt.config.PROFILES_DIR)
+        self.profiles_dir = getattr(self.args, 'profiles_dir', PROFILES_DIR)
         self.profile_path = os.path.join(self.profiles_dir, 'profiles.yml')
         try:
             self.project_dir = get_nearest_project_dir(self.args)
@@ -83,9 +81,7 @@ def __init__(self, args, config):
             else:
                 self.project_dir = os.getcwd()
         self.project_path = os.path.join(self.project_dir, 'dbt_project.yml')
-        self.cli_vars = dbt.utils.parse_cli_vars(
-            getattr(self.args, 'vars', '{}')
-        )
+        self.cli_vars = parse_cli_vars(getattr(self.args, 'vars', '{}'))
 
         # set by _load_*
         self.profile: Optional[Profile] = None
diff --git a/core/dbt/task/freshness.py b/core/dbt/task/freshness.py
index 1db06564440..3b766b98f22 100644
--- a/core/dbt/task/freshness.py
+++ b/core/dbt/task/freshness.py
@@ -1,13 +1,106 @@
 import os
-from dbt.task.runnable import GraphRunnableTask
-from dbt.node_runners import FreshnessRunner
+import threading
+import time
+from typing import Dict
+
+from .base import BaseRunner
+from .printer import (
+    print_start_line,
+    print_freshness_result_line,
+    print_run_result_error,
+)
+from .runnable import GraphRunnableTask
+
+from dbt.contracts.results import (
+    FreshnessExecutionResult,
+    SourceFreshnessResult,
+    PartialResult,
+)
+from dbt.exceptions import RuntimeException, InternalException
+from dbt.logger import print_timestamped_line
 from dbt.node_types import NodeType
-from dbt.ui.printer import print_timestamped_line, print_run_result_error
-from dbt.contracts.results import FreshnessExecutionResult
+
+from dbt import utils
 
 RESULT_FILE_NAME = 'sources.json'
 
 
+class FreshnessRunner(BaseRunner):
+    def on_skip(self):
+        raise RuntimeException(
+            'Freshness: nodes cannot be skipped!'
+        )
+
+    def get_result_status(self, result) -> Dict[str, str]:
+        if result.error:
+            return {'node_status': 'error', 'node_error': str(result.error)}
+        else:
+            return {'node_status': str(result.status)}
+
+    def before_execute(self):
+        description = 'freshness of {0.source_name}.{0.name}'.format(self.node)
+        print_start_line(description, self.node_index, self.num_nodes)
+
+    def after_execute(self, result):
+        print_freshness_result_line(result, self.node_index, self.num_nodes)
+
+    def _build_run_result(self, node, start_time, error, status, timing_info,
+                          skip=False, failed=None):
+        execution_time = time.time() - start_time
+        thread_id = threading.current_thread().name
+        status = utils.lowercase(status)
+        return PartialResult(
+            node=node,
+            status=status,
+            error=error,
+            execution_time=execution_time,
+            thread_id=thread_id,
+            timing=timing_info,
+        )
+
+    def from_run_result(self, result, start_time, timing_info):
+        result.execution_time = (time.time() - start_time)
+        result.timing.extend(timing_info)
+        return result
+
+    def execute(self, compiled_node, manifest):
+        # we should only be here if we compiled_node.has_freshness, and
+        # therefore loaded_at_field should be a str. If this invariant is
+        # broken, raise!
+        if compiled_node.loaded_at_field is None:
+            raise InternalException(
+                'Got to execute for source freshness of a source that has no '
+                'loaded_at_field!'
+            )
+
+        relation = self.adapter.Relation.create_from_source(compiled_node)
+        # given a Source, calculate its fresnhess.
+        with self.adapter.connection_for(compiled_node):
+            self.adapter.clear_transaction()
+            freshness = self.adapter.calculate_freshness(
+                relation,
+                compiled_node.loaded_at_field,
+                compiled_node.freshness.filter,
+                manifest=manifest
+            )
+
+        status = compiled_node.freshness.status(freshness['age'])
+
+        return SourceFreshnessResult(
+            node=compiled_node,
+            status=status,
+            thread_id=threading.current_thread().name,
+            **freshness
+        )
+
+    def compile(self, manifest):
+        if self.node.resource_type != NodeType.Source:
+            # should be unreachable...
+            raise RuntimeException('fresnhess runner: got a non-Source')
+        # we don't do anything interesting when we compile a source node
+        return self.node
+
+
 class FreshnessTask(GraphRunnableTask):
     def result_path(self):
         if self.args.output:
diff --git a/core/dbt/task/generate.py b/core/dbt/task/generate.py
index e78fe84649a..f973323e264 100644
--- a/core/dbt/task/generate.py
+++ b/core/dbt/task/generate.py
@@ -5,6 +5,9 @@
 
 from hologram import ValidationError
 
+from .compile import CompileTask
+from .runnable import write_manifest
+
 from dbt.adapters.factory import get_adapter
 from dbt.contracts.graph.compiled import CompileResultNode
 from dbt.contracts.graph.manifest import Manifest
@@ -14,16 +17,12 @@
 )
 from dbt.exceptions import InternalException
 from dbt.include.global_project import DOCS_INDEX_FILE_PATH
-from dbt.logger import GLOBAL_LOGGER as logger
+from dbt.logger import GLOBAL_LOGGER as logger, print_timestamped_line
 from dbt.perf_utils import get_full_manifest
-import dbt.ui.printer
 import dbt.utils
 import dbt.compilation
 import dbt.exceptions
 
-from dbt.task.compile import CompileTask
-from dbt.task.runnable import write_manifest
-
 
 CATALOG_FILENAME = 'catalog.json'
 
@@ -214,7 +213,7 @@ def run(self) -> CatalogResults:
         if self.args.compile:
             compile_results = CompileTask.run(self)
             if any(r.error is not None for r in compile_results):
-                dbt.ui.printer.print_timestamped_line(
+                print_timestamped_line(
                     'compile failed, cannot generate docs'
                 )
                 return CatalogResults(
@@ -238,7 +237,7 @@ def run(self) -> CatalogResults:
 
         adapter = get_adapter(self.config)
         with adapter.connection_named('generate_catalog'):
-            dbt.ui.printer.print_timestamped_line("Building catalog")
+            print_timestamped_line("Building catalog")
             catalog_table, exceptions = adapter.get_catalog(self.manifest)
 
         catalog_data: List[PrimitiveDict] = [
@@ -272,7 +271,7 @@ def run(self) -> CatalogResults:
                 .format(len(exceptions), (len(exceptions) != 1) * 's')
             )
 
-        dbt.ui.printer.print_timestamped_line(
+        print_timestamped_line(
             'Catalog written to {}'.format(os.path.abspath(path))
         )
 
diff --git a/core/dbt/ui/printer.py b/core/dbt/task/printer.py
similarity index 71%
rename from core/dbt/ui/printer.py
rename to core/dbt/task/printer.py
index 1f717ba51e6..938c3e5bcad 100644
--- a/core/dbt/ui/printer.py
+++ b/core/dbt/task/printer.py
@@ -1,58 +1,15 @@
-import textwrap
-import time
 from typing import Dict, Optional, Tuple
-from dbt.logger import GLOBAL_LOGGER as logger, DbtStatusMessage, TextOnly
+from dbt.logger import (
+    GLOBAL_LOGGER as logger,
+    DbtStatusMessage,
+    TextOnly,
+    get_timestamp,
+)
 from dbt.node_types import NodeType
-import dbt.ui.colors
 
-USE_COLORS = False
-
-COLOR_FG_RED = dbt.ui.colors.COLORS['red']
-COLOR_FG_GREEN = dbt.ui.colors.COLORS['green']
-COLOR_FG_YELLOW = dbt.ui.colors.COLORS['yellow']
-COLOR_RESET_ALL = dbt.ui.colors.COLORS['reset_all']
-
-PRINTER_WIDTH = 80
-
-
-def use_colors():
-    global USE_COLORS
-    USE_COLORS = True
-
-
-def printer_width(printer_width):
-    global PRINTER_WIDTH
-    PRINTER_WIDTH = printer_width
-
-
-def get_timestamp():
-    return time.strftime("%H:%M:%S")
-
-
-def color(text: str, color_code: str):
-    if USE_COLORS:
-        return "{}{}{}".format(color_code, text, COLOR_RESET_ALL)
-    else:
-        return text
-
-
-def green(text: str):
-    return color(text, COLOR_FG_GREEN)
-
-
-def yellow(text: str):
-    return color(text, COLOR_FG_YELLOW)
-
-
-def red(text: str):
-    return color(text, COLOR_FG_RED)
-
-
-def print_timestamped_line(msg: str, use_color: Optional[str] = None):
-    if use_color is not None:
-        msg = color(msg, use_color)
-
-    logger.info("{} | {}".format(get_timestamp(), msg))
+from dbt.tracking import InvocationProcessor
+from dbt import ui
+from dbt import utils
 
 
 def print_fancy_output_line(
@@ -68,8 +25,8 @@ def print_fancy_output_line(
         progress=progress,
         message=msg)
 
-    truncate_width = PRINTER_WIDTH - 3
-    justified = prefix.ljust(PRINTER_WIDTH, ".")
+    truncate_width = ui.PRINTER_WIDTH - 3
+    justified = prefix.ljust(ui.PRINTER_WIDTH, ".")
     if truncate and len(justified) > truncate_width:
         justified = justified[:truncate_width] + '...'
 
@@ -99,7 +56,7 @@ def get_counts(flat_nodes) -> str:
         counts[t] = counts.get(t, 0) + 1
 
     stat_line = ", ".join(
-        [dbt.utils.pluralize(v, k) for k, v in counts.items()])
+        [utils.pluralize(v, k) for k, v in counts.items()])
 
     return stat_line
 
@@ -119,7 +76,7 @@ def print_hook_end_line(
 ) -> None:
     msg = 'OK hook: {}'.format(statement)
     # hooks don't fail into this path, so always green
-    print_fancy_output_line(msg, green(status), index, total,
+    print_fancy_output_line(msg, ui.green(status), index, total,
                             execution_time=execution_time, truncate=True)
 
 
@@ -127,21 +84,21 @@ def print_skip_line(
     model, schema: str, relation: str, index: int, num_models: int
 ) -> None:
     msg = 'SKIP relation {}.{}'.format(schema, relation)
-    print_fancy_output_line(msg, yellow('SKIP'), index, num_models)
+    print_fancy_output_line(msg, ui.yellow('SKIP'), index, num_models)
 
 
 def print_cancel_line(model) -> None:
     msg = 'CANCEL query {}'.format(model)
-    print_fancy_output_line(msg, red('CANCEL'), index=None, total=None)
+    print_fancy_output_line(msg, ui.red('CANCEL'), index=None, total=None)
 
 
 def get_printable_result(result, success: str, error: str) -> Tuple[str, str]:
     if result.error is not None:
         info = 'ERROR {}'.format(error)
-        status = red(result.status)
+        status = ui.red(result.status)
     else:
         info = 'OK {}'.format(success)
-        status = green(result.status)
+        status = ui.green(result.status)
 
     return info, status
 
@@ -153,18 +110,18 @@ def print_test_result_line(
 
     if result.error is not None:
         info = "ERROR"
-        color = red
+        color = ui.red
     elif result.status == 0:
         info = 'PASS'
-        color = green
+        color = ui.green
 
     elif result.warn:
         info = 'WARN {}'.format(result.status)
-        color = yellow
+        color = ui.yellow
 
     elif result.fail:
         info = 'FAIL {}'.format(result.status)
-        color = red
+        color = ui.red
 
     else:
         raise RuntimeError("unexpected status: {}".format(result.status))
@@ -227,16 +184,16 @@ def print_seed_result_line(result, schema_name: str, index: int, total: int):
 def print_freshness_result_line(result, index: int, total: int) -> None:
     if result.error:
         info = 'ERROR'
-        color = red
+        color = ui.red
     elif result.status == 'error':
         info = 'ERROR STALE'
-        color = red
+        color = ui.red
     elif result.status == 'warn':
         info = 'WARN'
-        color = yellow
+        color = ui.yellow
     else:
         info = 'PASS'
-        color = green
+        color = ui.green
 
     if hasattr(result, 'node'):
         source_name = result.node.source_name
@@ -298,11 +255,11 @@ def print_run_result_error(
 
     if result.fail or (is_warning and result.warn):
         if is_warning:
-            color = yellow
+            color = ui.yellow
             info = 'Warning'
             logger_fn = logger.warning
         else:
-            color = red
+            color = ui.red
             info = 'Failure'
             logger_fn = logger.error
         logger_fn(color("{} in {} {} ({})").format(
@@ -316,7 +273,7 @@ def print_run_result_error(
         except ValueError:
             logger.error("  Status: {}".format(result.status))
         else:
-            status = dbt.utils.pluralize(result.status, 'result')
+            status = utils.pluralize(result.status, 'result')
             logger.error("  Got {}, expected 0.".format(status))
 
         if result.node.build_path is not None:
@@ -329,7 +286,7 @@ def print_run_result_error(
         first = True
         for line in result.error.split("\n"):
             if first:
-                logger.error(yellow(line))
+                logger.error(ui.yellow(line))
                 first = False
             else:
                 logger.error(line)
@@ -340,24 +297,24 @@ def print_skip_caused_by_error(
 ) -> None:
     msg = ('SKIP relation {}.{} due to ephemeral model error'
            .format(schema, relation))
-    print_fancy_output_line(msg, red('ERROR SKIP'), index, num_models)
+    print_fancy_output_line(msg, ui.red('ERROR SKIP'), index, num_models)
     print_run_result_error(result, newline=False)
 
 
 def print_end_of_run_summary(
     num_errors: int, num_warnings: int, keyboard_interrupt: bool = False
 ) -> None:
-    error_plural = dbt.utils.pluralize(num_errors, 'error')
-    warn_plural = dbt.utils.pluralize(num_warnings, 'warning')
+    error_plural = utils.pluralize(num_errors, 'error')
+    warn_plural = utils.pluralize(num_warnings, 'warning')
     if keyboard_interrupt:
-        message = yellow('Exited because of keyboard interrupt.')
+        message = ui.yellow('Exited because of keyboard interrupt.')
     elif num_errors > 0:
-        message = red("Completed with {} and {}:".format(
+        message = ui.red("Completed with {} and {}:".format(
             error_plural, warn_plural))
     elif num_warnings > 0:
-        message = yellow('Completed with {}:'.format(warn_plural))
+        message = ui.yellow('Completed with {}:'.format(warn_plural))
     else:
-        message = green('Completed successfully')
+        message = ui.green('Completed successfully')
 
     with TextOnly():
         logger.info('')
@@ -365,9 +322,6 @@ def print_end_of_run_summary(
 
 
 def print_run_end_messages(results, keyboard_interrupt: bool = False) -> None:
-    # Prevent import loop from happening by importing tracking here.
-    from dbt.tracking import InvocationProcessor  # noqa
-
     errors = [r for r in results if r.error is not None or r.fail]
     warnings = [r for r in results if r.warn]
     with DbtStatusMessage(), InvocationProcessor():
@@ -382,29 +336,3 @@ def print_run_end_messages(results, keyboard_interrupt: bool = False) -> None:
             print_run_result_error(warning, is_warning=True)
 
         print_run_status_line(results)
-
-
-def line_wrap_message(
-    msg: str, subtract: int = 0, dedent: bool = True, prefix: str = ''
-) -> str:
-    '''
-    Line wrap the given message to PRINTER_WIDTH - {subtract}. Convert double
-    newlines to newlines and avoid calling textwrap.fill() on them (like
-    markdown)
-    '''
-    width = PRINTER_WIDTH - subtract
-    if dedent:
-        msg = textwrap.dedent(msg)
-
-    if prefix:
-        msg = f'{prefix}{msg}'
-
-    # If the input had an explicit double newline, we want to preserve that
-    # (we'll turn it into a single line soon). Support windows, too.
-    splitter = '\r\n\r\n' if '\r\n\r\n' in msg else '\n\n'
-    chunks = msg.split(splitter)
-    return '\n'.join(textwrap.fill(chunk, width=width) for chunk in chunks)
-
-
-def warning_tag(msg: str) -> str:
-    return f'[{yellow("WARNING")}]: {msg}'
diff --git a/core/dbt/task/rpc/cli.py b/core/dbt/task/rpc/cli.py
index 622480470f9..d164b7c4e6f 100644
--- a/core/dbt/task/rpc/cli.py
+++ b/core/dbt/task/rpc/cli.py
@@ -4,6 +4,7 @@
 from typing import Type, Optional
 
 
+from dbt.config.utils import parse_cli_vars
 from dbt.contracts.rpc import RPCCliParameters
 
 from dbt.rpc.method import (
@@ -14,7 +15,6 @@
 )
 from dbt.exceptions import InternalException
 from dbt.perf_utils import get_full_manifest
-from dbt.utils import parse_cli_vars
 
 from .base import RPCTask
 
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index 61b9196cd6c..b9666970d17 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -2,6 +2,34 @@
 import time
 from typing import List, Dict, Any, Iterable, Set, Tuple, Optional
 
+from .compile import CompileRunner, CompileTask
+
+from .printer import (
+    print_start_line,
+    print_model_result_line,
+    print_hook_start_line,
+    print_hook_end_line,
+    print_run_end_messages,
+    get_counts,
+)
+
+from dbt import deprecations
+from dbt import tracking
+from dbt import utils
+from dbt.adapters.base import BaseRelation
+from dbt.clients.jinja import MacroGenerator
+from dbt.compilation import compile_node
+from dbt.context.providers import generate_runtime_model
+from dbt.contracts.graph.compiled import CompileResultNode
+from dbt.contracts.graph.model_config import Hook
+from dbt.contracts.graph.parsed import ParsedHookNode
+from dbt.contracts.results import RunModelResult
+from dbt.exceptions import (
+    CompilationException,
+    InternalException,
+    RuntimeException,
+    missing_materialization,
+)
 from dbt.logger import (
     GLOBAL_LOGGER as logger,
     TextOnly,
@@ -9,26 +37,10 @@
     UniqueID,
     TimestampNamed,
     DbtModelState,
+    print_timestamped_line,
 )
-from dbt.exceptions import InternalException
-from dbt.node_types import NodeType, RunHookType
-from dbt.node_runners import ModelRunner
-
-import dbt.exceptions
-import dbt.flags
 from dbt.hooks import get_hook_dict
-from dbt.ui.printer import \
-    print_hook_start_line, \
-    print_hook_end_line, \
-    print_timestamped_line, \
-    print_run_end_messages, \
-    get_counts
-
-from dbt.compilation import compile_node
-from dbt.contracts.graph.compiled import CompileResultNode
-from dbt.contracts.graph.model_config import Hook
-from dbt.contracts.graph.parsed import ParsedHookNode
-from dbt.task.compile import CompileTask
+from dbt.node_types import NodeType, RunHookType
 
 
 class Timer:
@@ -83,6 +95,150 @@ def get_hook(source, index):
     return Hook.from_dict(hook_dict)
 
 
+def track_model_run(index, num_nodes, run_model_result):
+    if tracking.active_user is None:
+        raise InternalException('cannot track model run with no active user')
+    invocation_id = tracking.active_user.invocation_id
+    tracking.track_model_run({
+        "invocation_id": invocation_id,
+        "index": index,
+        "total": num_nodes,
+        "execution_time": run_model_result.execution_time,
+        "run_status": run_model_result.status,
+        "run_skipped": run_model_result.skip,
+        "run_error": None,
+        "model_materialization": run_model_result.node.get_materialization(),
+        "model_id": utils.get_hash(run_model_result.node),
+        "hashed_contents": utils.get_hashed_contents(
+            run_model_result.node
+        ),
+        "timing": [t.to_dict() for t in run_model_result.timing],
+    })
+
+
+# make sure that we got an ok result back from a materialization
+def _validate_materialization_relations_dict(
+    inp: Dict[Any, Any], model
+) -> List[BaseRelation]:
+    try:
+        relations_value = inp['relations']
+    except KeyError:
+        msg = (
+            'Invalid return value from materialization, "relations" '
+            'not found, got keys: {}'.format(list(inp))
+        )
+        raise CompilationException(msg, node=model) from None
+
+    if not isinstance(relations_value, list):
+        msg = (
+            'Invalid return value from materialization, "relations" '
+            'not a list, got: {}'.format(relations_value)
+        )
+        raise CompilationException(msg, node=model) from None
+
+    relations: List[BaseRelation] = []
+    for relation in relations_value:
+        if not isinstance(relation, BaseRelation):
+            msg = (
+                'Invalid return value from materialization, '
+                '"relations" contains non-Relation: {}'
+                .format(relation)
+            )
+            raise CompilationException(msg, node=model)
+
+        assert isinstance(relation, BaseRelation)
+        relations.append(relation)
+    return relations
+
+
+class ModelRunner(CompileRunner):
+    def get_node_representation(self):
+        display_quote_policy = {
+            'database': False, 'schema': False, 'identifier': False
+        }
+        relation = self.adapter.Relation.create_from(
+            self.config, self.node, quote_policy=display_quote_policy
+        )
+        # exclude the database from output if it's the default
+        if self.node.database == self.config.credentials.database:
+            relation = relation.include(database=False)
+        return str(relation)
+
+    def describe_node(self):
+        return "{} model {}".format(self.node.get_materialization(),
+                                    self.get_node_representation())
+
+    def print_start_line(self):
+        description = self.describe_node()
+        print_start_line(description, self.node_index, self.num_nodes)
+
+    def print_result_line(self, result):
+        description = self.describe_node()
+        print_model_result_line(result, description, self.node_index,
+                                self.num_nodes)
+
+    def before_execute(self):
+        self.print_start_line()
+
+    def after_execute(self, result):
+        track_model_run(self.node_index, self.num_nodes, result)
+        self.print_result_line(result)
+
+    def _build_run_model_result(self, model, context):
+        result = context['load_result']('main')
+        return RunModelResult(model, status=result.status)
+
+    def _materialization_relations(
+        self, result: Any, model
+    ) -> List[BaseRelation]:
+        if isinstance(result, str):
+            deprecations.warn('materialization-return',
+                              materialization=model.get_materialization())
+            return [
+                self.adapter.Relation.create_from(self.config, model)
+            ]
+
+        if isinstance(result, dict):
+            return _validate_materialization_relations_dict(result, model)
+
+        msg = (
+            'Invalid return value from materialization, expected a dict '
+            'with key "relations", got: {}'.format(str(result))
+        )
+        raise CompilationException(msg, node=model)
+
+    def execute(self, model, manifest):
+        context = generate_runtime_model(
+            model, self.config, manifest
+        )
+
+        materialization_macro = manifest.find_materialization_macro_by_name(
+            self.config.project_name,
+            model.get_materialization(),
+            self.adapter.type())
+
+        if materialization_macro is None:
+            missing_materialization(model, self.adapter.type())
+
+        if 'config' not in context:
+            raise InternalException(
+                'Invalid materialization context generated, missing config: {}'
+                .format(context)
+            )
+        context_config = context['config']
+
+        hook_ctx = self.adapter.pre_model_hook(context_config)
+        try:
+            result = MacroGenerator(materialization_macro, context)()
+        finally:
+            self.adapter.post_model_hook(context_config, hook_ctx)
+
+        for relation in self._materialization_relations(result, model):
+            self.adapter.cache_added(relation.incorporate(dbt_created=True))
+
+        return self._build_run_model_result(model, context)
+
+
 class RunTask(CompileTask):
     def __init__(self, args, config):
         super().__init__(args, config)
@@ -178,7 +334,7 @@ def safe_run_hooks(
     ) -> None:
         try:
             self.run_hooks(adapter, hook_type, extra_context)
-        except dbt.exceptions.RuntimeException:
+        except RuntimeException:
             logger.info("Database error while running {}".format(hook_type))
             raise
 
diff --git a/core/dbt/task/run_operation.py b/core/dbt/task/run_operation.py
index 107fd60dd90..fc786724a10 100644
--- a/core/dbt/task/run_operation.py
+++ b/core/dbt/task/run_operation.py
@@ -3,15 +3,14 @@
 
 import agate
 
-from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.task.runnable import ManifestTask
+from .runnable import ManifestTask
+
+import dbt.exceptions
 from dbt.adapters.factory import get_adapter
+from dbt.config.utils import parse_cli_vars
 from dbt.contracts.results import RunOperationResult
 from dbt.exceptions import InternalException
-
-import dbt
-import dbt.utils
-import dbt.exceptions
+from dbt.logger import GLOBAL_LOGGER as logger
 
 
 class RunOperationTask(ManifestTask):
@@ -25,7 +24,7 @@ def _get_macro_parts(self):
         return package_name, macro_name
 
     def _get_kwargs(self) -> Dict[str, Any]:
-        return dbt.utils.parse_cli_vars(self.args.args)
+        return parse_cli_vars(self.args.args)
 
     def compile_manifest(self) -> None:
         if self.manifest is None:
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index d7c4fe7980d..39c36282418 100644
--- a/core/dbt/task/runnable.py
+++ b/core/dbt/task/runnable.py
@@ -5,6 +5,12 @@
 from multiprocessing.dummy import Pool as ThreadPool
 from typing import Optional, Dict, List, Set, Tuple, Iterable
 
+from .printer import (
+    print_run_result_error,
+    print_run_end_messages,
+    print_cancel_line,
+)
+from dbt import ui
 from dbt.task.base import ConfiguredTask
 from dbt.adapters.base import BaseRelation
 from dbt.adapters.factory import get_adapter
@@ -17,6 +23,7 @@
     DbtModelState,
     ModelMetadata,
     NodeCount,
+    print_timestamped_line,
 )
 from dbt.compilation import compile_manifest
 
@@ -34,8 +41,7 @@
 from dbt.perf_utils import get_full_manifest
 
 import dbt.exceptions
-import dbt.flags
-import dbt.ui.printer
+from dbt import flags
 import dbt.utils
 
 import dbt.graph.selector
@@ -46,7 +52,7 @@
 
 
 def write_manifest(config, manifest):
-    if dbt.flags.WRITE_JSON:
+    if flags.WRITE_JSON:
         manifest.write(os.path.join(config.target_path, MANIFEST_FILE_NAME))
 
 
@@ -293,8 +299,8 @@ def _cancel_connections(self, pool):
                    "cancellation. Some queries may still be "
                    "running!".format(adapter.type()))
 
-            yellow = dbt.ui.printer.COLOR_FG_YELLOW
-            dbt.ui.printer.print_timestamped_line(msg, yellow)
+            yellow = ui.COLOR_FG_YELLOW
+            print_timestamped_line(msg, yellow)
             raise
 
         for conn_name in adapter.cancel_open_connections():
@@ -303,7 +309,7 @@ def _cancel_connections(self, pool):
                 if node is not None and node.is_ephemeral_model:
                     continue
             # if we don't have a manifest/don't have a node, print anyway.
-            dbt.ui.printer.print_cancel_line(conn_name)
+            print_cancel_line(conn_name)
 
         pool.join()
 
@@ -314,9 +320,9 @@ def execute_nodes(self):
         text = "Concurrency: {} threads (target='{}')"
         concurrency_line = text.format(num_threads, target_name)
         with NodeCount(self.num_nodes):
-            dbt.ui.printer.print_timestamped_line(concurrency_line)
+            print_timestamped_line(concurrency_line)
         with TextOnly():
-            dbt.ui.printer.print_timestamped_line("")
+            print_timestamped_line("")
 
         pool = ThreadPool(num_threads)
         try:
@@ -324,13 +330,12 @@ def execute_nodes(self):
 
         except FailFastException as failure:
             self._cancel_connections(pool)
-            dbt.ui.printer.print_run_result_error(failure.result)
+            print_run_result_error(failure.result)
             raise
 
         except KeyboardInterrupt:
             self._cancel_connections(pool)
-            dbt.ui.printer.print_run_end_messages(self.node_results,
-                                                  keyboard_interrupt=True)
+            print_run_end_messages(self.node_results, keyboard_interrupt=True)
             raise
 
         pool.close()
@@ -407,7 +412,7 @@ def run(self):
         selected_uids = frozenset(n.unique_id for n in self._flattened_nodes)
         result = self.execute_with_hooks(selected_uids)
 
-        if dbt.flags.WRITE_JSON:
+        if flags.WRITE_JSON:
             result.write(self.result_path())
 
         self.task_end_messages(result.results)
@@ -515,4 +520,4 @@ def get_result(self, results, elapsed_time, generated_at):
         )
 
     def task_end_messages(self, results):
-        dbt.ui.printer.print_run_end_messages(results)
+        print_run_end_messages(results)
diff --git a/core/dbt/task/seed.py b/core/dbt/task/seed.py
index 217e94aa08e..8047c27068b 100644
--- a/core/dbt/task/seed.py
+++ b/core/dbt/task/seed.py
@@ -1,10 +1,37 @@
 import random
 
+from .run import ModelRunner, RunTask
+from .printer import (
+    print_start_line,
+    print_seed_result_line,
+    print_run_end_messages,
+)
+
 from dbt.logger import GLOBAL_LOGGER as logger, TextOnly
-from dbt.node_runners import SeedRunner
 from dbt.node_types import NodeType
-from dbt.task.run import RunTask
-import dbt.ui.printer
+
+
+class SeedRunner(ModelRunner):
+    def describe_node(self):
+        return "seed file {}".format(self.get_node_representation())
+
+    def before_execute(self):
+        description = self.describe_node()
+        print_start_line(description, self.node_index, self.num_nodes)
+
+    def _build_run_model_result(self, model, context):
+        result = super()._build_run_model_result(model, context)
+        agate_result = context['load_result']('agate_table')
+        result.agate_table = agate_result.table
+        return result
+
+    def compile(self, manifest):
+        return self.node
+
+    def print_result_line(self, result):
+        schema_name = self.node.schema
+        print_seed_result_line(result, schema_name, self.node_index,
+                               self.num_nodes)
 
 
 class SeedTask(RunTask):
@@ -25,7 +52,7 @@ def task_end_messages(self, results):
         if self.args.show:
             self.show_tables(results)
 
-        dbt.ui.printer.print_run_end_messages(results)
+        print_run_end_messages(results)
 
     def show_table(self, result):
         table = result.agate_table
diff --git a/core/dbt/task/snapshot.py b/core/dbt/task/snapshot.py
index 150ca520686..ba23dd050d5 100644
--- a/core/dbt/task/snapshot.py
+++ b/core/dbt/task/snapshot.py
@@ -1,6 +1,18 @@
-from dbt.node_runners import SnapshotRunner
+from .run import ModelRunner, RunTask
+from .printer import print_snapshot_result_line
+
 from dbt.node_types import NodeType
-from dbt.task.run import RunTask
+
+class SnapshotRunner(ModelRunner):
+    def describe_node(self):
+        return "snapshot {}".format(self.get_node_representation())
+
+    def print_result_line(self, result):
+        print_snapshot_result_line(
+            result,
+            self.get_node_representation(),
+            self.node_index,
+            self.num_nodes)
 
 
 class SnapshotTask(RunTask):
diff --git a/core/dbt/task/test.py b/core/dbt/task/test.py
index 62c87d12372..babf70f1ed8 100644
--- a/core/dbt/task/test.py
+++ b/core/dbt/task/test.py
@@ -1,8 +1,95 @@
 from typing import Dict, Any
 
-from dbt.node_runners import TestRunner
+from .compile import CompileRunner
+from .run import RunTask
+from .printer import print_start_line, print_test_result_line
+
+from dbt.contracts.graph.compiled import (
+    CompiledDataTestNode,
+    CompiledSchemaTestNode,
+    CompiledTestNode,
+)
+from dbt.contracts.graph.manifest import Manifest
+from dbt.contracts.results import RunModelResult
+from dbt.exceptions import raise_compiler_error, InternalException
+from dbt import flags
 from dbt.node_types import NodeType, RunHookType
-from dbt.task.run import RunTask
+
+
+class TestRunner(CompileRunner):
+    def describe_node(self):
+        node_name = self.node.name
+        return "test {}".format(node_name)
+
+    def print_result_line(self, result):
+        schema_name = self.node.schema
+        print_test_result_line(result, schema_name, self.node_index,
+                               self.num_nodes)
+
+    def print_start_line(self):
+        description = self.describe_node()
+        print_start_line(description, self.node_index, self.num_nodes)
+
+    def execute_data_test(self, test: CompiledDataTestNode):
+        sql = (
+            f'select count(*) as errors from (\n{test.injected_sql}\n) sbq'
+        )
+        res, table = self.adapter.execute(sql, auto_begin=True, fetch=True)
+
+        num_rows = len(table.rows)
+        if num_rows != 1:
+            num_cols = len(table.columns)
+            # since we just wrapped our query in `select count(*)`, we are in
+            # big trouble!
+            raise InternalException(
+                f"dbt internally failed to execute {test.unique_id}: "
+                f"Returned {num_rows} rows and {num_cols} cols, but expected "
+                f"1 row and 1 column"
+            )
+        return table[0][0]
+
+    def execute_schema_test(self, test: CompiledSchemaTestNode):
+        res, table = self.adapter.execute(
+            test.injected_sql,
+            auto_begin=True,
+            fetch=True,
+        )
+
+        num_rows = len(table.rows)
+        if num_rows != 1:
+            num_cols = len(table.columns)
+            raise_compiler_error(
+                f"Bad test {test.test_metadata.name}: "
+                f"Returned {num_rows} rows and {num_cols} cols, but expected "
+                f"1 row and 1 column"
+            )
+        return table[0][0]
+
+    def before_execute(self):
+        self.print_start_line()
+
+    def execute(self, test: CompiledTestNode, manifest: Manifest):
+        if isinstance(test, CompiledDataTestNode):
+            failed_rows = self.execute_data_test(test)
+        elif isinstance(test, CompiledSchemaTestNode):
+            failed_rows = self.execute_schema_test(test)
+        else:
+
+            raise InternalException(
+                f'Expected compiled schema test or compiled data test, got '
+                f'{type(test)}'
+            )
+        severity = test.config.severity.upper()
+
+        if failed_rows == 0:
+            return RunModelResult(test, status=failed_rows)
+        elif severity == 'ERROR' or flags.WARN_ERROR:
+            return RunModelResult(test, status=failed_rows, fail=True)
+        else:
+            return RunModelResult(test, status=failed_rows, warn=True)
+
+    def after_execute(self, result):
+        self.print_result_line(result)
 
 
 class TestTask(RunTask):
diff --git a/core/dbt/ui.py b/core/dbt/ui.py
new file mode 100644
index 00000000000..3efafd6c16c
--- /dev/null
+++ b/core/dbt/ui.py
@@ -0,0 +1,76 @@
+import textwrap
+from typing import Dict
+
+import colorama
+
+COLORS: Dict[str, str] = {
+    'red': colorama.Fore.RED,
+    'green': colorama.Fore.GREEN,
+    'yellow': colorama.Fore.YELLOW,
+    'reset_all': colorama.Style.RESET_ALL
+}
+
+
+USE_COLORS = False
+
+COLOR_FG_RED = COLORS['red']
+COLOR_FG_GREEN = COLORS['green']
+COLOR_FG_YELLOW = COLORS['yellow']
+COLOR_RESET_ALL = COLORS['reset_all']
+
+PRINTER_WIDTH = 80
+
+
+def use_colors():
+    global USE_COLORS
+    USE_COLORS = True
+
+
+def printer_width(printer_width):
+    global PRINTER_WIDTH
+    PRINTER_WIDTH = printer_width
+
+
+def color(text: str, color_code: str):
+    if USE_COLORS:
+        return "{}{}{}".format(color_code, text, COLOR_RESET_ALL)
+    else:
+        return text
+
+
+def green(text: str):
+    return color(text, COLOR_FG_GREEN)
+
+
+def yellow(text: str):
+    return color(text, COLOR_FG_YELLOW)
+
+
+def red(text: str):
+    return color(text, COLOR_FG_RED)
+
+
+def line_wrap_message(
+    msg: str, subtract: int = 0, dedent: bool = True, prefix: str = ''
+) -> str:
+    '''
+    Line wrap the given message to PRINTER_WIDTH - {subtract}. Convert double
+    newlines to newlines and avoid calling textwrap.fill() on them (like
+    markdown)
+    '''
+    width = PRINTER_WIDTH - subtract
+    if dedent:
+        msg = textwrap.dedent(msg)
+
+    if prefix:
+        msg = f'{prefix}{msg}'
+
+    # If the input had an explicit double newline, we want to preserve that
+    # (we'll turn it into a single line soon). Support windows, too.
+    splitter = '\r\n\r\n' if '\r\n\r\n' in msg else '\n\n'
+    chunks = msg.split(splitter)
+    return '\n'.join(textwrap.fill(chunk, width=width) for chunk in chunks)
+
+
+def warning_tag(msg: str) -> str:
+    return f'[{yellow("WARNING")}]: {msg}'
diff --git a/core/dbt/ui/__init__.py b/core/dbt/ui/__init__.py
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/core/dbt/ui/colors.py b/core/dbt/ui/colors.py
deleted file mode 100644
index 49b607ba812..00000000000
--- a/core/dbt/ui/colors.py
+++ /dev/null
@@ -1,10 +0,0 @@
-from typing import Dict
-
-import colorama
-
-COLORS: Dict[str, str] = {
-    'red': colorama.Fore.RED,
-    'green': colorama.Fore.GREEN,
-    'yellow': colorama.Fore.YELLOW,
-    'reset_all': colorama.Style.RESET_ALL
-}
diff --git a/core/dbt/utils.py b/core/dbt/utils.py
index 6e8ea6377d1..3b8e509c69e 100644
--- a/core/dbt/utils.py
+++ b/core/dbt/utils.py
@@ -17,10 +17,7 @@
 
 import dbt.exceptions
 
-from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.node_types import NodeType
-from dbt.clients import yaml_helper
-from dbt.ui import printer
 
 DECIMALS: Tuple[Type[Any], ...]
 try:
@@ -67,14 +64,6 @@ def get_model_name_or_none(model):
     return name
 
 
-def compiler_warning(model, msg, resource_type='model'):
-    name = get_model_name_or_none(model)
-    logger.info(
-        "* Compilation warning while compiling {} {}:\n* {}\n"
-        .format(resource_type, name, msg)
-    )
-
-
 MACRO_PREFIX = 'dbt_macro__'
 DOCS_PREFIX = 'dbt_docs__'
 
@@ -299,70 +288,6 @@ def __get__(self, obj, objtype):
         return functools.partial(self.__call__, obj)
 
 
-def invalid_ref_fail_unless_test(node, target_model_name,
-                                 target_model_package, disabled):
-
-    if node.resource_type == NodeType.Test:
-        msg = dbt.exceptions.get_target_not_found_or_disabled_msg(
-            node, target_model_name, target_model_package, disabled
-        )
-        if disabled:
-            logger.debug(printer.warning_tag(msg))
-        else:
-            dbt.exceptions.warn_or_error(
-                msg,
-                log_fmt=printer.warning_tag('{}')
-            )
-    else:
-        dbt.exceptions.ref_target_not_found(
-            node,
-            target_model_name,
-            target_model_package,
-            disabled=disabled,
-        )
-
-
-def invalid_source_fail_unless_test(
-    node, target_name, target_table_name, disabled
-):
-    if node.resource_type == NodeType.Test:
-        msg = dbt.exceptions.get_source_not_found_or_disabled_msg(
-            node, target_name, target_table_name, disabled
-        )
-        if disabled:
-            logger.debug(printer.warning_tag(msg))
-        else:
-            dbt.exceptions.warn_or_error(
-                msg,
-                log_fmt=printer.warning_tag('{}')
-            )
-    else:
-        dbt.exceptions.source_target_not_found(
-            node,
-            target_name,
-            target_table_name,
-            disabled=disabled
-        )
-
-
-def parse_cli_vars(var_string: str) -> Dict[str, Any]:
-    try:
-        cli_vars = yaml_helper.load_yaml_text(var_string)
-        var_type = type(cli_vars)
-        if var_type is dict:
-            return cli_vars
-        else:
-            type_name = var_type.__name__
-            dbt.exceptions.raise_compiler_error(
-                "The --vars argument must be a YAML dictionary, but was "
-                "of type '{}'".format(type_name))
-    except dbt.exceptions.ValidationException:
-        logger.error(
-            "The YAML provided in the --vars argument is not valid.\n"
-        )
-        raise
-
-
 K_T = TypeVar('K_T')
 V_T = TypeVar('V_T')
 
diff --git a/plugins/bigquery/dbt/adapters/bigquery/impl.py b/plugins/bigquery/dbt/adapters/bigquery/impl.py
index 325190aedfb..1ae97769553 100644
--- a/plugins/bigquery/dbt/adapters/bigquery/impl.py
+++ b/plugins/bigquery/dbt/adapters/bigquery/impl.py
@@ -7,8 +7,8 @@
 import dbt.flags as flags
 import dbt.clients.gcloud
 import dbt.clients.agate_helper
-import dbt.links
 
+from dbt import ui
 from dbt.adapters.base import (
     BaseAdapter, available, RelationType, SchemaSearchMap, AdapterConfig
 )
@@ -17,7 +17,7 @@
 from dbt.adapters.bigquery import BigQueryConnectionManager
 from dbt.contracts.connection import Connection
 from dbt.contracts.graph.manifest import Manifest
-from dbt.logger import GLOBAL_LOGGER as logger
+from dbt.logger import GLOBAL_LOGGER as logger, print_timestamped_line
 from dbt.utils import filter_null_values
 
 import google.auth
@@ -454,8 +454,9 @@ def _bq_table_to_relation(self, bq_table):
     @classmethod
     def warning_on_hooks(hook_type):
         msg = "{} is not supported in bigquery and will be ignored"
-        dbt.ui.printer.print_timestamped_line(msg.format(hook_type),
-                                              dbt.ui.printer.COLOR_FG_YELLOW)
+        print_timestamped_line(
+            msg.format(hook_type), ui.COLOR_FG_YELLOW
+        )
 
     @available
     def add_query(self, sql, auto_begin=True, bindings=None,
diff --git a/test/unit/test_main.py b/test/unit/test_main.py
index 651e2950cdd..ac15d1b7d27 100644
--- a/test/unit/test_main.py
+++ b/test/unit/test_main.py
@@ -16,7 +16,7 @@ def __init__(self, profiles_dir):
         self.profile = 'test'
 
 
-@mock.patch('dbt.ui.printer.use_colors')
+@mock.patch('dbt.ui.use_colors')
 @mock.patch('dbt.tracking.do_not_track')
 @mock.patch('dbt.tracking.initialize_tracking')
 class TestInitializeConfig(unittest.TestCase):
diff --git a/test/unit/utils.py b/test/unit/utils.py
index a3f198ab7c4..784b3803e5c 100644
--- a/test/unit/utils.py
+++ b/test/unit/utils.py
@@ -40,7 +40,7 @@ def profile_from_dict(profile, profile_name, cli_vars='{}'):
     from dbt.config import Profile
     from dbt.config.renderer import ProfileRenderer
     from dbt.context.base import generate_base_context
-    from dbt.utils import parse_cli_vars
+    from dbt.config.utils import parse_cli_vars
     if not isinstance(cli_vars, dict):
         cli_vars = parse_cli_vars(cli_vars)
 
@@ -56,7 +56,7 @@ def project_from_dict(project, profile, packages=None, cli_vars='{}'):
     from dbt.context.target import generate_target_context
     from dbt.config import Project
     from dbt.config.renderer import DbtProjectYamlRenderer
-    from dbt.utils import parse_cli_vars
+    from dbt.config.utils import parse_cli_vars
     if not isinstance(cli_vars, dict):
         cli_vars = parse_cli_vars(cli_vars)