diff --git a/dbt/adapters/bigquery/impl.py b/dbt/adapters/bigquery/impl.py
index f0722e40d7b..3f59636bb65 100644
--- a/dbt/adapters/bigquery/impl.py
+++ b/dbt/adapters/bigquery/impl.py
@@ -41,12 +41,12 @@ class BigQueryAdapter(PostgresAdapter):
         "expand_target_column_types",
         "load_dataframe",
         "get_missing_columns",
+        "cache_new_relation",
 
         "create_schema",
         "alter_table_add_columns",
 
         # versions of adapter functions that take / return Relations
-        "list_relations",
         "get_relation",
         "drop_relation",
         "rename_relation",
@@ -180,7 +180,10 @@ def close(cls, connection):
 
         return connection
 
-    def list_relations(self, schema, model_name=None):
+    def _link_cached_relations(self, manifest, schemas):
+        pass
+
+    def _list_relations(self, schema, model_name=None):
         connection = self.get_connection(model_name)
         client = connection.handle
 
@@ -201,27 +204,27 @@ def list_relations(self, schema, model_name=None):
         # This will 404 if the dataset does not exist. This behavior mirrors
         # the implementation of list_relations for other adapters
         try:
-            return [self.bq_table_to_relation(table) for table in all_tables]
+            return [self._bq_table_to_relation(table) for table in all_tables]
         except google.api_core.exceptions.NotFound as e:
             return []
 
-    def get_relation(self, schema=None, identifier=None,
-                     relations_list=None, model_name=None):
-        if schema is None and relations_list is None:
-            raise dbt.exceptions.RuntimeException(
-                'get_relation needs either a schema to query, or a list '
-                'of relations to use')
-
-        if relations_list is None and identifier is not None:
-            table = self.get_bq_table(schema, identifier)
+    def get_relation(self, schema, identifier, model_name=None):
+        if self._schema_is_cached(schema, model_name):
+            # if it's in the cache, use the parent's model of going through
+            # the relations cache and picking out the relation
+            return super(BigQueryAdapter, self).get_relation(
+                schema=schema,
+                identifier=identifier,
+                model_name=model_name
+            )
 
-            return self.bq_table_to_relation(table)
-
-        return super(BigQueryAdapter, self).get_relation(
-            schema, identifier, relations_list,
-            model_name)
+        table = self._get_bq_table(schema, identifier)
+        return self._bq_table_to_relation(table)
 
     def drop_relation(self, relation, model_name=None):
+        if self._schema_is_cached(relation.schema, model_name):
+            self.cache.drop(relation)
+
         conn = self.get_connection(model_name)
         client = conn.handle
 
@@ -515,7 +518,7 @@ def get_dataset(self, dataset_name, model_name=None):
         dataset_ref = conn.handle.dataset(dataset_name)
         return google.cloud.bigquery.Dataset(dataset_ref)
 
-    def bq_table_to_relation(self, bq_table):
+    def _bq_table_to_relation(self, bq_table):
         if bq_table is None:
             return None
 
@@ -529,7 +532,7 @@ def bq_table_to_relation(self, bq_table):
             },
             type=self.RELATION_TYPES.get(bq_table.table_type))
 
-    def get_bq_table(self, dataset_name, identifier, model_name=None):
+    def _get_bq_table(self, dataset_name, identifier, model_name=None):
         conn = self.get_connection(model_name)
 
         dataset = self.get_dataset(dataset_name, model_name)
