Custom Implementations

Implement a store and an HTTP destination. Start with First Pipeline or be familiar with ETL basics. Your implementation can live entirely in your own project; it does not need to be accepted as an official built-in destination. See Destinations for the distinction and upstream maintenance expectations.

Understanding the Destination Trait

ETL delivers data to destinations in two phases:

Phase	Method	When	Data Type
Initial sync	`write_table_rows()`	Startup	`Vec<TableRow>`
Ongoing replication	`write_events()`	During catch-up and ongoing replication	`Vec<Event>` including `Relation` schema events

Note

During initial sync, parallel table sync workers each process their own replication slot, so Begin and Commit transaction markers may appear multiple times. These repeated markers do not by themselves duplicate row events, but ETL is at-least-once, so destination writes should still be idempotent across retries and restarts.

Schema changes are surfaced through Event::Relation. If your destination keeps physical schemas, flush pending writes before handling a relation event, apply the supported schema diff, then process following row events with the new schema. The built-in destinations handle column adds, drops, renames, nullability changes, and supported default changes, but default expressions and backfill semantics remain destination-specific. See Schema Changes for the current semantics and limitations.

Create the project

Terminal

cargo new etl-custom --lib
cd etl-custom
rustup override set 1.95.0

Update Cargo.toml:

Cargo.toml

[package]
name = "etl-custom"
version = "0.1.0"
edition = "2021"

[[bin]]
name = "main"
path = "src/main.rs"

[dependencies]
etl = { git = "https://github.com/supabase/etl" }
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.12", features = ["json"] }
serde_json = "1.0"
tracing = "0.1"
tracing-subscriber = "0.3"

Check: cargo check succeeds.

Implement a custom store

Create src/custom_store.rs. A store must implement three traits (see Extension Points for full details):

SchemaStore - Versioned table schema storage, retrieval, and pruning
StateStore - Table state, persisted replication checkpoints, and destination table metadata tracking
TableStateLifecycleStore - Store lifecycle operations for table-copy preparation, resync resets, and publication changes

SharedStateStore, DestinationStore, and PipelineStore are blanket-implemented facades over these traits plus the required clone/thread-safety bounds, so custom stores do not implement those directly.

Instructional store

This example keeps state in memory so the trait contract is visible in one file, including rollback during an in-process retry. It does not survive a process restart. A production store must durably and atomically persist state history, checkpoints, schemas, and destination metadata before acknowledging each write.

src/custom_store.rs

use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::info;

use etl::{
    destination::{AppliedDestinationTableMetadata, DestinationTableMetadata},
    error::{ErrorKind, EtlResult},
    schema::{PgLsn, SnapshotId, TableId, TableSchema},
    store::{
        SchemaStore, StateStore, TableState, TableStateLifecycleStore, TableStateOperation,
        TableStates, WorkerType,
    },
};
use etl::etl_error;

#[derive(Debug, Clone, Default)]
struct TableEntry {
    schemas: HashMap<SnapshotId, Arc<TableSchema>>,
    state: Option<TableState>,
    state_history: Vec<TableState>,
    destination_metadata: Option<DestinationTableMetadata>,
}

#[derive(Debug, Clone)]
pub struct CustomStore {
    tables: Arc<Mutex<HashMap<TableId, TableEntry>>>,
    checkpoints: Arc<Mutex<HashMap<WorkerType, PgLsn>>>,
}

impl CustomStore {
    pub fn new() -> Self {
        info!("creating custom store");
        Self {
            tables: Arc::new(Mutex::new(HashMap::new())),
            checkpoints: Arc::new(Mutex::new(HashMap::new())),
        }
    }
}

// Compare SnapshotId values directly. Their display strings use variable-width
// decimal components and therefore do not have the same lexical ordering.
impl SchemaStore for CustomStore {
    async fn get_table_schema(
        &self,
        table_id: &TableId,
        snapshot_id: SnapshotId,
    ) -> EtlResult<Option<Arc<TableSchema>>> {
        let tables = self.tables.lock().await;
        Ok(tables.get(table_id).and_then(|entry| {
            entry
                .schemas
                .iter()
                .filter(|(sid, _)| **sid <= snapshot_id)
                .max_by_key(|(sid, _)| *sid)
                .map(|(_, schema)| Arc::clone(schema))
        }))
    }

