Build a Pipeline

A Pipeline passes source-table block data to a Function, then writes the results to a target table. The same configuration supports historical backfills and ongoing block processing. First follow Track liquidity pool reserves to create the state table demo.pool_reserves.eth and Function demo.read_pool_reserves.

Configure the source and function

Create a Notebook in demo with the following script. Sync events in the source table trigger reserve reads; ROW passes the current event to the saved Function.

from blockdb import Column
from blockdb.operator import filter
from blockx import Pipeline, ROW

SPACE = 'demo'

pipeline = Pipeline(
    sources=[{
        'table_id': 'chain.event.eth',
        'func_id': f'{SPACE}.read_pool_reserves',
        'args_tmpl': [ROW],
        'operator': filter(
            (Column('contract_id') == '0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc')
            & (Column('name') == 'Sync')
        ),
    }],
    target=f'{SPACE}.pool_reserves.eth',
)

pipeline.backfill(start=25939925, end=25939934)

Setting	Purpose
`sources`	Source tables that trigger computation; multiple sources are supported
`table_id`	Full source table ID
`func_id`	Saved Function ID
`args_tmpl`	Positional argument template; `ROW` is replaced by the source record
`operator`	Filters source records before computation, here by pool address and event name
`target`	Table receiving the Function’s returned records

The example Function queries the contract using the event’s block_id, keeping reserve reads at that event’s block state.

Run a historical backfill

Save the script, choose compute resources, and click Run. backfill() includes both boundary heights; this example processes ten blocks. When the run completes, check the Notebook output: this example processes 10 blocks and writes 2 records. Then open the target table to check its records and processed ranges.

Declare dependencies

If the function reads another Block State Table, add that table to requires. For example:

pipeline = Pipeline(
    sources=sources,
    target='your_space.output.eth',
    requires=['your_space.reference_state.eth'],
)

Replace sources with your configured source list and both table IDs with actual tables. A dependency is not passed to the function automatically and does not independently trigger computation. It requires data at the corresponding height to be ready; the function must still read that table explicitly.

Keep the table updated

After verifying a small backfill, replace the script’s final line with:

pipeline.update()

update() fills gaps based on the target table’s start height and processed ranges, then keeps processing new events.

The Pipeline fills unprocessed historical blocks before following new blocks

For ongoing execution, follow Create a continuous schedule. To stop maintenance, disable the schedule and confirm its instance has ended.

Overview

Quickstart

Guides

Reference

Configure the source and function

Run a historical backfill

Declare dependencies

Keep the table updated