diff --git a/dbt/adapters/cache.py b/dbt/adapters/cache.py
new file mode 100644
index 00000000000..4c562180ab7
--- /dev/null
+++ b/dbt/adapters/cache.py
@@ -0,0 +1,430 @@
+from collections import namedtuple
+import threading
+from copy import deepcopy
+import pprint
+from dbt.logger import CACHE_LOGGER as logger
+import dbt.exceptions
+
+
+_ReferenceKey = namedtuple('_ReferenceKey', 'schema identifier')
+
+
+def dot_separated(key):
+    """Return the key in dot-separated string form.
+
+    :param key _ReferenceKey: The key to stringify.
+    """
+    return '.'.join(key)
+
+
+class _CachedRelation(object):
+    """Nothing about _CachedRelation is guaranteed to be thread-safe!
+
+    :attr str schema: The schema of this relation.
+    :attr str identifier: The identifier of this relation.
+    :attr Dict[_ReferenceKey, _CachedRelation] referenced_by: The relations
+        that refer to this relation.
+    :attr DefaultRelation inner: The underlying dbt relation.
+    """
+    def __init__(self, inner):
+        self.referenced_by = {}
+        self.inner = inner
+
+    def __str__(self):
+        return (
+            '_CachedRelation(schema={}, identifier={}, inner={})'
+        ).format(self.schema, self.identifier, self.inner)
+
+    @property
+    def schema(self):
+        return self.inner.schema
+
+    @property
+    def identifier(self):
+        return self.inner.identifier
+
+    def __copy__(self):
+        new = self.__class__(self.inner)
+        new.__dict__.update(self.__dict__)
+        return new
+
+    def __deepcopy__(self, memo):
+        new = self.__class__(self.inner.incorporate())
+        new.__dict__.update(self.__dict__)
+        new.referenced_by = deepcopy(self.referenced_by, memo)
+
+    def is_referenced_by(self, key):
+        return key in self.referenced_by
+
+    def key(self):
+        """Get the _ReferenceKey that represents this relation
+
+        :return _ReferenceKey: A key for this relation.
+        """
+        return _ReferenceKey(self.schema, self.identifier)
+
+    def add_reference(self, referrer):
+        """Add a reference from referrer to self, indicating that if this node
+        were drop...cascaded, the referrer would be dropped as well.
+
+        :param _CachedRelation referrer: The node that refers to this node.
+        """
+        self.referenced_by[referrer.key()] = referrer
+
+    def collect_consequences(self):
+        """Recursively collect a set of _ReferenceKeys that would
+        consequentially get dropped if this were dropped via
+        "drop ... cascade".
+
+        :return Set[_ReferenceKey]: All the relations that would be dropped
+        """
+        consequences = {self.key()}
+        for relation in self.referenced_by.values():
+            consequences.update(relation.collect_consequences())
+        return consequences
+
+    def release_references(self, keys):
+        """Non-recursively indicate that an iterable of _ReferenceKey no longer
+        exist. Unknown keys are ignored.
+
+        :param Iterable[_ReferenceKey] keys: The keys to drop.
+        """
+        keys = set(self.referenced_by) & set(keys)
+        for key in keys:
+            self.referenced_by.pop(key)
+
+    def rename(self, new_relation):
+        """Rename this cached relation to new_relation.
+        Note that this will change the output of key(), all refs must be
+        updated!
+
+        :param _ReferenceKey new_relation: The new name to apply to the
+            relation
+        """
+        # Relations store this stuff inside their `path` dict. But they
+        # also store a table_name, and usually use it in their  .render(),
+        # so we need to update that as well. It doesn't appear that
+        # table_name is ever anything but the identifier (via .create())
+        self.inner = self.inner.incorporate(
+            path={
+                'schema': new_relation.schema,
+                'identifier': new_relation.identifier
+            },
+            table_name=new_relation.identifier
+        )
+
+    def rename_key(self, old_key, new_key):
+        """Rename a reference that may or may not exist. Only handles the
+        reference itself, so this is the other half of what `rename` does.
+
+        If old_key is not in referenced_by, this is a no-op.
+
+        :param _ReferenceKey old_key: The old key to be renamed.
+        :param _ReferenceKey new_key: The new key to rename to.
+        :raises InternalError: If the new key already exists.
+        """
+        if new_key in self.referenced_by:
+            dbt.exceptions.raise_cache_inconsistent(
+                'in rename of "{}" -> "{}", new name is in the cache already'
+                .format(old_key, new_key)
+            )
+
+        if old_key not in self.referenced_by:
+            return
+        value = self.referenced_by.pop(old_key)
+        self.referenced_by[new_key] = value
+
+    def dump_graph_entry(self):
+        """Return a key/value pair representing this key and its referents.
+
+        return List[str]: The dot-separated form of all referent keys.
+        """
+        return [dot_separated(r) for r in self.referenced_by]
+
+
+class RelationsCache(object):
+    """A cache of the relations known to dbt. Keeps track of relationships
+    declared between tables and handles renames/drops as a real database would.
+
+    :attr Dict[_ReferenceKey, _CachedRelation] relations: The known relations.
+    :attr threading.RLock lock: The lock around relations, held during updates.
+        The adapters also hold this lock while filling the cache.
+    :attr Set[str] schemas: The set of known/cached schemas, all lowercased.
+    """
+    def __init__(self):
+        self.relations = {}
+        self.lock = threading.RLock()
+        self.schemas = set()
+
+    def add_schema(self, schema):
+        """Add a schema to the set of known schemas (case-insensitive)
+
+        :param str schema: The schema name to add.
+        """
+        self.schemas.add(schema.lower())
+
+    def update_schemas(self, schemas):
+        """Add multiple schemas to the set of known schemas (case-insensitive)
+
+        :param Iterable[str] schemas: An iterable of the schema names to add.
+        """
+        self.schemas.update(s.lower() for s in schemas)
+
+    def __contains__(self, schema):
+        """A schema is 'in' the relations cache if it is in the set of cached
+        schemas.
+
+        :param str schema: The schema name to look up.
+        """
+        return schema in self.schemas
+
+    def dump_graph(self):
+        """Dump a key-only representation of the schema to a dictionary. Every
+        known relation is a key with a value of a list of keys it is referenced
+        by.
+        """
+        # we have to hold the lock for the entire dump, if other threads modify
+        # self.relations or any cache entry's referenced_by during iteration
+        # it's a runtime error!
+        with self.lock:
+            return {
+                dot_separated(k): v.dump_graph_entry()
+                for k, v in self.relations.items()
+            }
+
+    def _setdefault(self, relation):
+        """Add a relation to the cache, or return it if it already exists.
+
+        :param _CachedRelation relation: The relation to set or get.
+        :return _CachedRelation: The relation stored under the given relation's
+            key
+        """
+        self.schemas.add(relation.schema)
+        key = relation.key()
+        return self.relations.setdefault(key, relation)
+
+    def _add_link(self, referenced_key, dependent_key):
+        """Add a link between two relations to the database. Both the old and
+        new entries must alraedy exist in the database.
+
+        :param _ReferenceKey referenced_key: The key identifying the referenced
+            model (the one that if dropped will drop the dependent model).
+        :param _ReferenceKey dependent_key: The key identifying the dependent
+            model.
+        :raises InternalError: If either entry does not exist.
+        """
+        referenced = self.relations.get(referenced_key)
+        if referenced is None:
+            dbt.exceptions.raise_cache_inconsistent(
+                'in add_link, referenced link key {} not in cache!'
+                .format(referenced_key)
+            )
+
+        dependent = self.relations.get(dependent_key)
+        if dependent is None:
+            dbt.exceptions.raise_cache_inconsistent(
+                'in add_link, dependent link key {} not in cache!'
+                .format(dependent_key)
+            )
+
+        referenced.add_reference(dependent)
+
+    def add_link(self, referenced, dependent):
+        """Add a link between two relations to the database. Both the old and
+        new entries must already exist in the database.
+
+        The dependent model refers _to_ the referenced model. So, given
+        arguments of (jake_test, bar, jake_test, foo):
+        both values are in the schema jake_test and foo is a view that refers
+        to bar, so "drop bar cascade" will drop foo and all of foo's
+        dependents.
+
+        :param DefaultRelation referenced: The referenced model.
+        :param DefaultRelation dependent: The dependent model.
+        :raises InternalError: If either entry does not exist.
+        """
+        referenced = _ReferenceKey(
+            schema=referenced.schema,
+            identifier=referenced.name
+        )
+        if referenced.schema not in self:
+            # if we have not cached the referenced schema at all, we must be
+            # referring to a table outside our control. There's no need to make
+            # a link - we will never drop the referenced relation during a run.
+            logger.debug(
+                '{dep!s} references {ref!s} but {ref.schema} is not in the '
+                'cache, skipping assumed external relation'
+                .format(dep=dependent, ref=referenced)
+            )
+            return
+        dependent = _ReferenceKey(
+            schema=dependent.schema,
+            identifier=dependent.name
+        )
+        logger.debug(
+            'adding link, {!s} references {!s}'.format(dependent, referenced)
+        )
+        with self.lock:
+            self._add_link(referenced, dependent)
+
+    def add(self, relation):
+        """Add the relation inner to the cache, under the schema schema and
+        identifier identifier
+
+        :param DefaultRelation relation: The underlying relation.
+        """
+        cached = _CachedRelation(relation)
+        logger.debug('Adding relation: {!s}'.format(cached))
+        logger.debug('before adding: {}'.format(
+            pprint.pformat(self.dump_graph()))
+        )
+        with self.lock:
+            self._setdefault(cached)
+        logger.debug('after adding: {}'.format(
+            pprint.pformat(self.dump_graph()))
+        )
+
+    def _remove_refs(self, keys):
+        """Removes all references to all entries in keys. This does not
+        cascade!
+
+        :param Iterable[_ReferenceKey] keys: The keys to remove.
+        """
+        # remove direct refs
+        for key in keys:
+            del self.relations[key]
+        # then remove all entries from each child
+        for cached in self.relations.values():
+            cached.release_references(keys)
+
+    def _drop_cascade_relation(self, dropped):
+        """Drop the given relation and cascade it appropriately to all
+        dependent relations.
+
+        :param _CachedRelation dropped: An existing _CachedRelation to drop.
+        """
+        if dropped not in self.relations:
+            logger.debug('dropped a nonexistent relationship: {!s}'
+                         .format(dropped))
+            return
+        consequences = self.relations[dropped].collect_consequences()
+        logger.debug(
+            'drop {} is cascading to {}'.format(dropped, consequences)
+        )
+        self._remove_refs(consequences)
+
+    def drop(self, relation):
+        """Drop the named relation and cascade it appropriately to all
+        dependent relations.
+
+        Because dbt proactively does many `drop relation if exist ... cascade`
+        that are noops, nonexistent relation drops cause a debug log and no
+        other actions.
+
+        :param str schema: The schema of the relation to drop.
+        :param str identifier: The identifier of the relation to drop.
+        """
+        dropped = _ReferenceKey(schema=relation.schema,
+                                identifier=relation.identifier)
+        logger.debug('Dropping relation: {!s}'.format(dropped))
+        with self.lock:
+            self._drop_cascade_relation(dropped)
+
+    def _rename_relation(self, old_key, new_key):
+        """Rename a relation named old_key to new_key, updating references.
+        If the new key is already present, that is an error.
+        If the old key is absent, we only debug log and return, assuming it's a
+        temp table being renamed.
+
+        :param _ReferenceKey old_key: The existing key, to rename from.
+        :param _ReferenceKey new_key: The new key, to rename to.
+        :raises InternalError: If the new key is already present.
+        """
+        if old_key not in self.relations:
+            logger.debug(
+                'old key {} not found in self.relations, assuming temporary'
+                .format(old_key)
+            )
+            return
+
+        if new_key in self.relations:
+            dbt.exceptions.raise_cache_inconsistent(
+                'in rename, new key {} already in cache: {}'
+                .format(new_key, list(self.relations.keys()))
+            )
+
+        # On the database level, a rename updates all values that were
+        # previously referenced by old_name to be referenced by new_name.
+        # basically, the name changes but some underlying ID moves. Kind of
+        # like an object reference!
+        relation = self.relations.pop(old_key)
+
+        relation.rename(new_key)
+        # update all the relations that refer to it
+        for cached in self.relations.values():
+            if cached.is_referenced_by(old_key):
+                logger.debug(
+                    'updated reference from {0} -> {2} to {1} -> {2}'
+                    .format(old_key, new_key, cached.key())
+                )
+                cached.rename_key(old_key, new_key)
+
+        self.relations[new_key] = relation
+
+    def rename(self, old, new):
+        """Rename the old schema/identifier to the new schema/identifier and
+        update references.
+
+        If the new schema/identifier is already present, that is an error.
+        If the schema/identifier key is absent, we only debug log and return,
+        assuming it's a temp table being renamed.
+
+        :param DefaultRelation old: The existing relation name information.
+        :param DefaultRelation new: The new relation name information.
+        :raises InternalError: If the new key is already present.
+        """
+        old_key = _ReferenceKey(
+            schema=old.schema,
+            identifier=old.identifier
+        )
+        new_key = _ReferenceKey(
+            schema=new.schema,
+            identifier=new.identifier
+        )
+        logger.debug('Renaming relation {!s} to {!s}'.format(
+            old_key, new_key)
+        )
+        logger.debug('before rename: {}'.format(
+            pprint.pformat(self.dump_graph()))
+        )
+        with self.lock:
+            self._rename_relation(old_key, new_key)
+        logger.debug('after rename: {}'.format(
+            pprint.pformat(self.dump_graph()))
+        )
+
+    def get_relations(self, schema):
+        """Case-insensitively yield all relations matching the given schema.
+
+        :param str schema: The case-insensitive schema name to list from.
+        :return List[DefaultRelation]: The list of relations with the given
+            schema
+        """
+        schema = schema.lower()
+        with self.lock:
+            results = [
+                r.inner for r in self.relations.values()
+                if r.schema.lower() == schema
+            ]
+
+        if None in results:
+            dbt.exceptions.raise_cache_inconsistent(
+                'in get_relations, a None relation was found in the cache!'
+            )
+        return results
+
+    def clear(self):
+        """Clear the cache"""
+        with self.lock:
+            self.relations.clear()
+            self.schemas.clear()
diff --git a/dbt/adapters/default/impl.py b/dbt/adapters/default/impl.py
index 8fe29e4a684..004b68f51db 100644
--- a/dbt/adapters/default/impl.py
+++ b/dbt/adapters/default/impl.py
@@ -16,6 +16,7 @@
 from dbt.utils import filter_null_values
 
 from dbt.adapters.default.relation import DefaultRelation
