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

# bigqueryTarget

> BigQuery target for Pipes SDK

Write pipe output to Google BigQuery with fork-aware reorg handling. The target uses the BigQuery Storage Write API with committed streams: one long-lived stream per table, opened lazily on first write and reused for every batch.

```ts theme={"system"}
import { bigqueryTarget } from '@subsquid/pipes/targets/bigquery'
```

`@google-cloud/bigquery` and `@google-cloud/bigquery-storage` are optional peer dependencies. Install them alongside the SDK:

```bash theme={"system"}
npm install @google-cloud/bigquery @google-cloud/bigquery-storage
```

## `bigqueryTarget`

```ts theme={"system"}
bigqueryTarget<T>({
  client: { bigquery: BigQuery, writer?: WriterClient },
  dataset: string,
  tables: TrackedTable[],
  settings?: BigQuerySettings,
  onStart?: (ctx: { store: BigQueryWriter; logger: Logger }) => unknown | Promise<unknown>,
  onData: (ctx: { store: BigQueryWriter; data: T; ctx: HookContext }) => unknown | Promise<unknown>,
  onBeforeRollback?: (ctx: { cursor: BlockCursor }) => unknown | Promise<unknown>,
  onAfterRollback?: (ctx: { cursor: BlockCursor }) => unknown | Promise<unknown>,
})
```

| Parameter                              | Required | Description                                                                                                                                                                                         |
| -------------------------------------- | -------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `client`                               | Yes      | `{ bigquery }` from `@google-cloud/bigquery`. `writer` (a Storage Write API `WriterClient`) is optional; when omitted, the target constructs one with the same project ID and the default endpoint. |
| `dataset`                              | Yes      | BigQuery dataset that hosts both the tracked tables and the sync table. The target creates tables on demand but does not create the dataset.                                                        |
| `tables`                               | Yes      | Tracked tables (see below). Auto-created on first run if missing; validated against the declared schema on every restart. Writing to a non-listed table from `onData` throws.                       |
| `settings`                             | No       | State table and partitioning settings (see below).                                                                                                                                                  |
| `onStart`                              | No       | Runs once before processing starts.                                                                                                                                                                 |
| `onData`                               | Yes      | Called for each batch. Use `store.insert(table, rows)` to buffer rows; they are committed when `onData` returns.                                                                                    |
| `onBeforeRollback` / `onAfterRollback` | No       | Hooks around the fork `DELETE` phase, called with the safe cursor.                                                                                                                                  |

**`TrackedTable`:**

| Field               | Required | Description                                                                                                          |
| ------------------- | -------- | -------------------------------------------------------------------------------------------------------------------- |
| `table`             | Yes      | Unqualified table name.                                                                                              |
| `blockNumberColumn` | Yes      | Column used for partitioning and reorg `DELETE` scoping. Forced to `INT64 NOT NULL` regardless of the declared type. |
| `schema`            | Yes      | BigQuery field definitions (`TableField[]`) used for auto-creation and schema validation.                            |
| `clusterBy`         | No       | `CLUSTER BY` columns. Recommended for natural primary keys.                                                          |

**`settings`:**

| Field                     | Default       | Description                                                                                            |
| ------------------------- | ------------- | ------------------------------------------------------------------------------------------------------ |
| `state.table`             | `'sync'`      | Sync (cursor) table name.                                                                              |
| `state.id`                | source `id`   | Stream identifier within the sync table.                                                               |
| `state.maxRows`           | `10_000`      | Maximum sync rows retained per stream id.                                                              |
| `partitioning.bucketSize` | `10_000`      | Width of each `RANGE_BUCKET` partition, in blocks.                                                     |
| `partitioning.maxBlocks`  | `100_000_000` | Upper bound of the partition range.                                                                    |
| `partitioning`            |               | Set to `false` to disable partitioning DDL. Not recommended: fork `DELETE`s then scan the whole table. |

## Fork handling

On a reorg, the target opens an `IN_FLIGHT_ROLLBACK` row in the sync table, runs `DELETE FROM <table> WHERE <blockNumberColumn> BETWEEN safe+1 AND upper` on every tracked table in parallel, then marks the rollback complete. If the process dies between the two markers, the next startup re-runs the bounded `DELETE`s idempotently. See [Fork handling](../../../guides/architecture-deep-dives/fork-handling) for how the finalization watermark is resolved and enforced across targets.

## Example

```ts expandable theme={"system"}
import { BigQuery } from '@google-cloud/bigquery'
import { commonAbis, evmEventDecoder, evmPortalStream } from '@subsquid/pipes/evm'
import { bigqueryTarget } from '@subsquid/pipes/targets/bigquery'

const bigquery = new BigQuery({ projectId: 'my-gcp-project' })

await evmPortalStream({
  id: 'erc20-transfers',
  portal: 'https://portal.sqd.dev/datasets/ethereum-mainnet',
  outputs: evmEventDecoder({
    range: { from: '0' },
    events: { transfers: commonAbis.erc20.events.Transfer },
  }),
}).pipeTo(
  bigqueryTarget({
    client: { bigquery },
    dataset: 'eth_transfers',
    tables: [
      {
        table: 'transfers',
        blockNumberColumn: 'block_number',
        schema: [
          { name: 'block_number', type: 'INT64', mode: 'REQUIRED' },
          { name: 'log_index', type: 'INT64', mode: 'REQUIRED' },
          // TIMESTAMP wire format is INT64 microseconds since epoch
          { name: 'block_timestamp', type: 'TIMESTAMP', mode: 'REQUIRED' },
          { name: 'token', type: 'STRING', mode: 'REQUIRED' },
          { name: 'from', type: 'STRING', mode: 'REQUIRED' },
          { name: 'to', type: 'STRING', mode: 'REQUIRED' },
          { name: 'amount_raw', type: 'STRING', mode: 'REQUIRED' },
        ],
        clusterBy: ['token', 'from'],
      },
    ],
    onData: async ({ store, data }) => {
      store.insert(
        'transfers',
        data.transfers.map((t) => ({
          block_number: t.block.number,
          log_index: t.rawEvent.logIndex,
          // The Storage Write API does not parse Date/ISO strings; pass microseconds
          block_timestamp: t.timestamp ? t.timestamp.getTime() * 1000 : 0,
          token: t.rawEvent.address,
          from: t.event.from,
          to: t.event.to,
          amount_raw: t.event.value.toString(),
        })),
      )
    },
  }),
)
```

Full runnable example: [`16.bigquery.example.ts`](https://github.com/subsquid-labs/pipes-sdk/blob/main/docs/examples/evm/16.bigquery.example.ts).

## Notes

* `store.insert(table, rows)` is synchronous and buffers rows per table; the commit runs once `onData` returns. It throws immediately if the table is not declared in `tables`.
* The default `BIGNUMERIC` precision holds up to 38 integer digits. `uint256` values (e.g. the `2^256-1` "infinite approval" sentinel) overflow it; store the exact decimal as a `STRING` column, or clamp before insertion.
* `TIMESTAMP` columns take `INT64` microseconds since epoch on the write path; the Storage Write API JSONWriter does not parse `Date` objects or ISO strings.