    async fn get_table_schemas(&self) -> EtlResult<Vec<Arc<TableSchema>>> {
        let tables = self.tables.lock().await;
        Ok(tables
            .values()
            .flat_map(|entry| entry.schemas.values().cloned())
            .collect())
    }

    async fn load_table_schemas(&self) -> EtlResult<usize> {
        Ok(0)
    }

    async fn store_table_schema(&self, schema: TableSchema) -> EtlResult<Arc<TableSchema>> {
        let mut tables = self.tables.lock().await;
        let id = schema.id;
        let snapshot_id = schema.snapshot_id;
        let schema = Arc::new(schema);
        tables
            .entry(id)
            .or_default()
            .schemas
            .insert(snapshot_id, Arc::clone(&schema));
        Ok(schema)
    }

    async fn prune_table_schemas(
        &self,
        retention_snapshot_ids: BTreeMap<TableId, SnapshotId>,
    ) -> EtlResult<u64> {
        let mut tables = self.tables.lock().await;
        let mut removed_count = 0u64;

        for (table_id, entry) in tables.iter_mut() {
            let Some(retention_snapshot_id) = retention_snapshot_ids.get(table_id) else {
                continue;
            };

            let retained_snapshot_id = entry
                .schemas
                .keys()
                .filter(|snapshot_id| **snapshot_id <= *retention_snapshot_id)
                .max()
                .copied();
            let Some(retained_snapshot_id) = retained_snapshot_id else {
                continue;
            };

            let before_count = entry.schemas.len();
            entry.schemas.retain(|snapshot_id, _| *snapshot_id >= retained_snapshot_id);
            removed_count =
                removed_count.saturating_add(before_count.saturating_sub(entry.schemas.len()) as u64);
        }

        Ok(removed_count)
    }
}

impl StateStore for CustomStore {
    async fn get_table_state(&self, table_id: TableId) -> EtlResult<Option<TableState>> {
        let tables = self.tables.lock().await;
        Ok(tables.get(&table_id).and_then(|e| e.state.clone()))
    }

    async fn get_table_states(&self) -> EtlResult<TableStates> {
        let tables = self.tables.lock().await;
        Ok(Arc::new(
            tables
                .iter()
                .filter_map(|(id, e)| e.state.clone().map(|s| (*id, s)))
                .collect::<BTreeMap<_, _>>(),
        ))
    }

    async fn load_table_states(&self) -> EtlResult<usize> {
        Ok(0)
    }

    async fn update_table_states(&self, updates: Vec<(TableId, TableState)>) -> EtlResult<()> {
        let mut tables = self.tables.lock().await;
        for (table_id, state) in updates {
            info!("table {} -> {:?}", table_id.0, state);
            let entry = tables.entry(table_id).or_default();
            if let Some(current_state) = entry.state.replace(state) {
                entry.state_history.push(current_state);
            }
        }
        Ok(())
    }

    async fn rollback_table_state(&self, table_id: TableId) -> EtlResult<TableState> {
        let mut tables = self.tables.lock().await;
        let entry = tables.get_mut(&table_id).ok_or_else(|| {
            etl_error!(ErrorKind::StateRollbackError, "No table state available to roll back")
        })?;
        let previous_state = entry.state_history.pop().ok_or_else(|| {
            etl_error!(
                ErrorKind::StateRollbackError,
                "No previous state available to roll back to"
            )
        })?;
        entry.state = Some(previous_state.clone());
        Ok(previous_state)
    }

    async fn get_replication_checkpoint(
        &self,
        worker_type: WorkerType,
    ) -> EtlResult<Option<PgLsn>> {
        let checkpoints = self.checkpoints.lock().await;
        Ok(checkpoints.get(&worker_type).copied())
    }

