tomevault-io Substreams SQL Expert (SVM)
Expert assistant for building SQL database sinks from Solana/SVM Substreams data.
Prerequisites
- substreams-sink-sql: Required CLI tool. Install from releases.
Core Concepts
DatabaseChanges Approach
SVM Substreams use the CDC (Change Data Capture) approach via DatabaseChanges:
use substreams_database_change::pb::sf::substreams::sink::database::v1::DatabaseChanges;
use substreams_database_change::tables::Tables;
#[substreams::handlers::map]
pub fn db_out(
clock: Clock,
raydium_events: raydium::Events,
jupiter_events: jupiter::Events,
) -> Result<DatabaseChanges, Error> {
let mut tables = Tables::new();
// Insert block metadata
let seconds = clock.timestamp.as_ref().unwrap().seconds;
tables.create_row("blocks", [("block_num", clock.number.to_string())])
.set("block_hash", &clock.id)
.set("timestamp", seconds.to_string());
// Insert DEX swap events
for (tx_index, tx) in raydium_events.transactions.iter().enumerate() {
for (instr_index, instruction) in tx.instructions.iter().enumerate() {
let keys = common::db::common_key_v2(&clock, tx_index, instr_index);
let row = tables.create_row("raydium_amm_v4_swap", keys);
common::db::set_raydium_transaction_v2(tx, row);
common::db::set_raydium_instruction_v2(instruction, row);
// Set event-specific fields...
}
}
Ok(tables.to_database_changes())
}
Primary Key Pattern
SVM tables use a composite primary key of (block_hash, transaction_index, instruction_index):
pub fn common_key_v2(clock: &Clock, transaction_index: usize, instruction_index: usize) -> [(&'static str, String); 3] {
[
("block_hash", clock.id.to_string()),
("transaction_index", transaction_index.to_string()),
("instruction_index", instruction_index.to_string()),
]
}
ClickHouse Schema Patterns
Split Schema Files
The SVM repo uses split schema files for maintainability:
db-svm-dex-clickhouse/
├── schema.0.blocks.sql # Block metadata table
├── schema.0.functions.helpers.sql # Helper functions
├── schema.0.functions.programs.sql # Program ID mappings
├── schema.0.functions.tokens.sql # Token metadata functions
├── schema.0.template.sql # Template for new DEX tables
├── schema.1.events.raydium-amm-v4.sql # Raydium AMM V4 events
├── schema.1.events.jupiter.sql # Jupiter swap events
├── schema.1.events.pumpfun.sql # PumpFun events
├── schema.1.events.orca.sql # Orca Whirlpool events
├── schema.2.view.swaps.sql # Unified swaps view
├── schema.3.view.ohlc_prices.sql # OHLC price materialized view
└── schema.3.view.pool_activity_summary.sql
ClickHouse Table Template
CREATE TABLE IF NOT EXISTS raydium_amm_v4_swap
(
-- block --
block_num UInt64,
block_hash FixedString(64),
timestamp DateTime,
-- ordering --
transaction_index UInt32,
instruction_index UInt32,
-- transaction --
signature String,
fee_payer String,
signers_raw String DEFAULT '',
fee UInt64 DEFAULT 0,
compute_units_consumed UInt64 DEFAULT 0,
-- instruction --
program_id String,
stack_height UInt32,
-- event-specific --
amount_in UInt64,
minimum_amount_out UInt64,
pool_coin_token String,
pool_pc_token String,
-- versioning --
version UInt64 MATERIALIZED toUInt64(block_num) * 4294967296 + toUInt64(transaction_index) * 65536 + toUInt64(instruction_index)
)
ENGINE = ReplacingMergeTree(version)
PRIMARY KEY (block_hash, transaction_index, instruction_index)
ORDER BY (block_hash, transaction_index, instruction_index);
Common ClickHouse Indexes
CREATE INDEX IF NOT EXISTS idx_block_num ON raydium_amm_v4_swap (block_num) TYPE minmax;
CREATE INDEX IF NOT EXISTS idx_timestamp ON raydium_amm_v4_swap (timestamp) TYPE minmax;
CREATE INDEX IF NOT EXISTS idx_signature ON raydium_amm_v4_swap (signature) TYPE bloom_filter;
CREATE INDEX IF NOT EXISTS idx_fee_payer ON raydium_amm_v4_swap (fee_payer) TYPE bloom_filter;
Unified Swaps View
A materialized view combines all DEX swap tables into a single swaps view:
CREATE TABLE IF NOT EXISTS swaps (
block_num UInt64,
timestamp DateTime,
signature String,
fee_payer String,
program_id String,
dex String,
input_mint String,
input_amount UInt64,
output_mint String,
output_amount UInt64
) ENGINE = MergeTree()
ORDER BY (timestamp, signature);
PostgreSQL Schema Patterns
PostgreSQL Table Template
CREATE TABLE IF NOT EXISTS raydium_amm_v4_swap (
-- block --
block_num INTEGER NOT NULL,
block_hash TEXT NOT NULL,
timestamp TIMESTAMP NOT NULL,
-- ordering --
transaction_index INTEGER NOT NULL,
instruction_index INTEGER NOT NULL,
-- transaction --
signature TEXT NOT NULL,
fee_payer TEXT NOT NULL,
signers_raw TEXT NOT NULL DEFAULT '',
fee BIGINT NOT NULL DEFAULT 0,
compute_units_consumed BIGINT NOT NULL DEFAULT 0,
-- instruction --
program_id TEXT NOT NULL,
stack_height INTEGER NOT NULL,
-- event-specific --
amount_in BIGINT NOT NULL,
minimum_amount_out BIGINT NOT NULL,
PRIMARY KEY (block_hash, transaction_index, instruction_index)
);
CREATE INDEX IF NOT EXISTS idx_raydium_swap_block_num ON raydium_amm_v4_swap (block_num);
CREATE INDEX IF NOT EXISTS idx_raydium_swap_timestamp ON raydium_amm_v4_swap (timestamp);
CREATE INDEX IF NOT EXISTS idx_raydium_swap_signature ON raydium_amm_v4_swap (signature);
CREATE INDEX IF NOT EXISTS idx_raydium_swap_fee_payer ON raydium_amm_v4_swap (fee_payer);
Manifest Configuration
Base DB Module
imports:
database_changes: ../spkg/substreams-sink-database-changes-v4.0.0.spkg
sql: ../spkg/substreams-sink-sql-protodefs-v1.0.7.spkg
Sink Module
sink:
module: db_out
type: sf.substreams.sink.sql.v1.Service
config:
schema: "./schema.sql"
engine: clickhouse # or "postgres"
postgraphile_frontend:
enabled: false # true for postgres
Running the Sink
# ClickHouse
substreams-sink-sql run \
"clickhouse://default:@localhost:9000/solana" \
./db-svm-dex-clickhouse/substreams.yaml
# PostgreSQL
substreams-sink-sql run \
"postgres://user:pass@localhost:5432/solana?sslmode=disable" \
./db-svm-dex-postgres/substreams.yaml
Source: pinax-network/substreams-svm — distributed by TomeVault.