fix(bigquery): use job_id_prefix instead of job_id

Author: dtran-im

ibis/backends/bigquery/__init__.py

@@ -544,7 +544,7 @@ def create_database(
         catalog: str | None = None,
         force: bool = False,
         collate: str | None = None,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
         **options: Any,
     ) -> None:
         properties = [
@@ -564,7 +564,7 @@ def create_database(
             properties=sge.Properties(expressions=properties),
         )
 
-        self.raw_sql(stmt.sql(self.name), job_id=job_id)
+        self.raw_sql(stmt.sql(self.name), job_id_prefix=job_id_prefix)
 
     def drop_database(
         self,
@@ -574,7 +574,7 @@ def drop_database(
         catalog: str | None = None,
         force: bool = False,
         cascade: bool = False,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
     ) -> None:
         """Drop a BigQuery dataset."""
         stmt = sge.Drop(
@@ -584,7 +584,7 @@ def drop_database(
             cascade=cascade,
         )
 
-        self.raw_sql(stmt.sql(self.name), job_id=job_id)
+        self.raw_sql(stmt.sql(self.name), job_id_prefix=job_id_prefix)
 
     def table(
         self,
@@ -670,7 +670,7 @@ def _get_schema_using_query(self, query: str) -> sch.Schema:
         return BigQuerySchema.to_ibis(job.schema)
 
     def raw_sql(
-        self, query: str, params=None, job_id: str | None = None
+        self, query: str, params=None, job_id_prefix: str | None = None
     ) -> RowIterator:
         query_parameters = [
             bigquery_param(param.type(), value, param.get_name())
@@ -681,12 +681,12 @@ def raw_sql(
 
         job_config = bq.job.QueryJobConfig(query_parameters=query_parameters or [])
 
-        if job_id is not None:
+        if job_id_prefix is not None:
             return self.client.query(
                 query,
                 job_config=job_config,
                 project=self.billing_project,
-                job_id=job_id,
+                job_id_prefix=job_id_prefix,
             ).result()
         else:
             return self.client.query_and_wait(
@@ -781,14 +781,14 @@ def _to_query(
         *,
         params: Mapping[ir.Scalar, Any] | None = None,
         limit: int | str | None = None,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
         **kwargs: Any,
     ) -> RowIterator:
         self._run_pre_execute_hooks(table_expr)
         sql = self.compile(table_expr, limit=limit, params=params, **kwargs)
         self._log(sql)
 
-        return self.raw_sql(sql, params=params, job_id=job_id)
+        return self.raw_sql(sql, params=params, job_id_prefix=job_id_prefix)
 
     def to_pyarrow(
         self,
@@ -982,7 +982,7 @@ def create_table(
         partition_by: str | None = None,
         cluster_by: Iterable[str] | None = None,
         options: Mapping[str, Any] | None = None,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
     ) -> ir.Table:
         """Create a table in BigQuery.
 
@@ -1015,9 +1015,9 @@ def create_table(
         options
             BigQuery-specific table options; see the BigQuery documentation for
             details: https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#table_option_list
-        job_id
-            Optional custom job ID; when specified, bigquery will use this job ID instead
-            of a randomly generated one
+        job_id_prefix
+            Optional custom job ID prefix; when specified, bigquery will use this as a
+            prefix for the job ID it generates
 
         Returns
         -------
@@ -1109,7 +1109,7 @@ def create_table(
 
         sql = stmt.sql(self.name)
 
-        self.raw_sql(sql, job_id=job_id)
+        self.raw_sql(sql, job_id_prefix=job_id_prefix)
         return self.table(table.name, database=(table.catalog, table.db))
 
     def drop_table(
@@ -1119,7 +1119,7 @@ def drop_table(
         *,
         database: tuple[str | str] | str | None = None,
         force: bool = False,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
     ) -> None:
         table_loc = self._to_sqlglot_table(database)
         catalog, db = self._to_catalog_db_tuple(table_loc)
@@ -1132,7 +1132,7 @@ def drop_table(
             ),
             exists=force,
         )
-        self.raw_sql(stmt.sql(self.name), job_id=job_id)
+        self.raw_sql(stmt.sql(self.name), job_id_prefix=job_id_prefix)
 
     def create_view(
         self,
@@ -1142,7 +1142,7 @@ def create_view(
         *,
         database: str | None = None,
         overwrite: bool = False,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
     ) -> ir.Table:
         table_loc = self._to_sqlglot_table(database)
         catalog, db = self._to_catalog_db_tuple(table_loc)
@@ -1158,7 +1158,7 @@ def create_view(
             replace=overwrite,
         )
         self._run_pre_execute_hooks(obj)
-        self.raw_sql(stmt.sql(self.name), job_id=job_id)
+        self.raw_sql(stmt.sql(self.name), job_id_prefix=job_id_prefix)
         return self.table(name, database=(catalog, database))
 
     def drop_view(
@@ -1168,7 +1168,7 @@ def drop_view(
         *,
         database: str | None = None,
         force: bool = False,
-        job_id: str | None = None,
+        job_id_prefix: str | None = None,
     ) -> None:
         table_loc = self._to_sqlglot_table(database)
         catalog, db = self._to_catalog_db_tuple(table_loc)
@@ -1182,7 +1182,7 @@ def drop_view(
             ),
             exists=force,
         )
-        self.raw_sql(stmt.sql(self.name), job_id=job_id)
+        self.raw_sql(stmt.sql(self.name), job_id_prefix=job_id_prefix)
 
     def _drop_cached_table(self, name):
         self.drop_table(

ibis/backends/bigquery/tests/system/test_client.py

@@ -5,7 +5,6 @@
 import decimal
 import os
 from urllib.parse import urlparse
-from uuid import uuid4
 
 import pandas as pd
 import pandas.testing as tm
@@ -549,8 +548,8 @@ def test_window_with_count_distinct(tmp_table, expr, query):
     tm.assert_frame_equal(result, expected)
 
 
-def test_create_job_with_custom_job_id(con):
-    job_id = f"test_job_id_{uuid4()}"
+def test_create_job_with_custom_job_prefix(con):
+    job_id_prefix = "test_job_prefix_"
     query = "SELECT 1"
-    result = con.raw_sql(query, job_id=job_id)
-    assert result.job_id == job_id
+    result = con.raw_sql(query, job_id_prefix=job_id_prefix)
+    assert result.job_id.startswith(job_id_prefix)