Merge branch 'dev/stephen-girard' into dev/wilt-chamberlain

core/dbt/adapters/base/impl.py

@@ -1,7 +1,4 @@
 import abc
-import copy
-import multiprocessing
-import time
 
 import agate
 import pytz
@@ -13,11 +10,11 @@
 import dbt.clients.agate_helper
 
 from dbt.compat import abstractclassmethod, classmethod
-from dbt.contracts.connection import Connection
+from dbt.node_types import NodeType
 from dbt.loader import GraphLoader
 from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.schema import Column
-from dbt.utils import filter_null_values, translate_aliases
+from dbt.utils import filter_null_values
 
 from dbt.adapters.base.meta import AdapterMeta, available, available_raw, \
     available_deprecated
@@ -69,16 +66,76 @@ def test(row):
     return test
 
 
-def _utc(dt):
+def _utc(dt, source, field_name):
     """If dt has a timezone, return a new datetime that's in UTC. Otherwise,
     assume the datetime is already for UTC and add the timezone.
     """
-    if dt.tzinfo:
+    if dt is None:
+        raise dbt.exceptions.raise_database_error(
+            "Expected a non-null value when querying field '{}' of table "
+            " {} but received value 'null' instead".format(
+                field_name,
+                source))
+
+    elif not hasattr(dt, 'tzinfo'):
+        raise dbt.exceptions.raise_database_error(
+            "Expected a timestamp value when querying field '{}' of table "
+            "{} but received value of type '{}' instead".format(
+                field_name,
+                source,
+                type(dt).__name__))
+
+    elif dt.tzinfo:
         return dt.astimezone(pytz.UTC)
     else:
         return dt.replace(tzinfo=pytz.UTC)
 
 
+class SchemaSearchMap(dict):
+    """A utility class to keep track of what information_schema tables to
+    search for what schemas
+    """
+    def add(self, relation):
+        key = relation.information_schema_only()
+        if key not in self:
+            self[key] = set()
+        self[key].add(relation.schema.lower())
+
+    def search(self):
+        for information_schema_name, schemas in self.items():
+            for schema in schemas:
+                yield information_schema_name, schema
+
+    def schemas_searched(self):
+        result = set()
+        for information_schema_name, schemas in self.items():
+            result.update(
+                (information_schema_name.database, schema)
+                for schema in schemas
+            )
+        return result
+
+    def flatten(self):
+        new = self.__class__()
+
+        database = None
+        # iterate once to look for a database name
+        seen = {r.database.lower() for r in self if r.database}
+        if len(seen) > 1:
+            dbt.exceptions.raise_compiler_error(str(seen))
+        elif len(seen) == 1:
+            database = list(seen)[0]
+
+        for information_schema_name, schema in self.search():
+            new.add(information_schema_name.incorporate(
+                path={'database': database, 'schema': schema},
+                quote_policy={'database': False},
+                include_policy={'database': False},
+            ))
+
+        return new
+
+
 @six.add_metaclass(AdapterMeta)
 class BaseAdapter(object):
     """The BaseAdapter provides an abstract base class for adapters.
@@ -222,24 +279,44 @@ def _relations_filter_table(cls, table, schemas):
         """
         return table.where(_relations_filter_schemas(schemas))
 
+    def _get_cache_schemas(self, manifest, exec_only=False):
+        """Get a mapping of each node's "information_schema" relations to a
+        set of all schemas expected in that information_schema.
+
+        There may be keys that are technically duplicates on the database side,
+        for example all of '"foo", 'foo', '"FOO"' and 'FOO' could coexist as
+        databases, and values could overlap as appropriate. All values are
+        lowercase strings.
+        """
+        info_schema_name_map = SchemaSearchMap()
+        for node in manifest.nodes.values():
+            if exec_only and node.resource_type not in NodeType.executable():
+                continue
+            relation = self.Relation.create_from(self.config, node)
+            info_schema_name_map.add(relation)
+        # result is a map whose keys are information_schema Relations without
+        # identifiers that have appropriate database prefixes, and whose values
+        # are sets of lowercase schema names that are valid members of those
+        # schemas
+        return info_schema_name_map
+
     def _relations_cache_for_schemas(self, manifest):
         """Populate the relations cache for the given schemas. Returns an
         iteratble of the schemas populated, as strings.
         """
         if not dbt.flags.USE_CACHE:
             return
 
