> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Google BigQuery

> Client-mode REST API connector for Google BigQuery v2: streaming row inserts (tabledata.insertAll) from MQTT, SQL via jobs.query, query results and job status, plus dataset/table list and create

Technical Preview

## Overview

A client-mode connector for the [BigQuery API v2](https://cloud.google.com/bigquery/docs/reference/rest) that streams MQTT telemetry into BigQuery tables and runs SQL from MQTT:

| Catalog connector event | Endpoint | Notes |
| - | - | - |
| Streaming insert | `POST /bigquery/v2/projects/{p}/datasets/{d}/tables/{t}/insertAll` | [`tabledata.insertAll`](https://cloud.google.com/bigquery/docs/reference/rest/v2/tabledata/insertAll). Row-level failures come back as `HTTP 200` with `insertErrors`. |
| Run SQL | `POST /bigquery/v2/projects/{p}/queries` | [`jobs.query`](https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query). Synchronous up to `timeoutMs`; returns `jobReference`, `schema`, `rows`, `jobComplete`. |
| Query results | `GET /bigquery/v2/projects/{p}/queries/{jobId}?location=…&maxResults=…` | [`jobs.getQueryResults`](https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/getQueryResults). Poll when `jobComplete` was `false`, or page with more rows. |
| Job status | `GET /bigquery/v2/projects/{p}/jobs/{jobId}?location=…` | `jobs.get`: state, statistics, error result. |
| List datasets | `GET /bigquery/v2/projects/{p}/datasets` | Credential smoke. |
| Create dataset / table | `POST …/datasets`, `POST …/datasets/{d}/tables` | One-time setup: dataset location and table schema. |
| List tables / rows | `GET …/datasets/{d}/tables`, `GET …/tables/{t}/data?maxResults=…` | Read-back with `tabledata.list`. |

Default `BASE_ADDRESS` is `https://bigquery.googleapis.com`. Replace `YOUR_PROJECT_ID` in the endpoints with your GCP project id. Topics use `data/google-bigquery/<object>/<verb>`.

> [!WARNING]
>
> **Storage Write API is not included.** The BigQuery Storage Write API (and Storage Read API) is gRPC/protobuf only, so a `REST_API` connector can't call it. This template uses the legacy `tabledata.insertAll` streaming endpoint, which is plain HTTPS + JSON. Load jobs with media upload (`/upload/bigquery/v2/…`) and destructive deletes are also left out.

## Auth

BigQuery needs an OAuth 2.0 access token in `Authorization: Bearer …`. The template reads it from the MQTT payload field `access_token`; keep the token in a Hub secret such as `GET SECRET "GOOGLE_BIGQUERY_ACCESS_TOKEN"` and never commit it.

* **Minting the token is out of band.** Exchange a service-account key (signed JWT) for an access token with `gcloud auth print-access-token`, a Google auth SDK, or your own token service, then publish or refresh it. Access tokens expire after about **1 hour**.
* **Scopes:** `https://www.googleapis.com/auth/bigquery` (full), `https://www.googleapis.com/auth/bigquery.insertdata` (streaming insert only), or `https://www.googleapis.com/auth/cloud-platform`.
* **IAM:** the service account needs BigQuery Data Editor on the dataset and BigQuery Job User on the project to run queries.

A `401 UNAUTHENTICATED` response with `Request had invalid authentication credentials` means the token is missing, malformed, or expired.

## Basic Syntax

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE GoogleBigqueryRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://bigquery.googleapis.com"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '60'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        # Auth (prefer Hub secrets / GET SECRET "GOOGLE_BIGQUERY_ACCESS_TOKEN"):
        #   OAuth access_token on each MQTT payload → Bearer header below (refresh before ~1 h expiry).
        #   Aliases: BIGQUERY_ACCESS_TOKEN, GOOGLE_ACCESS_TOKEN.
        # Smoke: GET /bigquery/v2/projects/{p}/datasets → 200; POST /bigquery/v2/projects/{p}/queries
        #   {"query":"SELECT 1","useLegacySql":false} → 200 + jobComplete true.
    ADD EVENT ListDatasets
        WITH SOURCE_TOPIC "data/google-bigquery/datasets/list"
        WITH DESTINATION_TOPIC "data/google-bigquery/datasets/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CreateDataset
        WITH SOURCE_TOPIC "data/google-bigquery/dataset/create"
        WITH DESTINATION_TOPIC "data/google-bigquery/dataset/create/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT ListTables
        WITH SOURCE_TOPIC "data/google-bigquery/tables/list"
        WITH DESTINATION_TOPIC "data/google-bigquery/tables/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CreateTable
        WITH SOURCE_TOPIC "data/google-bigquery/table/create"
        WITH DESTINATION_TOPIC "data/google-bigquery/table/create/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT InsertRows
        WITH SOURCE_TOPIC "data/google-bigquery/rows/insert"
        WITH DESTINATION_TOPIC "data/google-bigquery/rows/insert/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables/{value.json.tableId}/insertAll"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT ListRows
        WITH SOURCE_TOPIC "data/google-bigquery/rows/list"
        WITH DESTINATION_TOPIC "data/google-bigquery/rows/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables/{value.json.tableId}/data?maxResults={value.json.maxResults}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT RunQuery
        WITH SOURCE_TOPIC "data/google-bigquery/query/run"
        WITH DESTINATION_TOPIC "data/google-bigquery/query/run/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/queries"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT GetQueryResults
        WITH SOURCE_TOPIC "data/google-bigquery/query/results"
        WITH DESTINATION_TOPIC "data/google-bigquery/query/results/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/queries/{value.json.jobId}?location={value.json.location}&maxResults={value.json.maxResults}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT GetJob
        WITH SOURCE_TOPIC "data/google-bigquery/job/get"
        WITH DESTINATION_TOPIC "data/google-bigquery/job/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/jobs/{value.json.jobId}?location={value.json.location}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
```

> [!NOTE]
>
> Request bodies come from `{value.json.body}`, not `{payload}`. The MQTT payload carries control fields (`access_token`, `datasetId`, `tableId`, `jobId`, `location`, `maxResults`) next to a `body` object, and only `body` is sent to Google. Google APIs reject unknown top-level fields with `400 Invalid JSON payload received`.

## Example MQTT payloads

**Create a dataset (topic `data/google-bigquery/dataset/create`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "body": { "datasetReference": { "projectId": "YOUR_PROJECT_ID", "datasetId": "coreflux_iiot" }, "location": "EU" }
}
```

**Create a telemetry table (topic `data/google-bigquery/table/create`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "datasetId": "coreflux_iiot",
  "body": {
    "tableReference": { "projectId": "YOUR_PROJECT_ID", "datasetId": "coreflux_iiot", "tableId": "telemetry" },
    "schema": { "fields": [
      { "name": "ts", "type": "TIMESTAMP" },
      { "name": "site", "type": "STRING" },
      { "name": "temperature", "type": "FLOAT" }
    ] }
  }
}
```

**Stream rows (topic `data/google-bigquery/rows/insert`).** `insertId` gives best-effort de-duplication on retries:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "datasetId": "coreflux_iiot",
  "tableId": "telemetry",
  "body": {
    "skipInvalidRows": false,
    "rows": [
      { "insertId": "porto-1", "json": { "ts": "2026-10-04T21:55:00Z", "site": "porto", "temperature": 21.5 } }
    ]
  }
}
```

A clean insert returns `{"kind":"bigquery#tableDataInsertAllResponse"}`. Rejected rows return `HTTP 200` with `insertErrors` (for example `no such field: pressure.`), so check the response body, not just the status.

**Run SQL (topic `data/google-bigquery/query/run`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "body": {
    "query": "SELECT site, AVG(temperature) AS avg_t FROM `YOUR_PROJECT_ID.coreflux_iiot.telemetry` GROUP BY site",
    "useLegacySql": false,
    "location": "EU",
    "timeoutMs": 10000
  }
}
```

**Fetch results or job status (topics `data/google-bigquery/query/results`, `data/google-bigquery/job/get`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{ "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>", "jobId": "<jobReference.jobId>", "location": "EU", "maxResults": 100 }
```

> [!TIP]
>
> Always send `location` and `maxResults` for the GET events, since the template puts them in the query string. Use the dataset's location (`US`, `EU`, `europe-west1`, …); jobs outside the US and EU multi-regions are not found without it.

## Quotas and limits

* `insertAll`: up to 10 MB per request; Google recommends about 500 rows per request. Streamed rows are queryable within seconds, but they can't be changed with DML for a short window after insert.
* `jobs.query` waits up to `timeoutMs` (default 10 s). If `jobComplete` is `false`, poll `GetQueryResults` with `jobReference.jobId`.
* On-demand queries are billed by bytes scanned. Use partitioned tables and `WHERE` filters on high-rate telemetry. See [BigQuery quotas](https://cloud.google.com/bigquery/quotas).

## Testing

**Status: blocked on credentials.** The factory box has no Google OAuth credentials, so the connector has not been smoke-tested against Google. Wire format was checked against a local [goccy/bigquery-emulator](https://github.com/goccy/bigquery-emulator) v0.8.1, which does not check OAuth: the request shapes for all nine events returned `200` (datasets list and create, table create and list, `insertAll` of two rows, `tabledata.list` read-back, `jobs.query` `SELECT 1` and an aggregate read-back, `getQueryResults`, `jobs.get`). Against `bigquery.googleapis.com` without a valid token, `GET …/datasets` returned `401 UNAUTHENTICATED`, as expected.

To clear the gate, run `GET /bigquery/v2/projects/{p}/datasets` (expect `200`) and `POST /bigquery/v2/projects/{p}/queries` with `{"query":"SELECT 1","useLegacySql":false}` (expect `200`, `jobComplete: true`) using a real access token, project id, and lab dataset.

## Related

* [REST API Connectors](/latest/lot-language/routes/rest-api-routes)
* [REST API Apps](/latest/lot-language/routes/rest-api-apps/index)
* [Google Cloud Storage](/latest/lot-language/routes/rest-api-apps/google-cloud-storage)
* [ClickHouse](/latest/lot-language/routes/rest-api-apps/clickhouse)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.