fix 409 duplicate batch issue (#1020)

* use dynamic schema in test_grant_access_to.py

* use dynamic schema in test_grant_access_to.py

* revert setup

* fix partitioninb

* skip TestPythonBatchIdModels

* add changie

* run python integration tests separately

* run python integration tests separately

* cleanup _get_batch_id

* add space to pipe

* fix integration.yml

* add extra changie

* use dynamic schema in test_grant_access_to.py

* use dynamic schema in test_grant_access_to.py

* revert setup

* replace deterministic default batch_id with uuid

* add changie

(cherry picked from commit 0c5422c)

.changes/unreleased/Fixes-20231111-150959.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: replace deterministic batch_id with uuid
+time: 2023-11-11T15:09:59.243797-08:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1006"

dbt/adapters/bigquery/python_submissions.py

@@ -1,3 +1,4 @@
+import uuid
 from typing import Dict, Union
 
 from dbt.events import AdapterLogger
@@ -126,8 +127,7 @@ def _get_job_client(self) -> dataproc_v1.BatchControllerClient:
 
     def _get_batch_id(self) -> str:
         model = self.parsed_model
-        default_batch_id = model["unique_id"].replace(".", "-").replace("_", "-")
-        default_batch_id += str(int(model["created_at"]))
+        default_batch_id = str(uuid.uuid4())
         return model["config"].get("batch_id", default_batch_id)
 
     def _submit_dataproc_job(self) -> Batch:

tests/functional/adapter/test_python_model.py

@@ -187,8 +187,6 @@ def model(dbt, spark):
 models:
   - name: python_array_batch_id
     description: A random table with a calculated column defined in python.
-    config:
-      batch_id: '{{ run_started_at.strftime("%Y-%m-%d-%H-%M-%S") }}-python-array'
     columns:
       - name: A
         description: Column A
@@ -216,7 +214,6 @@ def model(dbt, spark):
 """
 
 
-@pytest.mark.skip(reason="Currently failing as run_started_at is the same across dbt runs")
 class TestPythonBatchIdModels:
     @pytest.fixture(scope="class")
     def models(self):