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	`Option<TableCopyBatchId>` and `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 arrive as ordered Event::Relation values inside write_events() batches. If your destination keeps physical schemas, flush writes that still use the previous schema, apply the supported operations in the supplied order, then continue with following rows. See Schema Changes for publication-mask additions, defaults, recovery, and current 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 = "2024"

[[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::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 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 is at least once, so make row and event writes idempotent. Complete every async result handle: send DestinationWriteStatus::Durable from this immediate example after a successful write. ETL clears copy-scoped store state only after drop_table_for_copy() succeeds, so use the supplied schema to find the object to remove. See Extension Points for deferred Accepted writes, copy batch IDs, empty durability barriers, and startup/shutdown hooks.

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,
    TableCopyBatchId, 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,
        batch_id: Option<TableCopyBatchId>,
        rows: Vec<TableRow>,
        async_result: WriteTableRowsResult,
    ) -> EtlResult<()> {
        if rows.is_empty() {
            async_result.send(Ok(DestinationWriteStatus::Durable));
            return Ok(());
        }
        let batch_id = batch_id.ok_or_else(|| {
            etl_error!(ErrorKind::InvalidState, "Table copy row batch is missing its ID")
        })?;
        let table_name = replicated_table_schema.name().to_string();
        info!("writing {} rows to table {}", rows.len(), table_name);

        let payload = json!({
            "batch_id": batch_id.to_string(),
            "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,
        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: 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. The 20% batch-memory ratio is a global advisory target divided across positions that can simultaneously hold an accumulating or in-flight decoded batch. The 32 MiB value is the preferred ceiling for one batch, not a reservation. Emergency backpressure independently pauses new source polling at 85% usage and resumes below 75%, while already-owned destination work continues to drain. A single decoded row can exceed the target because rows are indivisible.

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