Feature/bq incremental and archive

dbt/adapters/bigquery/impl.py

@@ -16,6 +16,7 @@
 from dbt.logger import GLOBAL_LOGGER as logger
 
 import google.auth
+import google.api_core
 import google.oauth2
 import google.cloud.exceptions
 import google.cloud.bigquery
@@ -30,13 +31,18 @@ class BigQueryAdapter(PostgresAdapter):
         # deprecated -- use versions that take relations instead
         "query_for_existing",
         "execute_model",
+        "create_temporary_table",
         "drop",
         "execute",
         "quote_schema_and_table",
         "make_date_partitioned_table",
         "already_exists",
         "expand_target_column_types",
         "load_dataframe",
+        "get_missing_columns",
+
+        "create_schema",
+        "alter_table_add_column",
 
         # versions of adapter functions that take / return Relations
         "list_relations",
@@ -183,7 +189,6 @@ def close(cls, connection):
     @classmethod
     def list_relations(cls, profile, project_cfg, schema, model_name=None):
         connection = cls.get_connection(profile, model_name)
-        credentials = connection.get('credentials', {})
         client = connection.get('handle')
 
         bigquery_dataset = cls.get_dataset(
@@ -201,6 +206,15 @@ def list_relations(cls, profile, project_cfg, schema, model_name=None):
             #       won't need to do this
             max_results=100000)
 
+        # This will 404 if the dataset does not exist. It's lazily
+        # evaluated, so wrap it in a list to throw _now_. This behavior
+        # mirrors the implementation of list_relations for other adapters
+
+        try:
+            all_tables = list(all_tables)
+        except google.api_core.exceptions.NotFound as e:
+            all_tables = []
+
         return [cls.bq_table_to_relation(table) for table in all_tables]
 
     @classmethod
@@ -357,7 +371,7 @@ def execute_model(cls, profile, project_cfg, model,
         return res
 
     @classmethod
-    def execute(cls, profile, sql, model_name=None, fetch=False, **kwargs):
+    def raw_execute(cls, profile, sql, model_name=None, fetch=False, **kwargs):
         conn = cls.get_connection(profile, model_name)
         client = conn.get('handle')
 
@@ -372,6 +386,49 @@ def execute(cls, profile, sql, model_name=None, fetch=False, **kwargs):
         with cls.exception_handler(profile, sql, model_name):
             iterator = query_job.result()
 
+        return query_job, iterator
+
+    @classmethod
+    def create_temporary_table(cls, profile, project, sql, model_name=None,
+                               **kwargs):
+
+        # BQ queries always return a temp table with their results
+        query_job, _ = cls.raw_execute(profile, sql, model_name)
+        bq_table = query_job.destination
+
+        return cls.Relation.create(
+            project=bq_table.project,
+            schema=bq_table.dataset_id,
+            identifier=bq_table.table_id,
+            quote_policy={
+                'schema': True,
+                'identifier': True
+            },
+            type=BigQueryRelation.Table)
+
+    @classmethod
+    def alter_table_add_column(cls, profile, project, relation, column,
+                               model_name=None):
+
+        conn = cls.get_connection(profile, model_name)
+        client = conn.get('handle')
+
+        dataset = cls.get_dataset(profile, project, relation.schema,
+                                  model_name)
+
+        table_ref = dataset.table(relation.name)
+        table = client.get_table(table_ref)
+
+        new_schema = table.schema + [column.to_bq_schema_object()]
+
+        new_table = google.cloud.bigquery.Table(table_ref, schema=new_schema)
+        client.update_table(new_table, ['schema'])
+
+    @classmethod
+    def execute(cls, profile, sql, model_name=None, fetch=None, **kwargs):
+        _, iterator = cls.raw_execute(profile, sql, model_name, fetch,
+                                      **kwargs)
+
         if fetch:
             res = cls.get_table_from_response(iterator)
         else:
@@ -410,8 +467,13 @@ def create_schema(cls, profile, project_cfg, schema, model_name=None):
         client = conn.get('handle')
 
         dataset = cls.get_dataset(profile, project_cfg, schema, model_name)
-        with cls.exception_handler(profile, 'create dataset', model_name):
-            client.create_dataset(dataset)
+
+        # Emulate 'create schema if not exists ...'
+        try:
+            client.get_dataset(dataset)
+        except google.api_core.exceptions.NotFound:
+            with cls.exception_handler(profile, 'create dataset', model_name):
+                client.create_dataset(dataset)
 
     @classmethod
     def drop_tables_in_schema(cls, profile, project_cfg, dataset):
@@ -468,8 +530,10 @@ def get_columns_in_table(cls, profile, project_cfg,
 
         columns = []
         for col in table_schema:
+            # BigQuery returns type labels that are not valid type specifiers
+            dtype = cls.Column.translate_type(col.field_type)
             column = cls.Column(
-                col.name, col.field_type, col.fields, col.mode)
+                col.name, dtype, col.fields, col.mode)
             columns.append(column)
 
         return columns

dbt/adapters/default/impl.py

@@ -719,7 +719,7 @@ def already_exists(cls, profile, project_cfg,
 
     @classmethod
     def quote(cls, identifier):
-        return '"{}"'.format(identifier.replace('"', '""'))
+        return '"{}"'.format(identifier)
 
     @classmethod
     def quote_schema_and_table(cls, profile, project_cfg,

dbt/include/global_project/macros/adapters/bigquery.sql

@@ -27,3 +27,7 @@
     {{ sql }}
   );
 {% endmacro %}
+
+{% macro bigquery__create_schema(schema_name) %}
+    {{ adapter.create_schema(schema_name) }}
+{% endmacro %}

dbt/include/global_project/macros/adapters/common.sql

@@ -29,7 +29,13 @@
 {%- endmacro %}
 
 {% macro create_schema(schema_name) %}
-  create schema if not exists {{ schema_name }};
+  {{ adapter_macro('create_schema', schema_name) }}
+{% endmacro %}
+
+{% macro default__create_schema(schema_name) %}
+  {% call statement() %}
+    create schema if not exists {{ schema_name }};
+  {% endcall %}
 {% endmacro %}