    async fn upsert_replication_checkpoint(
        &self,
        worker_type: WorkerType,
        checkpoint_lsn: PgLsn,
    ) -> EtlResult<PgLsn> {
        let mut checkpoints = self.checkpoints.lock().await;
        let stored_lsn = checkpoints.entry(worker_type).or_insert(checkpoint_lsn);
        *stored_lsn = (*stored_lsn).max(checkpoint_lsn);
        Ok(*stored_lsn)
    }

    async fn delete_replication_checkpoint(&self, worker_type: WorkerType) -> EtlResult<()> {
        let mut checkpoints = self.checkpoints.lock().await;
        checkpoints.remove(&worker_type);
        Ok(())
    }

    async fn get_destination_table_metadata(
        &self,
        table_id: TableId,
    ) -> EtlResult<Option<DestinationTableMetadata>> {
        let tables = self.tables.lock().await;
        Ok(tables
            .get(&table_id)
            .and_then(|e| e.destination_metadata.clone()))
    }

    async fn get_applied_destination_table_metadata(
        &self,
        table_id: TableId,
    ) -> EtlResult<Option<AppliedDestinationTableMetadata>> {
        self.get_destination_table_metadata(table_id)
            .await?
            .map(|metadata| metadata.into_applied())
            .transpose()
    }

    async fn load_destination_tables_metadata(&self) -> EtlResult<usize> {
        Ok(0)
    }

    async fn store_destination_table_metadata(
        &self,
        table_id: TableId,
        metadata: DestinationTableMetadata,
    ) -> EtlResult<()> {
        let mut tables = self.tables.lock().await;
        tables.entry(table_id).or_default().destination_metadata = Some(metadata);
        Ok(())
    }
}

impl TableStateLifecycleStore for CustomStore {
    async fn apply_table_state_operation(
        &self,
        operation: TableStateOperation,
    ) -> EtlResult<usize> {
        match operation {
            TableStateOperation::PrepareForCopy { table_id } => {
                let mut tables = self.tables.lock().await;
                if let Some(entry) = tables.get_mut(&table_id) {
                    entry.schemas.clear();
                    entry.destination_metadata = None;
                }
                let mut checkpoints = self.checkpoints.lock().await;
                checkpoints.remove(&WorkerType::TableSync { table_id });
                Ok(0)
            }
            TableStateOperation::ResetForResync => {
                let mut tables = self.tables.lock().await;
                let reset_count = tables.len();
                for entry in tables.values_mut() {
                    if let Some(current_state) = entry.state.replace(TableState::Init) {
                        entry.state_history.push(current_state);
                    }
                }
                let mut checkpoints = self.checkpoints.lock().await;
                checkpoints.remove(&WorkerType::Apply);
                Ok(reset_count)
            }
            TableStateOperation::Delete { table_id } => {
                let mut tables = self.tables.lock().await;
                let removed = usize::from(tables.remove(&table_id).is_some());
                let mut checkpoints = self.checkpoints.lock().await;
                checkpoints.remove(&WorkerType::TableSync { table_id });
                Ok(removed)
            }
        }
    }
}

Check: cargo check succeeds.

Implement a custom destination

Create src/http_destination.rs. A destination implements the Destination trait with four required methods:

name() - Return an identifier for logging
drop_table_for_copy() - Idempotently drop destination objects and replay state before restarting a table copy using the previously stored replicated table schema
write_table_rows() - Receive rows during initial sync together with the current replicated table schema
write_events() - Receive change events (batches may span multiple tables)

There are also optional startup() and shutdown() methods with default no-op implementations. Override startup() if your destination needs to reconcile active durable ETL metadata with physical destination objects after a restart. Override shutdown() if your destination needs cleanup when the pipeline shuts down.

ETL clears its own schema versions, destination metadata, and table-sync progress only after drop_table_for_copy() succeeds. That lets the destination use the supplied previously stored replicated schema and any existing destination metadata to find the object that must be removed. Before returning success, also drain or invalidate writes accepted by an earlier copy attempt so stale work cannot mutate the recreated object. If the object is already gone and no stale write can recreate or mutate it, return success.