+from dbt.adapters.cache import RelationsCache
 
 GET_CATALOG_OPERATION_NAME = 'get_catalog_data'
 
@@ -24,25 +25,38 @@
 connections_available = []
 
 
-def _filter_schemas(manifest):
+def _expect_row_value(key, row):
+    if key not in row.keys():
+        raise dbt.exceptions.InternalException(
+            'Got a row without "{}" column, columns: {}'
+            .format(key, row.keys())
+        )
+    return row[key]
+
+
+def _relations_filter_schemas(schemas):
+    def test(row):
+        referenced_schema = _expect_row_value('referenced_schema', row)
+        dependent_schema = _expect_row_value('dependent_schema', row)
+        # handle the null schema
+        if referenced_schema is not None:
+            referenced_schema = referenced_schema.lower()
+        if dependent_schema is not None:
+            dependent_schema = dependent_schema.lower()
+        return referenced_schema in schemas or dependent_schema in schemas
+    return test
+
+
+def _catalog_filter_schemas(manifest):
     """Return a function that takes a row and decides if the row should be
     included in the catalog output.
     """
-    schemas = frozenset({
-        node.schema.lower()
-        for node in manifest.nodes.values()
-    })
+    schemas = frozenset(s.lower() for s in manifest.get_used_schemas())
 
     def test(row):
-        if 'table_schema' not in row.keys():
-            # this means the get catalog operation is somehow not well formed!
-            raise dbt.exceptions.InternalException(
-                'Got a row without "table_schema" column, columns: {}'
-                .format(row.keys())
-            )
+        table_schema = _expect_row_value('table_schema', row)
         # the schema may be present but None, which is not an error and should
         # be filtered out
-        table_schema = row['table_schema']
         if table_schema is None:
             return False
         return table_schema.lower() in schemas
@@ -58,6 +72,7 @@ class DefaultAdapter(object):
         "expand_target_column_types",
         "create_schema",
         "quote_as_configured",
+        "cache_new_relation",
 
         # deprecated -- use versions that take relations instead
         "already_exists",
@@ -70,7 +85,6 @@ class DefaultAdapter(object):
         "quote_schema_and_table",
 
         # versions of adapter functions that take / return Relations
-        "list_relations",
         "get_relation",
         "drop_relation",
         "rename_relation",
@@ -85,13 +99,14 @@ class DefaultAdapter(object):
         "get_status",
         "get_result_from_cursor",
         "quote",
-        "convert_type"
+        "convert_type",
     ]
     Relation = DefaultRelation
     Column = Column
 
     def __init__(self, config):
         self.config = config
+        self.cache = RelationsCache()
 
     ###
     # ADAPTER-SPECIFIC FUNCTIONS -- each of these must be overridden in
@@ -130,8 +145,7 @@ def query_for_existing(self, schemas, model_name=None):
         all_relations = []
 
         for schema in schemas:
-            all_relations.extend(
-                self.list_relations(schema, model_name))
+            all_relations.extend(self.list_relations(schema, model_name))
 
         return {relation.identifier: relation.type
                 for relation in all_relations}
@@ -151,6 +165,17 @@ def cancel_connection(self, connection):
     ###
     # FUNCTIONS THAT SHOULD BE ABSTRACT
     ###
+    def cache_new_relation(self, relation, model_name=None):
+        """Cache a new relation in dbt. It will show up in `list relations`."""
+        if relation is None:
+            dbt.exceptions.raise_compiler_error(
+                'Attempted to cache a null relation for {}'.format(model_name)
+            )
+        if dbt.flags.USE_CACHE:
+            self.cache.add(relation)
+        # so jinja doesn't render things
+        return ''
+
     @classmethod
     def get_result_from_cursor(cls, cursor):
         data = []
@@ -175,6 +200,8 @@ def drop(self, schema, relation, relation_type, model_name=None):
         return self.drop_relation(relation, model_name)
 
     def drop_relation(self, relation, model_name=None):
