fix(bigquery): prevent duplicate job submission (stable job_id + 409 recovery) (#2054)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tauhid621

dbt-bigquery/.changes/unreleased/Fixes-20260618-120000.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Hardened job submission against duplicate writes and spurious 409 Already Exists failures. Query and copy jobs now submit with a single predetermined job_id and attach to the in-flight job via get_job when BigQuery reports the job already exists, instead of resubmitting. This prevents a retryable error escaping the polling retry (or the client library's transport retry resubmitting after a lost response) from either re-running non-idempotent DML as a second job or failing the run with a 409.
+time: 2026-06-18T12:00:00.000000-05:00
+custom:
+    Author: tauhid621
+    Issue: "2006"

dbt-bigquery/src/dbt/adapters/bigquery/connections.py

@@ -6,7 +6,7 @@
 from multiprocessing.context import SpawnContext
 import re
 import time
-from typing import Dict, Hashable, List, Optional, Tuple, TYPE_CHECKING
+from typing import Callable, Dict, Hashable, List, Optional, Tuple, TYPE_CHECKING
 import uuid
 
 from google.auth.exceptions import RefreshError
@@ -22,7 +22,7 @@
     Table,
     TableReference,
 )
-from google.cloud.exceptions import BadRequest, Forbidden, NotFound
+from google.cloud.exceptions import BadRequest, Conflict, Forbidden, NotFound
 
 from dbt_common.events.contextvars import get_node_info
 from dbt_common.events.functions import fire_event
@@ -233,6 +233,22 @@ def generate_job_id(self) -> str:
         self.jobs_by_thread[thread_id].append(job_id)
         return job_id
 
+    def _submit_or_attach(self, client: Client, job_id: str, submit: Callable):
+        """Submit a job, or attach to the existing one via get_job on 409 Conflict.
+
+        A stable job_id makes jobs.insert idempotent: a resubmission (dbt's retry
+        or the client library's transport retry after a lost response) attaches to
+        the in-flight job instead of spawning a second one that re-runs work.
+        """
+        try:
+            return submit()
+        except Conflict:
+            logger.debug(
+                f"Job {job_id} already exists; attaching to the in-flight job "
+                "instead of resubmitting to avoid duplicate execution."
+            )
+            return client.get_job(job_id)
+
     def raw_execute(
         self,
         sql,
@@ -277,9 +293,11 @@ def raw_execute(
             job_params["job_timeout_ms"] = int(model_timeout * 1000)
 
         with self.exception_handler(sql):
+            # Mint the job_id once, outside the retry closure, so a re-entry
+            # resubmits the same job instead of spawning a duplicate.
+            job_id = self.generate_job_id()
 
             def _execute_with_retry():
-                job_id = self.generate_job_id()
                 return self._query_and_results(
                     conn,
                     sql,
@@ -484,11 +502,18 @@ def copy_bq_table(self, source, destination, write_disposition) -> None:
             destination_ref.path,
         )
         with self.exception_handler(msg):
-            copy_job = client.copy_table(
-                source_ref_array,
-                destination_ref,
-                job_config=CopyJobConfig(write_disposition=write_disposition),
-                retry=self._retry.create_reopen_with_deadline(conn),
+            # Stable job_id: copy_table has no built-in 409 recovery of its own.
+            job_id = self.generate_job_id()
+            copy_job = self._submit_or_attach(
+                client,
+                job_id,
+                lambda: client.copy_table(
+                    source_ref_array,
+                    destination_ref,
+                    job_config=CopyJobConfig(write_disposition=write_disposition),
+                    job_id=job_id,
+                    retry=self._retry.create_reopen_with_deadline(conn),
+                ),
             )
             model_timeout = getattr(conn, "_bq_model_timeout", None)
             copy_timeout = model_timeout or self._retry.create_job_execution_timeout(fallback=300)
@@ -626,13 +651,18 @@ def _query_and_results(
         polling_timeout = (
             timeout + 30 if timeout else None
         )  # buffer for polling after job execution timeout
-        # Cannot reuse job_config if destination is set and ddl is used
-        query_job = client.query(
-            query=sql,
-            job_config=query_job_config,
-            job_id=job_id,
-            job_retry=None,
-            timeout=self._retry.create_job_creation_timeout(),
+        # Cannot reuse job_config if destination is set and ddl is used.
+        # job_id is stable across retries (see raw_execute).
+        query_job = self._submit_or_attach(
+            client,
+            job_id,
+            lambda: client.query(
+                query=sql,
+                job_config=query_job_config,
+                job_id=job_id,
+                job_retry=None,
+                timeout=self._retry.create_job_creation_timeout(),
+            ),
         )
         if (
             query_job.location is not None

dbt-bigquery/tests/functional/adapter/test_stable_job_id_409.py

@@ -0,0 +1,85 @@
+import uuid
+
+import pytest
+
+from dbt.tests.util import run_dbt
+
+
+_SEED = "select 1 as id"
+
+
+class TestStableJobIdAttachesOnConflict:
+    """End-to-end proof that a re-used job_id triggers a real BigQuery 409 and
+    that dbt attaches to the in-flight/finished job (via get_job) instead of
+    resubmitting non-idempotent DML a second time (inc-6741 / PR #2054).
+
+    With the old behavior (fresh job_id per attempt) the INSERT would run twice
+    and the row count would be 2. With the stable job_id + 409-attach fix it
+    stays 1.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        # A trivial model just to get a schema/dataset created for the test.
+        return {"anchor.sql": _SEED}
+
+    def test_resubmit_same_job_id_does_not_duplicate(self, project):
+        run_dbt(["run"])
+
+        conns = project.adapter.connections
+        table = f"`{project.database}`.`{project.test_schema}`.`conflict_probe`"
+
+        with project.adapter.connection_named("__test_409"):
+            conns.raw_execute(f"create or replace table {table} (id int64)")
+
+            # Pin the job_id so the second submission collides in BigQuery.
+            fixed_id = f"dbt-conflict-test-{uuid.uuid4()}"
+            original = conns.generate_job_id
+            conns.generate_job_id = lambda: fixed_id
+            try:
+                dml = f"insert into {table} (id) values (1)"
+                conns.raw_execute(dml)  # 1st: real insert, job created
+                conns.raw_execute(dml)  # 2nd: same job_id -> 409 -> attach, no re-insert
+            finally:
+                conns.generate_job_id = original
+
+            _, iterator = conns.raw_execute(f"select count(*) as n from {table}")
+            count = list(iterator)[0][0]
+
+        assert count == 1, f"expected 1 row (attached to existing job), got {count}"
+
+    def test_copy_job_resubmit_attaches(self, project):
+        """copy_bq_table has no built-in 409 recovery; verify _submit_or_attach
+        keeps a resubmitted copy from failing the run."""
+        run_dbt(["run"])
+
+        conns = project.adapter.connections
+        src = project.adapter.Relation.create(
+            database=project.database, schema=project.test_schema, identifier="copy_src"
+        )
+        dst = project.adapter.Relation.create(
+            database=project.database, schema=project.test_schema, identifier="copy_dst"
+        )
+
+        with project.adapter.connection_named("__test_409_copy"):
+            conns.raw_execute(
+                f"create or replace table `{src.database}`.`{src.schema}`.`{src.identifier}` "
+                "as select 1 as id"
+            )
+
+            fixed_id = f"dbt-conflict-copy-{uuid.uuid4()}"
+            original = conns.generate_job_id
+            conns.generate_job_id = lambda: fixed_id
+            try:
+                conns.copy_bq_table(src, dst, "WRITE_TRUNCATE")
+                # Resubmit with the same job_id -> 409 -> attach, must not raise.
+                conns.copy_bq_table(src, dst, "WRITE_TRUNCATE")
+            finally:
+                conns.generate_job_id = original
+
+            _, iterator = conns.raw_execute(
+                f"select count(*) as n from `{dst.database}`.`{dst.schema}`.`{dst.identifier}`"
+            )
+            count = list(iterator)[0][0]
+
+        assert count == 1, f"expected 1 row in copy destination, got {count}"

dbt-bigquery/tests/unit/test_bigquery_connection_manager.py

@@ -110,6 +110,7 @@ def test_copy_bq_table_appends(self):
             [self._table_ref("project", "dataset", "table1")],
             self._table_ref("project", "dataset", "table2"),
             job_config=ANY,
+            job_id=ANY,
             retry=ANY,
         )
         args, kwargs = self.mock_client.copy_table.call_args
@@ -124,13 +125,36 @@ def test_copy_bq_table_truncates(self):
             [self._table_ref("project", "dataset", "table1")],
             self._table_ref("project", "dataset", "table2"),
             job_config=ANY,
+            job_id=ANY,
             retry=ANY,
         )
         args, kwargs = self.mock_client.copy_table.call_args
         self.assertEqual(
             kwargs["job_config"].write_disposition, dbt.adapters.bigquery.impl.WRITE_TRUNCATE
         )
 
+    def test_copy_bq_table_attaches_to_existing_job_on_conflict(self):
+        """A resubmitted copy job (e.g. a transport retry after a lost response)
+        must attach to the existing job via get_job rather than fail with a 409,
+        since copy_table has no built-in Conflict recovery."""
+        exceptions = dbt.adapters.bigquery.impl.google.cloud.exceptions
+        job_id = "job_x"
+        self.connections.generate_job_id = Mock(return_value=job_id)
+        self.mock_client.copy_table.side_effect = exceptions.Conflict(
+            f"Already Exists: Job project:{job_id}"
+        )
+        existing_job = Mock(job_id=job_id)
+        self.mock_client.get_job.return_value = existing_job
+
+        self._copy_table(write_disposition=dbt.adapters.bigquery.impl.WRITE_TRUNCATE)
+
+        self.assertEqual(self.mock_client.copy_table.call_count, 1)
+        # We must attach to the SAME job we tried to submit, not a different id.
+        self.assertEqual(self.mock_client.copy_table.call_args.kwargs["job_id"], job_id)
+        self.mock_client.get_job.assert_called_once_with(job_id)
+        # We wait on the attached job, not a resubmitted one.
+        existing_job.result.assert_called_once()
+
     def test_job_labels_valid_json(self):
         expected = {"key": "value"}
         labels = self.connections._labels_from_query_comment(json.dumps(expected))
@@ -161,7 +185,11 @@ def _copy_table(self, write_disposition):
         self.connections.copy_bq_table(source, destination, write_disposition)
 
     @patch("dbt.adapters.bigquery.connections.QueryJobConfig")
-    def test_raw_execute_retries_with_fresh_job_id(self, MockQueryJobConfig):
+    def test_raw_execute_reuses_job_id_on_retry(self, MockQueryJobConfig):
+        """The reopen-retry must reuse the SAME predetermined job_id across
+        attempts. Minting a fresh id per attempt double-executes non-idempotent
+        DML and duplicates rows (inc-6741). A stable id makes resubmission
+        idempotent via BigQuery's 409 Conflict path."""
         exceptions = dbt.adapters.bigquery.impl.google.cloud.exceptions
         job_ids_used = []
 
@@ -176,7 +204,33 @@ def capture_job_id(*args, **kwargs):
         self.mock_client.query.side_effect = capture_job_id
         self.connections.raw_execute("SELECT 1")
         self.assertEqual(self.mock_client.query.call_count, 2)
-        self.assertNotEqual(job_ids_used[0], job_ids_used[1])
+        self.assertEqual(job_ids_used[0], job_ids_used[1])
+
+    @patch("dbt.adapters.bigquery.connections.QueryJobConfig")
+    def test_query_and_results_attaches_to_existing_job_on_conflict(self, MockQueryJobConfig):
+        """If the job_id already exists (a prior attempt submitted it), recover
+        by attaching to the existing job via get_job instead of resubmitting,
+        so a single statement never spawns a duplicate BigQuery job."""
+        exceptions = dbt.adapters.bigquery.impl.google.cloud.exceptions
+        job_id = "job_x"
+        self.mock_client.query.side_effect = exceptions.Conflict(
+            f"Already Exists: Job project:{job_id}"
+        )
+        existing_job = Mock(job_id=job_id, location="US", project="project")
+        existing_job.result.return_value = iter([])
+        self.mock_client.get_job.return_value = existing_job
+
+        query_job, _ = self.connections._query_and_results(
+            self.mock_connection,
+            "MERGE INTO t USING s ON ...",
+            {"dry_run": False},
+            job_id=job_id,
+        )
+
+        self.mock_client.get_job.assert_called_once_with(job_id)
+        self.assertIs(query_job, existing_job)
+        # The DML must not be resubmitted.
+        self.assertEqual(self.mock_client.query.call_count, 1)
 
     @patch("dbt.adapters.bigquery.connections.QueryJobConfig")
     def test_raw_execute_no_retry_on_non_retryable_error(self, MockQueryJobConfig):