diff --git a/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py b/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py index 21fccce3..fe8f3875 100644 --- a/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py +++ b/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, diff --git a/packages/interloper-google-cloud/tests/test_bigquery.py b/packages/interloper-google-cloud/tests/test_bigquery.py index efdff3f7..ae346380 100644 --- a/packages/interloper-google-cloud/tests/test_bigquery.py +++ b/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