+        if dbt.flags.USE_CACHE:
+            self.cache.drop(relation)
         if relation.type is None:
             dbt.exceptions.raise_compiler_error(
                 'Tried to drop relation {}, but its type is null.'
@@ -216,6 +243,8 @@ def rename(self, schema, from_name, to_name, model_name=None):
 
     def rename_relation(self, from_relation, to_relation,
                         model_name=None):
+        if dbt.flags.USE_CACHE:
+            self.cache.rename(from_relation, to_relation)
         sql = 'alter table {} rename to {}'.format(
             from_relation, to_relation.include(schema=False))
 
@@ -324,10 +353,37 @@ def expand_target_column_types(self,
     ###
     # RELATIONS
     ###
-    def list_relations(self, schema, model_name=None):
+    def _schema_is_cached(self, schema, model_name=None,
+                          debug_on_missing=True):
+        """Check if the schema is cached, and by default logs if it is not."""
+        if dbt.flags.USE_CACHE is False:
+            return False
+        elif schema not in self.cache:
+            if debug_on_missing:
+                logger.debug(
+                    'On "{}": cache miss for schema "{}", this is inefficient'
+                    .format(model_name or '<None>', schema)
+                )
+            return False
+        else:
+            return True
+
+    def _list_relations(self, schema, model_name=None):
         raise dbt.exceptions.NotImplementedException(
             '`list_relations` is not implemented for this adapter!')
 
+    def list_relations(self, schema, model_name=None):
+        if self._schema_is_cached(schema, model_name):
+            return self.cache.get_relations(schema)
+
+        # we can't build the relations cache because we don't have a
+        # manifest so we can't run any operations.
+        relations = self._list_relations(schema, model_name=model_name)
+
+        logger.debug('with schema={}, model_name={}, relations={}'
+                     .format(schema, model_name, relations))
+        return relations
+
     def _make_match_kwargs(self, schema, identifier):
         quoting = self.config.quoting
         if identifier is not None and quoting['identifier'] is False:
@@ -339,15 +395,7 @@ def _make_match_kwargs(self, schema, identifier):
         return filter_null_values({'identifier': identifier,
                                    'schema': schema})
 
-    def get_relation(self, schema=None, identifier=None,
-                     relations_list=None, model_name=None):
-        if schema is None and relations_list is None:
-            raise dbt.exceptions.RuntimeException(
-                'get_relation needs either a schema to query, or a list '
-                'of relations to use')
-
-        if relations_list is None:
-            relations_list = self.list_relations(schema, model_name)
+    def _make_match(self, relations_list, schema, identifier):
 
         matches = []
 
@@ -357,6 +405,13 @@ def get_relation(self, schema=None, identifier=None,
             if relation.matches(**search):
                 matches.append(relation)
 
+        return matches
+
+    def get_relation(self, schema, identifier, model_name=None):
+        relations_list = self.list_relations(schema, model_name)
+
+        matches = self._make_match(relations_list, schema, identifier)
+
         if len(matches) > 1:
             dbt.exceptions.get_relation_returned_multiple_results(
                 {'identifier': identifier, 'schema': schema}, matches)
@@ -778,15 +833,56 @@ def run_operation(self, manifest, operation_name):
     # Abstract methods involving the manifest
     ###
     @classmethod
-    def _filter_table(cls, table, manifest):
-        return table.where(_filter_schemas(manifest))
+    def _catalog_filter_table(cls, table, manifest):
+        return table.where(_catalog_filter_schemas(manifest))
 
     def get_catalog(self, manifest):
         try:
-            table = self.run_operation(manifest,
-                                       GET_CATALOG_OPERATION_NAME)
+            table = self.run_operation(manifest, GET_CATALOG_OPERATION_NAME)
         finally:
             self.release_connection(GET_CATALOG_OPERATION_NAME)
 
-        results = self._filter_table(table, manifest)
+        results = self._catalog_filter_table(table, manifest)
         return results
+
+    @classmethod
+    def _relations_filter_table(cls, table, schemas):
+        return table.where(_relations_filter_schemas(schemas))
+
+    def _link_cached_relations(self, manifest, schemas):
+        """This method has to exist because BigQueryAdapter and SnowflakeAdapter
+        inherit from the PostgresAdapter, so they need something to override
+        in order to disable linking.
+        """
+        pass
+
+    def _relations_cache_for_schemas(self, manifest, schemas=None):
+        if not dbt.flags.USE_CACHE:
+            return
+
+        if schemas is None:
+            schemas = manifest.get_used_schemas()
+
+        relations = []
+        # add all relations
+        for schema in schemas:
+            # bypass the cache, of course!
+            for relation in self._list_relations(schema):
+                self.cache.add(relation)
+        self._link_cached_relations(manifest, schemas)
+        # it's possible that there were no relations in some schemas. We want
+        # to insert the schemas we query into the cache's `.schemas` attribute
+        # so we can check it later
+        self.cache.update_schemas(schemas)
+
+    def set_relations_cache(self, manifest, clear=False):
+        """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:
+            return
+
+        with self.cache.lock:
+            if clear:
+                self.cache.clear()
+            self._relations_cache_for_schemas(manifest)
diff --git a/dbt/adapters/postgres/impl.py b/dbt/adapters/postgres/impl.py
index 29355df5a76..a0a03967931 100644
--- a/dbt/adapters/postgres/impl.py
+++ b/dbt/adapters/postgres/impl.py
@@ -10,6 +10,9 @@
 from dbt.logger import GLOBAL_LOGGER as logger
 
 
+GET_RELATIONS_OPERATION_NAME = 'get_relations_data'
+
+
 class PostgresAdapter(dbt.adapters.default.DefaultAdapter):
 
     DEFAULT_TCP_KEEPALIVE = 0  # 0 means to use the default value
@@ -143,7 +146,21 @@ def alter_column_type(self, schema, table, column_name,
 
         return connection, cursor
 
-    def list_relations(self, schema, model_name=None):
+    def _link_cached_relations(self, manifest, schemas):
+        try:
+            table = self.run_operation(manifest, GET_RELATIONS_OPERATION_NAME)
+        finally:
+            self.release_connection(GET_RELATIONS_OPERATION_NAME)
+        table = self._relations_filter_table(table, schemas)
+
+        for (refed_schema, refed_name, dep_schema, dep_name) in table:
+            referenced = self.Relation.create(schema=refed_schema,
+                                              identifier=refed_name)
+            dependent = self.Relation.create(schema=dep_schema,
+                                             identifier=dep_name)
+            self.cache.add_link(dependent, referenced)
+
+    def _list_relations(self, schema, model_name=None):
         sql = """
         select tablename as name, schemaname as schema, 'table' as type from pg_tables
         where schemaname ilike '{schema}'
@@ -152,8 +169,7 @@ def list_relations(self, schema, model_name=None):
         where schemaname ilike '{schema}'
         """.format(schema=schema).strip()  # noqa
 
-        connection, cursor = self.add_query(sql, model_name,
-                                            auto_begin=False)
+        connection, cursor = self.add_query(sql, model_name, auto_begin=False)
 
         results = cursor.fetchall()
 
diff --git a/dbt/adapters/snowflake/impl.py b/dbt/adapters/snowflake/impl.py
index 20d088e59c4..99c874aed77 100644
--- a/dbt/adapters/snowflake/impl.py
+++ b/dbt/adapters/snowflake/impl.py
@@ -103,7 +103,10 @@ def open_connection(cls, connection):
 
         return connection
 
-    def list_relations(self, schema, model_name=None):
+    def _link_cached_relations(self, manifest, schemas):
+        pass
+
+    def _list_relations(self, schema, model_name=None):
         sql = """
         select
           table_name as name, table_schema as schema, table_type as type
@@ -133,6 +136,7 @@ def list_relations(self, schema, model_name=None):
 
     def rename_relation(self, from_relation, to_relation,
                         model_name=None):
+        self.cache.rename(from_relation, to_relation)
         sql = 'alter table {} rename to {}'.format(
             from_relation, to_relation)
 
@@ -209,13 +213,14 @@ def add_query(self, sql, model_name=None, auto_begin=True,
         return connection, cursor
 
     @classmethod
-    def _filter_table(cls, table, manifest):
+    def _catalog_filter_table(cls, table, manifest):
         # On snowflake, users can set QUOTED_IDENTIFIERS_IGNORE_CASE, so force
         # the column names to their lowercased forms.
         lowered = table.rename(
             column_names=[c.lower() for c in table.column_names]
         )
-        return super(SnowflakeAdapter, cls)._filter_table(lowered, manifest)
+        return super(SnowflakeAdapter, cls)._catalog_filter_table(lowered,
+                                                                  manifest)
 
     def _make_match_kwargs(self, schema, identifier):
         quoting = self.config.quoting
diff --git a/dbt/clients/jinja.py b/dbt/clients/jinja.py
index fa2fea4ecd7..fac6580db5d 100644
--- a/dbt/clients/jinja.py
+++ b/dbt/clients/jinja.py
@@ -1,3 +1,7 @@
+import codecs
+import linecache
+import os
+
 import jinja2
 import jinja2._compat
 import jinja2.ext
@@ -37,13 +41,28 @@ def _parse(self, source, name, filename):
             jinja2._compat.encode_filename(filename)
         ).parse()
 
+    def _compile(self, source, filename):
+        """Override jinja's compilation to stash the rendered source inside
+        the python linecache for debugging.
+        """
+        if filename == '<template>':
+            # make a better filename
+            filename = 'dbt-{}'.format(
+                codecs.encode(os.urandom(12), 'hex').decode('ascii')
+            )
+            # encode, though I don't think this matters
+            filename = jinja2._compat.encode_filename(filename)
+            # put ourselves in the cache using the 'lazycache' method
+            linecache.cache[filename] = (lambda: source,)
+
+        return super(MacroFuzzEnvironment, self)._compile(source, filename)
+
 
 def macro_generator(template, node):
     def apply_context(context):
         def call(*args, **kwargs):
             name = node.get('name')
-            module = template.make_module(
-                context, False, context)
+            module = template.make_module(context, False, context)
 
             if node['resource_type'] == NodeType.Operation:
                 macro = module.__dict__[dbt.utils.get_dbt_operation_name(name)]
@@ -188,7 +207,8 @@ def get_template(string, ctx, node=None, capture_macros=False):
 
         env = MacroFuzzEnvironment(**args)
 
-        return env.from_string(dbt.compat.to_string(string), globals=ctx)
+        template_source = dbt.compat.to_string(string)
+        return env.from_string(template_source, globals=ctx)
 
     except (jinja2.exceptions.TemplateSyntaxError,
             jinja2.exceptions.UndefinedError) as e:
diff --git a/dbt/compat.py b/dbt/compat.py
index 0e450552326..7f89e6d70c9 100644
--- a/dbt/compat.py
+++ b/dbt/compat.py
@@ -1,5 +1,6 @@
 import codecs
 import json
+import warnings
 
 WHICH_PYTHON = None
 
@@ -51,3 +52,12 @@ def write_file(path, s):
     else:
         with open(path, 'w') as f:
             return f.write(to_string(s))
+
+
+def suppress_warnings():
+    # in python 2, ResourceWarnings don't exist.
+    # in python 3, suppress ResourceWarnings about unclosed sockets, as the
+    # bigquery library never closes them.
+    if WHICH_PYTHON == 3:
+        warnings.filterwarnings("ignore", category=ResourceWarning,
+                                message="unclosed.*<socket.socket.*>")
diff --git a/dbt/contracts/graph/manifest.py b/dbt/contracts/graph/manifest.py
index 9f99e0b7a8c..c72a54b77e2 100644
--- a/dbt/contracts/graph/manifest.py
+++ b/dbt/contracts/graph/manifest.py
@@ -370,3 +370,9 @@ def __getattr__(self, name):
         raise AttributeError("'{}' object has no attribute '{}'".format(
             type(self).__name__, name)
         )
+
+    def get_used_schemas(self):
+        return frozenset({
+            node.schema
+            for node in self.nodes.values()
+        })
diff --git a/dbt/exceptions.py b/dbt/exceptions.py
index 511be3fc427..622db83a5d9 100644
--- a/dbt/exceptions.py
+++ b/dbt/exceptions.py
@@ -297,6 +297,10 @@ def bad_package_spec(repo, spec, error_message):
             spec, repo, error_message))
 
 
+def raise_cache_inconsistent(message):
+    raise InternalException('Cache inconsistency detected: {}'.format(message))
+
+
 def missing_config(model, name):
     raise_compiler_error(
         "Model '{}' does not define a required config parameter '{}'."
diff --git a/dbt/flags.py b/dbt/flags.py
index b1ec639aa97..bc909d5d64e 100644
--- a/dbt/flags.py
+++ b/dbt/flags.py
@@ -1,11 +1,15 @@
 STRICT_MODE = False
 NON_DESTRUCTIVE = False
 FULL_REFRESH = False
+LOG_CACHE_EVENTS = False
+USE_CACHE = True
 
 
 def reset():
-    global STRICT_MODE, NON_DESTRUCTIVE, FULL_REFRESH
+    global STRICT_MODE, NON_DESTRUCTIVE, FULL_REFRESH, LOG_CACHE_EVENTS
 
     STRICT_MODE = False
     NON_DESTRUCTIVE = False
     FULL_REFRESH = False
+    LOG_CACHE_EVENTS = False
+    USE_CACHE = True
diff --git a/dbt/include/global_project/macros/adapters/common.sql b/dbt/include/global_project/macros/adapters/common.sql
index 3ef6dc23341..843088e8045 100644
--- a/dbt/include/global_project/macros/adapters/common.sql
+++ b/dbt/include/global_project/macros/adapters/common.sql
@@ -33,6 +33,10 @@
 {% endmacro %}
 
 {% macro create_table_as(temporary, relation, sql) -%}
+  {%- if not temporary -%}
+    {{ adapter.cache_new_relation(relation) }}
+  {%- endif -%}
+
   {{ adapter_macro('create_table_as', temporary, relation, sql) }}
 {%- endmacro %}
 
@@ -46,6 +50,8 @@
 
 
 {% macro create_view_as(relation, sql) -%}
+  {{ adapter.cache_new_relation(relation) }}
+
   {{ adapter_macro('create_view_as', relation, sql) }}
 {%- endmacro %}
 
@@ -57,6 +63,8 @@
 
 
 {% macro create_archive_table(relation, columns) -%}
+  {{ adapter.cache_new_relation(relation) }}
+
   {{ adapter_macro('create_archive_table', relation, columns) }}
 {%- endmacro %}
 
@@ -80,3 +88,17 @@
 
   {{ exceptions.raise_compiler_error(msg) }}
 {% endmacro %}
+
+{% macro get_relations() -%}
+  {{ return(adapter_macro('get_relations')) }}
+{% endmacro %}
+
+
+{% macro default__get_relations() -%}
+  {% set typename = adapter.type() %}
+  {% set msg -%}
+    get_relations not implemented for {{ typename }}
+  {%- endset %}
+
+  {{ exceptions.raise_compiler_error(msg) }}
+{% endmacro %}
diff --git a/dbt/include/global_project/macros/materializations/incremental/bigquery_incremental.sql b/dbt/include/global_project/macros/materializations/incremental/bigquery_incremental.sql
index 0a75b9d1a79..e24a9caa132 100644
--- a/dbt/include/global_project/macros/materializations/incremental/bigquery_incremental.sql
+++ b/dbt/include/global_project/macros/materializations/incremental/bigquery_incremental.sql
@@ -13,9 +13,7 @@
 
   {%- set identifier = model['alias'] -%}
 
-  {%- set existing_relations = adapter.list_relations(schema=schema) -%}
-  {%- set old_relation = adapter.get_relation(relations_list=existing_relations,
-                                              schema=schema, identifier=identifier) -%}
+  {%- set old_relation = adapter.get_relation(schema=schema, identifier=identifier) -%}
 
   {%- set target_relation = api.Relation.create(identifier=identifier, schema=schema, type='table') -%}
 
diff --git a/dbt/include/global_project/macros/materializations/incremental/incremental.sql b/dbt/include/global_project/macros/materializations/incremental/incremental.sql
index 6d32e503631..8df0264a8a4 100644
--- a/dbt/include/global_project/macros/materializations/incremental/incremental.sql
+++ b/dbt/include/global_project/macros/materializations/incremental/incremental.sql
@@ -18,9 +18,7 @@
   {%- set identifier = model['alias'] -%}
   {%- set tmp_identifier = identifier + '__dbt_incremental_tmp' -%}
 
-  {%- set existing_relations = adapter.list_relations(schema=schema) -%}
-  {%- set old_relation = adapter.get_relation(relations_list=existing_relations,
-                                              schema=schema, identifier=identifier) -%}
+  {%- set old_relation = adapter.get_relation(schema=schema, identifier=identifier) -%}
   {%- set target_relation = api.Relation.create(identifier=identifier, schema=schema, type='table') -%}
   {%- set tmp_relation = api.Relation.create(identifier=tmp_identifier,
                                                  schema=schema, type='table') -%}
diff --git a/dbt/include/global_project/macros/materializations/seed/seed.sql b/dbt/include/global_project/macros/materializations/seed/seed.sql
index 08c06d7cde1..7754eebe055 100644
--- a/dbt/include/global_project/macros/materializations/seed/seed.sql
+++ b/dbt/include/global_project/macros/materializations/seed/seed.sql
@@ -14,6 +14,7 @@
 {% macro default__create_csv_table(model) %}
   {%- set agate_table = model['agate_table'] -%}
   {%- set column_override = model['config'].get('column_types', {}) -%}
+  {{ adapter.cache_new_relation(this) }}
 
   {% set sql %}
     create table {{ this.render(False) }} (
@@ -88,10 +89,8 @@
 
   {%- set identifier = model['alias'] -%}
   {%- set full_refresh_mode = (flags.FULL_REFRESH == True) -%}
-  {%- set existing_relations = adapter.list_relations(schema=schema) -%}
 
-  {%- set old_relation = adapter.get_relation(relations_list=existing_relations,
-                                              schema=schema, identifier=identifier) -%}
+  {%- set old_relation = adapter.get_relation(schema=schema, identifier=identifier) -%}
 
   {%- set exists_as_table = (old_relation is not none and old_relation.is_table) -%}
   {%- set exists_as_view = (old_relation is not none and old_relation.is_view) -%}
diff --git a/dbt/include/global_project/macros/materializations/table/snowflake_table.sql b/dbt/include/global_project/macros/materializations/table/snowflake_table.sql
index 5a753babfed..fa64dcab258 100644
--- a/dbt/include/global_project/macros/materializations/table/snowflake_table.sql
+++ b/dbt/include/global_project/macros/materializations/table/snowflake_table.sql
@@ -4,9 +4,7 @@
   {%- set backup_identifier = identifier + '__dbt_backup' -%}
   {%- set non_destructive_mode = (flags.NON_DESTRUCTIVE == True) -%}
 
-  {%- set existing_relations = adapter.list_relations(schema=schema) -%}
-  {%- set old_relation = adapter.get_relation(relations_list=existing_relations,
-                                              schema=schema, identifier=identifier) -%}
+  {%- set old_relation = adapter.get_relation(schema=schema, identifier=identifier) -%}
   {%- set target_relation = api.Relation.create(identifier=identifier,
                                                 schema=schema, type='table') -%}
   {%- set intermediate_relation = api.Relation.create(identifier=tmp_identifier,
diff --git a/dbt/include/global_project/macros/materializations/table/table.sql b/dbt/include/global_project/macros/materializations/table/table.sql
index c99df1121fe..24c21e992fe 100644
--- a/dbt/include/global_project/macros/materializations/table/table.sql
+++ b/dbt/include/global_project/macros/materializations/table/table.sql
@@ -4,9 +4,7 @@
   {%- set backup_identifier = identifier + '__dbt_backup' -%}
   {%- set non_destructive_mode = (flags.NON_DESTRUCTIVE == True) -%}
 
-  {%- set existing_relations = adapter.list_relations(schema=schema) -%}
-  {%- set old_relation = adapter.get_relation(relations_list=existing_relations,
-                                              schema=schema, identifier=identifier) -%}
+  {%- set old_relation = adapter.get_relation(schema=schema, identifier=identifier) -%}
   {%- set target_relation = api.Relation.create(identifier=identifier,
                                                 schema=schema, type='table') -%}
   {%- set intermediate_relation = api.Relation.create(identifier=tmp_identifier,
diff --git a/dbt/include/global_project/macros/materializations/view/view.sql b/dbt/include/global_project/macros/materializations/view/view.sql
index 489bf60fa63..dcbcbfea4a9 100644
--- a/dbt/include/global_project/macros/materializations/view/view.sql
+++ b/dbt/include/global_project/macros/materializations/view/view.sql
@@ -5,9 +5,7 @@
   {%- set backup_identifier = identifier + '__dbt_backup' -%}
   {%- set non_destructive_mode = (flags.NON_DESTRUCTIVE == True) -%}
 
-  {%- set existing_relations = adapter.list_relations(schema=schema) -%}
-  {%- set old_relation = adapter.get_relation(relations_list=existing_relations,
-                                              schema=schema, identifier=identifier) -%}
+  {%- set old_relation = adapter.get_relation(schema=schema, identifier=identifier) -%}
   {%- set target_relation = api.Relation.create(identifier=identifier, schema=schema,
                                                 type='view') -%}
   {%- set intermediate_relation = api.Relation.create(identifier=tmp_identifier,
diff --git a/dbt/include/global_project/macros/operations/relations/get_relations.sql b/dbt/include/global_project/macros/operations/relations/get_relations.sql
new file mode 100644
index 00000000000..ad943c2a84b
--- /dev/null
+++ b/dbt/include/global_project/macros/operations/relations/get_relations.sql
@@ -0,0 +1,4 @@
+{% operation get_relations_data %}
+    {% set relations = dbt.get_relations() %}
+    {{ return(relations) }}
+{% endoperation %}
diff --git a/dbt/include/global_project/macros/relations/postgres_relations.sql b/dbt/include/global_project/macros/relations/postgres_relations.sql
new file mode 100644
index 00000000000..e6ef11f93b0
--- /dev/null
+++ b/dbt/include/global_project/macros/relations/postgres_relations.sql
@@ -0,0 +1,64 @@
+{% macro postgres__get_relations () -%}
+  {%- call statement('relations', fetch_result=True) -%}
+    -- {#
+    -- in pg_depend, objid is the dependent, refobjid is the referenced object
+    -- "a pg_depend entry indicates that the referenced object cannot be dropped without also dropping the dependent object."
+    -- #}
+    with relation as (
+        select
+            pg_rewrite.ev_class as class,
+            pg_rewrite.oid as id
+        from pg_rewrite
+    ),
+    class as (
+        select
+            oid as id,
+            relname as name,
+            relnamespace as schema,
+            relkind as kind
+        from pg_class
+    ),
+    dependency as (
+        select
+            pg_depend.objid as id,
+            pg_depend.refobjid as ref
+        from pg_depend
+    ),
+    schema as (
+        select
+            pg_namespace.oid as id,
+            pg_namespace.nspname as name
+        from pg_namespace
+        where nspname != 'information_schema' and nspname not like 'pg_%'
+    ),
+    relationships as (
+        select
+            referenced_class.name as referenced_name,
+            referenced_class.schema as referenced_schema_id,
+            dependent_class.name as dependent_name,
+            dependent_class.schema as dependent_schema_id,
+            referenced_class.kind as kind
+        from relation
+        join class as referenced_class on relation.class=referenced_class.id
+        join dependency on relation.id=dependency.id
+        join class as dependent_class on dependency.ref=dependent_class.id
+        where
+            referenced_class.kind in ('r', 'v') and
+            (referenced_class.name != dependent_class.name or
+             referenced_class.schema != dependent_class.schema)
+    )
+
+    select
+        referenced_schema.name as referenced_schema,
+        relationships.referenced_name as referenced_name,
+        dependent_schema.name as dependent_schema,
+        relationships.dependent_name as dependent_name
+    from relationships
+    join schema as dependent_schema on relationships.dependent_schema_id=dependent_schema.id
+    join schema as referenced_schema on relationships.referenced_schema_id=referenced_schema.id
+    group by referenced_schema, referenced_name, dependent_schema, dependent_name
+    order by referenced_schema, referenced_name, dependent_schema, dependent_name;
+  {%- endcall -%}
+
+  {{ return(load_result('relations').table) }}
+{% endmacro %}
diff --git a/dbt/include/global_project/macros/relations/redshift_relations.sql b/dbt/include/global_project/macros/relations/redshift_relations.sql
new file mode 100644
index 00000000000..ed682ae7d63
--- /dev/null
+++ b/dbt/include/global_project/macros/relations/redshift_relations.sql
@@ -0,0 +1,3 @@
+{% macro redshift__get_relations () -%}
+  {{ return(dbt.postgres__get_relations()) }}
+{% endmacro %}
diff --git a/dbt/logger.py b/dbt/logger.py
index 58b9d2fb8dd..11f19ec0b07 100644
--- a/dbt/logger.py
+++ b/dbt/logger.py
@@ -1,8 +1,10 @@
 import dbt.compat
+import dbt.flags
 import logging
 import logging.handlers
 import os
 import sys
+import warnings
 
 import colorama
 
@@ -42,9 +44,13 @@
 logging.getLogger('snowflake.connector').setLevel(logging.INFO)
 logging.getLogger('parsedatetime').setLevel(logging.INFO)
 
+# provide this for the cache.
+CACHE_LOGGER = logging.getLogger('dbt.cache')
+
 # Redirect warnings through our logging setup
 # They will be logged to a file below
 logging.captureWarnings(True)
+dbt.compat.suppress_warnings()
 
 initialized = False
 
@@ -101,6 +107,8 @@ def initialize_logger(debug_mode=False, path=None):
         warning_logger.addHandler(logdir_handler)
         warning_logger.setLevel(logging.DEBUG)
 
+        CACHE_LOGGER.propagate = dbt.flags.LOG_CACHE_EVENTS
+
     initialized = True
 
 
diff --git a/dbt/main.py b/dbt/main.py
index 1f2ba98da53..a28cf825b59 100644
--- a/dbt/main.py
+++ b/dbt/main.py
@@ -251,6 +251,8 @@ def invoke_dbt(parsed):
         return None
 
     flags.NON_DESTRUCTIVE = getattr(parsed, 'non_destructive', False)
+    flags.LOG_CACHE_EVENTS = getattr(parsed, 'log_cache_events', False)
+    flags.USE_CACHE = getattr(parsed, 'use_cache', True)
 
     arg_drop_existing = getattr(parsed, 'drop_existing', False)
     arg_full_refresh = getattr(parsed, 'full_refresh', False)
@@ -331,6 +333,20 @@ def parse_args(args):
             should be a YAML string, eg. '{my_variable: my_value}'"""
     )
 
+    # if set, log all cache events. This is extremely verbose!
+    base_subparser.add_argument(
+        '--log-cache-events',
+        action='store_true',
+        help=argparse.SUPPRESS,
+    )
+
+    base_subparser.add_argument(
+        '--bypass-cache',
+        action='store_false',
+        dest='use_cache',
+        help='If set, bypass the adapter-level cache of database state',
+    )
+
     sub = subs.add_parser('init', parents=[base_subparser])
     sub.add_argument('project_name', type=str, help='Name of the new project')
     sub.set_defaults(cls=init_task.InitTask, which='init')
diff --git a/dbt/node_runners.py b/dbt/node_runners.py
index 20a25a835e1..78bfba7211f 100644
--- a/dbt/node_runners.py
+++ b/dbt/node_runners.py
@@ -345,8 +345,13 @@ def create_schemas(cls, config, adapter, manifest):
         for schema in (required_schemas - existing_schemas):
             adapter.create_schema(schema)
 
+    @classmethod
+    def populate_adapter_cache(cls, config, adapter, manifest):
+        adapter.set_relations_cache(manifest)
+
     @classmethod
     def before_run(cls, config, adapter, manifest):
+        cls.populate_adapter_cache(config, adapter, manifest)
         cls.safe_run_hooks(config, adapter, manifest, RunHookType.Start)
         cls.create_schemas(config, adapter, manifest)
 
diff --git a/test/integration/037_external_reference_test/models/my_model.sql b/test/integration/037_external_reference_test/models/my_model.sql
new file mode 100644
index 00000000000..5d10e607ed7
--- /dev/null
+++ b/test/integration/037_external_reference_test/models/my_model.sql
@@ -0,0 +1,7 @@
+{{
+  config(
+    materialized = "view"
+  )
+}}
+
+select * from "{{ this.schema + 'z' }}"."external"
diff --git a/test/integration/037_external_reference_test/test_external_reference.py b/test/integration/037_external_reference_test/test_external_reference.py
new file mode 100644
index 00000000000..5a71898e7ec
--- /dev/null
+++ b/test/integration/037_external_reference_test/test_external_reference.py
@@ -0,0 +1,39 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+
+class TestExternalReference(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "external_reference_037"
+
+    @property
+    def models(self):
+        return "test/integration/037_external_reference_test/models"
+
+    def setUp(self):
+        super(TestExternalReference, self).setUp()
+        self.use_default_project()
+        self.external_schema = self.unique_schema()+'z'
+        self.run_sql(
+            'create schema "{}"'.format(self.external_schema)
+        )
+        self.run_sql(
+            'create table "{}"."external" (id integer)'
+            .format(self.external_schema)
+        )
+        self.run_sql(
+            'insert into "{}"."external" values (1), (2)'
+            .format(self.external_schema)
+        )
+
+    def tearDown(self):
+        # This has to happen before we drop the external schema, because
+        # otherwise postgres hangs forever.
+        self._drop_schema()
+        self.adapter.drop_schema(self.external_schema, '__test')
+        super(TestExternalReference, self).tearDown()
+
+    @use_profile('postgres')
+    def test__postgres__external_reference(self):
+        self.assertEquals(len(self.run_dbt()), 1)
+        # running it again should succeed
+        self.assertEquals(len(self.run_dbt()), 1)
diff --git a/test/integration/base.py b/test/integration/base.py
index ca86f1b1df7..a72b75f110f 100644
--- a/test/integration/base.py
+++ b/test/integration/base.py
@@ -350,6 +350,7 @@ def run_dbt(self, args=None, expect_pass=True, strict=True):
 
         if strict:
             args = ["--strict"] + args
+        args.append('--log-cache-events')
         logger.info("Invoking dbt with {}".format(args))
 
         res, success = dbt.handle_and_check(args)
diff --git a/test/unit/test_cache.py b/test/unit/test_cache.py
new file mode 100644
index 00000000000..9350b958914
--- /dev/null
+++ b/test/unit/test_cache.py
@@ -0,0 +1,327 @@
+from unittest import TestCase
+from dbt.adapters.cache import RelationsCache
+from dbt.adapters.default.relation import DefaultRelation
+from multiprocessing.dummy import Pool as ThreadPool
+import dbt.exceptions
+
+import random
+import time
+
+
+def make_relation(schema, identifier):
+    return DefaultRelation.create(schema=schema, identifier=identifier)
+
+def make_mock_relationship(schema, identifier):
+    return DefaultRelation.create(
+        database='test_db', schema=schema, identifier=identifier,
+        table_name=identifier, type='view'
+    )
+
+
+class TestCache(TestCase):
+    def setUp(self):
+        self.cache = RelationsCache()
+
+    def test_empty(self):
+        self.assertEqual(len(self.cache.relations), 0)
+        relations = self.cache.get_relations('test')
+        self.assertEqual(len(relations), 0)
+
+    def test_bad_drop(self):
+        self.cache.drop(make_relation('foo', 'bar'))
+
+    def test_bad_link(self):
+        self.cache.add(make_relation('schema', 'foo'))
+        # src does not exist
+        with self.assertRaises(dbt.exceptions.InternalException):
+            self.cache.add_link(make_relation('schema', 'bar'),
+                                make_relation('schema', 'foo'))
+
+        # dst does not exist
+        with self.assertRaises(dbt.exceptions.InternalException):
+            self.cache.add_link(make_relation('schema', 'foo'),
+                                make_relation('schema', 'bar'))
+
+    def test_bad_rename(self):
+        # foo does not exist - should be ignored
+        self.cache.rename(make_relation('schema', 'foo'),
+                          make_relation('schema', 'bar'))
+
+        self.cache.add(make_relation('schema', 'foo'))
+        self.cache.add(make_relation('schema', 'bar'))
+        # bar exists
+        with self.assertRaises(dbt.exceptions.InternalException):
+            self.cache.rename(make_relation('schema', 'foo'),
+                              make_relation('schema', 'bar'))
+
+    def test_get_relations(self):
+        relation = make_relation('foo', 'bar')
+        self.cache.add(relation)
+        self.assertEqual(len(self.cache.relations), 1)
+
+        relations = self.cache.get_relations('foo')
+        self.assertEqual(len(relations), 1)
+        self.assertIs(relations[0], relation)
+
+        relations = self.cache.get_relations('FOO')
+        self.assertEqual(len(relations), 1)
+        self.assertIs(relations[0], relation)
+
+    def test_add(self):
+        rel = make_relation('foo', 'bar')
+        self.cache.add(rel)
+
+        relations = self.cache.get_relations('foo')
+        self.assertEqual(len(relations), 1)
+        self.assertIs(relations[0], rel)
+
+        # add a new relation with same name
+        self.cache.add(make_relation('foo', 'bar'))
+        self.assertEqual(len(self.cache.relations), 1)
+        self.assertEqual(self.cache.schemas, {'foo'})
+
+        relations = self.cache.get_relations('foo')
+        self.assertEqual(len(relations), 1)
+        self.assertIs(relations[0], rel)
+
+        self.cache.add(make_relation('FOO', 'baz'))
+        self.assertEqual(len(self.cache.relations), 2)
+
+        relations = self.cache.get_relations('foo')
+        self.assertEqual(len(relations), 2)
+
+        self.assertEqual(self.cache.schemas, {'foo', 'FOO'})
+        self.assertIsNot(self.cache.relations[('foo', 'bar')].inner, None)
+        self.assertIsNot(self.cache.relations[('FOO', 'baz')].inner, None)
+
+    def test_rename(self):
+        self.cache.add(make_relation('foo', 'bar'))
+        self.assertIsNot(self.cache.relations[('foo', 'bar')].inner, None)
+        self.cache.rename(make_relation('foo', 'bar'),
+                          make_relation('foo', 'baz'))
+
+        relations = self.cache.get_relations('foo')
+        self.assertEqual(len(relations), 1)
+        self.assertEqual(relations[0].schema, 'foo')
+        self.assertEqual(relations[0].identifier, 'baz')
+
+        relation = self.cache.relations[('foo', 'baz')]
+        self.assertEqual(relation.inner.schema, 'foo')
+        self.assertEqual(relation.inner.identifier, 'baz')
+        self.assertEqual(relation.schema, 'foo')
+        self.assertEqual(relation.identifier, 'baz')
+
+        with self.assertRaises(KeyError):
+            self.cache.relations[('foo', 'bar')]
+
+
+class TestLikeDbt(TestCase):
+    def setUp(self):
+        self.cache = RelationsCache()
+        self._sleep = True
+
+        # add a bunch of cache entries
+        for ident in 'abcdef':
+            self.cache.add(make_relation('schema', ident))
+        # 'b' references 'a'
+        self.cache.add_link(make_relation('schema', 'a'),
+                            make_relation('schema', 'b'))
+        # and 'c' references 'b'
+        self.cache.add_link(make_relation('schema', 'b'),
+                            make_relation('schema', 'c'))
+        # and 'd' references 'b'
+        self.cache.add_link(make_relation('schema', 'b'),
+                            make_relation('schema', 'd'))
+        # and 'e' references 'a'
+        self.cache.add_link(make_relation('schema', 'a'),
+                            make_relation('schema', 'e'))
+        # and 'f' references 'd'
+        self.cache.add_link(make_relation('schema', 'd'),
+                            make_relation('schema', 'f'))
+        # so drop propagation goes (a -> (b -> (c (d -> f))) e)
+
+    def assert_has_relations(self, expected):
+        current = set(r.identifier for r in self.cache.get_relations('schema'))
+        self.assertEqual(current, expected)
+
+    def test_drop_inner(self):
+        self.assert_has_relations(set('abcdef'))
+        self.cache.drop(make_relation('schema', 'b'))
+        self.assert_has_relations({'a', 'e'})
+
+    def test_rename_and_drop(self):
+        self.assert_has_relations(set('abcdef'))
+        # drop the backup/tmp
+        self.cache.drop(make_relation('schema', 'b__backup'))
+        self.cache.drop(make_relation('schema', 'b__tmp'))
+        self.assert_has_relations(set('abcdef'))
+        # create a new b__tmp
+        self.cache.add(make_relation('schema', 'b__tmp',))
+        self.assert_has_relations(set('abcdef') | {'b__tmp'})
+        # rename b -> b__backup
+        self.cache.rename(make_relation('schema', 'b'),
+                          make_relation('schema', 'b__backup'))
+        self.assert_has_relations(set('acdef') | {'b__tmp', 'b__backup'})
+        # rename temp to b
+        self.cache.rename(make_relation('schema', 'b__tmp'),
+                          make_relation('schema', 'b'))
+        self.assert_has_relations(set('abcdef') | {'b__backup'})
+
+        # drop backup, everything that used to depend on b should be gone, but
+        # b itself should still exist
+        self.cache.drop(make_relation('schema', 'b__backup'))
+        self.assert_has_relations(set('abe'))
+        relation = self.cache.relations[('schema', 'a')]
+        self.assertEqual(len(relation.referenced_by), 1)
+
+    def _rand_sleep(self):
+        if not self._sleep:
+            return
+        time.sleep(random.random() * 0.1)
+
+    def _target(self, ident):
+        self._rand_sleep()
+        self.cache.rename(make_relation('schema', ident),
+                          make_relation('schema', ident+'__backup'))
+        self._rand_sleep()
+        self.cache.add(make_relation('schema', ident+'__tmp')
+        )
+        self._rand_sleep()
+        self.cache.rename(make_relation('schema', ident+'__tmp'),
+                          make_relation('schema', ident))
+        self._rand_sleep()
+        self.cache.drop(make_relation('schema', ident+'__backup'))
+        return ident, self.cache.get_relations('schema')
+
+    def test_threaded(self):
+        # add three more short subchains for threads to test on
+        for ident in 'ghijklmno':
+            obj = make_mock_relationship('schema', ident)
+            self.cache.add(make_relation('schema', ident))
+
+        self.cache.add_link(make_relation('schema', 'a'),
+                            make_relation('schema', 'g'))
+        self.cache.add_link(make_relation('schema', 'g'),
+                            make_relation('schema', 'h'))
+        self.cache.add_link(make_relation('schema', 'h'),
+                            make_relation('schema', 'i'))
+
+        self.cache.add_link(make_relation('schema', 'a'),
+                            make_relation('schema', 'j'))
+        self.cache.add_link(make_relation('schema', 'j'),
+                            make_relation('schema', 'k'))
+        self.cache.add_link(make_relation('schema', 'k'),
+                            make_relation('schema', 'l'))
+
+        self.cache.add_link(make_relation('schema', 'a'),
+                            make_relation('schema', 'm'))
+        self.cache.add_link(make_relation('schema', 'm'),
+                            make_relation('schema', 'n'))
+        self.cache.add_link(make_relation('schema', 'n'),
+                            make_relation('schema', 'o'))
+
+        pool = ThreadPool(4)
+        results = list(pool.imap_unordered(self._target, ('b', 'g', 'j', 'm')))
+        pool.close()
+        pool.join()
+        # at a minimum, we expect each table to "see" itself, its parent ('a'),
+        # and the unrelated table ('a')
+        min_expect = {
+            'b': {'a', 'b', 'e'},
+            'g': {'a', 'g', 'e'},
+            'j': {'a', 'j', 'e'},
+            'm': {'a', 'm', 'e'},
+        }
+
+        for ident, relations in results:
+            seen = set(r.identifier for r in relations)
+            self.assertTrue(min_expect[ident].issubset(seen))
+
+        self.assert_has_relations(set('abgjme'))
+
+    def test_threaded_repeated(self):
+        for _ in range(10):
+            self.setUp()
+            self._sleep = False
+            self.test_threaded()
+
+
+
+class TestComplexCache(TestCase):
+    def setUp(self):
+        self.cache = RelationsCache()
+        inputs = [
+            ('foo', 'table1'),
+            ('bar', 'table2'),
+            ('foo', 'table3'),
+            ('foo', 'table4'),
+            ('bar', 'table3'),
+        ]
+        self.inputs = [make_relation(s, i) for s, i in inputs]
+        for relation in self.inputs:
+            self.cache.add(relation)
+
+        # foo.table3 references foo.table1
+        # (create view table3 as (select * from table1...))
+        self.cache.add_link(
+            make_relation('foo', 'table1'),
+            make_relation('foo', 'table3')
+        )
+        # bar.table3 references foo.table3
+        # (create view bar.table5 as (select * from foo.table3...))
+        self.cache.add_link(
+            make_relation('foo', 'table3'),
+            make_relation('bar', 'table3')
+        )
+
+        # foo.table2 also references foo.table1
+        self.cache.add_link(
+            make_relation('foo', 'table1'),
+            make_relation('foo', 'table4')
+        )
+
+    def test_get_relations(self):
+        self.assertEqual(len(self.cache.get_relations('foo')), 3)
+        self.assertEqual(len(self.cache.get_relations('bar')), 2)
+        self.assertEqual(len(self.cache.relations), 5)
+
+    def test_drop_one(self):
+        # dropping bar.table2 should only drop itself
+        self.cache.drop(make_relation('bar', 'table2'))
+        self.assertEqual(len(self.cache.get_relations('foo')), 3)
+        self.assertEqual(len(self.cache.get_relations('bar')), 1)
+        self.assertEqual(len(self.cache.relations), 4)
+
+    def test_drop_many(self):
+        # dropping foo.table1 should drop everything but bar.table2.
+        self.cache.drop(make_relation('foo', 'table1'))
+        self.assertEqual(len(self.cache.get_relations('foo')), 0)
+        self.assertEqual(len(self.cache.get_relations('bar')), 1)
+        self.assertEqual(len(self.cache.relations), 1)
+
+    def test_rename_root(self):
+        self.cache.rename(make_relation('foo', 'table1'),
+                          make_relation('bar', 'table1'))
+        retrieved = self.cache.relations[('bar', 'table1')].inner
+        self.assertEqual(retrieved.schema, 'bar')
+        self.assertEqual(retrieved.identifier, 'table1')
+        self.assertEqual(len(self.cache.get_relations('foo')), 2)
+        self.assertEqual(len(self.cache.get_relations('bar')), 3)
+
+        # make sure drops still cascade from the renamed table
+        self.cache.drop(make_relation('bar', 'table1'))
+        self.assertEqual(len(self.cache.get_relations('foo')), 0)
+        self.assertEqual(len(self.cache.get_relations('bar')), 1)
+        self.assertEqual(len(self.cache.relations), 1)
+
+    def test_rename_branch(self):
+        self.cache.rename(make_relation('foo', 'table3'),
+                          make_relation('foo', 'table2'))
+        self.assertEqual(len(self.cache.get_relations('foo')), 3)
+        self.assertEqual(len(self.cache.get_relations('bar')), 2)
+
+        # make sure drops still cascade through the renamed table
+        self.cache.drop(make_relation('foo', 'table1'))
+        self.assertEqual(len(self.cache.get_relations('foo')), 0)
+        self.assertEqual(len(self.cache.get_relations('bar')), 1)
+        self.assertEqual(len(self.cache.relations), 1)
diff --git a/test/unit/test_postgres_adapter.py b/test/unit/test_postgres_adapter.py
index 7dea6a40a3c..8c8e789bbc6 100644
--- a/test/unit/test_postgres_adapter.py
+++ b/test/unit/test_postgres_adapter.py
@@ -116,15 +116,8 @@ def test_get_catalog_various_schemas(self, mock_run):
         mock_run.return_value = agate.Table(rows=rows,
                                             column_names=column_names)
 
-        # we should accept the lowercase matching 'foo's only.
-        mock_nodes = [
-            mock.MagicMock(spec_set=['schema'], schema='foo')
-            for k in range(2)
-        ]
-        mock_nodes.append(mock.MagicMock(spec_set=['schema'], schema='quux'))
-        nodes = {str(idx): n for idx, n in enumerate(mock_nodes)}
-        # give manifest the dict it wants
-        mock_manifest = mock.MagicMock(spec_set=['nodes'], nodes=nodes)
+        mock_manifest = mock.MagicMock()
+        mock_manifest.get_used_schemas.return_value = {'foo', 'quux'}
 
         catalog = self.adapter.get_catalog(mock_manifest)
         self.assertEqual(