Repository: dbt-labs/dbt-bigquery
Pull request #550 (2023-02-22 18:24:43)

[Backport] 404
resolves #https://github.com/dbt-labs/dbt-bigquery/issues/522

<!---
  Include the number of the issue addressed by this PR above if applicable.
  PRs for code changes without an associated issue *will not be merged*.
  See CONTRIBUTING.md for more information.

  Example:
    resolves #1234
-->

### Description
fix duplicate authorized views in 1.4

As noted in #522 following the steps to reproduce would indeed throw the `duplicate authorized views` error 

however if we backport #404 the run looks like

```
(env) ➜  jaffle-shop git:(master) ✗ (env) ➜  jaffle-shop git:(master) ✗ dbt run
18:39:09  Running with dbt=1.4.2-rc1
18:39:09  Found 2 models, 1 test, 0 snapshots, 0 analyses, 571 macros, 0 operations, 2 seed files, 0 sources, 0 exposures, 0 metrics
18:39:09
18:39:11  Concurrency: 1 threads (target='dev')
18:39:11
18:39:11  1 of 2 START sql view model dbt_mcknight.my_model .............................. [RUN]
18:39:12  1 of 2 OK created sql view model dbt_mcknight.my_model ......................... [CREATE VIEW (0 processed) in 1.10s]
18:39:12  2 of 2 START sql view model dbt_mcknight.second_model .......................... [RUN]
18:39:13  BigQuery adapter: Access entry <AccessEntry: role=None, view={'projectId': 'dbt-dev-168022', 'datasetId': 'dbt_mcknight', 'tableId': 'second_model'}> already exists in dataset
18:39:14  2 of 2 OK created sql view model dbt_mcknight.second_model ..................... [CREATE VIEW (0 processed) in 1.47s]
18:39:14
18:39:14  Finished running 2 view models in 0 hours 0 minutes and 5.09 seconds (5.09s).
18:39:14
18:39:14  Completed successfully
18:39:14
18:39:14  Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2
(env) ➜  jaffle-shop git:(master) ✗
```

throwing a log warn showing that the grants have already been given to the view that exists. 

running on 1.3.latest locally does allow for the model to run


