> ## 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.
# Dynamic Tables
> Create real-time lookup tables for filtering and enrichment
Dynamic tables are a powerful feature that allows you to maintain updatable lookup tables within your pipeline. Unlike static SQL transforms, dynamic tables can be modified in real-time without redeploying your pipeline.
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
Monitor transfers to/from specific wallet addresses
Track processed records to avoid duplicates
Track contracts created by a factory
## Basic Configuration
```yaml theme={null}
transforms:
my_dynamic_table:
type: dynamic_table
backend_type: Postgres
backend_entity_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
```
### Parameters
Must be `dynamic_table`
Storage backend: `Postgres`
The table name in the backend storage. For Postgres, this creates a table in
the `streamling` schema (configurable via the `schema` field).
The name of a Goldsky secret containing Postgres credentials. Required for
the Postgres backend.
Optional. SQL query to automatically populate the table from pipeline data.
Optional. PostgreSQL schema name for the table. Defaults to `streamling`.
Optional. Name of the primary key column storing values. Defaults to `value`.
Optional. Name of the timestamp column. Defaults to `updated_at`.
## Backend Types
### PostgreSQL Backend (Recommended)
Best for production deployments requiring persistence:
```yaml theme={null}
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:
```yaml theme={null}
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`:
```sql theme={null}
CREATE TABLE my_app.tracked_contracts (
contract_addr TEXT PRIMARY KEY,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
```
Use custom schemas to organize dynamic tables by application or environment. For example, use `production` and `staging` schemas to separate data.
## PostgreSQL setup
Dynamic tables with a Postgres backend need a database Goldsky can connect to and a [secret](/turbo-pipelines/pipeline-config#secrets) with valid credentials. Use the same secret for your `dynamic_table` transform and any Postgres sinks in the same pipeline.
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](/turbo-pipelines/sinks/postgres#overview) 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:
```yaml theme={null}
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.`).
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](/mirror/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](mailto: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:
```sql theme={null}
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;
```
```bash theme={null}
goldsky secret create MY_POSTGRES_WHITELISTED
```
```yaml theme={null}
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.
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.
```sql theme={null}
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.
```bash theme={null}
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
```
No `use_dedicated_ip` flag is needed. Reference the secret in your `dynamic_table` transform (and sinks if applicable), then deploy:
```bash theme={null}
goldsky turbo validate wallet-tracker.yaml
goldsky turbo apply wallet-tracker.yaml
```
### Verify the connection
After deployment, confirm the dynamic table was created:
```sql theme={null}
SELECT * FROM streamling. LIMIT 10;
```
If the pipeline fails to start, check logs for connection errors:
```bash theme={null}
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:
```yaml theme={null}
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
```sql theme={null}
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.
See also [`dynamic_table_check` in SQL functions reference](/turbo-pipelines/reference/sql-functions#dynamic-table-check).
## Auto-Population with SQL
You can automatically populate a dynamic table from your pipeline data:
```yaml theme={null}
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.)
```sql theme={null}
-- 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:
```yaml theme={null}
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:**
```sql theme={null}
-- 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:
```yaml theme={null}
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:**
```sql theme={null}
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:
```yaml theme={null}
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 allows you to control which contracts to track without redeploying your pipeline.
```sql theme={null}
-- 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:
```yaml theme={null}
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:
# Decode PoolCreated events from factory
pool_created_events:
type: sql
primary_key: id
sql: |
SELECT
id,
address,
evm_decode_log(
'[{"anonymous":false,"inputs":[{"indexed":true,"name":"token0","type":"address"},{"indexed":true,"name":"token1","type":"address"},{"indexed":true,"name":"fee","type":"uint24"},{"indexed":false,"name":"tickSpacing","type":"int24"},{"indexed":false,"name":"pool","type":"address"}],"name":"PoolCreated","type":"event"}]',
topics,
data
) as decoded,
block_number,
block_timestamp
FROM base_logs
WHERE
address = lower('0x33128a8fC17869897dcE68Ed026d694621f6FDfD')
AND SPLIT_INDEX(topics, ',', 0) = '0x783cca1c0412dd0d695e784568c96da2e9c22ff989357a2e8b1d9b2b4e6b7118'
# Auto-populate dynamic table with factory-created pools
factory_pools:
type: dynamic_table
backend_type: Postgres
backend_entity_name: factory_addresses
secret_name: MY_POSTGRES
sql: |
SELECT lower(decoded.event_params[5]) as contract_address
FROM pool_created_events
# 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:
```yaml theme={null}
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:
```yaml theme={null}
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
* 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.
* Postgres backend: each `dynamic_table_check()` call queries the table directly (no in-process cache), so changes take effect on the next batch — typically within a second or two.
* Auto-population via SQL: updates flow through with normal pipeline latency.
* No hard row limit is enforced, but lookup cost scales with table size — 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
Always use the Postgres backend for production deployments to ensure data persistence and external updatability.
Use specific filters in auto-population SQL to ensure data consistency and proper synchronization.
Store addresses and other identifiers in lowercase for consistent matching:
```sql theme={null}
INSERT INTO streamling.my_table (value) VALUES (lower('0x...'));
```
Periodically check table sizes and clean up old entries by `updated_at` if needed.
## Related topics
- [Turbo Pipeline Configuration](/turbo-pipelines/pipeline-config.md)
- [Benefits](/benefits.md)
- [Turbo SQL Functions Reference](/turbo-pipelines/reference/sql-functions.md)
- [SQL](/turbo-pipelines/transforms/sql.md)
- [EVM Sources](/turbo-pipelines/sources/evm.md)