All write-like methods must complete their async result handle. Treat the method return value as the place for immediate dispatch or setup failures, and send the final write result through async_result. Immediate destinations should send DestinationWriteStatus::Durable after successful write_table_rows() and write_events() calls. A copy destination that returns DestinationWriteStatus::Accepted must take ownership of the rows, bound its accepted-but-not-durable backlog, and delay Accepted until it has capacity. It must later treat an empty write_table_rows() call as a same-table durability barrier: return Durable only when every row accepted during that copy attempt is durable. For ongoing-replication writes, WriteEventsDurability::MayDefer permits either Accepted or Durable. ETL may issue an empty RequireDurable write as a durability-only barrier for earlier accepted work, but never sends an empty MayDefer write. The empty vector carries no new replication events, but the call may flush or wait for earlier accepted work. It cannot return Accepted and must not return Durable until every earlier Accepted write in the originating ordered apply-loop stream is durable; a stronger barrier scope is valid. ETL is at least once, so make row and event writes idempotent. write_events() preserves per-table ordering, but batches can include multiple tables and transaction markers are not a complete all-tables boundary during initial sync and catch-up.

src/http_destination.rs

use reqwest::Client;
use serde_json::json;
use std::time::Duration;
use tracing::{info, warn};

use etl::destination::{
    Destination, DestinationWriteStatus, DropTableForCopyResult, WriteEventsDurability,
    WriteEventsResult, WriteTableRowsResult,
};
use etl::error::{ErrorKind, EtlResult};
use etl::{data::TableRow, event::Event, schema::ReplicatedTableSchema};
use etl::{bail, etl_error};

#[derive(Debug, Clone)]
pub struct HttpDestination {
    client: Client,
    base_url: String,
}

impl HttpDestination {
    pub fn new(base_url: String) -> EtlResult<Self> {
        let client = Client::builder()
            .timeout(Duration::from_secs(30))
            .build()
            .map_err(|e| etl_error!(ErrorKind::Unknown, "HTTP client error", source: e))?;
        Ok(Self { client, base_url })
    }

    async fn post(&self, path: &str, body: serde_json::Value) -> EtlResult<()> {
        let url = format!("{}/{}", self.base_url.trim_end_matches('/'), path);

        for attempt in 1..=3 {
            match self.client.post(&url).json(&body).send().await {
                Ok(resp) if resp.status().is_success() => return Ok(()),
                Ok(resp) if resp.status().is_client_error() => {
                    bail!(ErrorKind::Unknown, "Client error", resp.status());
                }
                Ok(resp) => warn!("attempt {}/3: status {}", attempt, resp.status()),
                Err(e) => warn!("attempt {}/3: {}", attempt, e),
            }
            if attempt < 3 {
                tokio::time::sleep(Duration::from_millis(500 * attempt as u64)).await;
            }
        }
        bail!(ErrorKind::Unknown, "Request failed after retries");
    }
}

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

    async fn drop_table_for_copy(
        &self,
        replicated_table_schema: &ReplicatedTableSchema,
        async_result: DropTableForCopyResult<()>,
    ) -> EtlResult<()> {
        let table_name = replicated_table_schema.name().to_string();
        info!("dropping table before copy {}", table_name);
        let result = self
            .post(&format!("tables/{table_name}/drop-for-copy"), json!({}))
            .await;
        async_result.send(result);
        Ok(())
    }

    async fn write_table_rows(
        &self,
        replicated_table_schema: &ReplicatedTableSchema,
        rows: Vec<TableRow>,
        async_result: WriteTableRowsResult,
    ) -> EtlResult<()> {
        if rows.is_empty() {
            async_result.send(Ok(DestinationWriteStatus::Durable));
            return Ok(());
        }
        let table_name = replicated_table_schema.name().to_string();
        info!("writing {} rows to table {}", rows.len(), table_name);

        let payload = json!({
            "table_name": table_name,
            "rows": rows.iter().map(|r| {
                json!({ "values": r.values().iter().map(|v| format!("{:?}", v)).collect::<Vec<_>>() })
            }).collect::<Vec<_>>()
        });

        let result = self
            .post("rows", payload)
            .await
            .map(|_| DestinationWriteStatus::Durable);
        async_result.send(result);
        Ok(())
    }

    async fn write_events(
        &self,
        events: Vec<Event>,
        _durability: WriteEventsDurability,
        async_result: WriteEventsResult,
    ) -> EtlResult<()> {
        if events.is_empty() {
            // This immediate destination never returns Accepted, so it has no
            // earlier ongoing-replication durability debt to settle.
            async_result.send(Ok(DestinationWriteStatus::Durable));
            return Ok(());
        }
        info!("writing {} events", events.len());

        let payload = json!({
            "events": events.iter().map(|e| {
                match e {
                    Event::Insert(i) => json!({"type": "insert", "table": i.replicated_table_schema.name().to_string()}),
                    Event::Update(u) => json!({"type": "update", "table": u.replicated_table_schema.name().to_string()}),
                    Event::Delete(d) => json!({"type": "delete", "table": d.replicated_table_schema.name().to_string()}),
                    Event::Begin(_) => json!({"type": "begin"}),
                    Event::Commit(_) => json!({"type": "commit"}),
                    Event::Relation(r) => json!({"type": "relation", "table": r.replicated_table_schema.name().to_string()}),
                    Event::Truncate(t) => json!({"type": "truncate", "tables": t.truncated_tables.iter().map(|table| table.name().to_string()).collect::<Vec<_>>() }),
                    Event::Unsupported => json!({"type": "unsupported"}),
                }
            }).collect::<Vec<_>>()
        });

        let result = self.post("events", payload).await;
        async_result.send(result.map(|_| DestinationWriteStatus::Durable));
        Ok(())
    }
}