-        schemas = manifest.get_used_schemas()
-
-        relations = []
-        # add all relations
-        for db, schema in schemas:
+        info_schema_name_map = self._get_cache_schemas(manifest,
+                                                       exec_only=True)
+        for db, schema in info_schema_name_map.search():
             for relation in self.list_relations_without_caching(db, schema):
                 self.cache.add(relation)
+
         # 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)
+        self.cache.update_schemas(info_schema_name_map.schemas_searched())
 
     def set_relations_cache(self, manifest, clear=False):
         """Run a query that gets a populated cache of the relations in the
@@ -400,13 +477,14 @@ def expand_column_types(self, goal, current, model_name=None):
         )
 
     @abc.abstractmethod
-    def list_relations_without_caching(self, database, schema,
+    def list_relations_without_caching(self, information_schema, schema,
                                        model_name=None):
         """List relations in the given schema, bypassing the cache.
 
         This is used as the underlying behavior to fill the cache.
 
-        :param str database: The name of the database to list relations from.
+        :param Relation information_schema: The information schema to list
+            relations from.
         :param str schema: The name of the schema to list relations from.
         :param Optional[str] model_name: The name of the model to use for the
             connection.
@@ -480,10 +558,15 @@ def list_relations(self, database, schema, model_name=None):
         if self._schema_is_cached(database, schema, model_name):
             return self.cache.get_relations(database, schema)
 
+        information_schema = self.Relation.create(
+            database=database,
+            schema=schema,
+            model_name='').information_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_without_caching(
-            database, schema, model_name=model_name
+            information_schema, schema, model_name=model_name
         )
 
         logger.debug('with schema={}, model_name={}, relations={}'
@@ -787,10 +870,11 @@ def get_catalog(self, manifest):
         """Get the catalog for this manifest by running the get catalog macro.
         Returns an agate.Table of catalog information.
         """
+        information_schemas = list(self._get_cache_schemas(manifest).keys())
         # make it a list so macros can index into it.
-        context = {'databases': list(manifest.get_used_databases())}
+        kwargs = {'information_schemas': information_schemas}
         table = self.execute_macro(GET_CATALOG_MACRO_NAME,
-                                   context_override=context,
+                                   kwargs=kwargs,
                                    release=True)
 
         results = self._catalog_filter_table(table, manifest)
@@ -823,11 +907,12 @@ def calculate_freshness(self, source, loaded_at_field, manifest=None,
             dbt.exceptions.raise_compiler_error(
                 'Got an invalid result from "{}" macro: {}'.format(
                     FRESHNESS_MACRO_NAME, [tuple(r) for r in table]
-                ),
-                node=node
+                )
             )
 
-        max_loaded_at, snapshotted_at = map(_utc, table[0])
+        max_loaded_at = _utc(table[0][0], source, loaded_at_field)
+        snapshotted_at = _utc(table[0][1], source, loaded_at_field)
+
         age = (snapshotted_at - max_loaded_at).total_seconds()
         return {
             'max_loaded_at': max_loaded_at,

core/dbt/adapters/base/relation.py

@@ -1,5 +1,6 @@
 from dbt.api import APIObject
 from dbt.utils import filter_null_values
+from dbt.node_types import NodeType
 
 import dbt.exceptions
 
@@ -30,15 +31,15 @@ class BaseRelation(APIObject):
             'database': True,
             'schema': True,
             'identifier': True
-        }
+        },
     }
 
     PATH_SCHEMA = {
         'type': 'object',
         'properties': {
             'database': {'type': ['string', 'null']},
             'schema': {'type': ['string', 'null']},
-            'identifier': {'type': 'string'},
+            'identifier': {'type': ['string', 'null']},
         },
         'required': ['database', 'schema', 'identifier'],
     }
@@ -135,6 +136,36 @@ def include(self, database=None, schema=None, identifier=None):
 
         return self.incorporate(include_policy=policy)
 
+    def information_schema(self, identifier=None):
+        include_db = self.database is not None
+        include_policy = filter_null_values({
+            'database': include_db,
+            'schema': True,
+            'identifier': identifier is not None
+        })
+        quote_policy = filter_null_values({
+            'database': self.quote_policy['database'],
+            'schema': False,
+            'identifier': False,
+        })
+
+        path_update = {
+            'schema': 'information_schema',
+            'identifier': identifier
+        }
+
+        return self.incorporate(
+            quote_policy=quote_policy,
+            include_policy=include_policy,
+            path=path_update,
+            table_name=identifier)
+
+    def information_schema_only(self):
+        return self.information_schema()
+
+    def information_schema_table(self, identifier):
+        return self.information_schema(identifier)
+
     def render(self, use_table_name=True):
         parts = []
 
