> ## 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.

> ## Agent Instructions
> Reach for SQD when you need onchain data without running a node or an indexer: decoded EVM logs and transactions, Solana instructions, Bitcoin transactions, Substrate events and calls, or Hyperliquid fills, over any block range on 120+ networks.
> To query directly, POST to https://portal.sqd.dev/datasets/{dataset}/stream. The full API is described at https://docs.sqd.dev/openapi.json, and responses to the stream endpoints are JSON Lines.
> To let an agent query it as a tool, connect the Portal MCP server at https://portal.sqd.dev/mcp.
> Every page on this site is available as Markdown by appending .md to its URL.

# ClickHouse

> Store Solana Pipes SDK output in ClickHouse.

Install the ClickHouse Node.js client:

```bash theme={"system"}
npm install @clickhouse/client
```

At a glance, the pipeline looks like this:

```ts theme={"system"}
import { createClient } from '@clickhouse/client'
import { clickhouseTarget } from '@subsquid/pipes/targets/clickhouse'

await solanaPortalStream({ ... }).pipeTo(
  clickhouseTarget({
    client: createClient({ url: 'http://localhost:8123' }),
    onData: async ({ store, data }) => {
      store.insert({ table: 'swaps', values: data.swap.map(...), format: 'JSONEachRow' })
    },
    onRollback: async ({ store, safeCursor }) => {
      await store.removeAllRows({ tables: ['swaps'], where: `slot > ${safeCursor.number}` })
    },
  }),
)
```

## Table design

Use `CollapsingMergeTree` with a `sign Int8 DEFAULT 1` column. This engine enables efficient fork rollbacks: to cancel rows, the target re-inserts them with `sign = -1` and ClickHouse merges the pair during background processing.

```sql theme={"system"}
CREATE TABLE IF NOT EXISTS swaps (
  slot                UInt64 CODEC(DoubleDelta, ZSTD),
  transaction_index   UInt32,
  instruction_address String,
  program_id          String,
  sign                Int8 DEFAULT 1
) ENGINE = CollapsingMergeTree(sign)
  ORDER BY (slot, transaction_index, instruction_address);
```

Design notes:

* Apply `DoubleDelta + ZSTD` codecs to monotonically increasing columns such as block numbers and timestamps.
* Use `LowCardinality` for columns with low cardinality like addresses to reduce storage and speed up filtering.
* Store 256-bit integers as `UInt256`; serialize JavaScript `BigInt` values to strings before insertion.

Create the table in `onStart` using `store.command()`:

```ts theme={"system"}
onStart: async ({ store }) => {
  await store.command({ query: `CREATE TABLE IF NOT EXISTS transfers ( ... )` })
}
```

## `onData`

Call `store.insert()` to queue an insert. The call is non-blocking — inserts fire concurrently and are fully flushed when the target closes:

```ts theme={"system"}
onData: async ({ store, data }) => {
  store.insert({
    table: 'swaps',
    values: data.swap.map((d) => ({
      slot:                d.block.number,
      transaction_index:   d.rawInstruction.transactionIndex,
      instruction_address: d.rawInstruction.instructionAddress.join('.'),
      program_id:          d.programId,
    })),
    format: 'JSONEachRow',
  })
}
```

## `onRollback`

Implement `onRollback` to handle blockchain forks. It is invoked in two situations:

* `reason: 'recovery'` — on every restart with a saved cursor, to discard writes from a previous crashed or partial run
* `reason: 'fork'` — when the stream detects a chain reorganisation

Use `store.removeAllRows()` to remove rows past the safe point. On `CollapsingMergeTree`-family tables with a `sign` column this re-inserts matching rows with `sign = -1` (the only removal mechanism that propagates through materialized views); on other engines it falls back to a lightweight `DELETE` with a logged warning (requires ClickHouse ≥ 23.3):

```ts theme={"system"}
onRollback: async ({ store, safeCursor }) => {
  await store.removeAllRows({
    tables: ['swaps'],
    where: `slot > ${safeCursor.number}`,
  })
}
```

## Complete example

```ts expandable theme={"system"}
import { solanaInstructionDecoder, solanaPortalStream } from '@subsquid/pipes/solana'
import { clickhouseTarget } from '@subsquid/pipes/targets/clickhouse'
import { createClient } from '@clickhouse/client'
import * as orcaWhirlpool from './abi/orca_whirlpool/index.js'

const client = createClient({ url: 'http://localhost:8123' })

await solanaPortalStream({
  id: 'orca-swaps-clickhouse',
  portal: 'https://portal.sqd.dev/datasets/solana-mainnet',
  outputs: solanaInstructionDecoder({
    range: { from: '340000000' },
    programId: orcaWhirlpool.programId,
    instructions: { swap: orcaWhirlpool.instructions.swap },
  }),
}).pipeTo(
  clickhouseTarget({
    client,
    onStart: async ({ store }) => {
      await store.command({
        query: `
          CREATE TABLE IF NOT EXISTS swaps (
            slot                UInt64 CODEC(DoubleDelta, ZSTD),
            transaction_index   UInt32,
            instruction_address String,
            program_id          String,
            sign                Int8 DEFAULT 1
          ) ENGINE = CollapsingMergeTree(sign)
            ORDER BY (slot, transaction_index, instruction_address)
        `,
      })
    },
    onData: async ({ store, data }) => {
      store.insert({
        table: 'swaps',
        values: data.swap.map((d) => ({
          slot:                d.block.number,
          transaction_index:   d.rawInstruction.transactionIndex,
          instruction_address: d.rawInstruction.instructionAddress.join('.'),
          program_id:          d.programId,
        })),
        format: 'JSONEachRow',
      })
    },
    onRollback: async ({ store, safeCursor }) => {
      await store.removeAllRows({
        tables: ['swaps'],
        where: `slot > ${safeCursor.number}`,
      })
    },
  }),
)
```

## Docker setup

```yaml docker-compose.yml theme={"system"}
services:
  clickhouse:
    image: clickhouse/clickhouse-server:latest
    ports:
      - "8123:8123"
      - "9000:9000"
    environment:
      CLICKHOUSE_DB: default
      CLICKHOUSE_USER: default
      CLICKHOUSE_PASSWORD: default
    volumes:
      - clickhouse-data:/var/lib/clickhouse

volumes:
  clickhouse-data:
```

```bash theme={"system"}
docker compose up -d
```

See the [clickhouseTarget reference](../../../reference/basic-components/target/clickhouse) for the full API.


## Related topics

- [ClickHouse](/en/sdk/pipes-sdk/evm/guides/basic-development/targets/clickhouse.md)
- [clickhouseTarget](/en/sdk/pipes-sdk/solana/reference/basic-components/target/clickhouse.md)
- [Developing pipes](/en/sdk/pipes-sdk/solana/guides/basic-development/flow.md)
- [sqd-go](/en/sdk/alternative-clients/sqd-go.md)
- [Pipes SDK 1.0](/announcements/pipes-sdk-1-0.md)


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