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

# Pipeline architecture patterns

> How to shape a Turbo pipeline: linear, fan-out, fan-in, multi-chain, dynamic tables, and aggregate sinks

Every Turbo pipeline has the same three building blocks (sources, transforms, and sinks), but how you arrange them determines how easy the pipeline is to operate, scale, and debug. This guide shows six patterns that cover most real-world pipelines, each with a complete, runnable example.

Examples use Ethereum Mainnet (and Base for multi-chain). The same patterns work on any chain; see [supported networks](/chains/supported-networks#turbo).

## Choosing a pattern

| Pattern                                                         | Use it when                                                          |
| --------------------------------------------------------------- | -------------------------------------------------------------------- |
| [Linear](#linear-pipeline)                                      | One source, one destination, straightforward processing              |
| [Fan-out](#fan-out-one-source-multiple-sinks)                   | One source feeding multiple destinations with different views        |
| [Fan-in](#fan-in-multiple-inputs-one-sink)                      | Multiple event types or chains combined into one table               |
| [Multi-chain templated](#multi-chain-templated-deployment)      | The same logic deployed per chain, with independent lifecycles       |
| [Dynamic tables](#dynamic-table-architecture)                   | Filtering or enrichment against lookup data that changes at runtime  |
| [PostgreSQL aggregate sink](#postgresql-aggregate-sink-pattern) | Real-time running totals (balances, counters) maintained in Postgres |

Two decisions apply to every pattern:

* **Streaming or job mode**: all examples below are streaming pipelines. For one-time backfills or exports, add `job: true`; see [Job mode](/turbo-pipelines/job-mode).
* **Resource size**: each section suggests a starting `resource_size`. Start small and scale up if the pipeline lags; see the [resource size reference](/turbo-pipelines/pipeline-config#top-level-fields) for CPU and memory per tier.

Validate any pipeline before deploying it:

```bash theme={"dark"}
goldsky turbo validate pipeline.yaml
goldsky turbo apply pipeline.yaml -i
```

## Linear pipeline

The simplest shape: one source, one or more chained transforms, one sink.

```text theme={"dark"}
source → transform_a → transform_b → sink
```

**Use when** you have a single data source, a single destination, and straightforward processing: decode, filter, reshape.

This example decodes `OrderFilled` events from an exchange contract and writes clean, typed rows to PostgreSQL:

```yaml theme={"dark"}
name: exchange-trades
resource_size: s

sources:
  raw_logs:
    type: dataset
    dataset_name: ethereum.raw_logs
    version: 1.0.0
    start_at: earliest
    # Fast scan: pre-filter at the source so the backfill skips irrelevant blocks.
    # Replace the address with your contract.
    filter: >-
      address = '0x1111111111111111111111111111111111111111'
      AND block_number >= 21000000

transforms:
  # Step 1: decode raw logs with an inline ABI
  decoded_events:
    type: sql
    primary_key: id
    sql: |
      SELECT
        _gs_log_decode(
          '[{"anonymous":false,"inputs":[{"indexed":true,"name":"maker","type":"address"},{"indexed":true,"name":"taker","type":"address"},{"indexed":false,"name":"amount","type":"uint256"},{"indexed":false,"name":"price","type":"uint256"}],"name":"OrderFilled","type":"event"}]',
          topics,
          data
        ) AS decoded,
        id, block_number, transaction_hash, address, block_timestamp
      FROM raw_logs

  # Step 2: extract typed columns from the decoded event
  trades:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        block_number,
        block_timestamp,
        transaction_hash,
        address AS contract_address,
        decoded.event_params[1] AS maker,
        decoded.event_params[2] AS taker,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount,
        (CAST(decoded.event_params[4] AS DOUBLE) / 1e6) AS price_usdc
      FROM decoded_events
      WHERE decoded.event_signature = 'OrderFilled'

sinks:
  database:
    type: postgres
    from: trades
    schema: public
    table: trades
    secret_name: MY_POSTGRES_SECRET
    primary_key: id
```

**Resource size:** `s` for filtered streams; move to `m` if you backfill a busy contract from `earliest`.

Related pages:

* [SQL transforms](/turbo-pipelines/transforms/sql) for the transform syntax used here
* [EVM sources](/turbo-pipelines/sources/evm#fast-scan) for how `filter` speeds up backfills
* [PostgreSQL sink](/turbo-pipelines/sinks/postgres) for sink options and secret format

## Fan-out (one source, multiple sinks)

One source feeds multiple transforms, each writing a different view of the data to a different sink.

```text theme={"dark"}
              ┌─→ transform_a ─→ sink_1 (ClickHouse)
source ───────┤
              └─→ transform_b ─→ sink_2 (webhook)
```

**Use when** different consumers need different views or subsets of the same data: for example, full history in a warehouse plus real-time alerts.

<Note>
  Default to one pipeline with multiple sinks when you send the same source data to several destinations. Sinks run independently: one failing does not block the others, and each can have its own batching settings. Splitting into one pipeline per destination duplicates source ingestion and wastes resources. Split only when the destinations need different resource sizes or genuinely independent lifecycles.
</Note>

This example reshapes USDC transfers for an analytics warehouse and sends high-value transfers of any token to an alerting webhook:

```yaml theme={"dark"}
name: transfer-fan-out
resource_size: m

sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # View 1: USDC transfers, reshaped for analytics
  usdc_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        sender,
        recipient,
        (CAST(amount AS DOUBLE) / 1e6) AS amount_usdc,
        block_number,
        block_timestamp
      FROM ethereum_transfers
      WHERE address = lower('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48')

  # View 2: high-value transfers of any token, for alerting
  whale_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        address AS token_address,
        sender,
        recipient,
        amount,
        transaction_hash,
        block_timestamp
      FROM ethereum_transfers
      WHERE amount > 1000000000000000000000000

sinks:
  # Sink 1: analytics warehouse
  warehouse:
    type: clickhouse
    from: usdc_transfers
    table: usdc_transfers
    secret_name: MY_CLICKHOUSE
    primary_key: id
    batch_size: 100000
    batch_flush_interval: 10s

  # Sink 2: real-time alerts
  alerts:
    type: webhook
    from: whale_transfers
    url: https://alerts.example.com/whale-transfer
    one_row_per_request: true
```

**Resource size:** `m`; multiple sinks mean more concurrent work than a linear pipeline.

Related pages:

* [Multiple sinks](/turbo-pipelines/sinks/overview#multiple-sinks) for how sinks behave independently
* [ClickHouse sink](/turbo-pipelines/sinks/clickhouse) and [webhook sink](/turbo-pipelines/sinks/webhook) for sink-specific options

## Fan-in (multiple inputs, one sink)

Multiple event types are decoded from the same source, normalized to a common schema, and combined with `UNION ALL` into a single sink.

```text theme={"dark"}
                  ┌─→ event_type_a ──┐
source → decode ──┤                  ├─→ UNION ALL → sink
                  └─→ event_type_b ──┘
```

**Use when** you want a unified table combining several event types (trades, deposits, withdrawals, transfers) as one activity feed.

This example builds a WETH activity feed from four event types. Every branch projects the same columns in the same order, which `UNION ALL` requires:

```yaml theme={"dark"}
name: weth-activity-feed
resource_size: l

sources:
  raw_logs:
    type: dataset
    dataset_name: ethereum.raw_logs
    version: 1.0.0
    start_at: earliest
    # Fast scan: only WETH logs, starting from a recent block
    filter: >-
      address = '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2'
      AND block_number >= 21000000

transforms:
  # Decode all four event types in a single transform
  decoded_events:
    type: sql
    primary_key: id
    sql: |
      SELECT
        _gs_log_decode(
          '[{"anonymous":false,"inputs":[{"indexed":true,"name":"from","type":"address"},{"indexed":true,"name":"to","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Transfer","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"owner","type":"address"},{"indexed":true,"name":"spender","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Approval","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"dst","type":"address"},{"indexed":false,"name":"wad","type":"uint256"}],"name":"Deposit","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"src","type":"address"},{"indexed":false,"name":"wad","type":"uint256"}],"name":"Withdrawal","type":"event"}]',
          topics,
          data
        ) AS decoded,
        id, block_number, transaction_hash, address, block_timestamp
      FROM raw_logs

  # Event type 1: transfers
  transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        decoded.event_params[2] AS counterparty,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount,
        'TRANSFER' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Transfer'

  # Event type 2: approvals
  approvals:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        decoded.event_params[2] AS counterparty,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount,
        'APPROVAL' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Approval'

  # Event type 3: deposits (ETH wrapped into WETH)
  deposits:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        '' AS counterparty,
        (CAST(decoded.event_params[2] AS DOUBLE) / 1e18) AS amount,
        'DEPOSIT' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Deposit'

  # Event type 4: withdrawals (WETH unwrapped back to ETH)
  withdrawals:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        '' AS counterparty,
        (CAST(decoded.event_params[2] AS DOUBLE) / 1e18) AS amount,
        'WITHDRAWAL' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Withdrawal'

  # Combine all events into a unified activity feed
  all_activity:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM transfers
      UNION ALL
      SELECT * FROM approvals
      UNION ALL
      SELECT * FROM deposits
      UNION ALL
      SELECT * FROM withdrawals

sinks:
  warehouse:
    type: clickhouse
    from: all_activity
    table: weth_activity
    secret_name: MY_CLICKHOUSE
    primary_key: id
    batch_size: 1000
    batch_flush_interval: 1s
```

**Resource size:** `l`; decoding plus many transforms is the heaviest shape in this guide.

### Multi-chain fan-in

The same shape works with multiple sources instead of multiple event types: combine chains into one output with `UNION ALL`.

```text theme={"dark"}
ethereum_transfers ──┐
                     ├─→ UNION ALL → sink
base_transfers ──────┘
```

```yaml theme={"dark"}
sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

  base_transfers:
    type: dataset
    dataset_name: base.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  combined:
    type: sql
    primary_key: id
    sql: |
      SELECT *, 'ethereum' AS chain FROM ethereum_transfers
      UNION ALL
      SELECT *, 'base' AS chain FROM base_transfers
```

See the [multi-chain recipe](/turbo-pipelines/reference/pipeline-cookbook#multi-chain-combine-ethereum-and-base) in the cookbook for the complete pipeline. Use `m` for two chains and `l` as you add more. If the chains do not need to land in the same table, prefer the [templated deployment](#multi-chain-templated-deployment) below.

## Multi-chain templated deployment

When you need the **same pipeline logic** across multiple chains, deploy one pipeline file per chain rather than a single multi-source pipeline.

```text theme={"dark"}
ethereum-transfers.yaml ─→ [pipeline 1] ─→ ethereum_token_transfers
base-transfers.yaml     ─→ [pipeline 2] ─→ base_token_transfers
```

**Use when** you want per-chain independence:

* Independent lifecycle: deploy, pause, or delete one chain without touching the others
* Independent checkpointing: one chain failing or lagging doesn't block the others
* Clearer monitoring: each chain has its own pipeline status and logs

The pattern is a template: write the pipeline once, then copy it and swap the chain-specific values. Here's the Ethereum version:

```yaml ethereum-transfers.yaml theme={"dark"}
name: ethereum-transfer-streaming
resource_size: m

sources:
  ethereum_erc20_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Tag every row with its chain so downstream tables can be unioned later
  chain_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        'ethereum' AS chain,
        address AS token_address,
        sender,
        recipient,
        amount,
        block_number,
        block_timestamp
      FROM ethereum_erc20_transfers

sinks:
  warehouse:
    type: clickhouse
    from: chain_transfers
    table: ethereum_token_transfers
    secret_name: MY_CLICKHOUSE
    primary_key: id
    batch_size: 100000
    batch_flush_interval: 10s
```

To create the Base version, copy the file and swap five values:

| Field          | Ethereum pipeline             | Base pipeline             |
| -------------- | ----------------------------- | ------------------------- |
| `name`         | `ethereum-transfer-streaming` | `base-transfer-streaming` |
| Source key     | `ethereum_erc20_transfers`    | `base_erc20_transfers`    |
| `dataset_name` | `ethereum.erc20_transfers`    | `base.erc20_transfers`    |
| Transform SQL  | `'ethereum' AS chain`         | `'base' AS chain`         |
| Sink `table`   | `ethereum_token_transfers`    | `base_token_transfers`    |

### Templated vs. multi-source

| Approach                | Pros                                          | Cons                               |
| ----------------------- | --------------------------------------------- | ---------------------------------- |
| Templated (per-chain)   | Independent lifecycle, clear monitoring       | More files to manage               |
| Multi-source (one file) | Single deployment, cross-chain UNION possible | Coupled lifecycle, harder to debug |

**Resource size:** `m` per chain. Because each chain runs in its own pipeline, you can also size each one independently: a busy chain can run `l` while a quiet one runs `s`.

## Dynamic table architecture

Dynamic tables give a pipeline runtime-updatable lookup data, the Turbo answer to "no joins in streaming SQL". A `dynamic_table` transform is backed by a table (typically PostgreSQL) that you can update at any time without restarting the pipeline, and SQL transforms query it with the `dynamic_table_check()` function.

### Pattern: dynamic allowlist or blocklist

```text theme={"dark"}
                    ┌──────────────────────┐
                    │  External updates    │
                    │  (Postgres / API)    │
                    └──────────┬───────────┘
                               ▼
source ──→ sql transform ──→ [dynamic_table_check()] ──→ sink
```

The SQL transform filters records against the dynamic table. Change the table contents externally (from a backend, an admin tool, or plain SQL) and the pipeline picks up the change within seconds.

**Use when** the set of things you filter on (wallets, contracts, tokens) changes over time, or is too large for a hardcoded `WHERE ... IN` list.

This example only keeps transfers that touch a tracked wallet:

```yaml theme={"dark"}
name: tracked-wallet-transfers
resource_size: s

sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Postgres-backed lookup table; update it externally at any time
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: user_wallets
    secret_name: MY_POSTGRES

  # Keep only transfers involving a tracked wallet, and tag the direction
  wallet_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        *,
        CASE
          WHEN dynamic_table_check('tracked_wallets', sender) THEN 'outgoing'
          ELSE 'incoming'
        END AS direction
      FROM ethereum_transfers
      WHERE dynamic_table_check('tracked_wallets', sender)
         OR dynamic_table_check('tracked_wallets', recipient)

sinks:
  postgres_output:
    type: postgres
    from: wallet_transfers
    schema: public
    table: wallet_activity
    secret_name: MY_POSTGRES
    primary_key: id
```

Add a wallet from any Postgres client, and the pipeline starts matching it within seconds:

```sql theme={"dark"}
INSERT INTO streamling.user_wallets (value) VALUES (lower('0x...your-wallet...'));
```

### Pattern: lookup enrichment

```text theme={"dark"}
source ──→ decode ──→ filter ──→ sql (with dynamic_table_check) ──→ sink
                                        ▲
                              [token_metadata table]
                              (Postgres-backed)
```

Store lookup data (token allowlists, protocol contract sets, factory-created pool addresses) in the dynamic table and reference it from transforms. A dynamic table can even populate itself from pipeline data using an inline `sql:` query; see the [factory pattern example](/turbo-pipelines/transforms/dynamic-tables#example-factory-pattern).

### Choosing a backend

| Backend    | `backend_type` | When to use                                                                |
| ---------- | -------------- | -------------------------------------------------------------------------- |
| PostgreSQL | `Postgres`     | Data managed by external systems; must persist across pipeline restarts    |
| In-memory  | `InMemory`     | Auto-populated from pipeline data; ephemeral; avoids a database round trip |

Use `Postgres` for production: it persists across restarts and can be updated externally.

### Sizing considerations

* Dynamic tables add memory overhead proportional to table size, so for large lookup tables (over \~100K rows) use the `Postgres` backend.
* Lookups are batched and indexed, but cost still scales with table size, so keep tables as small as your use case allows.
* **Resource size:** `s` is usually enough for allowlist filtering; size for your transform complexity, not for the dynamic table itself.

For full configuration syntax, table management, and more examples, see [Dynamic tables](/turbo-pipelines/transforms/dynamic-tables) and the [`dynamic_table_check` reference](/turbo-pipelines/reference/sql-functions#dynamic-table-check).

## PostgreSQL aggregate sink pattern

The `postgres_aggregate` sink maintains real-time running aggregations (balances, counters, totals) using a two-table pattern: a landing table receives raw events, and a database trigger incrementally updates an aggregation table.

```text theme={"dark"}
source ──→ transform ──→ landing table ──→ trigger ──→ aggregation table
```

**Use when** you need running totals updated on every event, without recomputing aggregates from scratch. Your application reads the small aggregation table instead of scanning event history.

This example turns every ERC-20 transfer into two balance changes (negative for the sender, positive for the recipient) and lets the trigger maintain per-account, per-token balances:

```yaml theme={"dark"}
name: token-balances
resource_size: m

sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Each transfer produces two rows: -amount for the sender, +amount for the recipient
  balance_changes:
    type: sql
    primary_key: change_id
    sql: |
      SELECT
        CONCAT(id, '-out') AS change_id,
        lower(sender) AS account,
        lower(address) AS token_address,
        CAST(amount AS DECIMAL(38, 0)) * -1 AS amount
      FROM ethereum_transfers
      UNION ALL
      SELECT
        CONCAT(id, '-in') AS change_id,
        lower(recipient) AS account,
        lower(address) AS token_address,
        CAST(amount AS DECIMAL(38, 0)) AS amount
      FROM ethereum_transfers

sinks:
  balances:
    type: postgres_aggregate
    from: balance_changes
    schema: public
    landing_table: balance_change_log
    agg_table: account_balances
    primary_key: change_id
    secret_name: MY_POSTGRES
    group_by:
      account:
        type: text
      token_address:
        type: text
    aggregate:
      balance:
        from: amount
        fn: sum
        type: numeric(38,0)
      change_count:
        fn: count
```

Query the result like any Postgres table:

```sql theme={"dark"}
SELECT balance FROM public.account_balances
WHERE account = lower('0x...') AND token_address = lower('0x...');
```

**Supported aggregation functions:** `sum`, `count`, `avg`, `min`, `max`. Not every function supports every operation type: `sum` and `avg` cannot handle updates, and `min`/`max` are insert-only. See [supported aggregation functions](/turbo-pipelines/sinks/postgres-aggregate#supported-aggregation-functions) before choosing.

<Note>
  This example tracks every token and holder on Ethereum Mainnet, which is a high write volume for a Postgres database. For production, add a `WHERE` clause to track specific tokens, or size your database accordingly.
</Note>

**Resource size:** `m`. Throughput is usually bounded by the Postgres database rather than the pipeline, so scale the database before scaling the pipeline.

For landing-table deduplication, trigger internals, and update/delete semantics, see the [PostgreSQL aggregation sink](/turbo-pipelines/sinks/postgres-aggregate) reference.

## Next steps

* Copy-paste starting points for all of these patterns (and more) live in the [pipeline cookbook](/turbo-pipelines/reference/pipeline-cookbook).
* Full YAML field reference: [pipeline configuration](/turbo-pipelines/pipeline-config).
* Deploy and monitor with the [CLI reference](/turbo-pipelines/cli-reference).


## Related topics

- [Pipeline cookbook](/turbo-pipelines/reference/pipeline-cookbook.md)
- [Stream onchain data with Turbo](/turbo-pipelines/introduction.md)
- [Agent Skills](/ai-skills.md)
- [Troubleshooting Turbo pipelines](/turbo-pipelines/guides/troubleshooting.md)
- [Solana sources for Turbo pipelines](/turbo-pipelines/sources/solana.md)