@@ -174,15 +205,16 @@ def quoted(self, identifier):
 
     @classmethod
     def create_from_source(cls, source, **kwargs):
+        quote_policy = dbt.utils.deep_merge(
+            cls.DEFAULTS['quote_policy'],
+            source.quoting,
+            kwargs.get('quote_policy', {})
+        )
         return cls.create(
             database=source.database,
             schema=source.schema,
             identifier=source.identifier,
-            quote_policy={
-                'database': True,
-                'schema': True,
-                'identifier': True,
-            },
+            quote_policy=quote_policy,
             **kwargs
         )
 
@@ -202,6 +234,13 @@ def create_from_node(cls, config, node, table_name=None, quote_policy=None,
             quote_policy=quote_policy,
             **kwargs)
 
+    @classmethod
+    def create_from(cls, config, node, **kwargs):
+        if node.resource_type == NodeType.Source:
+            return cls.create_from_source(node, **kwargs)
+        else:
+            return cls.create_from_node(config, node, **kwargs)
+
     @classmethod
     def create(cls, database=None, schema=None,
                identifier=None, table_name=None,

core/dbt/adapters/sql/impl.py

@@ -1,15 +1,10 @@
-import abc
-import time
-
 import agate
-import six
 
 import dbt.clients.agate_helper
 import dbt.exceptions
 import dbt.flags
 from dbt.adapters.base import BaseAdapter, available
 from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.compat import abstractclassmethod
 
 
 LIST_RELATIONS_MACRO_NAME = 'list_relations_without_caching'
@@ -196,11 +191,12 @@ def drop_schema(self, database, schema, model_name=None):
                            kwargs=kwargs,
                            connection_name=model_name)
 
-    def list_relations_without_caching(self, database, schema,
+    def list_relations_without_caching(self, information_schema, schema,
                                        model_name=None):
+        kwargs = {'information_schema': information_schema, 'schema': schema}
         results = self.execute_macro(
             LIST_RELATIONS_MACRO_NAME,
-            kwargs={'database': database, 'schema': schema},
+            kwargs=kwargs,
             connection_name=model_name,
             release=True
         )
@@ -236,9 +232,14 @@ def list_schemas(self, database, model_name=None):
         return [row[0] for row in results]
 
     def check_schema_exists(self, database, schema, model_name=None):
+        information_schema = self.Relation.create(
+            database=database, schema=schema
+        ).information_schema()
+
+        kwargs = {'information_schema': information_schema, 'schema': schema}
         results = self.execute_macro(
             CHECK_SCHEMA_EXISTS_MACRO_NAME,
-            kwargs={'database': database, 'schema': schema},
+            kwargs=kwargs,
             connection_name=model_name
         )
         return results[0][0] > 0

core/dbt/context/common.py

@@ -36,6 +36,11 @@ def __init__(self, adapter):
     def __getattr__(self, key):
         return getattr(self.relation_type, key)
 
+    def create_from_source(self, *args, **kwargs):
+        # bypass our create when creating from source so as not to mess up
+        # the source quoting
+        return self.relation_type.create_from_source(*args, **kwargs)
+
     def create(self, *args, **kwargs):
         kwargs['quote_policy'] = dbt.utils.merge(
             self.quoting_config,

core/dbt/contracts/graph/manifest.py

@@ -4,7 +4,7 @@
     PARSED_MACRO_CONTRACT, PARSED_DOCUMENTATION_CONTRACT, \
     PARSED_SOURCE_DEFINITION_CONTRACT
 from dbt.contracts.graph.compiled import COMPILED_NODE_CONTRACT, CompiledNode
-from dbt.exceptions import ValidationException
+from dbt.exceptions import raise_duplicate_resource_name
 from dbt.node_types import NodeType
 from dbt.logger import GLOBAL_LOGGER as logger
 from dbt import tracking
@@ -401,10 +401,11 @@ def __getattr__(self, name):
             type(self).__name__, name)
         )
 
-    def get_used_schemas(self):
+    def get_used_schemas(self, resource_types=None):
         return frozenset({
             (node.database, node.schema)
             for node in self.nodes.values()
+            if not resource_types or node.resource_type in resource_types
         })
 
     def get_used_databases(self):