Supabase ETL
Supabase ETL documentation

First Pipeline

Learn Supabase ETL by building a working Postgres replication pipeline.

Ask ChatGPTAsk Claude

Build a pipeline that performs an initial sync of a Postgres table, then replicates row changes through a small custom Destination.

Want a real destination immediately?

The fastest full pipeline is local ClickHouse. After cloning this repository:

cargo x init
cargo x setup replicator
cargo x seed
cargo x run replicator

No cloud account is required. See Standalone Replicator for that path. This tutorial embeds the etl crate with a tiny custom destination instead.

Active development

Supabase ETL is under active development. APIs and setup steps may change before the first stable release.

Prerequisites

  • Rust toolchain 1.95.0, matching rust-toolchain.toml
  • PostgreSQL 14 through 18 with wal_level = logical
  • A Postgres user with LOGIN and REPLICATION. A local superuser such as postgres is enough for this tutorial.
  • Basic familiarity with Rust and SQL

New to Postgres logical replication? Read Logical Replication first. For production settings, publications, and a dedicated replication user, see Configure Postgres.

Create the project

Terminal
cargo new etl-tutorial
cd etl-tutorial
rustup override set 1.95.0

Add dependencies to Cargo.toml:

Cargo.toml
[dependencies]
etl = { git = "https://github.com/supabase/etl" }
tokio = { version = "1", features = ["full"] }
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

Check: Run cargo check and confirm it compiles without errors.

Set up Postgres

Connect to Postgres and create a test database, table, seed rows, and publication:

psql
CREATE DATABASE etl_tutorial;
\c etl_tutorial

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

INSERT INTO users (name, email) VALUES
    ('Alice Johnson', 'alice@example.com'),
    ('Bob Smith', 'bob@example.com');

CREATE PUBLICATION my_publication FOR TABLE users;

Check: SHOW wal_level; returns logical, and SELECT * FROM pg_publication WHERE pubname = 'my_publication'; returns one row.

Write the pipeline

Replace src/main.rs:

src/main.rs
use etl::{
    config::{
        BatchConfig, InvalidatedSlotBehavior, PgConnectionConfig, PipelineConfig,
        TableSyncCopyConfig, TcpKeepaliveConfig, TlsConfig,
    },
    data::TableRow,
    destination::{
        Destination, DestinationWriteStatus, DropTableForCopyResult, WriteEventsDurability,
        TableCopyBatchId, WriteEventsResult, WriteTableRowsResult,
    },
    error::EtlResult,
    event::Event,
    pipeline::Pipeline,
    schema::ReplicatedTableSchema,
    store::MemoryStore,
};
use std::error::Error;

#[derive(Clone)]
struct LoggingDestination;

impl Destination for LoggingDestination {
    fn name() -> &'static str {
        "logging"
    }

    async fn drop_table_for_copy(
        &self,
        _replicated_table_schema: &ReplicatedTableSchema,
        async_result: DropTableForCopyResult<()>,
    ) -> EtlResult<()> {
        println!("preparing fresh table copy");
        async_result.send(Ok(()));
        Ok(())
    }

    async fn write_table_rows(
        &self,
        _replicated_table_schema: &ReplicatedTableSchema,
        _batch_id: Option<TableCopyBatchId>,
        rows: Vec<TableRow>,
        async_result: WriteTableRowsResult,
    ) -> EtlResult<()> {
        println!("copied {} rows", rows.len());
        async_result.send(Ok(DestinationWriteStatus::Durable));
        Ok(())
    }

    async fn write_events(
        &self,
        events: Vec<Event>,
        _durability: WriteEventsDurability,
        async_result: WriteEventsResult,
    ) -> EtlResult<()> {
        println!("received {} ongoing replication events", events.len());
        async_result.send(Ok(DestinationWriteStatus::Durable));
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    tracing_subscriber::fmt::init();

    let pg_config = PgConnectionConfig {
        host: std::env::var("PGHOST").unwrap_or_else(|_| "localhost".to_string()),
        hostaddr: None,
        port: std::env::var("PGPORT")
            .ok()
            .and_then(|port| port.parse().ok())
            .unwrap_or(5432),
        name: std::env::var("PGDATABASE").unwrap_or_else(|_| "etl_tutorial".to_string()),
        username: std::env::var("PGUSER").unwrap_or_else(|_| "postgres".to_string()),
        password: std::env::var("PGPASSWORD").ok().map(Into::into),
        tls: TlsConfig {
            enabled: false,
            trusted_root_certs: String::new(),
        },
        keepalive: TcpKeepaliveConfig::default(),
    };

    let config = PipelineConfig {
        id: 1, // Unique per Postgres cluster. Local `cargo x setup replicator` also uses 1.
        publication_name: "my_publication".to_string(),
        pg_connection: pg_config,
        store_pg_connection: None,
        replication_slot: Default::default(),
        run_source_migrations: true,
        batch: BatchConfig {
            max_fill_ms: 5000,
            memory_budget_ratio: 0.2,
            max_bytes: 32 * 1024 * 1024,
        },
        table_error_retry_delay_ms: 10_000,
        table_error_retry_max_attempts: 5,
        max_table_sync_workers: 4,
        max_copy_connections_per_table: PipelineConfig::DEFAULT_MAX_COPY_CONNECTIONS_PER_TABLE,
        memory_refresh_interval_ms: 100,
        table_sync_monitor_refresh_interval_ms: 10_000,
        memory_backpressure: None,
        table_sync_copy: TableSyncCopyConfig::default(),
        invalidated_slot_behavior: InvalidatedSlotBehavior::default(),
    };

    let store = MemoryStore::new();
    let destination = LoggingDestination;

    println!("Starting pipeline...");
    let mut pipeline = Pipeline::new(config, store, destination);
    pipeline.start().await?;
    pipeline.wait().await?;

    Ok(())
}

Note

The example reads the standard PGHOST, PGPORT, PGDATABASE, PGUSER, and PGPASSWORD environment variables, with local defaults where possible. Pipeline::start() installs the ETL source-side schema helpers before replication begins, even when the tutorial keeps runtime state in MemoryStore. The 20% batch-memory ratio is one global advisory target that ETL divides across positions that can simultaneously hold an accumulating or in-flight decoded batch. The 32 MiB value is the preferred ceiling for each batch, not a reservation. This example disables emergency memory backpressure, but dynamic batch sizing remains active. A single row can exceed the target because ETL checks its decoded size afterward. MemoryStore is for this tutorial only: it does not survive process restart. Production pipelines should use PostgresStore or another durable store.

Run the pipeline

Terminal
PGPASSWORD=your_password RUST_LOG=info cargo run

You should see ETL startup logs plus messages from LoggingDestination during the initial sync. The running pipeline then prints each ongoing replication batch size.

Test ongoing replication

In another terminal, make changes to the database:

psql
\c etl_tutorial

INSERT INTO users (name, email) VALUES ('Charlie Brown', 'charlie@example.com');
UPDATE users SET name = 'Alice Cooper' WHERE email = 'alice@example.com';
DELETE FROM users WHERE email = 'bob@example.com';

Your pipeline terminal should show new ongoing replication batches.

Cleanup

Stop the pipeline with Ctrl+C, then clean up the database:

psql
-- Connect to a different database first (e.g., postgres)
\c postgres
DROP DATABASE etl_tutorial;

Next Steps

Standalone Replicator

Build, configure, and run Supabase ETL as a standalone process.

On this page

Prerequisites1Create the project2Set up Postgres3Write the pipeline4Run the pipeline5Test ongoing replicationCleanupNext Steps