Implement relations api

Makefile

@@ -17,8 +17,7 @@ test-integration:
 	@echo "Integration test run starting..."
 	@time docker-compose run test tox -e integration-postgres-py27,integration-postgres-py36,integration-snowflake-py27,integration-snowflake-py36,integration-bigquery-py27,integration-bigquery-py36
 
-test-new:
-	@echo "Test run starting..."
-	@echo "Changed test files:"
-	@echo "${changed_tests}"
-	@docker-compose run test /usr/src/app/test/runner.sh ${changed_tests}
+
+test-quick:
+	@echo "Integration test run starting..."
+	@time docker-compose run test tox -e integration-postgres-py36 -- -x

TODO.md

@@ -0,0 +1,16 @@
+- [ ] rip out query_for_existing > replace with get_relation
+  - maybe part of the relation object is metadata about when the relation was pulled
+    from the db (relation type, when we asked the db about it, columns, ??)
+- [ ] add get_relation, list_relations
+- fns / macros
+  - [x] query_for_existing
+  - [ ] get_columns_in_table
+  - [x] rename
+  - [x] truncate
+  - [ ] drop
+  - [ ] reset_csv_table
+  - [ ] load_csv_rows
+  - [ ] macro: get_existing_relation_type
+  - [ ] macro: create_table_as
+  - (go back through these after BQ pr is merged)
+- [ ] deprecation warnings

dbt/adapters/bigquery/__init__.py

@@ -0,0 +1,5 @@
+from dbt.adapters.bigquery.impl import BigQueryAdapter
+
+__all__ = [
+    BigQueryAdapter,
+]

dbt/adapters/bigquery.py → dbt/adapters/bigquery/impl.py

@@ -10,6 +10,7 @@
 import dbt.clients.agate_helper
 
 from dbt.adapters.postgres import PostgresAdapter
+from dbt.adapters.bigquery.relation import BigQueryRelation
 from dbt.contracts.connection import validate_connection
 from dbt.logger import GLOBAL_LOGGER as logger
 
@@ -28,6 +29,7 @@ class BigQueryAdapter(PostgresAdapter):
         "query_for_existing",
         "execute_model",
         "drop",
+        "drop_relation",
         "execute",
         "quote_schema_and_table",
         "make_date_partitioned_table",
@@ -37,6 +39,8 @@ class BigQueryAdapter(PostgresAdapter):
         "get_columns_in_table"
     ]
 
+    Relation = BigQueryRelation
+
     SCOPE = ('https://www.googleapis.com/auth/bigquery',
              'https://www.googleapis.com/auth/cloud-platform',
              'https://www.googleapis.com/auth/drive')
@@ -159,24 +163,36 @@ def query_for_existing(cls, profile, schemas, model_name=None):
         if not isinstance(schemas, (list, tuple)):
             schemas = [schemas]
 
-        conn = cls.get_connection(profile, model_name)
-        client = conn.get('handle')
+        all_relations = []
 
-        all_tables = []
         for schema in schemas:
-            dataset = cls.get_dataset(profile, schema, model_name)
-            all_tables.extend(client.list_tables(dataset))
+            all_relations.extend(cls.list_relations(
+                profile, schema, model_name))
+
+        return {relation.identifier: relation.type
+                for relation in all_relations}
+
+    @classmethod
+    def list_relations(cls, profile, dataset, model_name=None):
+        connection = cls.get_connection(profile, model_name)
+        credentials = connection.get('credentials', {})
+        client = connection.get('handle')
+
+        bigquery_dataset = cls.get_dataset(profile, dataset, model_name)
+        all_tables = client.list_tables(bigquery_dataset)
 
         relation_types = {
             'TABLE': 'table',
             'VIEW': 'view',
             'EXTERNAL': 'external'
         }
 
-        existing = [(table.table_id, relation_types.get(table.table_type))
-                    for table in all_tables]
-
-        return dict(existing)
+        return [cls.Relation.create(
+            project=credentials.get('project'),
+            dataset=dataset,
+            identifier=table.table_id,
+            type=relation_types.get(table.table_type))
+                for table in all_tables]
 
     @classmethod
     def table_exists(cls, profile, schema, table, model_name=None):
@@ -185,19 +201,25 @@ def table_exists(cls, profile, schema, table, model_name=None):
         return exists
 
     @classmethod
-    def drop(cls, profile, schema, relation, relation_type, model_name=None):
+    def drop_relation(cls, profile, relation, model_name=None):
         conn = cls.get_connection(profile, model_name)
         client = conn.get('handle')
 
