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

# Postgres via Drizzle

> Store Solana Pipes SDK output in PostgreSQL with Drizzle ORM.

Install Drizzle ORM and the PostgreSQL driver:

```bash theme={"system"}
npm install drizzle-orm pg
npm install -D drizzle-kit @types/pg
```

At a glance, the pipeline looks like this:

```ts theme={"system"}
await solanaPortalStream({ ... }).pipeTo(
  drizzleTarget({
    db: drizzle('postgresql://...'),
    tables: [swapsTable],
    onData: async ({ tx, data }) => {
      for (const batch of chunkForInsert(data.swap)) {
        await tx.insert(swapsTable).values(batch.map(...))
      }
    },
  }),
)
```

## Schema

Define your tables with Drizzle ORM. Every table needs a primary key, and every table written to in [`onData`](#ondata) must appear in [`tables`](#tables).

```ts theme={"system"}
import { bigint, integer, pgTable, primaryKey, varchar } from 'drizzle-orm/pg-core'

const swapsTable = pgTable('swaps', {
  slot:               bigint({ mode: 'bigint' }).notNull(),
  transactionIndex:   integer().notNull(),
  instructionAddress: varchar().notNull(),
  programId:          varchar().notNull(),
}, (t) => [primaryKey({ columns: [t.slot, t.transactionIndex, t.instructionAddress] })])
```

<h2 id="ondata">
  `onData` and `chunkForInsert`
</h2>

`onData` runs inside a serializable transaction. Use `chunkForInsert` to split data arrays into chunks that fit within PostgreSQL's 32,767-parameter limit — chunk size is calculated automatically from the number of columns:

```ts theme={"system"}
import { chunkForInsert, drizzleTarget } from '@subsquid/pipes/targets/drizzle/node-postgres'

onData: async ({ tx, data }) => {
  for (const batch of chunkForInsert(data.swap)) {
    await tx.insert(swapsTable).values(
      batch.map((d) => ({
        slot:               BigInt(d.block.number),
        transactionIndex:   d.rawInstruction.transactionIndex,
        instructionAddress: d.rawInstruction.instructionAddress.join('.'),
        programId:          d.programId,
      })),
    )
  }
}
```

Pass an explicit second argument to `chunkForInsert` to cap chunk size:

```ts theme={"system"}
for (const batch of chunkForInsert(data.transfers, 100)) { ... }
```

## `tables`

Every table written to in `onData` must be listed in `tables`. At startup, the target installs PostgreSQL trigger functions on these tables to track row-level changes for automatic fork handling. Inserting into an unlisted table throws at runtime.

## Schema migrations

Use [Drizzle Kit](https://orm.drizzle.team/docs/kit-overview) to generate and apply migrations:

```bash theme={"system"}
npx drizzle-kit generate
npx drizzle-kit migrate
```

Alternatively, run migrations automatically on startup via `onStart`:

```ts theme={"system"}
import { migrate } from 'drizzle-orm/node-postgres/migrator'

drizzleTarget({
  db,
  tables: [...],
  onStart: async ({ db }) => {
    await migrate(db, { migrationsFolder: './drizzle' })
  },
  onData: async ({ tx, data }) => { ... },
})
```

## Rollback handling

Fork handling is fully automatic. Each batch runs inside a transaction that snapshots row-level changes. When the stream detects a fork, the target replays those snapshots in reverse to restore the pre-fork state.

Use `onBeforeRollback` and `onAfterRollback` to run custom logic around a rollback. Both callbacks receive the Drizzle transaction and the `cursor` (`BlockCursor`) to which state was rolled back:

```ts theme={"system"}
drizzleTarget({
  db,
  tables: [...],
  onBeforeRollback: async ({ tx, cursor }) => { /* e.g. log or acquire an external lock */ },
  onAfterRollback:  async ({ tx, cursor }) => { /* e.g. invalidate a cache */ },
  onData: async ({ tx, data }) => { ... },
})
```

## Complete example

```ts expandable theme={"system"}
import { solanaInstructionDecoder, solanaPortalStream } from '@subsquid/pipes/solana'
import { chunkForInsert, drizzleTarget } from '@subsquid/pipes/targets/drizzle/node-postgres'
import { drizzle } from 'drizzle-orm/node-postgres'
import { bigint, integer, pgTable, primaryKey, varchar } from 'drizzle-orm/pg-core'
import * as orcaWhirlpool from './abi/orca_whirlpool/index.js'

const swapsTable = pgTable('swaps', {
  slot:               bigint({ mode: 'bigint' }).notNull(),
  transactionIndex:   integer().notNull(),
  instructionAddress: varchar().notNull(),
  programId:          varchar().notNull(),
}, (t) => [primaryKey({ columns: [t.slot, t.transactionIndex, t.instructionAddress] })])

await solanaPortalStream({
  id: 'orca-swaps-drizzle',
  portal: 'https://portal.sqd.dev/datasets/solana-mainnet',
  outputs: solanaInstructionDecoder({
    range: { from: '340000000' },
    programId: orcaWhirlpool.programId,
    instructions: { swap: orcaWhirlpool.instructions.swap },
  }),
}).pipeTo(
  drizzleTarget({
    db: drizzle('postgresql://postgres:postgres@localhost:5432/postgres'),
    tables: [swapsTable],
    onData: async ({ tx, data }) => {
      for (const batch of chunkForInsert(data.swap)) {
        await tx.insert(swapsTable).values(
          batch.map((d) => ({
            slot:               BigInt(d.block.number),
            transactionIndex:   d.rawInstruction.transactionIndex,
            instructionAddress: d.rawInstruction.instructionAddress.join('.'),
            programId:          d.programId,
          })),
        )
      }
    },
  }),
)
```

See the [drizzleTarget reference](../../../reference/basic-components/target/postgres-drizzle) for the full API.


## Related topics

- [Postgres via Drizzle](/en/sdk/pipes-sdk/evm/guides/basic-development/targets/postgres-drizzle.md)
- [drizzleTarget](/en/sdk/pipes-sdk/solana/reference/basic-components/target/postgres-drizzle.md)
- [Developing pipes](/en/sdk/pipes-sdk/solana/guides/basic-development/flow.md)
- [Pipes SDK 1.0](/announcements/pipes-sdk-1-0.md)
- [Pipe anatomy](/en/sdk/pipes-sdk/solana/guides/basic-development/anatomy.md)


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