Dynamic tables for blockchain data filtering and enrichment

Dynamic tables are lookup tables you can update while the pipeline runs. Change a watchlist or an allowlist and the running pipeline picks it up, no redeploy needed. Dynamic tables work by maintaining a lookup table in a Postgres database. The lookup table can be updated in the pipeline or externally, and the pipeline can filter on the values in the lookup table using the dynamic_table_check() function. This feature can be used with a Goldsky Hosted Postgres database, or with your own Postgres database, allowing for clean compliance with GDPR and other data protection laws.

Key Features

Real-time Updates: Add or remove values without pipeline restart by updating a postgres table.
Unbounded Filters: Filter on unlimited values from onchain data or backend updates.
High Throughput: Sub-microsecond lookups, with dynamically batched queries for performance.
Pipeline Updates: Update the table with on-chain data in the same pipeline.

Use Cases

Wallet Tracking

Monitor transfers to/from specific wallet addresses

Deduplication

Track processed records to avoid duplicates

Factory Pattern

Track contracts created by a factory

Basic Configuration

transforms:
  my_dynamic_table:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: <table_name>
    secret_name: <secret_name> # A Goldsky Postgres secret
    sql: | # Optional - auto-populate from pipeline data
      SELECT column FROM source
    # Optional PostgreSQL customization
    schema: custom_schema # Default: streamling
    column: custom_column # Default: value
    time_column: custom_time # Default: updated_at
    # Optional in-memory caching
    cache: true # Optional. Default: false
    cache_refresh_debounce_ms: 1000 # Optional. Only set when cache is true. Default: 1000

Parameters

string

required

Must be dynamic_table

string

required

Storage backend: Postgres

string

required

The table name in the backend storage. For Postgres, this creates a table in the streamling schema (configurable via the schema field).

string

required

The name of a Goldsky secret containing Postgres credentials. Required for the Postgres backend.

string

Optional. SQL query to automatically populate the table from pipeline data. Must be a single projection/filter (SELECT ... FROM ... WHERE ...; no subqueries, joins, or aggregations) reading directly from a source; when the table is consumed via dynamic_table_check, it is planned before the pipeline’s transforms, so its SQL cannot reference a transform.

string

Optional. PostgreSQL schema name for the table. Defaults to streamling.

string

Optional. Name of the primary key column storing values. Defaults to value.

string

Optional. Name of the timestamp column. Defaults to updated_at. The in-memory cache uses this column to find new rows.

boolean

Optional. Set to true to load the table into memory and serve lookups from the cache instead of querying Postgres for each batch. Defaults to false. See In-memory caching.

integer

Optional. How often, in milliseconds, the pipeline checks Postgres for new rows. Only set this field when cache is true. Defaults to 1000.

Backend Types

PostgreSQL Backend (Recommended)

Best for production deployments requiring persistence:

transforms:
  tracked_contracts:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: tracked_contracts
    secret_name: MY_POSTGRES

Benefits:

Data persists across pipeline restarts and failures
Can be updated externally via direct SQL, no redeploy needed
Indexed primary-key lookups scale to millions of rows

Table Structure: PostgreSQL dynamic tables are created with two columns:

A primary key column (default: value) storing the lookup values
A timestamp column (default: updated_at) automatically set to the insertion time

Table Location: By default, tables are created in the streamling schema: streamling.tracked_contracts

Custom schema and column names

You can customize the schema, column name, and timestamp column name:

transforms:
  tracked_contracts:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: tracked_contracts
    secret_name: MY_POSTGRES
    schema: my_app          # Custom schema (default: streamling)
    column: contract_addr   # Custom column name (default: value)
    time_column: created_at # Custom timestamp column (default: updated_at)

This creates a table at my_app.tracked_contracts:

CREATE TABLE my_app.tracked_contracts (
  contract_addr TEXT PRIMARY KEY,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

The table Goldsky creates for you sets the timestamp column with DEFAULT NOW(). If you plan to use in-memory caching, create the table yourself before you deploy the pipeline and set the timestamp column to DEFAULT clock_timestamp(). See Requirements for cached tables.

Use custom schemas to organize dynamic tables by application or environment. For example, use production and staging schemas to separate data.

In-memory caching

By default, every dynamic_table_check() call sends a lookup query to Postgres for each batch. With caching enabled, the pipeline loads the whole table into memory once at startup and serves lookups from memory. This removes the per-batch database round trip, which lowers latency on busy pipelines. After startup, the pipeline checks MAX(time_column) on an interval set by cache_refresh_debounce_ms (default: once per second). It then loads only the rows newer than the last value it saw. To enable caching, set cache: true and make sure time_column points to the table’s timestamp column:

transforms:
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: tracked_wallets
    secret_name: MY_POSTGRES
    time_column: updated_at         # Default: updated_at
    cache: true                     # Required to enable caching
    cache_refresh_debounce_ms: 1000 # Optional. Only set when cache is true

Both fields are optional. cache defaults to false. Only set cache_refresh_debounce_ms when cache is true.

Requirements for cached tables

The cache only adds rows. It never re-reads rows it has already loaded. Follow these rules when you write to a cached table:

Set an always-increasing timestamp on every insert

Every insert must set time_column, and its value must always increase. Use clock_timestamp(), for example as the column default. The table Goldsky creates automatically uses NOW(), so create the table yourself before you deploy the pipeline:

CREATE TABLE streamling.tracked_wallets (
  value TEXT PRIMARY KEY,
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp()
);

Do not use NOW() or CURRENT_TIMESTAMP. Both are fixed when the transaction starts, so a row can commit with a timestamp older than one the pipeline has already seen. The pipeline never loads that row.

Serialize writes from multiple processes

If more than one process writes to the table, each insert must first take a transaction-level advisory lock on the table in the same transaction:

BEGIN;
SELECT pg_advisory_xact_lock(hashtextextended('streamling.tracked_wallets', 0));
INSERT INTO streamling.tracked_wallets (value) VALUES (lower('0x...'));
COMMIT;

Replace streamling.tracked_wallets with your <schema>.<table>.

Restart the pipeline to apply removals and updates

The cache does not pick up deleted or updated rows. If you delete a row, for example when you stop tracking a wallet, the pipeline keeps matching it until you restart the pipeline:

goldsky turbo restart <pipeline>

Turbo pipelines restart quickly. If you need removals to take effect right away, leave caching off.

PostgreSQL setup

Dynamic tables with a Postgres backend need a database Goldsky can connect to and a secret with valid credentials. Use the same secret for your dynamic_table transform and any Postgres sinks in the same pipeline.

Goldsky hosted PostgreSQL
IP allowlisted PostgreSQL
Public PostgreSQL

Goldsky can provision and manage a PostgreSQL database for you. This is the fastest path for dynamic tables: no IP allowlisting and no database administration on your side.Follow Goldsky-hosted Postgres on the Turbo Postgres sink page to provision a database from the web app and store the connection string as a Goldsky secret. Reference that secret in your dynamic_table transform:

transforms:
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: user_wallets
    secret_name: MY_HOSTED_POSTGRES

You can reuse the same secret for Postgres sinks in the same pipeline. After deploy, update lookup values directly in SQL (default table location: streamling.<backend_entity_name>).

Use this when your Postgres instance only accepts connections from specific IP addresses. Goldsky routes pipeline traffic through static egress IPs so you can allowlist Goldsky without opening the database to the internet.Follow Static IPs to enable the feature on your account, allowlist Goldsky’s egress IPs on your database firewall, and set use_dedicated_ip: true at the top level of your pipeline YAML. Turbo uses the same setting as Mirror.

Static IPs are an enterprise feature. Contact support@goldsky.com to enable use_dedicated_ip on your account before deploying.

Then create a database role, secret, and reference both in your pipeline:

CREATE ROLE goldsky_writer WITH LOGIN PASSWORD 'your_secure_password';

GRANT CREATE ON DATABASE your_database TO goldsky_writer;
GRANT USAGE, CREATE ON SCHEMA streamling TO goldsky_writer;

goldsky secret create MY_POSTGRES_WHITELISTED

name: wallet-tracker
resource_size: s
use_dedicated_ip: true  # Required for IP-allowlisted Postgres

transforms:
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: user_wallets
    secret_name: MY_POSTGRES_WHITELISTED

Without use_dedicated_ip: true, the pipeline connects from dynamic IPs and your database firewall will reject the connection.

Use this when your database accepts connections from the public internet, or when you use a managed provider (Neon, Supabase, RDS with a public endpoint, etc.) that does not restrict by source IP.

Configure your database for external access

Depending on your provider:

Neon / Supabase / similar: Use the connection string from your project dashboard. SSL is typically required (?sslmode=require in the URL).
Self-hosted Postgres: Set listen_addresses = '*' in postgresql.conf, add a pg_hba.conf rule for your Goldsky writer role, and ensure your host firewall allows inbound TCP on the Postgres port.

Create a database role

CREATE ROLE goldsky_writer WITH LOGIN PASSWORD 'your_secure_password';

GRANT CREATE ON DATABASE your_database TO goldsky_writer;
GRANT USAGE, CREATE ON SCHEMA streamling TO goldsky_writer;

If you use a custom schema instead of the default streamling schema, replace streamling with your schema name.

Create a secret

goldsky secret create MY_POSTGRES

When prompted, enter your connection string:

postgres://goldsky_writer:your_secure_password@db.example.com:5432/your_database?sslmode=require

Reference the secret and deploy

No use_dedicated_ip flag is needed. Reference the secret in your dynamic_table transform (and sinks if applicable), then deploy:

goldsky turbo validate wallet-tracker.yaml
goldsky turbo apply wallet-tracker.yaml

Verify the connection

After deployment, confirm the dynamic table was created:

SELECT * FROM streamling.<backend_entity_name> LIMIT 10;

If the pipeline fails to start, check logs for connection errors:

goldsky turbo logs wallet-tracker

Common issues:

Connection refused or timeout: Firewall, wrong host/port, or missing IP allowlist entry (use the IP allowlisted tab if your database blocks unknown IPs).
Permission denied: The database role lacks CREATE on the database or USAGE/CREATE on the target schema.
SSL required: Add ?sslmode=require to your connection string for providers that enforce TLS.

Using Dynamic Tables in SQL

Once defined, use the dynamic_table_check() function in SQL transforms:

transforms:
  # Define the dynamic table
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: user_wallets
    secret_name: MY_POSTGRES

  # Use it in SQL
  filtered_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT *
      FROM erc20_transfers
      WHERE dynamic_table_check('tracked_wallets', to_address)

Function Signature

dynamic_table_check(table_name, value) -> BOOLEAN

table_name (TEXT): The transform name of the dynamic table in your pipeline (not the backend_entity_name). Must be a string literal: the same value on every row.
value (TEXT): The value to check for existence.
Returns: true if the value exists in the table, false otherwise.

Auto-Population with SQL

You can automatically populate a dynamic table from your pipeline data:

transforms:
  # Auto-populate from pipeline
  usdc_senders:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: usdc_senders
    secret_name: MY_POSTGRES
    sql: |
      SELECT DISTINCT from_address
      FROM erc20_transfers
      WHERE contract_address = lower('0x...')

  # Use in subsequent transform
  all_usdc_activity:
    type: sql
    primary_key: id
    sql: |
      SELECT *
      FROM erc20_transfers
      WHERE dynamic_table_check('usdc_senders', from_address)
         OR dynamic_table_check('usdc_senders', to_address)

When using SQL to populate, the query only supports projections and filters (no joins or aggregations).

Manual Updates

For Postgres backends, you can update the table directly using any Postgres client. (Substitute the schema and column names if you customized them.)

-- Add values
INSERT INTO streamling.tracked_contracts (value) VALUES
  (lower('0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174')),
  (lower('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'));

-- Remove values
DELETE FROM streamling.tracked_contracts
WHERE value = lower('0x...');

-- Check contents (includes updated_at timestamp)
SELECT * FROM streamling.tracked_contracts;

-- Query recently added entries
SELECT * FROM streamling.tracked_contracts
WHERE updated_at > NOW() - INTERVAL '1 hour';

Changes take effect immediately - within seconds, your pipeline will start filtering based on the updated values!

Example: Track Specific Token Contracts

Monitor transfers for specific ERC-20 tokens like USDC:

name: token-tracker
resource_size: s

sources:
  polygon_transfers:
    type: dataset
    dataset_name: matic.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Dynamic table to store token contract addresses we want to track
  tracked_tokens:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: tracked_tokens
    secret_name: MY_POSTGRES

  # Filter to only transfers of tracked tokens
  filtered_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT *
      FROM polygon_transfers
      WHERE dynamic_table_check('tracked_tokens', address)

sinks:
  postgres_output:
    type: postgres
    from: filtered_transfers
    schema: public
    table: tracked_token_transfers
    secret_name: MY_POSTGRES
    primary_key: id

To start tracking USDC on Polygon:

-- Add USDC contract address (Polygon)
INSERT INTO streamling.tracked_tokens (value) VALUES (lower('0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174'));

-- Add USDT contract address (Polygon)
INSERT INTO streamling.tracked_tokens (value) VALUES (lower('0xc2132D05D31c914a87C6611C10748AEb04B58e8F'));

Within seconds, your pipeline will start capturing transfers for these tokens!

Example: Track Wallet Activity

Monitor all ERC-20 transfers for specific wallets:

name: wallet-tracker
resource_size: s

sources:
  polygon_transfers:
    type: dataset
    dataset_name: matic.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Dynamic table for wallets we're tracking
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: user_wallets
    secret_name: MY_POSTGRES

  # Filter to only transfers involving tracked wallets
  wallet_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        *,
        CASE
          WHEN dynamic_table_check('tracked_wallets', sender) THEN 'outgoing'
          WHEN dynamic_table_check('tracked_wallets', recipient) THEN 'incoming'
          ELSE 'unknown'
        END as direction
      FROM polygon_transfers
      WHERE
        dynamic_table_check('tracked_wallets', sender)
        OR dynamic_table_check('tracked_wallets', recipient)

sinks:
  postgres_sink:
    type: postgres
    from: wallet_transfers
    schema: public
    table: wallet_activity
    secret_name: MY_POSTGRES

To add a wallet to track:

INSERT INTO streamling.user_wallets (value) VALUES (lower('0x...your-wallet...'));

Example: Complete Pipeline with Dynamic Tables

This example shows a complete pipeline that uses dynamic tables to filter ERC-20 transfers to specific contracts:

name: erc20-filtered-transfers
resource_size: s

sources:
  polygon_erc20_transfers:
    type: dataset
    dataset_name: matic.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Define a dynamic table to store contract addresses we want to track
  tracked_contracts:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: tracked_contracts
    secret_name: MY_POSTGRES_SECRET

  # Filter transfers to only include our tracked contracts
  filtered_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        *,
        'polygon' as network
      FROM polygon_erc20_transfers
      WHERE dynamic_table_check('tracked_contracts', address)

sinks:
  postgres_sink:
    type: postgres
    schema: public
    table: erc20_transfers
    secret_name: MY_POSTGRES_SECRET
    from: filtered_transfers
    primary_key: id

Add contracts to track: The dynamic table lets you change which contracts to track without redeploying your pipeline.

-- Add USDC contract on Polygon
INSERT INTO streamling.tracked_contracts (value) VALUES (
  lower('0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174')
);

-- Add USDT contract on Polygon
INSERT INTO streamling.tracked_contracts (value) VALUES (
  lower('0xc2132D05D31c914a87C6611C10748AEb04B58e8F')
);

-- View tracked contracts (includes updated_at timestamp)
SELECT * FROM streamling.tracked_contracts;

Within seconds, your pipeline will start processing transfers for these contracts!

Example: Factory Pattern

Track all contracts created by a factory and filter events from those contracts:

name: uniswap-pool-tracker
resource_size: m

sources:
  base_logs:
    type: dataset
    dataset_name: base.raw_logs
    version: 1.0.0
    start_at: earliest

transforms:
  # Auto-populate dynamic table with factory-created pools.
  # A dynamic table's population SQL must be a single projection/filter
  # reading directly from a source, so the pool address is extracted
  # positionally instead of via evm_decode_log: the PoolCreated data payload
  # is two 32-byte words (tickSpacing, then pool), and the address is the
  # last 20 bytes (40 hex chars) of the second word.
  factory_pools:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: factory_addresses
    secret_name: MY_POSTGRES
    sql: |
      SELECT lower(concat('0x', substr(data, 91, 40))) as contract_address
      FROM base_logs
      WHERE
        address = lower('0x33128a8fC17869897dcE68Ed026d694621f6FDfD')
        AND SPLIT_INDEX(topics, ',', 0) = '0x783cca1c0412dd0d695e784568c96da2e9c22ff989357a2e8b1d9b2b4e6b7118'

  # Decode swap events from raw logs
  decoded_swaps:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        transaction_hash,
        address as pool_address,
        evm_decode_log(
          '[{"anonymous":false,"inputs":[{"indexed":true,"name":"sender","type":"address"},{"indexed":true,"name":"recipient","type":"address"},{"indexed":false,"name":"amount0","type":"int256"},{"indexed":false,"name":"amount1","type":"int256"},{"indexed":false,"name":"sqrtPriceX96","type":"uint160"},{"indexed":false,"name":"liquidity","type":"uint128"},{"indexed":false,"name":"tick","type":"int24"}],"name":"Swap","type":"event"}]',
          topics,
          data
        ) as decoded,
        block_timestamp
      FROM base_logs
      WHERE
        SPLIT_INDEX(topics, ',', 0) = '0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67'
        AND dynamic_table_check('factory_pools', address)

  # Extract swap data
  pool_swaps:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        transaction_hash,
        pool_address,
        decoded.event_params[1] as sender,
        decoded.event_params[2] as recipient,
        decoded.event_params[3] as amount0,
        decoded.event_params[4] as amount1,
        decoded.event_params[5] as sqrt_price_x96,
        decoded.event_params[6] as liquidity,
        decoded.event_params[7] as tick,
        block_timestamp
      FROM decoded_swaps

sinks:
  clickhouse_sink:
    type: clickhouse
    from: pool_swaps
    primary_key: id
    table: uniswap_swaps
    secret_name: MY_CLICKHOUSE

This pattern automatically tracks new pools as they’re created and immediately starts capturing their swap events!

Source Validation

When a dynamic table uses SQL to auto-populate, both the dynamic table and any SQL transform using it must reference the same source or upstream transform.This ensures data consistency and proper synchronization.

Good example:

transforms:
  my_table:
    sql: SELECT value FROM source_1 # Uses source_1

  my_transform:
    sql: |
      SELECT * FROM source_1  # Also uses source_1
      WHERE dynamic_table_check('my_table', value)

Bad example:

transforms:
  my_table:
    sql: SELECT value FROM source_1 # Uses source_1

  my_transform:
    sql: |
      SELECT * FROM source_2  # Uses source_2 - ERROR!
      WHERE dynamic_table_check('my_table', value)

Performance Considerations

Lookup Performance

Lookups are batched and executed in parallel against Postgres (ANY(ARRAY[...]) queries).
The value column is PRIMARY KEY, so Postgres uses its unique index automatically.
Large tables (millions of entries) work fine as long as the Postgres instance has adequate resources.

Update Latency

Postgres backend: by default, each dynamic_table_check() call queries the table directly, so changes take effect on the next batch, typically within a second or two.
With in-memory caching enabled, new rows take effect after the next cache refresh (cache_refresh_debounce_ms, default 1000 ms). Deletes and updates require a pipeline restart.
Auto-population via SQL: updates flow through with normal pipeline latency.

Table Size

No hard row limit is enforced, but lookup cost scales with table size, so keep tables as small as your use case allows.
Use specific filters in auto-population SQL to avoid unbounded growth.
For long-running pipelines, consider a cleanup strategy (DELETE old rows by updated_at).

Best Practices

Use Postgres for production

Always use the Postgres backend for production deployments to ensure data persistence and external updatability.

Use specific filters in auto-population SQL

Use specific filters in auto-population SQL to ensure data consistency and proper synchronization.

Lowercase string values

Store addresses and other identifiers in lowercase for consistent matching:

INSERT INTO streamling.my_table (value) VALUES (lower('0x...'));

Monitor table growth

Periodically check table sizes and clean up old entries by updated_at if needed.

Turbo

Guides

Reference

Mirror (legacy)

Dynamic Tables

Key Features

Use Cases

Wallet Tracking

Deduplication

Factory Pattern

Basic Configuration

Parameters

Backend Types

PostgreSQL Backend (Recommended)

Custom schema and column names

In-memory caching

Requirements for cached tables

PostgreSQL setup

Verify the connection

Using Dynamic Tables in SQL

Function Signature

Auto-Population with SQL

Manual Updates

Example: Track Specific Token Contracts

Example: Track Wallet Activity

Example: Complete Pipeline with Dynamic Tables

Example: Factory Pattern

Source Validation

Performance Considerations

Best Practices

​Key Features

​Use Cases

Wallet Tracking

Deduplication

Factory Pattern

​Basic Configuration

​Parameters

​Backend Types

​PostgreSQL Backend (Recommended)

​Custom schema and column names

​In-memory caching

​Requirements for cached tables

​PostgreSQL setup

​Verify the connection

​Using Dynamic Tables in SQL

​Function Signature

​Auto-Population with SQL

​Manual Updates

​Example: Track Specific Token Contracts

​Example: Track Wallet Activity

​Example: Complete Pipeline with Dynamic Tables

​Example: Factory Pattern

​Source Validation

​Performance Considerations

​Best Practices

Key Features

Use Cases

Basic Configuration

Parameters

Backend Types

PostgreSQL Backend (Recommended)

Custom schema and column names

In-memory caching

Requirements for cached tables

PostgreSQL setup

Verify the connection

Using Dynamic Tables in SQL

Function Signature

Auto-Population with SQL

Manual Updates

Example: Track Specific Token Contracts

Example: Track Wallet Activity

Example: Complete Pipeline with Dynamic Tables

Example: Factory Pattern

Source Validation

Performance Considerations

Best Practices