-        dataset = cls.get_dataset(profile, schema, model_name)
-        relation_object = dataset.table(relation)
+        dataset = cls.get_dataset(profile, relation._schema, model_name)
+        relation_object = dataset.table(relation._identifier)
         client.delete_table(relation_object)
 
     @classmethod
     def rename(cls, profile, schema, from_name, to_name, model_name=None):
         raise dbt.exceptions.NotImplementedException(
             '`rename` is not implemented for this adapter!')
 
+    @classmethod
+    def rename_relation(cls, profile, from_relation, to_relation,
+                        model_name=None):
+        raise dbt.exceptions.NotImplementedException(
+            '`rename_relation` is not implemented for this adapter!')
+
     @classmethod
     def get_timeout(cls, conn):
         credentials = conn['credentials']
@@ -239,8 +261,8 @@ def poll_until_job_completes(cls, job, timeout):
             raise job.exception()
 
     @classmethod
-    def make_date_partitioned_table(cls, profile, dataset_name, identifier,
-                                    model_name=None):
+    def make_date_partitioned_table(cls, profile, dataset_name,
+                                    identifier, model_name=None):
         conn = cls.get_connection(profile, model_name)
         client = conn.get('handle')
 
@@ -459,12 +481,16 @@ def quote(cls, identifier):
 
     @classmethod
     def quote_schema_and_table(cls, profile, schema, table, model_name=None):
+        return cls.render_relation(profile,
+                                   cls.quote(schema),
+                                   cls.quote(table))
+
+    @classmethod
+    def render_relation(cls, profile, schema, table):
         connection = cls.get_connection(profile)
         credentials = connection.get('credentials', {})
         project = credentials.get('project')
-        return '{}.{}.{}'.format(cls.quote(project),
-                                 cls.quote(schema),
-                                 cls.quote(table))
+        return '{}.{}.{}'.format(cls.quote(project), schema, table)
 
     @classmethod
     def convert_text_type(cls, agate_table, col_idx):
@@ -489,8 +515,10 @@ def create_csv_table(cls, profile, schema, table_name, agate_table):
 
     @classmethod
     def reset_csv_table(cls, profile, schema, table_name, agate_table,
-                        full_refresh=False):
-        cls.drop(profile, schema, table_name, "table")
+                        full_refresh=False, model_name=None):
+        relation = cls.Relation(dataset=schema, identifier=table_name)
+
+        cls.drop_relation(profile, relation, model_name)
 
     @classmethod
     def _agate_to_schema(cls, agate_table):

dbt/adapters/bigquery/relation.py

@@ -0,0 +1,77 @@
+from dbt.adapters.default.relation import DefaultRelation
+from dbt.utils import filter_null_values
+
+
+class BigQueryRelation(DefaultRelation):
+
+    DEFAULTS = {
+        'metadata': {
+            '_type': 'DefaultRelation'
+        },
+        'quote_character': '"',
+        'quote_policy': {
+            'project': True,
+            'dataset': True,
+            'identifier': True
+        },
+        'include_policy': {
+            'project': False,
+            'dataset': True,
+            'identifier': True
+        }
+    }
+
+    PATH_DATASET = {
+        'type': 'object',
+        'properties': {
+            'project': {'type': ['string', 'null']},
+            'dataset': {'type': ['string', 'null']},
+            'identifier': {'type': 'string'},
+        },
+        'required': ['project', 'dataset', 'identifier'],
+    }
+
+    POLICY_DATASET = {
+        'type': 'object',
+        'properties': {
+            'project': {'type': 'boolean'},
+            'dataset': {'type': 'boolean'},
+            'identifier': {'type': 'boolean'},
+        },
+        'required': ['project', 'dataset', 'identifier'],
+    }
+
+    def matches(self, project=None, dataset=None, identifier=None):
+        search = filter_null_values({
+            'project': project,
+            'dataset': dataset,
+            'identifier': identifier
+        })
+
+        if not search:
+            # nothing was passed in
+            pass
+
+        for k, v in search.items():
+            if self.get_path_part(k) != v:
+                return False
+
+        return True
+
+    def quote(self, project=None, dataset=None, identifier=None):
+        policy = filter_null_values({
+            'project': project,
+            'dataset': dataset,
+            'identifier': identifier
+        })
+
+        return self.incorporate(quote_policy=policy)
+
+    def include(self, project=None, dataset=None, identifier=None):
+        policy = filter_null_values({
+            'project': project,
+            'dataset': dataset,
+            'identifier': identifier
+        })
+
+        return self.incorporate(include_policy=policy)

dbt/adapters/default/__init__.py

@@ -0,0 +1,7 @@
+from dbt.adapters.default.impl import DefaultAdapter
+from dbt.adapters.default.relation import DefaultRelation
+
+__all__ = [
+    DefaultAdapter,
+    DefaultRelation,
+]