Repository: dbt-labs/dbt-bigquery (python)

Issue:
[ADAP-622] [Regression] Authorised views `grant_access_to` exceeded rate limit error 
### Is this a regression in a recent version of dbt-bigquery?

- [X] I believe this is a regression in dbt-bigquery functionality
- [X] I have searched the existing issues, and I could not find an existing issue for this regression

### Current Behavior

When creating many authorized views on a single dataset in a short amount of time, the [bigquery update dataset limit](https://cloud.google.com/bigquery/quotas#dataset_limits) of 5 update operations per 10 seconds is quickly reached. 

This is unavoidable and expected the first time the authorized views are created, but on subsequent dbt runs when the authorized views already exist, I am now also seeing this limit being hit which previously was not happening. 

When the dbt logs show this warning:
```
BigQuery adapter: Access entry <AccessEntry: role=None, view={'projectId': '...', 'datasetId': '...', 'tableId': '...'}> already exists in dataset
```
BigQuery logs are giving this error
```
NOTICE 2023-06-13T20:26:42.048907Z [protoPayload.serviceName: bigquery.googleapis.com] [protoPayload.methodName: google.cloud.bigquery.v2.DatasetService.PatchDataset] [protoPayload.resourceName:...] [protoPayload.authenticationInfo.principalEmail:...] audit_log, method: "google.cloud.bigquery.v2.DatasetService.PatchDataset", principal_email: "..."
ERROR 2023-06-13T20:26:42.439870Z [protoPayload.serviceName: bigquery.googleapis.com] [protoPayload.methodName: google.cloud.bigquery.v2.DatasetService.PatchDataset] [protoPayload.resourceName:...] [protoPayload.authenticationInfo.principalEmail:...] Exceeded rate limits: too many dataset metadata update operations for this dataset. For more information, see https://cloud.google.com/bigquery/docs/troubleshoot-quotas
ERROR 2023-06-13T20:26:42.630266Z [protoPayload.serviceName: bigquery.googleapis.com] [protoPayload.methodName: google.cloud.bigquery.v2.DatasetService.PatchDataset] [protoPayload.resourceName:...] [protoPayload.authenticationInfo.principalEmail:...] Exceeded rate limits: too many dataset metadata update operations for this dataset. For more information, see https://cloud.google.com/bigquery/docs/troubleshoot-quotas
```

### Expected/Previous Behavior

Previously when an authorized view already existed in a dataset, using `grant_access_to` would show the `already exists in dataset` warning but not cause an additional unnecessary dataset update. 

### Steps To Reproduce

With dbt-bigquery >= v1.4.2, create a lot of authorized views (>5) on a single dataset and do two dbt runs.

After the second run, in gcp logs explorer run the following query to view rate exceeded errors 
```
resource.type = "bigquery_dataset"
protoPayload.methodName:"google.cloud.bigquery.v2.DatasetService.PatchDataset"
```





### Relevant log output

_No response_

### Environment

```markdown
- OS: macos 13.4
- Python:3.9.6
- dbt-core (working version):1.4.6
- dbt-bigquery (working version):1.4.1
- dbt-core (regression version):1.4.6
- dbt-bigquery (regression version):1.4.2
```


### Additional Context

This issue is due to a [recent change](https://github.com/dbt-labs/dbt-bigquery/pull/404) with the BigQueryAdapter grant_access_to() method which now will call the bigquery patch.dataset api even when the authorized view already exists in the dataset. 

I don't believe this is expected behaviour because of the relatively low limit for this operation, calling it when unnecessary has the potential to significantly impact performance.


Pull request:
Avoid unnecessary grant call
resolves #770 

### Description
 remove call to dataset update if dataset has not changed
<!---
  Describe the Pull Request here. Add any references and info to help reviewers
  understand your changes. Include any tradeoffs you considered.
-->

### Checklist

- [ ] I have read [the contributing guide](https://github.com/dbt-labs/dbt-bigquery/blob/main/CONTRIBUTING.md) and understand what's expected of me
- [ ] 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
- [ ] 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-20230626-105156.yaml b/.changes/unreleased/Fixes-20230626-105156.yaml
new file mode 100644
index 0000000..d1c6b9e
--- /dev/null
+++ b/.changes/unreleased/Fixes-20230626-105156.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: remove call to dataset update if dataset has not changed
+time: 2023-06-26T10:51:56.698483-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "770"
diff --git a/dbt/adapters/bigquery/dataset.py b/dbt/adapters/bigquery/dataset.py
index ebffe10..c886637 100644
--- a/dbt/adapters/bigquery/dataset.py
+++ b/dbt/adapters/bigquery/dataset.py
@@ -6,15 +6,15 @@ 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
+def is_access_entry_in_dataset(dataset: Dataset, access_entry: AccessEntry) -> bool:
+    """Check if the access entry already exists in the dataset.
 
     Args:
         dataset (Dataset): the dataset to be updated
         access_entry (AccessEntry): the access entry to be added to the dataset
 
     Returns:
-        Dataset
+        bool: True if entry exists in dataset, False otherwise
     """
     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
@@ -24,8 +24,22 @@ def add_access_entry_to_dataset(dataset: Dataset, access_entry: AccessEntry) ->
         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
+            return True
+    return False
+
+
+def add_access_entry_to_dataset(dataset: Dataset, access_entry: AccessEntry) -> Dataset:
+    """Adds an access entry to a dataset, always use access_entry_present_in_dataset to check
+    if the access entry already exists before calling this function.
+
+    Args:
+        dataset (Dataset): the dataset to be updated
+        access_entry (AccessEntry): the access entry to be added to the dataset
+
+    Returns:
+        Dataset: the updated dataset
+    """
+    access_entries: List[AccessEntry] = dataset.access_entries
     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 37ed2c2..5c965ca 100644
--- a/dbt/adapters/bigquery/impl.py
+++ b/dbt/adapters/bigquery/impl.py
@@ -25,7 +25,7 @@ from dbt.adapters.cache import _make_ref_key_dict  # type: ignore
 
 from dbt.adapters.bigquery.column import get_nested_column_data_types
 from dbt.adapters.bigquery.relation import BigQueryRelation
-from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset
+from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset, is_access_entry_in_dataset
 from dbt.adapters.bigquery import BigQueryColumn
 from dbt.adapters.bigquery import BigQueryConnectionManager
 from dbt.adapters.bigquery.python_submissions import (
@@ -899,8 +899,12 @@ class BigQueryAdapter(BaseAdapter):
             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"])
+            # only perform update if access entry not in dataset
+            if is_access_entry_in_dataset(dataset, access_entry):
+                logger.warning(f"Access entry {access_entry} " f"already exists in dataset")
+            else:
+                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/unit/test_dataset.py b/tests/unit/test_dataset.py
index 6815931..adb3964 100644
--- a/tests/unit/test_dataset.py
+++ b/tests/unit/test_dataset.py
@@ -1,10 +1,10 @@
-from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset
+from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset, is_access_entry_in_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():
+def test_add_access_entry_to_dataset_updates_dataset():
     database = "someDb"
     dataset = "someDataset"
     entity = BigQueryRelation.from_dict(
@@ -23,11 +23,9 @@ def test_add_access_entry_to_dataset_idempotently_adds_entries():
     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():
+def test_add_access_entry_to_dataset_updates_with_pre_existing_entries():
     database = "someOtherDb"
     dataset = "someOtherDataset"
     entity_2 = BigQueryRelation.from_dict(
@@ -48,4 +46,45 @@ def test_add_access_entry_to_dataset_does_not_add_with_pre_existing_entries():
     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
+    assert len(dataset.access_entries) == 2
+
+
+def test_is_access_entry_in_dataset_returns_true_if_entry_in_dataset():
+    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 is_access_entry_in_dataset(dataset, access_entry)
+
+
+def test_is_access_entry_in_dataset_returns_false_if_entry_not_in_dataset():
+    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)
+    assert not is_access_entry_in_dataset(dataset, access_entry)