Check: cargo check succeeds.

Wire it together

Create src/main.rs:

The custom store owns ETL runtime state. Pipeline::start() still prepares the source database with ETL's schema snapshot helpers and schema-change event trigger before replication begins.

src/main.rs

mod custom_store;
mod http_destination;

use custom_store::CustomStore;
use etl::config::{
    BatchConfig, InvalidatedSlotBehavior, MemoryBackpressureConfig, PgConnectionConfig,
    PipelineConfig, TableSyncCopyConfig, TcpKeepaliveConfig, TlsConfig,
};
use etl::pipeline::Pipeline;
use http_destination::HttpDestination;
use std::error::Error;

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

    let pg_config = PgConnectionConfig {
        host: "localhost".to_string(),
        hostaddr: None,
        port: 5432,
        name: "your_database".to_string(),
        username: "postgres".to_string(),
        password: Some("your_password".to_string().into()),
        tls: TlsConfig {
            enabled: false,
            trusted_root_certs: String::new(),
        },
        keepalive: TcpKeepaliveConfig::default(),
    };

    let store = CustomStore::new();
    let destination = HttpDestination::new("https://your-endpoint.example.com".to_string())?;

    let config = PipelineConfig {
        id: 1,
        publication_name: "my_publication".to_string(),
        pg_connection: pg_config,
        store_pg_connection: None,
        run_source_migrations: true,
        batch: BatchConfig {
            max_fill_ms: 5000,
            memory_budget_ratio: 0.2,
            max_bytes: 8 * 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,
        replication_lag_refresh_interval_ms: 10_000,
        memory_backpressure: Some(MemoryBackpressureConfig::default()),
        table_sync_copy: TableSyncCopyConfig::default(),
        invalidated_slot_behavior: InvalidatedSlotBehavior::default(),
    };

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

    Ok(())
}

Note

Update the database name, password, and HTTP endpoint to match your setup.

Test the pipeline

Terminal

cargo run

The pipeline will connect to Postgres and start replicating. You'll see your custom store logging state transitions and your destination receiving HTTP calls.

Next Steps

Extension Points - Full trait API documentation
Events - Details on all events your destination receives
Configure Postgres - Production database setup
Architecture - How ETL works internally

Understanding the Destination Trait

Create the project

Implement a custom store

Implement a custom destination

Wire it together

Test the pipeline

Next Steps

On this page