original author: @colin-rogers-dbt 
backport author: @McKnight-42 
lead: @VersusFacit 
### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-bigquery/blob/main/CONTRIBUTING.md) and understand what's expected of me
- [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
- [ ] I have run this code in development and it appears to resolve the stated issue
- [ ] This PR includes tests, or tests are not required/relevant for this PR
- [ ] I have [opened an issue to add/update docs](https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose), or docs changes are not required/relevant for this PR
- [x] I have run `changie new` to [create a changelog entry](https://github.com/dbt-labs/dbt-bigquery/blob/main/CONTRIBUTING.md#Adding-CHANGELOG-Entry)



Patch:
diff --git a/.changes/unreleased/Fixes-20221213-102005.yaml b/.changes/unreleased/Fixes-20221213-102005.yaml
new file mode 100644
index 0000000..5e4c486
--- /dev/null
+++ b/.changes/unreleased/Fixes-20221213-102005.yaml
@@ -0,0 +1,7 @@
+kind: Fixes
+body: Make BQ access_grant updates thread safe
+time: 2022-12-13T10:20:05.714134-08:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "266"
+  PR: "404"
diff --git a/dbt/adapters/bigquery/dataset.py b/dbt/adapters/bigquery/dataset.py
new file mode 100644
index 0000000..ebffe10
--- /dev/null
+++ b/dbt/adapters/bigquery/dataset.py
@@ -0,0 +1,31 @@
+from typing import List
+from google.cloud.bigquery import Dataset, AccessEntry
+
+from dbt.events import AdapterLogger
+
+logger = AdapterLogger("BigQuery")
+
+
+def add_access_entry_to_dataset(dataset: Dataset, access_entry: AccessEntry) -> Dataset:
+    """Idempotently adds an access entry to a dataset
+
+    Args:
+        dataset (Dataset): the dataset to be updated
+        access_entry (AccessEntry): the access entry to be added to the dataset
+
+    Returns:
+        Dataset
+    """
+    access_entries: List[AccessEntry] = dataset.access_entries
+    # we can't simply check if an access entry is in the list as the current equality check
+    # does not work because the locally created AccessEntry can have extra properties.
+    for existing_entry in access_entries:
+        role_match = existing_entry.role == access_entry.role
+        entity_type_match = existing_entry.entity_type == access_entry.entity_type
+        property_match = existing_entry._properties.items() <= access_entry._properties.items()
+        if role_match and entity_type_match and property_match:
+            logger.warning(f"Access entry {access_entry} " f"already exists in dataset")
+            return dataset
+    access_entries.append(access_entry)
+    dataset.access_entries = access_entries
+    return dataset
diff --git a/dbt/adapters/bigquery/impl.py b/dbt/adapters/bigquery/impl.py
index 920b992..c41df44 100644
--- a/dbt/adapters/bigquery/impl.py
+++ b/dbt/adapters/bigquery/impl.py
@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+import threading
 from typing import Dict, List, Optional, Any, Set, Union, Type
 from dbt.dataclass_schema import dbtClassMixin, ValidationError
 
@@ -11,6 +12,7 @@ from dbt.adapters.base import (
     BaseAdapter,
     available,
     RelationType,
+    BaseRelation,
     SchemaSearchMap,
     AdapterConfig,
     PythonJobHelper,
@@ -19,6 +21,7 @@ from dbt.adapters.base import (
 from dbt.adapters.cache import _make_ref_key_msg
 
 from dbt.adapters.bigquery.relation import BigQueryRelation
+from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset
 from dbt.adapters.bigquery import BigQueryColumn
 from dbt.adapters.bigquery import BigQueryConnectionManager
 from dbt.adapters.bigquery.python_submissions import (
@@ -51,6 +54,7 @@ WRITE_APPEND = google.cloud.bigquery.job.WriteDisposition.WRITE_APPEND
 WRITE_TRUNCATE = google.cloud.bigquery.job.WriteDisposition.WRITE_TRUNCATE
 
 CREATE_SCHEMA_MACRO_NAME = "create_schema"
+_dataset_lock = threading.Lock()
 
 
 def sql_escape(string):
@@ -287,14 +291,16 @@ class BigQueryAdapter(BaseAdapter):
         # 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]  # type: ignore[misc]
         except google.api_core.exceptions.NotFound:
             return []
         except google.api_core.exceptions.Forbidden as exc:
             logger.debug("list_relations_without_caching error: {}".format(str(exc)))
             return []
 
-    def get_relation(self, database: str, schema: str, identifier: str) -> BigQueryRelation:
+    def get_relation(
+        self, database: str, schema: str, identifier: str
+    ) -> Optional[BigQueryRelation]:
         if self._schema_is_cached(database, schema):
             # if it's in the cache, use the parent's model of going through
             # the relations cache and picking out the relation
@@ -477,7 +483,7 @@ class BigQueryAdapter(BaseAdapter):
             message = "\n".join(error["message"].strip() for error in job.errors)
             raise dbt.exceptions.DbtRuntimeError(message)
 
-    def _bq_table_to_relation(self, bq_table):
+    def _bq_table_to_relation(self, bq_table) -> Union[BigQueryRelation, None]:
         if bq_table is None:
             return None
 
@@ -597,7 +603,7 @@ class BigQueryAdapter(BaseAdapter):
         """
         return PartitionConfig.parse(raw_partition_by)
 
-    def get_table_ref_from_relation(self, relation):
+    def get_table_ref_from_relation(self, relation: BaseRelation):
         return self.connections.table_ref(relation.database, relation.schema, relation.identifier)
 
     def _update_column_dict(self, bq_column_dict, dbt_columns, parent=""):
@@ -798,29 +804,20 @@ class BigQueryAdapter(BaseAdapter):
     @available.parse_none
     def grant_access_to(self, entity, entity_type, role, grant_target_dict):
         """
-        Given an entity, grants it access to a permissioned dataset.
+        Given an entity, grants it access to a dataset.
         """
-        conn = self.connections.get_thread_connection()
+        conn: BigQueryConnectionManager = self.connections.get_thread_connection()
         client = conn.handle
-
         GrantTarget.validate(grant_target_dict)
         grant_target = GrantTarget.from_dict(grant_target_dict)
-        dataset_ref = self.connections.dataset_ref(grant_target.project, grant_target.dataset)
-        dataset = client.get_dataset(dataset_ref)
-
         if entity_type == "view":
             entity = self.get_table_ref_from_relation(entity).to_api_repr()
-
-        access_entry = AccessEntry(role, entity_type, entity)
-        access_entries = dataset.access_entries
-
-        if access_entry in access_entries:
-            logger.debug(f"Access entry {access_entry} " f"already exists in dataset")
-            return
-
-        access_entries.append(AccessEntry(role, entity_type, entity))
-        dataset.access_entries = access_entries
-        client.update_dataset(dataset, ["access_entries"])
+        with _dataset_lock:
+            dataset_ref = self.connections.dataset_ref(grant_target.project, grant_target.dataset)
+            dataset = client.get_dataset(dataset_ref)
+            access_entry = AccessEntry(role, entity_type, entity)
+            dataset = add_access_entry_to_dataset(dataset, access_entry)
+            client.update_dataset(dataset, ["access_entries"])
 
     @available.parse_none
     def get_dataset_location(self, relation):


Test patch:
diff --git a/tests/functional/adapter/test_grant_access_to.py b/tests/functional/adapter/test_grant_access_to.py
new file mode 100644
index 0000000..7699647
--- /dev/null
+++ b/tests/functional/adapter/test_grant_access_to.py
@@ -0,0 +1,65 @@
+from abc import abstractmethod
+import pytest
+import os
+from dbt.tests.util import run_dbt
+
+SELECT_1 = """
+{{ config(
+    materialized='view',
+    grant_access_to=[
+      {'project': 'dbt-test-env-alt', 'dataset': 'GrantAccessTest'},
+    ]
+) }}
+SELECT 1 as one
+"""
+
+SELECT_1_TABLE = """
+{{ config(
+    materialized='table',
+    grant_access_to=[
+      {'project': 'dbt-test-env-alt', 'dataset': 'GrantAccessTest'},
+    ]
+) }}
+SELECT 1 as one
+"""
+BAD_CONFIG_TABLE_NAME = "bad_view"
+BAD_CONFIG_TABLE = """
+{{ config(
+    materialized='view',
+    grant_access_to=[
+      {'project': 'dbt-test-env-alt', 'dataset': 'NonExistentDataset'},
+    ]
+) }}
+SELECT 1 as one
+"""
+
+BAD_CONFIG_CHILD_TABLE = "SELECT 1 as one FROM {{ref('" + BAD_CONFIG_TABLE_NAME + "')}}"
+
+
+class TestAccessGrantSucceeds:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"select_1.sql": SELECT_1, "select_1_table.sql": SELECT_1_TABLE}
+
+    def test_grant_access_succeeds(self, project):
+        # Need to run twice to validate idempotency
+        results = run_dbt(["run"])
+        assert len(results) == 2
+        results = run_dbt(["run"])
+        assert len(results) == 2
+
+
+class TestAccessGrantFails:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "bad_config_table_child.sql": BAD_CONFIG_CHILD_TABLE,
+            f"{BAD_CONFIG_TABLE_NAME}.sql": BAD_CONFIG_TABLE,
+        }
+
+    def test_grant_access_fails_without_running_child_table(self, project):
+        # Need to run twice to validate idempotency
+        results = run_dbt(["run"], expect_pass=False)
+        assert results[0].status == "error"
+        assert results[1].status == "skipped"
+        assert results[0].message.startswith("404 GET https://bigquery.googleapis.com/")
diff --git a/tests/unit/test_bigquery_adapter.py b/tests/unit/test_bigquery_adapter.py
index 29c7a9e..6ac77ee 100644
--- a/tests/unit/test_bigquery_adapter.py
+++ b/tests/unit/test_bigquery_adapter.py
@@ -1,3 +1,4 @@
+from unicodedata import name
 import agate
 import decimal
 import json
@@ -27,6 +28,7 @@ from dbt.logger import GLOBAL_LOGGER as logger  # noqa
 from dbt.context.providers import RuntimeConfigObject
 
 import google.cloud.bigquery
+from google.cloud.bigquery import AccessEntry
 
 from .utils import config_from_parts_or_dicts, inject_adapter, TestAdapterConversions
 
@@ -38,85 +40,84 @@ def _bq_conn():
 
 
 class BaseTestBigQueryAdapter(unittest.TestCase):
-
     def setUp(self):
         self.raw_profile = {
-            'outputs': {
-                'oauth': {
-                    'type': 'bigquery',
-                    'method': 'oauth',
-                    'project': 'dbt-unit-000000',
-                    'schema': 'dummy_schema',
-                    'threads': 1,
+            "outputs": {
+                "oauth": {
+                    "type": "bigquery",
+                    "method": "oauth",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "threads": 1,
                 },
-                'service_account': {
-                    'type': 'bigquery',
-                    'method': 'service-account',
-                    'project': 'dbt-unit-000000',
-                    'schema': 'dummy_schema',
-                    'keyfile': '/tmp/dummy-service-account.json',
-                    'threads': 1,
+                "service_account": {
+                    "type": "bigquery",
+                    "method": "service-account",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "keyfile": "/tmp/dummy-service-account.json",
+                    "threads": 1,
                 },
-                'loc': {
-                    'type': 'bigquery',
-                    'method': 'oauth',
-                    'project': 'dbt-unit-000000',
-                    'schema': 'dummy_schema',
-                    'threads': 1,
-                    'location': 'Luna Station',
-                    'priority': 'batch',
-                    'maximum_bytes_billed': 0,
+                "loc": {
+                    "type": "bigquery",
+                    "method": "oauth",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "threads": 1,
+                    "location": "Luna Station",
+                    "priority": "batch",
+                    "maximum_bytes_billed": 0,
                 },
-                'impersonate': {
-                    'type': 'bigquery',
-                    'method': 'oauth',
-                    'project': 'dbt-unit-000000',
-                    'schema': 'dummy_schema',
-                    'threads': 1,
-                    'impersonate_service_account': 'dummyaccount@dbt.iam.gserviceaccount.com'
+                "impersonate": {
+                    "type": "bigquery",
+                    "method": "oauth",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "threads": 1,
+                    "impersonate_service_account": "dummyaccount@dbt.iam.gserviceaccount.com",
                 },
-                'oauth-credentials-token': {
-                    'type': 'bigquery',
-                    'method': 'oauth-secrets',
-                    'token': 'abc',
-                    'project': 'dbt-unit-000000',
-                    'schema': 'dummy_schema',
-                    'threads': 1,
-                    'location': 'Luna Station',
-                    'priority': 'batch',
-                    'maximum_bytes_billed': 0,
+                "oauth-credentials-token": {
+                    "type": "bigquery",
+                    "method": "oauth-secrets",
+                    "token": "abc",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "threads": 1,
+                    "location": "Luna Station",
+                    "priority": "batch",
+                    "maximum_bytes_billed": 0,
                 },
-                'oauth-credentials': {
-                    'type': 'bigquery',
-                    'method': 'oauth-secrets',
-                    'client_id': 'abc',
-                    'client_secret': 'def',
-                    'refresh_token': 'ghi',
-                    'token_uri': 'jkl',
-                    'project': 'dbt-unit-000000',
-                    'schema': 'dummy_schema',
-                    'threads': 1,
-                    'location': 'Luna Station',
-                    'priority': 'batch',
-                    'maximum_bytes_billed': 0,
+                "oauth-credentials": {
+                    "type": "bigquery",
+                    "method": "oauth-secrets",
+                    "client_id": "abc",
+                    "client_secret": "def",
+                    "refresh_token": "ghi",
+                    "token_uri": "jkl",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "threads": 1,
+                    "location": "Luna Station",
+                    "priority": "batch",
+                    "maximum_bytes_billed": 0,
                 },
-                'oauth-no-project': {
-                    'type': 'bigquery',
-                    'method': 'oauth',
-                    'schema': 'dummy_schema',
-                    'threads': 1,
-                    'location': 'Solar Station',
+                "oauth-no-project": {
+                    "type": "bigquery",
+                    "method": "oauth",
+                    "schema": "dummy_schema",
+                    "threads": 1,
+                    "location": "Solar Station",
                 },
             },
-            'target': 'oauth',
+            "target": "oauth",
         }
 
         self.project_cfg = {
-            'name': 'X',
-            'version': '0.1',
-            'project-root': '/tmp/dbt/does-not-exist',
-            'profile': 'default',
-            'config-version': 2,
+            "name": "X",
+            "version": "0.1",
+            "project-root": "/tmp/dbt/does-not-exist",
+            "profile": "default",
+            "config-version": 2,
         }
         self.qh_patch = None
 
@@ -125,7 +126,7 @@ class BaseTestBigQueryAdapter(unittest.TestCase):
             self.qh_patch.stop()
         super().tearDown()
 
-    def get_adapter(self, target):
+    def get_adapter(self, target) -> BigQueryAdapter:
         project = self.project_cfg.copy()
         profile = self.raw_profile.copy()
         profile['target'] = target
@@ -187,8 +188,8 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
     def test_acquire_connection_service_account_validations(self, mock_open_connection):
         adapter = self.get_adapter('service_account')
         try:
-            connection = adapter.acquire_connection('dummy')
-            self.assertEqual(connection.type, 'bigquery')
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
 
         except dbt.exceptions.DbtValidationError as e:
             self.fail('got DbtValidationError: {}'.format(str(e)))
@@ -200,12 +201,12 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
         connection.handle
         mock_open_connection.assert_called_once()
 
-    @patch('dbt.adapters.bigquery.BigQueryConnectionManager.open', return_value=_bq_conn())
+    @patch("dbt.adapters.bigquery.BigQueryConnectionManager.open", return_value=_bq_conn())
     def test_acquire_connection_oauth_token_validations(self, mock_open_connection):
-        adapter = self.get_adapter('oauth-credentials-token')
+        adapter = self.get_adapter("oauth-credentials-token")
         try:
-            connection = adapter.acquire_connection('dummy')
-            self.assertEqual(connection.type, 'bigquery')
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
 
         except dbt.exceptions.DbtValidationError as e:
             self.fail('got DbtValidationError: {}'.format(str(e)))
@@ -217,12 +218,12 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
         connection.handle
         mock_open_connection.assert_called_once()
 
-    @patch('dbt.adapters.bigquery.BigQueryConnectionManager.open', return_value=_bq_conn())
+    @patch("dbt.adapters.bigquery.BigQueryConnectionManager.open", return_value=_bq_conn())
     def test_acquire_connection_oauth_credentials_validations(self, mock_open_connection):
-        adapter = self.get_adapter('oauth-credentials')
+        adapter = self.get_adapter("oauth-credentials")
         try:
-            connection = adapter.acquire_connection('dummy')
-            self.assertEqual(connection.type, 'bigquery')
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
 
         except dbt.exceptions.DbtValidationError as e:
             self.fail('got DbtValidationError: {}'.format(str(e)))
@@ -234,12 +235,14 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
         connection.handle
         mock_open_connection.assert_called_once()
 
-    @patch('dbt.adapters.bigquery.BigQueryConnectionManager.open', return_value=_bq_conn())
-    def test_acquire_connection_impersonated_service_account_validations(self, mock_open_connection):
-        adapter = self.get_adapter('impersonate')
+    @patch("dbt.adapters.bigquery.BigQueryConnectionManager.open", return_value=_bq_conn())
+    def test_acquire_connection_impersonated_service_account_validations(
+        self, mock_open_connection
+    ):
+        adapter = self.get_adapter("impersonate")
         try:
-            connection = adapter.acquire_connection('dummy')
-            self.assertEqual(connection.type, 'bigquery')
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
 
         except dbt.exceptions.DbtValidationError as e:
             self.fail('got DbtValidationError: {}'.format(str(e)))
@@ -251,13 +254,13 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
         connection.handle
         mock_open_connection.assert_called_once()
 
-    @patch('dbt.adapters.bigquery.BigQueryConnectionManager.open', return_value=_bq_conn())
+    @patch("dbt.adapters.bigquery.BigQueryConnectionManager.open", return_value=_bq_conn())
     def test_acquire_connection_priority(self, mock_open_connection):
-        adapter = self.get_adapter('loc')
+        adapter = self.get_adapter("loc")
         try:
-            connection = adapter.acquire_connection('dummy')
-            self.assertEqual(connection.type, 'bigquery')
-            self.assertEqual(connection.credentials.priority, 'batch')
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
+            self.assertEqual(connection.credentials.priority, "batch")
 
         except dbt.exceptions.DbtValidationError as e:
             self.fail('got DbtValidationError: {}'.format(str(e)))
@@ -266,12 +269,12 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
         connection.handle
         mock_open_connection.assert_called_once()
 
-    @patch('dbt.adapters.bigquery.BigQueryConnectionManager.open', return_value=_bq_conn())
+    @patch("dbt.adapters.bigquery.BigQueryConnectionManager.open", return_value=_bq_conn())
     def test_acquire_connection_maximum_bytes_billed(self, mock_open_connection):
-        adapter = self.get_adapter('loc')
+        adapter = self.get_adapter("loc")
         try:
-            connection = adapter.acquire_connection('dummy')
-            self.assertEqual(connection.type, 'bigquery')
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
             self.assertEqual(connection.credentials.maximum_bytes_billed, 0)
 
         except dbt.exceptions.DbtValidationError as e:
@@ -282,63 +285,66 @@ class TestBigQueryAdapterAcquire(BaseTestBigQueryAdapter):
         mock_open_connection.assert_called_once()
 
     def test_cancel_open_connections_empty(self):
-        adapter = self.get_adapter('oauth')
+        adapter = self.get_adapter("oauth")
         self.assertEqual(adapter.cancel_open_connections(), None)
 
     def test_cancel_open_connections_master(self):
-        adapter = self.get_adapter('oauth')
+        adapter = self.get_adapter("oauth")
         adapter.connections.thread_connections[0] = object()
         self.assertEqual(adapter.cancel_open_connections(), None)
 
     def test_cancel_open_connections_single(self):
-        adapter = self.get_adapter('oauth')
-        adapter.connections.thread_connections.update({
-            0: object(),
-            1: object(),
-        })
+        adapter = self.get_adapter("oauth")
+        adapter.connections.thread_connections.update(
+            {
+                0: object(),
+                1: object(),
+            }
+        )
         # actually does nothing
         self.assertEqual(adapter.cancel_open_connections(), None)
 
-    @patch('dbt.adapters.bigquery.impl.google.auth.default')
-    @patch('dbt.adapters.bigquery.impl.google.cloud.bigquery')
+    @patch("dbt.adapters.bigquery.impl.google.auth.default")
+    @patch("dbt.adapters.bigquery.impl.google.cloud.bigquery")
     def test_location_user_agent(self, mock_bq, mock_auth_default):
         creds = MagicMock()
         mock_auth_default.return_value = (creds, MagicMock())
-        adapter = self.get_adapter('loc')
+        adapter = self.get_adapter("loc")
 
-        connection = adapter.acquire_connection('dummy')
+        connection = adapter.acquire_connection("dummy")
         mock_client = mock_bq.Client
 
         mock_client.assert_not_called()
         connection.handle
-        mock_client.assert_called_once_with('dbt-unit-000000', creds,
-                                            location='Luna Station',
-                                            client_info=HasUserAgent())
+        mock_client.assert_called_once_with(
+            "dbt-unit-000000", creds, location="Luna Station", client_info=HasUserAgent()
+        )
 
 
 class HasUserAgent:
-    PAT = re.compile(r'dbt-\d+\.\d+\.\d+((a|b|rc)\d+)?')
+    PAT = re.compile(r"dbt-\d+\.\d+\.\d+((a|b|rc)\d+)?")
 
     def __eq__(self, other):
-        compare = getattr(other, 'user_agent', '')
+        compare = getattr(other, "user_agent", "")
         return bool(self.PAT.match(compare))
 
 
 class TestConnectionNamePassthrough(BaseTestBigQueryAdapter):
-
     def setUp(self):
         super().setUp()
-        self._conn_patch = patch.object(BigQueryAdapter, 'ConnectionManager')
+        self._conn_patch = patch.object(BigQueryAdapter, "ConnectionManager")
         self.conn_manager_cls = self._conn_patch.start()
 
-        self._relation_patch = patch.object(BigQueryAdapter, 'Relation')
+        self._relation_patch = patch.object(BigQueryAdapter, "Relation")
         self.relation_cls = self._relation_patch.start()
 
         self.mock_connection_manager = self.conn_manager_cls.return_value
-        self.conn_manager_cls.TYPE = 'bigquery'
-        self.relation_cls.get_default_quote_policy.side_effect = BigQueryRelation.get_default_quote_policy
+        self.conn_manager_cls.TYPE = "bigquery"
+        self.relation_cls.get_default_quote_policy.side_effect = (
+            BigQueryRelation.get_default_quote_policy
+        )
 
-        self.adapter = self.get_adapter('oauth')
+        self.adapter = self.get_adapter("oauth")
 
     def tearDown(self):
         super().tearDown()
@@ -346,23 +352,25 @@ class TestConnectionNamePassthrough(BaseTestBigQueryAdapter):
         self._relation_patch.stop()
 
     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')
+        self.adapter.get_relation("db", "schema", "my_model")
+        self.mock_connection_manager.get_bq_table.assert_called_once_with(
+            "db", "schema", "my_model"
+        )
 
-    @patch.object(BigQueryAdapter, 'check_schema_exists')
+    @patch.object(BigQueryAdapter, "check_schema_exists")
     def test_drop_schema(self, mock_check_schema):
         mock_check_schema.return_value = True
-        relation = BigQueryRelation.create(database='db', schema='schema')
+        relation = BigQueryRelation.create(database="db", schema="schema")
         self.adapter.drop_schema(relation)
-        self.mock_connection_manager.drop_dataset.assert_called_once_with('db', 'schema')
+        self.mock_connection_manager.drop_dataset.assert_called_once_with("db", "schema")
 
     def test_get_columns_in_relation(self):
         self.mock_connection_manager.get_bq_table.side_effect = ValueError
         self.adapter.get_columns_in_relation(
-            MagicMock(database='db', schema='schema', identifier='ident'),
+            MagicMock(database="db", schema="schema", identifier="ident"),
         )
         self.mock_connection_manager.get_bq_table.assert_called_once_with(
-            database='db', schema='schema', identifier='ident'
+            database="db", schema="schema", identifier="ident"
         )
 
 
@@ -372,75 +380,49 @@ class TestBigQueryRelation(unittest.TestCase):
 
     def test_view_temp_relation(self):
         kwargs = {
-            'type': None,
-            'path': {
-                'database': 'test-project',
-                'schema': 'test_schema',
-                'identifier': 'my_view'
-            },
-            'quote_policy': {
-                'identifier': False
-            }
+            "type": None,
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "my_view"},
+            "quote_policy": {"identifier": False},
         }
         BigQueryRelation.validate(kwargs)
 
     def test_view_relation(self):
         kwargs = {
-            'type': 'view',
-            'path': {
-                'database': 'test-project',
-                'schema': 'test_schema',
-                'identifier': 'my_view'
-            },
-            'quote_policy': {
-                'identifier': True,
-                'schema': True
-            }
+            "type": "view",
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "my_view"},
+            "quote_policy": {"identifier": True, "schema": True},
         }
         BigQueryRelation.validate(kwargs)
 
     def test_table_relation(self):
         kwargs = {
-            'type': 'table',
-            'path': {
-                'database': 'test-project',
-                'schema': 'test_schema',
-                'identifier': 'generic_table'
+            "type": "table",
+            "path": {
+                "database": "test-project",
+                "schema": "test_schema",
+                "identifier": "generic_table",
             },
-            'quote_policy': {
-                'identifier': True,
-                'schema': True
-            }
+            "quote_policy": {"identifier": True, "schema": True},
         }
         BigQueryRelation.validate(kwargs)
 
     def test_external_source_relation(self):
         kwargs = {
-            'type': 'external',
-            'path': {
-                'database': 'test-project',
-                'schema': 'test_schema',
-                'identifier': 'sheet'
-            },
-            'quote_policy': {
-                'identifier': True,
-                'schema': True
-            }
+            "type": "external",
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "sheet"},
+            "quote_policy": {"identifier": True, "schema": True},
         }
         BigQueryRelation.validate(kwargs)
 
     def test_invalid_relation(self):
         kwargs = {
-            'type': 'invalid-type',
-            'path': {
-                'database': 'test-project',
-                'schema': 'test_schema',
-                'identifier': 'my_invalid_id'
+            "type": "invalid-type",
+            "path": {
+                "database": "test-project",
+                "schema": "test_schema",
+                "identifier": "my_invalid_id",
             },
-            'quote_policy': {
-                'identifier': False,
-                'schema': True
-            }
+            "quote_policy": {"identifier": False, "schema": True},
         }
         with self.assertRaises(dbt.dataclass_schema.ValidationError):
             BigQueryRelation.validate(kwargs)
@@ -453,28 +435,22 @@ class TestBigQueryInformationSchema(unittest.TestCase):
     def test_replace(self):
 
         kwargs = {
-            'type': None,
-            'path': {
-                'database': 'test-project',
-                'schema': 'test_schema',
-                'identifier': 'my_view'
-            },
+            "type": None,
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "my_view"},
             # test for #2188
-            'quote_policy': {
-                'database': False
+            "quote_policy": {"database": False},
+            "include_policy": {
+                "database": True,
+                "schema": True,
+                "identifier": True,
             },
-            'include_policy': {
-                'database': True,
-                'schema': True,
-                'identifier': True,
-            }
         }
         BigQueryRelation.validate(kwargs)
         relation = BigQueryRelation.from_dict(kwargs)
         info_schema = relation.information_schema()
 
-        tables_schema = info_schema.replace(information_schema_view='__TABLES__')
-        assert tables_schema.information_schema_view == '__TABLES__'
+        tables_schema = info_schema.replace(information_schema_view="__TABLES__")
+        assert tables_schema.information_schema_view == "__TABLES__"
         assert tables_schema.include_policy.schema is True
         assert tables_schema.include_policy.identifier is False
         assert tables_schema.include_policy.database is True
@@ -482,8 +458,8 @@ class TestBigQueryInformationSchema(unittest.TestCase):
         assert tables_schema.quote_policy.identifier is False
         assert tables_schema.quote_policy.database is False
 
-        schemata_schema = info_schema.replace(information_schema_view='SCHEMATA')
-        assert schemata_schema.information_schema_view == 'SCHEMATA'
+        schemata_schema = info_schema.replace(information_schema_view="SCHEMATA")
+        assert schemata_schema.information_schema_view == "SCHEMATA"
         assert schemata_schema.include_policy.schema is False
         assert schemata_schema.include_policy.identifier is True
         assert schemata_schema.include_policy.database is True
@@ -491,8 +467,8 @@ class TestBigQueryInformationSchema(unittest.TestCase):
         assert schemata_schema.quote_policy.identifier is False
         assert schemata_schema.quote_policy.database is False
 
-        other_schema = info_schema.replace(information_schema_view='SOMETHING_ELSE')
-        assert other_schema.information_schema_view == 'SOMETHING_ELSE'
+        other_schema = info_schema.replace(information_schema_view="SOMETHING_ELSE")
+        assert other_schema.information_schema_view == "SOMETHING_ELSE"
         assert other_schema.include_policy.schema is True
         assert other_schema.include_policy.identifier is True
         assert other_schema.include_policy.database is True
@@ -502,14 +478,12 @@ class TestBigQueryInformationSchema(unittest.TestCase):
 
 
 class TestBigQueryConnectionManager(unittest.TestCase):
-
     def setUp(self):
         credentials = Mock(BigQueryCredentials)
         profile = Mock(query_comment=None, credentials=credentials)
         self.connections = BigQueryConnectionManager(profile=profile)
 
-        self.mock_client = Mock(
-          dbt.adapters.bigquery.impl.google.cloud.bigquery.Client)
+        self.mock_client = Mock(dbt.adapters.bigquery.impl.google.cloud.bigquery.Client)
         self.mock_connection = MagicMock()
 
         self.mock_connection.handle = self.mock_client
@@ -518,8 +492,7 @@ class TestBigQueryConnectionManager(unittest.TestCase):
         self.connections.get_job_retry_deadline_seconds = lambda x: None
         self.connections.get_job_retries = lambda x: 1
 
-    @patch(
-        'dbt.adapters.bigquery.connections._is_retryable', return_value=True)
+    @patch("dbt.adapters.bigquery.connections._is_retryable", return_value=True)
     def test_retry_and_handle(self, is_retryable):
         self.connections.DEFAULT_MAXIMUM_DELAY = 2.0
 
@@ -531,6 +504,7 @@ class TestBigQueryConnectionManager(unittest.TestCase):
 
         class DummyException(Exception):
             """Count how many times this exception is raised"""
+
             count = 0
 
             def __init__(self):
@@ -541,12 +515,11 @@ class TestBigQueryConnectionManager(unittest.TestCase):
 
         with self.assertRaises(DummyException):
             self.connections._retry_and_handle(
-                 "some sql", Mock(credentials=Mock(retries=8)),
-                 raiseDummyException)
+                "some sql", Mock(credentials=Mock(retries=8)), raiseDummyException
+            )
             self.assertEqual(DummyException.count, 9)
 
-    @patch(
-        'dbt.adapters.bigquery.connections._is_retryable', return_value=True)
+    @patch("dbt.adapters.bigquery.connections._is_retryable", return_value=True)
     def test_retry_connection_reset(self, is_retryable):
         self.connections.open = MagicMock()
         self.connections.close = MagicMock()
@@ -563,20 +536,20 @@ class TestBigQueryConnectionManager(unittest.TestCase):
 
         mock_conn = Mock(credentials=Mock(retries=1))
         with self.assertRaises(ConnectionResetError):
-            self.connections._retry_and_handle(
-              "some sql", mock_conn,
-              raiseConnectionResetError)
+            self.connections._retry_and_handle("some sql", mock_conn, raiseConnectionResetError)
         self.connections.close.assert_called_once_with(mock_conn)
         self.connections.open.assert_called_once_with(mock_conn)
 
     def test_is_retryable(self):
         _is_retryable = dbt.adapters.bigquery.connections._is_retryable
         exceptions = dbt.adapters.bigquery.impl.google.cloud.exceptions
-        internal_server_error = exceptions.InternalServerError('code broke')
-        bad_request_error = exceptions.BadRequest('code broke')
-        connection_error = ConnectionError('code broke')
-        client_error = exceptions.ClientError('bad code')
-        rate_limit_error = exceptions.Forbidden("code broke", errors=[{"reason": "rateLimitExceeded"}])
+        internal_server_error = exceptions.InternalServerError("code broke")
+        bad_request_error = exceptions.BadRequest("code broke")
+        connection_error = ConnectionError("code broke")
+        client_error = exceptions.ClientError("bad code")
+        rate_limit_error = exceptions.Forbidden(
+            "code broke", errors=[{"reason": "rateLimitExceeded"}]
+        )
 
         self.assertTrue(_is_retryable(internal_server_error))
         self.assertTrue(_is_retryable(bad_request_error))
@@ -587,34 +560,38 @@ class TestBigQueryConnectionManager(unittest.TestCase):
     def test_drop_dataset(self):
 
         mock_table = Mock()
-        mock_table.reference = 'table1'
+        mock_table.reference = "table1"
         self.mock_client.list_tables.return_value = [mock_table]
 
-        self.connections.drop_dataset('project', 'dataset')
+        self.connections.drop_dataset("project", "dataset")
 
         self.mock_client.list_tables.assert_not_called()
         self.mock_client.delete_table.assert_not_called()
         self.mock_client.delete_dataset.assert_called_once()
 
-    @patch('dbt.adapters.bigquery.impl.google.cloud.bigquery')
+    @patch("dbt.adapters.bigquery.impl.google.cloud.bigquery")
     def test_query_and_results(self, mock_bq):
         self.connections._query_and_results(
-            self.mock_client, 'sql', {'job_param_1': 'blah'},
+            self.mock_client,
+            "sql",
+            {"job_param_1": "blah"},
             job_creation_timeout=15,
-            job_execution_timeout=100)
+            job_execution_timeout=100,
+        )
 
         mock_bq.QueryJobConfig.assert_called_once()
         self.mock_client.query.assert_called_once_with(
-            query='sql', job_config=mock_bq.QueryJobConfig(), timeout=15)
+            query="sql", job_config=mock_bq.QueryJobConfig(), timeout=15
+        )
 
     def test_copy_bq_table_appends(self):
-        self._copy_table(
-            write_disposition=dbt.adapters.bigquery.impl.WRITE_APPEND)
+        self._copy_table(write_disposition=dbt.adapters.bigquery.impl.WRITE_APPEND)
         args, kwargs = self.mock_client.copy_table.call_args
         self.mock_client.copy_table.assert_called_once_with(
-            [self._table_ref('project', 'dataset', 'table1')],
-            self._table_ref('project', 'dataset', 'table2'),
-            job_config=ANY)
+            [self._table_ref("project", "dataset", "table1")],
+            self._table_ref("project", "dataset", "table2"),
+            job_config=ANY,
+        )
         args, kwargs = self.mock_client.copy_table.call_args
         self.assertEqual(
             kwargs['job_config'].write_disposition,
@@ -654,25 +631,24 @@ class TestBigQueryConnectionManager(unittest.TestCase):
 
 
 class TestBigQueryAdapter(BaseTestBigQueryAdapter):
-
     def test_copy_table_materialization_table(self):
-        adapter = self.get_adapter('oauth')
+        adapter = self.get_adapter("oauth")
         adapter.connections = MagicMock()
-        adapter.copy_table('source', 'destination', 'table')
+        adapter.copy_table("source", "destination", "table")
         adapter.connections.copy_bq_table.assert_called_once_with(
-            'source', 'destination',
-            dbt.adapters.bigquery.impl.WRITE_TRUNCATE)
+            "source", "destination", dbt.adapters.bigquery.impl.WRITE_TRUNCATE
+        )
 
     def test_copy_table_materialization_incremental(self):
-        adapter = self.get_adapter('oauth')
+        adapter = self.get_adapter("oauth")
         adapter.connections = MagicMock()
-        adapter.copy_table('source', 'destination', 'incremental')
+        adapter.copy_table("source", "destination", "incremental")
         adapter.connections.copy_bq_table.assert_called_once_with(
-            'source', 'destination',
-            dbt.adapters.bigquery.impl.WRITE_APPEND)
+            "source", "destination", dbt.adapters.bigquery.impl.WRITE_APPEND
+        )
 
     def test_parse_partition_by(self):
-        adapter = self.get_adapter('oauth')
+        adapter = self.get_adapter("oauth")
 
         with self.assertRaises(dbt.exceptions.DbtValidationError):
             adapter.parse_partition_by("date(ts)")
@@ -681,163 +657,150 @@ class TestBigQueryAdapter(BaseTestBigQueryAdapter):
             adapter.parse_partition_by("ts")
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {
+                    "field": "ts",
+                }
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "date",
                 "granularity": "day",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "date",
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {
+                    "field": "ts",
+                    "data_type": "date",
+                }
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "date",
                 "granularity": "day",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "date",
-                "granularity": "MONTH"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "date", "granularity": "MONTH"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "date",
                 "granularity": "MONTH",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "date",
-                "granularity": "YEAR"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "date", "granularity": "YEAR"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "date",
                 "granularity": "YEAR",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "timestamp",
-                "granularity": "HOUR"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "timestamp", "granularity": "HOUR"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "timestamp",
                 "granularity": "HOUR",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "timestamp",
-                "granularity": "MONTH"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "timestamp", "granularity": "MONTH"}
+            ).to_dict(omit_none=True), {
                 "field": "ts",
                 "data_type": "timestamp",
                 "granularity": "MONTH",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "timestamp",
-                "granularity": "YEAR"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "timestamp", "granularity": "YEAR"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "timestamp",
                 "granularity": "YEAR",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "datetime",
-                "granularity": "HOUR"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "datetime", "granularity": "HOUR"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "datetime",
                 "granularity": "HOUR",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "datetime",
-                "granularity": "MONTH"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "datetime", "granularity": "MONTH"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "datetime",
                 "granularity": "MONTH",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "data_type": "datetime",
-                "granularity": "YEAR"
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "data_type": "datetime", "granularity": "YEAR"}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "datetime",
                 "granularity": "YEAR",
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "ts",
-                "time_ingestion_partitioning": True,
-                "copy_partitions": True
-
-            }).to_dict(omit_none=True), {
+            adapter.parse_partition_by(
+                {"field": "ts", "time_ingestion_partitioning": True, "copy_partitions": True}
+            ).to_dict(omit_none=True),
+            {
                 "field": "ts",
                 "data_type": "date",
                 "granularity": "day",
                 "time_ingestion_partitioning": True,
-                "copy_partitions": True
-            }
+                "copy_partitions": True,
+            },
         )
 
         # Invalid, should raise an error
@@ -846,38 +809,31 @@ class TestBigQueryAdapter(BaseTestBigQueryAdapter):
 
         # passthrough
         self.assertEqual(
-            adapter.parse_partition_by({
-                "field": "id",
-                "data_type": "int64",
-                "range": {
-                    "start": 1,
-                    "end": 100,
-                    "interval": 20
+            adapter.parse_partition_by(
+                {
+                    "field": "id",
+                    "data_type": "int64",
+                    "range": {"start": 1, "end": 100, "interval": 20},
                 }
-            }).to_dict(omit_none=True
-                ), {
+            ).to_dict(omit_none=True),
+            {
                 "field": "id",
                 "data_type": "int64",
                 "granularity": "day",
-                "range": {
-                    "start": 1,
-                    "end": 100,
-                    "interval": 20
-                },
+                "range": {"start": 1, "end": 100, "interval": 20},
                 "time_ingestion_partitioning": False,
-                "copy_partitions": False
-            }
+                "copy_partitions": False,
+            },
         )
 
     def test_hours_to_expiration(self):
-        adapter = self.get_adapter('oauth')
-        mock_config = create_autospec(
-            RuntimeConfigObject)
-        config = {'hours_to_expiration': 4}
+        adapter = self.get_adapter("oauth")
+        mock_config = create_autospec(RuntimeConfigObject)
+        config = {"hours_to_expiration": 4}
         mock_config.get.side_effect = lambda name: config.get(name)
 
         expected = {
-            'expiration_timestamp': 'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 4 hour)',
+            "expiration_timestamp": "TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 4 hour)",
         }
         actual = adapter.get_table_options(mock_config, node={}, temporary=False)
         self.assertEqual(expected, actual)
@@ -890,8 +846,7 @@ class TestBigQueryAdapter(BaseTestBigQueryAdapter):
         mock_config.get.side_effect = lambda name: config.get(name)
 
         expected = {
-            'expiration_timestamp': (
-                'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 12 hour)'),
+            "expiration_timestamp": ("TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 12 hour)"),
         }
         actual = adapter.get_table_options(mock_config, node={}, temporary=True)
         self.assertEqual(expected, actual)
@@ -903,9 +858,7 @@ class TestBigQueryAdapter(BaseTestBigQueryAdapter):
         config = {'kms_key_name': 'some_key'}
         mock_config.get.side_effect = lambda name: config.get(name)
 
-        expected = {
-            'kms_key_name': "'some_key'"
-        }
+        expected = {"kms_key_name": "'some_key'"}
         actual = adapter.get_table_options(mock_config, node={}, temporary=False)
         self.assertEqual(expected, actual)
 
@@ -924,80 +877,80 @@ class TestBigQueryAdapter(BaseTestBigQueryAdapter):
 class TestBigQueryFilterCatalog(unittest.TestCase):
     def test__catalog_filter_table(self):
         manifest = MagicMock()
-        manifest.get_used_schemas.return_value = [['a', 'B'], ['a', '1234']]
-        column_names = ['table_name', 'table_database', 'table_schema', 'something']
+        manifest.get_used_schemas.return_value = [["a", "B"], ["a", "1234"]]
+        column_names = ["table_name", "table_database", "table_schema", "something"]
         rows = [
-            ['foo', 'a', 'b', '1234'],  # include
-            ['foo', 'a', '1234', '1234'],  # include, w/ table schema as str
-            ['foo', 'c', 'B', '1234'],  # skip
-            ['1234', 'A', 'B', '1234'],  # include, w/ table name as str
+            ["foo", "a", "b", "1234"],  # include
+            ["foo", "a", "1234", "1234"],  # include, w/ table schema as str
+            ["foo", "c", "B", "1234"],  # skip
+            ["1234", "A", "B", "1234"],  # include, w/ table name as str
         ]
-        table = agate.Table(
-            rows, column_names, agate_helper.DEFAULT_TYPE_TESTER
-        )
+        table = agate.Table(rows, column_names, agate_helper.DEFAULT_TYPE_TESTER)
 
         result = BigQueryAdapter._catalog_filter_table(table, manifest)
         assert len(result) == 3
         for row in result.rows:
-            assert isinstance(row['table_schema'], str)
-            assert isinstance(row['table_database'], str)
-            assert isinstance(row['table_name'], str)
-            assert isinstance(row['something'], decimal.Decimal)
+            assert isinstance(row["table_schema"], str)
+            assert isinstance(row["table_database"], str)
+            assert isinstance(row["table_name"], str)
+            assert isinstance(row["something"], decimal.Decimal)
 
 
 class TestBigQueryAdapterConversions(TestAdapterConversions):
     def test_convert_text_type(self):
         rows = [
-            ['', 'a1', 'stringval1'],
-            ['', 'a2', 'stringvalasdfasdfasdfa'],
-            ['', 'a3', 'stringval3'],
+            ["", "a1", "stringval1"],
+            ["", "a2", "stringvalasdfasdfasdfa"],
+            ["", "a3", "stringval3"],
         ]
         agate_table = self._make_table_of(rows, agate.Text)
-        expected = ['string', 'string', 'string']
+        expected = ["string", "string", "string"]
         for col_idx, expect in enumerate(expected):
             assert BigQueryAdapter.convert_text_type(agate_table, col_idx) == expect
 
     def test_convert_number_type(self):
         rows = [
-            ['', '23.98', '-1'],
-            ['', '12.78', '-2'],
-            ['', '79.41', '-3'],
+            ["", "23.98", "-1"],
+            ["", "12.78", "-2"],
+            ["", "79.41", "-3"],
         ]
         agate_table = self._make_table_of(rows, agate.Number)
-        expected = ['int64', 'float64', 'int64']
+        expected = ["int64", "float64", "int64"]
         for col_idx, expect in enumerate(expected):
             assert BigQueryAdapter.convert_number_type(agate_table, col_idx) == expect
 
     def test_convert_boolean_type(self):
         rows = [
-            ['', 'false', 'true'],
-            ['', 'false', 'false'],
-            ['', 'false', 'true'],
+            ["", "false", "true"],
+            ["", "false", "false"],
+            ["", "false", "true"],
         ]
         agate_table = self._make_table_of(rows, agate.Boolean)
-        expected = ['bool', 'bool', 'bool']
+        expected = ["bool", "bool", "bool"]
         for col_idx, expect in enumerate(expected):
             assert BigQueryAdapter.convert_boolean_type(agate_table, col_idx) == expect
 
     def test_convert_datetime_type(self):
         rows = [
-            ['', '20190101T01:01:01Z', '2019-01-01 01:01:01'],
-            ['', '20190102T01:01:01Z', '2019-01-01 01:01:01'],
-            ['', '20190103T01:01:01Z', '2019-01-01 01:01:01'],
+            ["", "20190101T01:01:01Z", "2019-01-01 01:01:01"],
+            ["", "20190102T01:01:01Z", "2019-01-01 01:01:01"],
+            ["", "20190103T01:01:01Z", "2019-01-01 01:01:01"],
         ]
-        agate_table = self._make_table_of(rows, [agate.DateTime, agate_helper.ISODateTime, agate.DateTime])
-        expected = ['datetime', 'datetime', 'datetime']
+        agate_table = self._make_table_of(
+            rows, [agate.DateTime, agate_helper.ISODateTime, agate.DateTime]
+        )
+        expected = ["datetime", "datetime", "datetime"]
         for col_idx, expect in enumerate(expected):
             assert BigQueryAdapter.convert_datetime_type(agate_table, col_idx) == expect
 
     def test_convert_date_type(self):
         rows = [
-            ['', '2019-01-01', '2019-01-04'],
-            ['', '2019-01-02', '2019-01-04'],
-            ['', '2019-01-03', '2019-01-04'],
+            ["", "2019-01-01", "2019-01-04"],
+            ["", "2019-01-02", "2019-01-04"],
+            ["", "2019-01-03", "2019-01-04"],
         ]
         agate_table = self._make_table_of(rows, agate.Date)
-        expected = ['date', 'date', 'date']
+        expected = ["date", "date", "date"]
         for col_idx, expect in enumerate(expected):
             assert BigQueryAdapter.convert_date_type(agate_table, col_idx) == expect
 
@@ -1005,16 +958,57 @@ class TestBigQueryAdapterConversions(TestAdapterConversions):
         # dbt's default type testers actually don't have a TimeDelta at all.
         agate.TimeDelta
         rows = [
-            ['', '120s', '10s'],
-            ['', '3m', '11s'],
-            ['', '1h', '12s'],
+            ["", "120s", "10s"],
+            ["", "3m", "11s"],
+            ["", "1h", "12s"],
         ]
         agate_table = self._make_table_of(rows, agate.TimeDelta)
-        expected = ['time', 'time', 'time']
+        expected = ["time", "time", "time"]
         for col_idx, expect in enumerate(expected):
             assert BigQueryAdapter.convert_time_type(agate_table, col_idx) == expect
 
 
+class TestBigQueryGrantAccessTo(BaseTestBigQueryAdapter):
+    entity = BigQueryRelation.from_dict(
+        {
+            "type": None,
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "my_view"},
+            "quote_policy": {"identifier": False},
+        }
+    )
+
+    def setUp(self):
+        super().setUp()
+        self.mock_dataset: MagicMock = MagicMock(name="GrantMockDataset")
+        self.mock_dataset.access_entries = [AccessEntry(None, "table", self.entity)]
+        self.mock_client: MagicMock = MagicMock(name="MockBQClient")
+        self.mock_client.get_dataset.return_value = self.mock_dataset
+        self.mock_connection = MagicMock(name="MockConn")
+        self.mock_connection.handle = self.mock_client
+        self.mock_connection_mgr = MagicMock(
+            name="GrantAccessMockMgr",
+        )
+        self.mock_connection_mgr.get_thread_connection.return_value = self.mock_connection
+        _adapter = self.get_adapter("oauth")
+        _adapter.connections = self.mock_connection_mgr
+        self.adapter = _adapter
+
+    def test_grant_access_to_calls_update_with_valid_access_entry(self):
+        a_different_entity = BigQueryRelation.from_dict(
+            {
+                "type": None,
+                "path": {"database": "another-test-project",
+                 "schema": "test_schema_2", "identifier": "my_view"},
+                "quote_policy": {"identifier": True},
+            }
+        )
+        grant_target_dict = {"dataset": "someOtherDataset", "project": "someProject"}
+        self.adapter.grant_access_to(
+            entity=a_different_entity, entity_type="view", role=None, grant_target_dict=grant_target_dict
+        )
+        self.mock_client.update_dataset.assert_called_once()
+
+
 @pytest.mark.parametrize(
     ["input", "output"],
     [
diff --git a/tests/unit/test_dataset.py b/tests/unit/test_dataset.py
new file mode 100644
index 0000000..53109e5
--- /dev/null
+++ b/tests/unit/test_dataset.py
@@ -0,0 +1,43 @@
+from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset
+from dbt.adapters.bigquery import BigQueryRelation
+
+from google.cloud.bigquery import Dataset, AccessEntry, DatasetReference
+
+
+def test_add_access_entry_to_dataset_idempotently_adds_entries():
+    database = "someDb"
+    dataset = "someDataset"
+    entity = BigQueryRelation.from_dict(
+        {
+            "type": None,
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "my_table"},
+            "quote_policy": {"identifier": False},
+        }
+    ).to_dict()
+    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
+    dataset = Dataset(dataset_ref)
+    access_entry = AccessEntry(None, "table", entity)
+    dataset = add_access_entry_to_dataset(dataset, access_entry)
+    assert access_entry in dataset.access_entries
+    dataset = add_access_entry_to_dataset(dataset, access_entry)
+    assert len(dataset.access_entries) == 1
+
+
+def test_add_access_entry_to_dataset_does_not_add_with_pre_existing_entries():
+    database = "someOtherDb"
+    dataset = "someOtherDataset"
+    entity_2 = BigQueryRelation.from_dict(
+        {
+            "type": None,
+            "path": {"database": "test-project", "schema": "test_schema", "identifier": "some_other_view"},
+            "quote_policy": {"identifier": False},
+        }
+    ).to_dict()
+    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
+    dataset = Dataset(dataset_ref)
+    initial_entry = AccessEntry(None, "view", entity_2)
+    initial_entry._properties.pop("role")
+    dataset.access_entries = [initial_entry]
+    access_entry = AccessEntry(None, "view", entity_2)
+    dataset = add_access_entry_to_dataset(dataset, access_entry)
+    assert len(dataset.access_entries) == 1