Try using SQL for create_schema

CHANGELOG.md

@@ -1,3 +1,8 @@
+## dbt-bigquery 1.2.0 (Release TBD)
+
+### Features
+- Implement `create_schema` via SQL, instead of Python method, allowing users to override if desired. `drop_schema` remains a Python method for the time being. ([#182](https://github.com/dbt-labs/dbt-bigquery/issues/182), [#183](https://github.com/dbt-labs/dbt-bigquery/pull/183))
+
 ## dbt-bigquery 1.2.0b1 (June 24, 2022)
 
 ### Fixes

dbt/adapters/bigquery/impl.py

@@ -8,11 +8,15 @@
 
 from dbt import ui  # type: ignore
 from dbt.adapters.base import BaseAdapter, available, RelationType, SchemaSearchMap, AdapterConfig
+from dbt.adapters.cache import _make_key
+
 from dbt.adapters.bigquery.relation import BigQueryRelation
 from dbt.adapters.bigquery import BigQueryColumn
 from dbt.adapters.bigquery import BigQueryConnectionManager
 from dbt.contracts.graph.manifest import Manifest
 from dbt.events import AdapterLogger
+from dbt.events.functions import fire_event
+from dbt.events.types import SchemaCreation, SchemaDrop
 from dbt.utils import filter_null_values
 
 import google.auth
@@ -33,6 +37,8 @@
 WRITE_APPEND = google.cloud.bigquery.job.WriteDisposition.WRITE_APPEND
 WRITE_TRUNCATE = google.cloud.bigquery.job.WriteDisposition.WRITE_TRUNCATE
 
+CREATE_SCHEMA_MACRO_NAME = "create_schema"
+
 
 def sql_escape(string):
     if not isinstance(string, str):
@@ -273,16 +279,29 @@ def get_relation(self, database: str, schema: str, identifier: str) -> BigQueryR
             table = None
         return self._bq_table_to_relation(table)
 
+    # BigQuery added SQL support for 'create schema' + 'drop schema' in March 2021
+    # Unfortunately, 'drop schema' runs into permissions issues during tests
+    # Most of the value here comes from user overrides of 'create_schema'
+
+    # TODO: the code below is copy-pasted from SQLAdapter.create_schema. Is there a better way?
     def create_schema(self, relation: BigQueryRelation) -> None:
-        database = relation.database
-        schema = relation.schema
-        logger.debug('Creating schema "{}.{}".', database, schema)
-        self.connections.create_dataset(database, schema)
+        # use SQL 'create schema'
+        relation = relation.without_identifier()
+        fire_event(SchemaCreation(relation=_make_key(relation)))
+        kwargs = {
+            "relation": relation,
+        }
+        self.execute_macro(CREATE_SCHEMA_MACRO_NAME, kwargs=kwargs)
+        self.commit_if_has_connection()
+        # we can't update the cache here, as if the schema already existed we
+        # don't want to (incorrectly) say that it's empty
 
     def drop_schema(self, relation: BigQueryRelation) -> None:
+        # still use a client method, rather than SQL 'drop schema ... cascade'
         database = relation.database
         schema = relation.schema
-        logger.debug('Dropping schema "{}.{}".', database, schema)
+        logger.debug('Dropping schema "{}.{}".', database, schema)  # in lieu of SQL
+        fire_event(SchemaDrop(relation=_make_key(relation)))
         self.connections.drop_dataset(database, schema)
         self.cache.drop_schema(database, schema)
 

dbt/include/bigquery/macros/adapters.sql

@@ -76,10 +76,6 @@
 
 {% endmacro %}
 
-{% macro bigquery__create_schema(relation) -%}
-  {{ adapter.create_schema(relation) }}
-{% endmacro %}
-
 {% macro bigquery__drop_schema(relation) -%}
   {{ adapter.drop_schema(relation) }}
 {% endmacro %}

tests/unit/test_bigquery_adapter.py

@@ -349,11 +349,6 @@ def test_get_relation(self):
         self.adapter.get_relation('db', 'schema', 'my_model')
         self.mock_connection_manager.get_bq_table.assert_called_once_with('db', 'schema', 'my_model')
 
-    def test_create_schema(self):
-        relation = BigQueryRelation.create(database='db', schema='schema')
-        self.adapter.create_schema(relation)
-        self.mock_connection_manager.create_dataset.assert_called_once_with('db', 'schema')
-
     @patch.object(BigQueryAdapter, 'check_schema_exists')
     def test_drop_schema(self, mock_check_schema):
         mock_check_schema.return_value = True