fix(google-cloud): a BigQuery table created by a concurrent run is not an error

packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py

@@ -156,7 +156,12 @@ def _create_table(
         time_partitioning: bigquery.TimePartitioning | None = None,
         description: str | None = None,
     ) -> None:
-        """Create a BigQuery table with an explicit schema.
+        """Create a BigQuery table with an explicit schema, unless one just appeared.
+
+        Two runs writing different partitions of the same new table race on its
+        creation (a backfill, or a queue burst after downtime): whichever loses
+        gets a 409 from BigQuery, which is not an error for it, since the table
+        it wanted now exists with the same schema, so it loads into that one.
 
         Args:
             table: Target table name.
@@ -168,7 +173,10 @@ def _create_table(
         bq_table = bigquery.Table(self._table_ref(table, schema), schema=bq_schema)
         bq_table.time_partitioning = time_partitioning
         bq_table.description = description
-        self.client.create_table(bq_table)
+        try:
+            self.client.create_table(bq_table)
+        except Conflict:
+            pass  # Created by a concurrent run of the same asset
 
     def _sync_table_metadata(
         self,

packages/interloper-google-cloud/tests/test_bigquery.py

@@ -388,6 +388,18 @@ def test_table_created_from_inference_without_schema(self):
         created = mock_client.create_table.call_args.args[0]
         assert [(f.name, f.field_type) for f in created.schema] == [("a", "INTEGER")]
 
+    def test_a_table_created_by_a_concurrent_run_is_loaded_into_not_an_error(self):
+        from google.cloud.exceptions import Conflict, NotFound
+
+        dest, mock_client = _make_destination(dataset="ds")
+        mock_client.get_table.side_effect = NotFound("nope")
+        mock_client.create_table.side_effect = Conflict("Already Exists: Table ds.tbl")
+
+        dest._insert_data("tbl", "ds", [{"id": 1, "cost": 2.0, "day": None}], _ctx(_plain_asset(), _RowSchema))
+
+        assert mock_client.create_table.called
+        assert mock_client.load_table_from_json.call_args.args[0] == [{"id": 1, "cost": 2.0, "day": None}]
+
     def test_dataframe_without_schema_lets_load_job_create_table(self):
         import pandas as pd
         from google.cloud.exceptions import NotFound