Add BigQuery query client and dialect support

datajunction-query/config.djqs.yml

@@ -16,6 +16,12 @@ engines:
     version: ''
     type: sqlalchemy
     uri: postgresql+psycopg://readonly_user:readonly_pass@postgres_metadata:5432/dj
+  # - name: bigquery
+  #   version: ''
+  #   type: bigquery
+  #   uri: bigquery:///
+  #   extra_params:
+  #     project: <your-gcp-project>
 catalogs:
   - name: warehouse
     engines:
@@ -26,3 +32,6 @@ catalogs:
   - name: dj
     engines:
       - dj_system
+  # - name: <your-gcp-project>
+  #   engines:
+  #     - bigquery

datajunction-query/djqs/config.py

@@ -28,6 +28,7 @@ class EngineType(Enum):
     SQLALCHEMY = "sqlalchemy"
     SNOWFLAKE = "snowflake"
     TRINO = "trino"
+    BIGQUERY = "bigquery"
 
 
 class EngineInfo:  # pylint: disable=too-few-public-methods

datajunction-query/djqs/engine.py

@@ -11,6 +11,7 @@
 
 import duckdb
 import snowflake.connector
+from google.cloud import bigquery
 from psycopg_pool import AsyncConnectionPool
 from sqlalchemy import create_engine, text
 
@@ -117,6 +118,25 @@ def run_query(  # pylint: disable=R0914
         cur = conn.cursor()
 
         return run_snowflake_query(query, cur)
+    elif engine.type == EngineType.BIGQUERY:
+        _logger.info("Creating BigQuery client")
+        project = engine.extra_params.get("project")
+        credentials_path = engine.extra_params.get("credentials_path")
+        location = engine.extra_params.get("location")
+
+        client_kwargs = {}
+        if credentials_path:
+            from google.oauth2 import service_account
+
+            credentials = service_account.Credentials.from_service_account_file(
+                credentials_path,
+            )
+            client_kwargs["credentials"] = credentials
+        if location:
+            client_kwargs["location"] = location
+
+        bq_client = bigquery.Client(project=project, **client_kwargs)
+        return run_bigquery_query(query, bq_client)
 
     _logger.info(
         "Creating sqlalchemy engine using engine name and version defined on query",
@@ -164,6 +184,29 @@ def run_snowflake_query(
     return output
 
 
+def run_bigquery_query(
+    query: Query,
+    client: bigquery.Client,
+) -> List[Tuple[str, List[ColumnMetadata], Stream]]:
+    """
+    Run a query against BigQuery.
+
+    DJ prefixes table references with the catalog name (e.g. your-gcp-project.dataset.table),
+    but BigQuery interprets three-part names as project.dataset.table. Since the client
+    already has the correct project configured, we strip the catalog prefix so BigQuery
+    receives dataset.table references instead.
+    """
+    sql = query.submitted_query
+    if query.catalog_name:
+        sql = sql.replace(f"{query.catalog_name}.", "")
+    output: List[Tuple[str, List[ColumnMetadata], Stream]] = []
+    result = client.query(sql).result()
+    rows = iter([tuple(row.values()) for row in result])
+    columns: List[ColumnMetadata] = []
+    output.append((sql, columns, rows))
+    return output
+
+
 def serialize_for_json(obj):
     """
     Handle serialization of date/datetimes for JSON output.

datajunction-query/pyproject.toml

@@ -24,6 +24,7 @@ dependencies = [
     "rich>=10.16.2",
     "toml>=0.10.2",
     "snowflake-connector-python>=3.3.1",
+    "google-cloud-bigquery>=3.11.0",
     "pyyaml>=6.0.1",
     "trino>=0.324.0",
     "psycopg[pool]>=3.2.1",