# Supabase ETL Documentation > A high-performance Postgres replication engine written in Rust. Embed it in your Rust application or run it as a standalone binary. - Project status: Supabase ETL is under active development. APIs and setup steps may change before the first stable release. - Canonical documentation: https://supabase.github.io/etl/ - Repository: https://github.com/supabase/etl - Managed product: https://supabase.com/docs/guides/database/replication/pipelines Terminology: the two replication phases are **initial sync** and **ongoing replication**. Supabase ETL initially syncs existing table rows, then replicates subsequent changes and delivers them to the destination. Streaming is a transfer mode, not a replication phase. --- # Home > A high-performance Postgres replication engine written in Rust. Embed it in your Rust application or run it as a standalone binary. - Canonical HTML: https://supabase.github.io/etl/ - Agent-readable Markdown: https://supabase.github.io/etl/index.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/index.mdx A high-performance Postgres replication engine written in Rust. Embed it in your Rust application or run it as a standalone binary. For each published table, it performs an initial sync of existing rows, then replicates changes to a destination. Supabase ETL is under active development. APIs and setup steps may change before the first stable release. For the managed Supabase product, use [Supabase Pipelines](https://supabase.com/docs/guides/database/replication/pipelines). ## Documentation map ### Get started - [First Pipeline](https://supabase.github.io/etl/guides/first-pipeline.md): Learn Supabase ETL by building a working Postgres replication pipeline. - [Standalone Replicator](https://supabase.github.io/etl/guides/standalone-replicator.md): Build, configure, and run Supabase ETL as a standalone process. ### Guides - [Configure Postgres](https://supabase.github.io/etl/guides/configure-postgres.md): Set up Postgres with the permissions and settings required by Supabase ETL. - [Custom Implementations](https://supabase.github.io/etl/guides/custom-implementations.md): Implement your own stores and destinations. ### Concepts - [Logical Replication](https://supabase.github.io/etl/explanation/concepts.md): Essential Postgres logical replication concepts for working with Supabase ETL. - [Architecture](https://supabase.github.io/etl/explanation/architecture.md): How Supabase ETL initially syncs tables and replicates changes to destinations. - [Schema Changes](https://supabase.github.io/etl/explanation/schema-changes.md): How Supabase ETL handles DDL and evolving table schemas. ### Reference - [Destinations](https://supabase.github.io/etl/reference/destinations.md): Official built-in destinations, maturity, requirements, and limitations. - [Events](https://supabase.github.io/etl/explanation/events.md): Understand the events Supabase ETL delivers during ongoing replication. - [Extension Points](https://supabase.github.io/etl/explanation/traits.md): Traits you implement to customize Supabase ETL behavior. ## Replication phases 1. **Initial sync:** Copy the existing rows selected by the publication. 2. **Ongoing replication:** Capture subsequent inserts, updates, deletes, and truncates, then deliver those changes as ordered events. Streaming describes a transfer mode that may be used within either phase; it is not a separate replication phase. Across both phases, ETL persists checkpoints and table state so replication can recover safely after a restart. --- # Architecture > How Supabase ETL initially syncs tables and replicates changes to destinations. - Canonical HTML: https://supabase.github.io/etl/explanation/architecture/ - Agent-readable Markdown: https://supabase.github.io/etl/explanation/architecture.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/explanation/architecture.mdx Supabase ETL uses **Postgres logical replication** to initially sync published tables, then replicate subsequent changes to a destination in near real time. ## Overview [#overview] ## How It Works [#how-it-works] ETL operates in **two phases**: ### Phase 1: Initial Sync [#phase-1-initial-sync] When a pipeline starts, it copies all existing data from each table in the publication. Multiple **Table Sync Workers** run in parallel to copy tables concurrently. Each worker: 1. Creates a replication slot to capture a consistent snapshot 2. Copies all rows using Postgres `COPY` 3. Sends rows to the destination via `write_table_rows()` > **Note** > > During this phase, `Begin` and `Commit` events may be delivered multiple times > because workers consume slots in parallel. This is expected: destinations > should rely on **per-table event ordering** and **idempotent writes** rather > than treating transaction markers as a complete global transaction boundary. ### Phase 2: Ongoing Replication [#phase-2-ongoing-replication] Once tables are copied, the **Apply Worker** handles ongoing replication from the Postgres WAL. It: 1. Receives change events (inserts, updates, deletes) 2. Batches events for efficiency 3. Sends batches to the destination via `write_events()` ### Schema Changes [#schema-changes] ETL supports simple column evolution from `ALTER TABLE` and supported `ALTER PUBLICATION` changes. A source-side event trigger emits internal schema messages for published permanent tables, ETL stores a new schema snapshot, and destinations observe the change through a fresh `Relation` event before following row events. See [Schema Changes](https://supabase.github.io/etl/explanation/schema-changes.md) for the supported operations and limitations. ## Core Components [#core-components] ### Pipeline [#pipeline] The **central orchestrator** that manages the entire replication process. It spawns workers, coordinates state transitions, and handles shutdown. ### Destination [#destination] The `Destination` trait receives bulk rows during initial sync and event batches during catch-up and ongoing replication. Result handles let a destination distinguish accepted work from durable work without blocking dispatch. See [Extension Points](https://supabase.github.io/etl/explanation/traits.md#destination) for the method contract and [Custom Stores and Destinations](https://supabase.github.io/etl/guides/custom-implementations.md) for an implementation. ### Store [#store] Three store traits persist the state needed to resume after restarts: * **StateStore**: Tracks table state, persisted replication checkpoints, and destination table metadata * **SchemaStore**: Stores versioned table schema information (columns, types, primary keys, snapshot IDs) and prunes obsolete schema versions behind persisted checkpoints while preserving the retained boundary schema and newer versions * **TableStateLifecycleStore**: Prepares table-copy state, resets table states for resync, and deletes all ETL-owned state when a table leaves the publication See [Extension Points](https://supabase.github.io/etl/explanation/traits.md#schemastore) for the cache and durability contracts. ## Delivery Guarantees [#delivery-guarantees] ETL provides **at-least-once delivery**. If restarts occur, some events may be delivered more than once. This is a deliberate design choice. ### Why Not Exactly-Once? [#why-not-exactly-once] Exactly-once delivery requires distributed transactions between Postgres and the destination, adding complexity and latency. Instead, ETL optimizes for throughput and simplicity while minimizing duplicates through: * **Controlled shutdown**: The pipeline attempts to finish in-flight work before its shutdown deadline; interrupted work can be replayed after restart * **Frequent status updates**: Progress is reported to Postgres regularly, reducing the replay window after restarts ### Handling Duplicates [#handling-duplicates] Destinations should make writes **idempotent** using the source table's replica identity or primary key plus ETL's event ordering metadata. For append-style CDC tables, persist a sequence key derived from `commit_lsn` and `tx_ordinal`. For current-state tables, upsert by the destination's chosen row key so replayed events converge to the same state. The `commit_lsn` and `tx_ordinal` fields on sequenced events provide stable **ordering and checkpointing**. For example, BigQuery destinations use the pair to maintain correct event order in destination tables. See [Event Types](https://supabase.github.io/etl/explanation/events.md#understanding-event-sequence-keys) for details. ## Table States [#table-states] Each table progresses through these states: | State | Set By | Description | | ---------------- | ----------------- | -------------------------------------------------------------------------------- | | **Init** | Pipeline | Table discovered, ready for initial sync | | **DataSync** | Table Sync Worker | Initial table copy in progress | | **FinishedCopy** | Table Sync Worker | Initial sync complete | | **SyncWait** | Table Sync Worker | Waiting for Apply Worker to pause (in-memory only) | | **Catchup** | Apply Worker | Apply Worker paused; Table Sync Worker catching up to its LSN (in-memory only) | | **SyncDone** | Table Sync Worker | Catch-up complete; durable decoder retained until Apply materializes local state | | **Ready** | Apply Worker | Apply Worker now handles this table exclusively | | **Errored** | Either | Error occurred; contains reason, solution hint, and retry policy | --- # Logical Replication > Essential Postgres logical replication concepts for working with Supabase ETL. - Canonical HTML: https://supabase.github.io/etl/explanation/concepts/ - Agent-readable Markdown: https://supabase.github.io/etl/explanation/concepts.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/explanation/concepts.md Read this first if Postgres logical replication is new to you. ## What is Logical Replication? [#what-is-logical-replication] Postgres supports two types of replication: | Type | What it copies | Use case | | ------------ | -------------------------------------------- | -------------------------------- | | **Physical** | Exact byte-for-byte copy of data files | Disaster recovery, read replicas | | **Logical** | Decoded row changes (INSERT, UPDATE, DELETE) | Data integration, ETL, CDC | ETL uses logical replication because decoded row changes can be sent to systems other than Postgres. ## The Write-Ahead Log (WAL) [#the-write-ahead-log-wal] Before Postgres modifies data on disk, it first writes the change to the **Write-Ahead Log (WAL)**. This guarantees durability: if Postgres crashes, it can replay the WAL to recover. For logical replication, Postgres decodes the WAL back into **logical changes**: ETL receives these decoded events and forwards them to downstream consumers. ### WAL Level [#wal-level] Postgres must be configured to record enough information for logical decoding: ```ini # In postgresql.conf wal_level = logical ``` With `wal_level = logical`, Postgres records additional metadata needed to reconstruct row changes. Lower levels (`replica`, `minimal`) **do not capture enough detail**. ## Publications [#publications] A **publication** defines which tables to replicate. Think of it as a filter that says "replicate changes from these tables." ```sql -- Replicate specific tables CREATE PUBLICATION my_publication FOR TABLE users, orders; -- Replicate all tables (use with caution) CREATE PUBLICATION my_publication FOR ALL TABLES; ``` When you create an ETL pipeline, you specify which publication to consume. **Only tables and operations selected by that publication are replicated.** ### What Publications Control [#what-publications-control] * **Which tables**: Only tables in the publication are replicated * **Which operations**: You can filter to only INSERT, UPDATE, or DELETE * **Which columns** (Postgres 15+): Replicate only specific columns * **Which rows** (Postgres 15+): Filter rows with a WHERE clause ## Replication Slots [#replication-slots] A **replication slot** is a bookmark that tracks how far a consumer has read in the WAL. ### Why Slots Exist [#why-slots-exist] Without slots, Postgres would delete old WAL files when it no longer needs them for crash recovery. If ETL disconnects temporarily, it needs those WAL files to catch up when it reconnects. Replication slots tell Postgres: **"Don't delete WAL files until this consumer has processed them."** ```sql -- View existing slots SELECT slot_name, confirmed_flush_lsn, active FROM pg_replication_slots; ``` ### How ETL Uses Slots [#how-etl-uses-slots] ETL creates replication slots automatically: | Slot | Purpose | | -------------------------------------------------- | --------------------------------- | | `supabase_etl_apply_{pipeline_id}` | Main slot for ongoing replication | | `supabase_etl_table_sync_{pipeline_id}_{table_id}` | Temporary slots for initial sync | The Apply Worker uses one persistent slot. Table Sync Workers create temporary slots during initial sync, then delete them. ### Slot Risks [#slot-risks] Slots prevent WAL cleanup. If ETL stops consuming because of crashes, network issues, or a slow consumer, WAL files accumulate on disk. **This can fill your disk.** To mitigate this risk: * Monitor slot lag with `pg_replication_slots` * Set `max_slot_wal_keep_size` to limit WAL retention * Alert when slots fall behind See [Configure Postgres](https://supabase.github.io/etl/guides/configure-postgres.md#wal-buildup-and-disk-usage) for details. ## The pgoutput Decoder [#the-pgoutput-decoder] When Postgres decodes WAL for logical replication, it uses a **decoder plugin**. ETL uses `pgoutput`, Postgres's built-in decoder. The decoder transforms binary WAL records into structured messages: | Message | Meaning | | ---------- | ----------------------------- | | `BEGIN` | Transaction started | | `RELATION` | Table schema (columns, types) | | `INSERT` | Row added | | `UPDATE` | Row modified | | `DELETE` | Row removed | | `TRUNCATE` | Table cleared | | `COMMIT` | Transaction completed | ETL receives these messages and converts them to events. ## Why Two Phases? [#why-two-phases] ETL replicates data in two phases: **initial sync** and **ongoing replication**. ### Phase 1: Initial Sync [#phase-1-initial-sync] Logical replication only captures **changes**. It does not know about data that existed before replication started. So ETL first copies all existing rows using Postgres's `COPY` command: 1. Create replication slot (captures consistent snapshot point) 2. COPY all rows from the table 3. Begin ongoing replication from the snapshot point The slot ensures **no changes are lost** between the snapshot and ongoing replication. ### Phase 2: Ongoing Replication [#phase-2-ongoing-replication] After initial sync, ETL begins ongoing replication. It captures subsequent changes from the WAL and delivers them to the destination: Each change is delivered as an `Event` through `write_events()`. Large tables can spend significant time in initial sync. ETL exposes separate `write_table_rows()` and `write_events()` methods so destinations can optimize initial-copy rows and change events independently. ## Replica Identity [#replica-identity] **REPLICA IDENTITY** controls what data Postgres includes in UPDATE and DELETE events. ### The Problem [#the-problem] When a row is updated or deleted, downstream systems need enough old-row information to identify which source row changed. The important nuance is that PostgreSQL does **not** always send an old-side tuple for `UPDATE`. Under key-based replica identity, it only sends a key image when it is needed. For `DELETE`, PostgreSQL sends an old-side tuple whenever the delete is publishable. This means replica identity is both a **PostgreSQL logging rule** and a **consumer contract** for downstream consumers. It determines whether an event contains enough old-row data to match an existing row, detect key changes, or compare before-and-after values. ### Settings [#settings] ```sql -- See current setting (d=default, f=full, n=nothing, i=index) SELECT relname, relreplident FROM pg_class WHERE relname = 'your_table'; -- Change setting ALTER TABLE your_table REPLICA IDENTITY FULL; ``` | Setting | Published `UPDATE` payload | Published `DELETE` payload | Notes | | ------------------------------- | --------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------ | ----------------------------------------------------------------- | | `DEFAULT` with a primary key | Old primary-key columns only when PostgreSQL determines the old key must be logged; otherwise no old tuple | Old primary-key columns | Most tables with a primary key | | `DEFAULT` without a primary key | Source `UPDATE` is rejected when the table publishes updates | Source `DELETE` is rejected when the table publishes deletes | Equivalent to having no usable replica identity for update/delete | | `FULL` | Full old row | Full old row | Use when consumers need full old-row images | | `NOTHING` | Source `UPDATE` is rejected when the table publishes updates | Source `DELETE` is rejected when the table publishes deletes | Suitable only when updates/deletes are not published | | `USING INDEX` | Old replica-identity index columns only when PostgreSQL determines the old key must be logged; otherwise no old tuple | Old replica-identity index columns | Tables whose replication identity differs from the primary key | ### Impact on ETL [#impact-on-etl] ETL preserves PostgreSQL's old-row semantics in update and delete events: ```rust pub old_table_row: Option ``` * `Some(OldTableRow::Key(row))` means PostgreSQL sent only the replica-identity columns, normalized into replicated table-column order. * `Some(OldTableRow::Full(row))` means PostgreSQL sent the full old row. * `None` means PostgreSQL did not send an old-side tuple for that update. This is normal under `DEFAULT` or `USING INDEX` when PostgreSQL determines no old-side image is required. For `FULL`, PostgreSQL sends a full old row for every published update and delete. For `DELETE`, valid pgoutput messages always include either a full old row or a key image. `REPLICA IDENTITY NOTHING`, and `DEFAULT` on a table without a primary key, do not produce update/delete events when those actions are published; the source statement is rejected instead. The Rust event API keeps the old-row fields optional at the boundary, but those `None` cases are broader than the PostgreSQL pgoutput shapes described here. **TOAST adds one more wrinkle.** PostgreSQL can mark unchanged toasted update values as `UnchangedToast` instead of resending the value. ETL can reconstruct those values only if the old-side row image contains them, so tables with toasted columns can produce partial update rows unless they use `REPLICA IDENTITY FULL` or the missing values are present in a logged key image. If you need **old values** for auditing, comparison, complete replacement rows, or reliable reconstruction of unchanged toasted columns, set `REPLICA IDENTITY FULL` on those tables. If a consumer only needs stable key values, `DEFAULT` with a primary key or `USING INDEX` is usually enough, but update events will not always include `old_table_row`. ## LSN (Log Sequence Number) [#lsn-log-sequence-number] Every position in the WAL has a unique **LSN** - a monotonically increasing pointer. ```text Format: 0/16B3748 (segment/offset) ``` ### LSNs in Events [#lsns-in-events] Sequenced ETL events include a commit LSN and transaction-local ordinal: | Field | Meaning | | ------------ | --------------------------------------------- | | `commit_lsn` | LSN of the commit message in the WAL | | `tx_ordinal` | Zero-based event order within the transaction | Multiple events in the same transaction share the same `commit_lsn`; their `tx_ordinal` values distinguish their order. Relation events are connection-local metadata and do not have an event sequence key. ## Persisted State [#persisted-state] ETL persists the state needed to resume safely after a restart: ETL stores: | State | Purpose | | -------------------------------- | ---------------------------------------------------------------------------- | | Table state | Track each table from initial sync through ongoing replication | | Persisted replication checkpoint | Resume workers from a safe replay frontier | | Table schemas | Decode events against the correct versioned schema | | Destination table metadata | Track destination table IDs, applied schema snapshots, and replication masks | The built-in `PostgresStore` persists to your Postgres database and runs its state-store migrations when it is created. If the pipeline reads from a read-only replica, configure `store_pg_connection` to point at a writable Postgres endpoint for this state. `MemoryStore` is for testing only - state is lost on restart. `Pipeline::start()` runs the ETL source migrations that install schema helpers and the DDL event trigger before replication begins. See [Architecture](https://supabase.github.io/etl/explanation/architecture.md) for the worker lifecycle and [Configure Postgres](https://supabase.github.io/etl/guides/configure-postgres.md) for production settings. --- # Events > Understand the events Supabase ETL delivers during ongoing replication. - Canonical HTML: https://supabase.github.io/etl/explanation/events/ - Agent-readable Markdown: https://supabase.github.io/etl/explanation/events.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/explanation/events.mdx During ongoing replication, Supabase ETL delivers events decoded from Postgres logical replication via `write_events()`. This page documents their Rust shapes and PostgreSQL semantics. ## Event Overview [#event-overview] | Event | Description | Has Table ID | | ------------- | ----------------- | ------------ | | `Begin` | Transaction start | No | | `Commit` | Transaction end | No | | `Insert` | New row added | Yes | | `Update` | Row modified | Yes | | `Delete` | Row removed | Yes | | `Truncate` | Table cleared | Yes | | `Relation` | Table schema | Yes | | `Unsupported` | Unknown event | No | ## Data Modification Events [#data-modification-events] These events carry **row data** and are associated with specific tables. ### Row Images [#row-images] Data modification events use **row-image helper types**: ```rust pub enum UpdatedTableRow { Full(TableRow), Partial(PartialTableRow), } pub enum OldTableRow { Full(TableRow), Key(TableRow), } ``` `TableRow` is a **complete dense row**. Its values are ordered to match the replicated table-column order. `PartialTableRow` is used when an update row is **not complete**. It exposes: * `total_columns()`: the number of replicated columns in the table schema; * `values()`: present values in replicated table-column order, excluding missing columns; * `missing_column_indexes()`: zero-based replicated-column indexes for values ETL could not reconstruct. `OldTableRow::Full(row)` contains a **complete old row**. `OldTableRow::Key(row)` contains only replica-identity columns, densely packed in replicated table-column order. ### Insert [#insert] A new row was added to a table. ```rust pub struct InsertEvent { pub commit_lsn: PgLsn, pub tx_ordinal: u64, pub replicated_table_schema: ReplicatedTableSchema, pub table_row: TableRow, } ``` ### Update [#update] An existing row was modified. ```rust pub struct UpdateEvent { pub commit_lsn: PgLsn, pub tx_ordinal: u64, pub replicated_table_schema: ReplicatedTableSchema, pub updated_table_row: UpdatedTableRow, pub old_table_row: Option, } ``` `updated_table_row` is the **authoritative post-update payload**: * `UpdatedTableRow::Full` when ETL knows every replicated column value after decoding the update. * `UpdatedTableRow::Partial` when PostgreSQL emitted `UnchangedToast` fields that ETL could not reconstruct safely. `old_table_row` provides optional context for row matching, key changes, and reconstruction. #### Unchanged Toast [#unchanged-toast] PostgreSQL may encode an unchanged toasted value as `UnchangedToast` in the update's new tuple instead of resending the full column value. ETL can turn the update into `UpdatedTableRow::Full` only when that value can be recovered from the old-side row image: * a `FULL` old row can recover unchanged toasted values for any replicated column; * a key-only old row can recover unchanged toasted values only for replica-identity columns included in that key image; * no old row means unchanged toasted values cannot be recovered from the event. When any `UnchangedToast` field cannot be recovered, ETL emits `UpdatedTableRow::Partial` with known values and missing column indexes instead of pretending the unknown value is `NULL` or a replacement value. Destinations that need complete replacement rows, full before/after comparison, or complete audit records should require `REPLICA IDENTITY FULL` for tables with toasted columns, or keep enough prior state to fill missing values themselves. #### Old Row Mapping [#old-row-mapping] ETL maps PostgreSQL pgoutput update tuple markers directly: | pgoutput marker | ETL field | | ----------------------- | ------------------------------ | | `O` old tuple | `Some(OldTableRow::Full(row))` | | `K` old key | `Some(OldTableRow::Key(row))` | | no old tuple/key marker | `None` | | `N` new tuple | `updated_table_row` | PostgreSQL chooses which old-side marker to emit from the table's replica identity: | REPLICA IDENTITY | `old_table_row` contains for published updates | | ------------------------------- | ----------------------------------------------------------------------------------------------------- | | `FULL` | `Some(OldTableRow::Full(row))` | | `DEFAULT` with a primary key | `Some(OldTableRow::Key(row))` when PostgreSQL determines the old key must be logged, otherwise `None` | | `DEFAULT` without a primary key | Source `UPDATE` is rejected when the table publishes updates | | `USING INDEX` | `Some(OldTableRow::Key(row))` when PostgreSQL determines the old key must be logged, otherwise `None` | | `NOTHING` | Source `UPDATE` is rejected when the table publishes updates | For `UPDATE`, PostgreSQL only sends an old key image under `DEFAULT` or `USING INDEX` when it determines the old key must be logged. In practice, that happens when: * any replica-identity column changed, or * any replica-identity column contains external data, such as a toasted value that must be available from the old tuple. That means non-identity updates under `DEFAULT` or `USING INDEX` often arrive with `old_table_row = None`. Under `FULL`, PostgreSQL sends a full old row for every published update. A `FULL` update with `old_table_row = None` is not a pgoutput shape. When handling updates: 1. Treat `updated_table_row` as the post-update payload and handle `UpdatedTableRow::Partial` explicitly. 2. Treat `OldTableRow::Key` as replica-identity values, not necessarily the primary key. Use `OldTableRow::Full` for complete before-images. 3. If `old_table_row` is `None`, match or upsert from the new row according to the destination's keying model. ### Delete [#delete] A row was removed from a table. ```rust pub struct DeleteEvent { pub commit_lsn: PgLsn, pub tx_ordinal: u64, pub replicated_table_schema: ReplicatedTableSchema, pub old_table_row: Option, } ``` For `DELETE`, valid PostgreSQL publications send an old-side image: | REPLICA IDENTITY | `old_table_row` contains for published deletes | | ------------------------------- | ------------------------------------------------------------ | | `FULL` | `Some(OldTableRow::Full(row))` | | `DEFAULT` with a primary key | `Some(OldTableRow::Key(row))` | | `DEFAULT` without a primary key | Source `DELETE` is rejected when the table publishes deletes | | `USING INDEX` | `Some(OldTableRow::Key(row))` | | `NOTHING` | Source `DELETE` is rejected when the table publishes deletes | **Important implications:** * Deletes do not carry a new row image. `old_table_row` is the delete payload. * `OldTableRow::Key(row)` again means replica-identity columns only, not necessarily the table's primary key. * Consumers that require full old rows for deletes need `REPLICA IDENTITY FULL`. * The Rust field is optional at the event API boundary, but PostgreSQL pgoutput populates it for every published delete. Tables without usable replica identity fail at the source when they publish deletes. * Destination implementations should decide up front whether key-only deletes are enough. If they are not, require `REPLICA IDENTITY FULL` for those tables. ### Truncate [#truncate] One or more tables were truncated (all rows deleted). ```rust pub struct TruncateEvent { pub commit_lsn: PgLsn, pub tx_ordinal: u64, pub options: i8, pub truncated_tables: Vec, } ``` > **Note** > > A single `Truncate` event can affect multiple tables when using > `TRUNCATE ... CASCADE`. ## Transaction Events [#transaction-events] These events mark transaction boundaries. ### Begin [#begin] Marks the start of a transaction. ```rust pub struct BeginEvent { pub commit_lsn: PgLsn, pub tx_ordinal: u64, pub timestamp: i64, pub xid: u32, } ``` ### Commit [#commit] Marks successful transaction completion. ```rust pub struct CommitEvent { pub commit_lsn: PgLsn, pub tx_ordinal: u64, pub flags: i8, pub end_lsn: PgLsn, pub timestamp: i64, } ``` ## Schema Events [#schema-events] ### Relation [#relation] Provides **table schema information**. Sent before data events for a table and again after supported schema changes. ```rust pub struct RelationEvent { pub replicated_table_schema: ReplicatedTableSchema, } ``` Relation messages are connection-local protocol metadata. PostgreSQL can emit the same relation again after a reconnect even when the source schema did not change. A `RelationEvent` therefore has no durable event sequence key and does not consume a transaction ordinal. PostgreSQL pgoutput builds relation messages by walking the table descriptor in `pg_attribute.attnum` order and skipping columns that are not published. It sends tuple data in the same order as the relation message. ETL builds replication and identity masks from relation-message column names, so the order is not needed to decide which columns are included. That name-based matching is sound because PostgreSQL live column names are unique within a table schema version. The order matters only after the masks are applied: stored table schemas are also ordered by `attnum`, so `ReplicatedTableSchema::column_schemas()` becomes a positional view that matches the tuple payloads exactly, even when a publication filters columns. For DDL behavior, including add/drop/rename semantics, default and nullability handling, and current limitations, see [Schema Changes](https://supabase.github.io/etl/explanation/schema-changes.md). ## Begin/Commit Behavior [#begincommit-behavior] During initial sync, parallel Table Sync Workers may deliver `Begin` and `Commit` markers more than once. Ignore them if you do not need transaction markers; otherwise deduplicate them by `commit_lsn` plus `tx_ordinal`. Row writes must still be idempotent because ETL provides at-least-once delivery. ## Understanding Event Sequence Keys [#understanding-event-sequence-keys] Transaction and data events include a commit **LSN (Log Sequence Number)** plus a transaction-local ordinal. Together, these fields provide stable ordering for idempotent destination writes. Relation events are not sequenced because they are connection-local protocol metadata. ### `commit_lsn` and `tx_ordinal` [#commit_lsn-and-tx_ordinal] | Field | Meaning | Use Case | | ------------ | ---------------------------------------------------- | --------------------------------------------- | | `commit_lsn` | Position where the transaction will commit | Transaction grouping, recovery checkpoints | | `tx_ordinal` | Zero-based order of the event within its transaction | Ordering and idempotency within a transaction | Events in one transaction share a `commit_lsn`; `tx_ordinal` distinguishes their order. Use the pair as a sequence key for ordered writes and replay detection. See [Logical Replication](https://supabase.github.io/etl/explanation/concepts.md#lsn-log-sequence-number) for LSN background. ## Event Batching [#event-batching] ETL batches events before calling `write_events()`. A batch may contain events from **multiple tables, multiple transactions, and mixed event types**. Schema changes do not end a batch: because batching is driven by size and time, one batch may contain multiple `Relation` events and multiple schema changes. **Ordering requirement:** Process events for the same destination row in order. Events for independent rows may be processed concurrently. --- # Schema Changes > How Supabase ETL handles DDL and evolving table schemas. - Canonical HTML: https://supabase.github.io/etl/explanation/schema-changes/ - Agent-readable Markdown: https://supabase.github.io/etl/explanation/schema-changes.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/explanation/schema-changes.mdx Schema-change support is in public beta and is being expanded incrementally. The current implementation is intentionally conservative: the source-side event trigger captures a rich PostgreSQL-shaped snapshot, while ETL currently models well-understood column changes: **adds, drops, renames, and column default and nullability changes**, plus publication column-list changes for tables the running pipeline already tracks. A few known edge cases remain. Built-in destination support varies by destination DDL capabilities. **BigQuery, ClickHouse, DuckLake, and Snowflake** apply supported schema changes automatically; Iceberg is deprecated for new deployments and does not support schema-change DDL. ## Short Version [#short-version] For published permanent tables, ETL currently models these `ALTER TABLE` changes: | Source change | ETL interpretation | | ---------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------- | | Add a replicated column | Add column | | Drop a replicated column | Drop column | | Rename a replicated column | Rename column | | Change a replicated column default | Column default modification | | Drop a replicated column default | Column default removal | | Drop `NOT NULL` from a replicated column | BigQuery relaxes an existing `REQUIRED` column to `NULLABLE`; other built-in destinations currently leave nullability unchanged | | Set `NOT NULL` on a replicated column | Detected in the schema snapshot, but not applied to built-in destinations | | Several of the above in one statement | One schema snapshot, diffed into column additions, removals, and grouped column modifications | When several attributes of the same logical column change at once, ETL groups them into one column change with multiple modifications. For example, renaming a column and changing its default in one `ALTER TABLE` statement is treated as a single logical column change. ## How It Works [#how-it-works] ETL installs a PostgreSQL `ddl_command_end` event trigger named `supabase_etl_ddl_message_trigger`. When an `ALTER TABLE` statement or supported `ALTER PUBLICATION` change affects a published permanent table, the trigger emits a transactional logical message with prefix `supabase_etl_ddl`. PostgreSQL does not pass user-defined arguments to an event-trigger function. The function reads `TG_TAG` and the object addresses returned by `pg_event_trigger_ddl_commands()`, then resolves the post-DDL catalogs: * For `ALTER TABLE`, the object address identifies the table. The message has no publication name because a physical table change applies to every publication containing that table. * For per-table `ALTER PUBLICATION` changes, the object address identifies a surviving `pg_publication_rel` row, which links the explicitly named table to its publication. This path does not expand a named partition root into its effective leaves. * For publication-level `ALTER PUBLICATION` changes, the object address identifies a `pg_publication` row. The trigger does not parse which parameter changed; it expands the publication's complete post-command effective table set through `pg_publication_tables`. The publication name is therefore optional in the payload for table DDL but required for publication DDL. Logical decoding exposes custom messages to a slot independently of pgoutput's table filtering, so ETL accepts an `ALTER PUBLICATION` message only when its resolved name exactly matches the pipeline's configured publication. A missing name fails closed. That message is **internal plumbing**. Destinations do not receive it directly. Instead, ETL: 1. Parses the schema-change message. 2. Stores a new versioned table schema using a composite snapshot ID ordered by commit LSN and then message LSN. 3. Invalidates the in-memory relation state for that table. 4. Waits for PostgreSQL pgoutput to emit a fresh `RELATION` message before the next row event for that table. 5. Sends destinations a public `Event::Relation` with the new `ReplicatedTableSchema`. Snapshot IDs display both LSNs as decimal unsigned 64-bit values in `commit_lsn:message_lsn` form. Their ordering is the numeric tuple ordering of those two components, not the lexical ordering of the displayed string. `ALTER PUBLICATION ... DROP TABLE` intentionally emits no schema snapshot for the removed table because it is no longer part of the publication. PostgreSQL 14 may still send an empty `BEGIN`/`COMMIT` pair for that transaction; PostgreSQL 15 and later suppress empty logical-replication transactions. The empty pair contains no schema or row event and does not affect snapshot ordering. See the [PostgreSQL 15 release notes](https://www.postgresql.org/docs/15/release-15.html) and the [upstream change](https://github.com/postgres/postgres/commit/d5a9d86d8ffcadc52ff3729cd00fbd83bc38643c). For a table already known to the pipeline, a supported publication change creates a new snapshot ID even when the full physical table schema is unchanged. The following `RELATION` message supplies the current publication and identity masks. This gives successive mask changes distinct snapshot IDs and preserves their ordering. Multiple schema messages sharing one commit LSN are ordered by their message LSN. A newly published table that the running pipeline does not know is ignored until startup publication reconciliation creates its table state and initial sync. ### `ALTER PUBLICATION` support boundary [#alter-publication-support-boundary] The trigger's ability to observe an `ALTER PUBLICATION` command is not the same as runtime support for that publication change. The trigger emits schema snapshots; it does not add or remove ETL table state, start an initial sync, or change which relation OIDs the running apply worker owns. | Publication change | Trigger behavior | Runtime behavior | | ---------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------- | | Change the column list of an already tracked table | Emits a snapshot scoped to that table and publication | Supported. The next `RELATION` message installs the new replication mask before following row events. | | `ADD TABLE` or a membership-changing `SET TABLE` | Emits snapshots for surviving, explicitly identified `pg_publication_rel` rows. A named partition root is not expanded into leaves on this path. | Newly effective relation OIDs are ignored until startup reconciliation initializes and copies them. | | `DROP TABLE` | Emits no snapshot because the `pg_publication_rel` row is gone at `ddl_command_end`. | Removed table state is purged during startup reconciliation. Destination data is not automatically removed. | | Add, set, or drop `TABLES IN SCHEMA` | The current trigger does not resolve `pg_publication_namespace` events into table snapshots. | Effective membership is loaded during startup reconciliation. | | Change a row filter | May emit a per-table snapshot, but row-filter mutation is not part of the supported live-update contract. | Do not rely on changing row filters while the pipeline is running. | | Change a publication-level setting such as `publish` | A `pg_publication` event expands to one snapshot for every post-command effective table, even if physical schemas did not change. | Publication settings are not a supported live schema-update interface. | | Change `publish_via_partition_root` | Emits snapshots for the complete post-change effective set: the root when enabled, or the leaves when disabled. | Unsupported while running. The new relation OIDs are not dynamically initialized, and the previous destination relation layout is not migrated. | | Rename the publication or change its owner | These are publication-level events and may expand across the effective table set. A rename scopes messages to the new name. | Not supported as live schema changes. A configured pipeline does not automatically adopt a new publication name. | At startup, ETL calls `pg_get_publication_tables()` to load the same effective relation identities PostgreSQL uses for pgoutput. New identities enter initial sync and identities no longer returned by PostgreSQL have their ETL state purged. This startup process does not rename, merge, or delete destination tables created for an earlier root or leaf identity. > **Warning** > > Only publication column-list changes for already tracked tables are supported > while a pipeline is running. Other `ALTER PUBLICATION` changes can produce > skipped events, stale destination tables, or missing or duplicated data. Plan > a controlled change with source writes paused and the pipeline stopped, > followed by an explicit restart, table resynchronization, or new pipeline > instead of relying on live behavior. The important public boundary is: ```text ... -> internal DDL message -> Relation(new schema) -> Insert/Update/Delete ... ``` Destinations should treat `Event::Relation` as the point where the **active schema changes** for following row events. `Relation` is an ordered event, not a batch boundary. ETL batches calls to `write_events()` based on size and time, so a single destination batch may contain zero, one, or many schema changes, including multiple relation events for the same table. ## Destination-Specific DDL Behavior [#destination-specific-ddl-behavior] ETL has one shared schema-change signal, but **DDL behavior is implemented per destination**. A destination may choose to apply DDL automatically, reject a schema change, or require operator handling. For every built-in destination, a relation whose snapshot ID and replication mask exactly match the applied destination metadata is idempotent. BigQuery, ClickHouse, DuckLake, and Snowflake currently reject an older snapshot, or the same snapshot with a different replication mask, instead of attempting to infer schema ordering from later row events. A relation has no DML sequence key of its own, so destination row-replay deduplication does not prove that reverse or ambiguously ordered DDL is safe. Recovering from either rejection currently requires resynchronizing the table. This comparison applies only to metadata already marked `Applied`; it neither defines nor initiates recovery from an interrupted `Applying` state. | Destination | Current DDL behavior | | ------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | BigQuery | Supports add, drop, rename, `REQUIRED` to `NULLABLE` relaxation, and supported literal default metadata. BigQuery requires added columns to be nullable and does not backfill existing rows for `ADD COLUMN ... DEFAULT`. PostgreSQL remains responsible for enforcing later `SET NOT NULL` changes because BigQuery cannot tighten an existing column in place. | | ClickHouse | Supports add, drop, rename, and supported literal defaults. `ReplacingMergeTree` rejects primary-key drops or renames because the ordering expression cannot be rewritten safely. ClickHouse default expressions are metadata-only unless explicitly materialized; ETL does not issue `MATERIALIZE COLUMN`. | | DuckLake | Supports add, drop, rename, and supported literal defaults. DuckLake records supported add-time defaults as metadata without rewriting existing data files. | | Snowflake | Supports add, drop, rename, create-table literal defaults, and literal add-column defaults. Literal defaults are included in `ADD COLUMN` so Snowflake can expose add-time default values for existing rows; non-literal defaults and later default changes are skipped with a warning. | | Iceberg | Deprecated. An identical relation is idempotent, but schema-change DDL is not supported and any newer schema is rejected. Older or ambiguously masked relations are rejected before row writes. | | Custom destinations | Destination authors decide which `Event::Relation` changes to apply, reject, or handle manually. | See [Destinations](https://supabase.github.io/etl/reference/destinations.md) for the canonical maturity status and broader limitations of every built-in implementation. ## Default Backfills [#default-backfills] When a source table adds a replicated column with a default, PostgreSQL can make pre-existing source rows read as though they already contain that default. ETL deliberately avoids **physical destination backfills** for those existing rows. Built-in destinations avoid operations such as `UPDATE`, `MERGE`, CTAS/swap rewrites, or ClickHouse `MATERIALIZE COLUMN` because those operations can rewrite large tables, block replication progress, and create destination-specific cost spikes. Instead, destinations apply the schema change in the cheapest safe form they support: * BigQuery adds the column as nullable, then sets supported literal default metadata for future writes. * ClickHouse may expose default values for pre-existing rows through default metadata, but ETL does not issue `MATERIALIZE COLUMN`. * DuckLake uses add-time initial default metadata for supported defaults; its schema evolution does not rewrite data files. * Snowflake uses `ADD COLUMN ... DEFAULT` for supported literal defaults. Snowflake exposes default values for existing rows and does not document this as a physical row rewrite, but defaults created this way cannot later be dropped. As a result, pre-existing destination rows might not match PostgreSQL's historical `ADD COLUMN ... DEFAULT` view unless the destination has a metadata-only initial-default mechanism and the source default is supported. This does **not** mean future replicated tuples lose their default values: PostgreSQL sends evaluated column values in row data after the relation change, and ETL writes those values normally. The limitation is only that unsupported defaults are not installed as destination schema default metadata, and existing destination rows are not rewritten by ETL. ## Supported Column Defaults [#supported-column-defaults] Column defaults are **best-effort metadata translations**, not a PostgreSQL expression evaluator. ETL reads the source default from PostgreSQL's `pg_get_expr` output, parses only deterministic literal defaults, and asks each destination whether that parsed default can be rendered safely in that destination's SQL dialect. If a default is not in the supported subset, ETL skips the destination default metadata with a warning. Replication does not fail, and future tuples still carry evaluated values from PostgreSQL. This is intentional: runtime-generated defaults can evaluate at different times or under different session settings in different systems, so installing a similar-looking destination default can create silent mismatches. ETL can add support for specific additional defaults later when their semantics can be preserved for the destination. The shared parser currently recognizes only these source default shapes: | Source default shape | Examples | | ---------------------------- | ---------------------------------------------------------------------------- | | String literals | `'pending'::text`, `('don''t'::text)` | | Numeric literals | `42`, `-1`, `'42.10'::numeric(10,2)` | | Boolean literals | `true`, `false`, `'true'::boolean` | | Date/time/timestamp literals | `'2026-01-01'::date`, `'12:30:00'::time`, `'2026-01-01 12:30:00'::timestamp` | | JSON literals | `'{}'::jsonb`, `'{"enabled": true}'::json` | | UUID literals | `'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid` | The parser is intentionally conservative. These PostgreSQL defaults are examples of **unsupported** expressions: ```sql default nextval('users_id_seq') default 'a' || 'b' default lower('USER' || '_ID') default concat('a', 'b') default md5('x') default random() default clock_timestamp() default now() default current_timestamp default current_user default gen_random_uuid() default uuid_generate_v4() default timezone('UTC', now()) default array['a', 'b'] default (select 'x') default current_setting('app.tenant_id') default 1e6 default interval '1 day 2 hours' ``` Unsupported defaults are skipped because translating arbitrary PostgreSQL expressions would require both a PostgreSQL parser and a destination-specific expression translator. Most destinations do not accept arbitrary PostgreSQL expressions as column defaults, and even similar-looking SQL can have different volatility, time zone, type coercion, or evaluation semantics. Skipping the destination schema default does not drop actual row values emitted by PostgreSQL. Destination support may be narrower than parser support: | Destination | Supported default behavior | | ----------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | BigQuery | Supports compatible string, numeric, boolean, date, time, timestamp, JSON, and UUID literals. Added columns are created nullable and supported defaults are set afterward for future writes. | | ClickHouse | Supports compatible string, numeric, boolean, date, time, timestamp, JSON, and UUID literals. Defaults are metadata only unless separately materialized. | | DuckLake | Supports compatible string, numeric, date, time, timestamp, JSON, and UUID literals. Boolean defaults are currently skipped by the DuckLake destination. | | Snowflake | `CREATE TABLE` supports compatible string, numeric, boolean, date, time, timestamp, JSON, and UUID literals. `ADD COLUMN` only receives the literal subset Snowflake allows for add-column defaults: string, numeric, and boolean literals. Later default changes on existing columns are skipped. | When changing a default from one supported expression to an unsupported expression, destinations that can safely remove defaults drop the old supported default to avoid leaving stale destination behavior behind. Snowflake is the exception for defaults introduced by `ADD COLUMN ... DEFAULT`, because Snowflake does not allow those defaults to be dropped safely. ## Diff Semantics [#diff-semantics] ETL stores schemas in **PostgreSQL column ordinal order** (`pg_attribute.attnum`) and computes destination schema diffs over replicated columns only. The current diff rules are: * Same ordinal position, different name: column rename. * Same ordinal position, different default expression: column default change. * Same ordinal position, different nullability: detected schema metadata change. * Old ordinal position missing from the new schema: column drop. * New ordinal position missing from the old schema: column add. This matches PostgreSQL's normal behavior for simple column operations: renames keep the same `attnum`, dropped columns disappear from the visible schema, and newly added columns receive new ordinal positions. ## Destination Handling [#destination-handling] Custom destinations should handle schema changes in `write_events()` by watching for `Event::Relation`. A practical flow is: 1. Iterate through the batch in order, treating each relation event as a possible schema transition for that table. 2. Flush any buffered rows/events for the old schema before processing the relation event. 3. Compare the old destination schema with the relation event's new `ReplicatedTableSchema`. 4. Mark destination metadata as `Applying` if the destination needs recovery bookkeeping for the DDL transition. 5. Apply supported destination DDL for adds, drops, renames, and default changes. 6. Mark destination metadata as `Applied` only after the destination schema is actually ready for following row events. 7. Process following row events with the new schema. The built-in BigQuery, ClickHouse, DuckLake, and Snowflake destinations follow this shape: they mark destination schema metadata as `Applying`, apply the supported DDL operations, then mark the schema as `Applied`. Because destination DDL is not always transactional, recovery is destination-specific. ClickHouse and DuckLake can retry an interrupted `Applying` operation. When an arriving relation drives that retry, it must exactly match the snapshot ID and replication mask recorded as the target; DuckLake startup recovery can instead reconstruct that exact target from durable schema state. BigQuery does not automatically repair `Applying` schema-change metadata, and Snowflake only automatically retries interrupted initial setup, not an interrupted schema change. Other interrupted states require resynchronization. Other destination modules may support a narrower schema-change surface. Treat `Event::Relation` as the stable ETL contract, then check the destination's status and implementation before relying on automatic destination DDL. ## Supported Scope [#supported-scope] The source event trigger intentionally observes a broad schema snapshot, but ETL currently supports the simplest safe cases: * `ALTER TABLE ... ADD COLUMN` for replicated columns. * `ALTER TABLE ... DROP COLUMN` for replicated columns. * `ALTER TABLE ... RENAME COLUMN` for replicated columns. * `ALTER TABLE ... ALTER COLUMN ... SET DEFAULT` where the destination supports setting compatible default metadata. * `ALTER TABLE ... ALTER COLUMN ... DROP DEFAULT` where the destination supports removing default metadata safely. * Multi-subcommand `ALTER TABLE` statements composed of those simple changes. * Changes to published permanent tables only. The trigger ignores temporary tables, unpublished tables, generated columns, dropped-column catalog tombstones, extension-owned DDL, and non-logical-WAL databases. ## Known Beta Limitations [#known-beta-limitations] These behaviors are **not full destination DDL semantics** yet: * Only `ALTER TABLE` and supported `ALTER PUBLICATION` changes are captured by the ETL DDL trigger today. * Type changes, constraint changes, identity changes, and replica-identity changes may be visible in the emitted snapshot, but they are not yet interpreted as destination DDL operations. * Table create/drop/rename operations are outside the current schema-change contract. Publication membership and table cleanup remain separate pipeline lifecycle concerns. * Live publication membership, row-filter, operation-setting, publication-name, and `publish_via_partition_root` changes are unsupported. Although the source trigger may emit schema snapshots for some of these commands, it does not dynamically reconcile table ownership or migrate destination table identity. * If a table-sync worker decodes a DDL or publication-column change during catch-up but receives no following relation before its handover boundary, it cannot construct the complete decoder required by `SyncDone`. The DDL message supplies the physical schema, but only the relation supplies the exact publication and replica-identity masks for that WAL position. ETL cannot safely reuse older masks or read newer catalog state, so this correctness edge case fails the table sync closed. Retry or resynchronize the table after schema activity has settled. * A drop and re-add is not treated as a rename. It becomes a drop plus an add because PostgreSQL assigns a new ordinal position to the new column. * Destination defaults are best-effort metadata translations. Unsupported defaults are skipped with a warning instead of failing replication. This does not remove values PostgreSQL emits in future row events; it only means the destination schema default metadata is not set. If a previously supported default becomes unsupported, ETL removes the old destination default where the destination supports that operation so stale behavior is not left behind. Snowflake default changes on existing columns are skipped with a warning because `ALTER COLUMN SET DEFAULT` is documented only for existing sequence defaults, and defaults introduced by `ALTER TABLE ADD COLUMN ... DEFAULT` cannot be dropped safely. * Runtime-generated defaults are intentionally unsupported as destination schema defaults. Examples include `now()`, `clock_timestamp()`, `gen_random_uuid()`, `random()`, sequence defaults, and session-dependent expressions such as `current_user`. Those expressions can evaluate at different times or under different session settings in each destination. PostgreSQL still sends evaluated values for future row events, but existing destination rows are not physically backfilled by ETL. Avoid runtime-generated defaults for replicated schema changes when exact historical values matter. * `ADD COLUMN ... DEFAULT` semantics differ by destination. ETL intentionally avoids destination DDL that rewrites all existing rows. BigQuery leaves pre-existing destination rows null for newly added defaulted columns, while ClickHouse, DuckLake, and Snowflake can expose supported add-time defaults without ETL issuing a materialization rewrite. Snowflake only receives add-column defaults for source defaults that can be rendered as Snowflake literals. * BigQuery applies ongoing-replication `DROP NOT NULL` changes so future source NULL values remain writable. BigQuery cannot change an existing `NULLABLE` column to `REQUIRED`, so PostgreSQL enforces later `SET NOT NULL` changes while the destination column remains nullable. Other built-in destinations currently leave ongoing-replication nullability changes unchanged. Newly added columns remain nullable where the destination requires that for historical rows. * The trigger payload includes `current_query` for debugging only. It can contain literals and multiple statements, so it must not be treated as replayable DDL. * BigQuery, ClickHouse, DuckLake, and Snowflake reject stale or ambiguously ordered relation schemas instead of rewinding. An older snapshot could drive reverse DDL that drops newer columns and their data. An equal snapshot with a different replication mask is also rejected: supported publication column-list changes always receive a new composite snapshot ID, so the conflicting masks have no ordering with which to choose a safe winner. The pipeline fails with a schema-rewind error and the affected table must be resynchronized. * Sessions can set `supabase_etl.skip_ddl_log = 'true'` as an emergency opt-out while recovering a system. DDL executed with that setting enabled is not logged for ETL. ## Event Ordering [#event-ordering] Schema changes are transactional logical messages, so they appear in WAL order relative to row changes. ETL updates its stored schema when it decodes the DDL message, then waits for the next `RELATION` message to rebuild the runtime replication and identity masks for row decoding. This matters for custom destinations: row events after a relation event should be decoded and written using that relation event's schema. Row events before it belong to the previous schema. --- # Extension Points > Traits you implement to customize Supabase ETL behavior. - Canonical HTML: https://supabase.github.io/etl/explanation/traits/ - Agent-readable Markdown: https://supabase.github.io/etl/explanation/traits.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/explanation/traits.md Implement these traits to control where replicated data goes and how ETL state is stored. ## Destination [#destination] Receives replicated data. This is the **primary extension point** for sending data to custom systems. ETL is **at least once**, so destinations must tolerate duplicate writes and concurrent calls. ```rust pub trait Destination { fn name() -> &'static str; fn shutdown(&self) -> impl Future> + Send { async { Ok(()) } } fn startup(&self) -> impl Future> + Send { async { Ok(()) } } fn drop_table_for_copy(&self, replicated_table_schema: &ReplicatedTableSchema, async_result: DropTableForCopyResult<()>) -> impl Future> + Send; fn write_table_rows(&self, replicated_table_schema: &ReplicatedTableSchema, table_rows: Vec, async_result: WriteTableRowsResult) -> impl Future> + Send; fn write_events(&self, events: Vec, durability: WriteEventsDurability, async_result: WriteEventsResult) -> impl Future> + Send; } ``` ### Methods [#methods] | Method | Purpose | | ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `name()` | Returns identifier for logging and diagnostics | | `shutdown()` | Called when the pipeline shuts down. Default is a no-op. Override for cleanup or bookkeeping | | `startup()` | Called after store caches are loaded, removed-publication tables are purged, and before workers start. Default is a no-op. Override to reconcile destination state after restarts | | `drop_table_for_copy()` | Drops the existing destination object and destination-private replay state before restarting a table copy. Receives the previously stored replicated schema for locating the old object | | `write_table_rows()` | Writes rows during initial table copy. Receives the current replicated schema and may get an empty vector for an empty table or a deferred durability barrier | | `write_events()` | Processes ongoing replication events (inserts, updates, deletes, truncates, relations, and transaction markers). Batches may span multiple tables, or be empty for a required durability barrier | ### Implementation Notes [#implementation-notes] * `drop_table_for_copy()` should be **idempotent**. ETL calls it before clearing copy-scoped store state, so implementations can still use the supplied schema and existing destination metadata to locate the old object. Before returning success, it must also drain or invalidate writes accepted by an earlier copy attempt so stale work cannot mutate the recreated table. * `write_table_rows()` is called even for empty source tables so destinations can prepare initial destination state before ongoing replication begins. * An immediate `write_table_rows()` implementation returns `DestinationWriteStatus::Durable` after the batch is durable. A deferred implementation may return `Accepted` after taking ownership of a nonempty batch. It must bound its accepted-but-not-durable backlog and delay `Accepted` when no capacity is available. If any batch returns `Accepted`, ETL sends an empty batch after all copy workers finish; the destination must return `Durable` from that call only after all rows accepted during the current copy attempt are durable. * `WriteEventsDurability::MayDefer` permits `write_events()` to return `Accepted` or `Durable`. ETL may issue `write_events(Vec::new(), WriteEventsDurability::RequireDurable, ...)` as a durability-only barrier, but never an empty `MayDefer` write. The empty vector carries no new replication events, but the call may flush or wait for earlier accepted work and must return `Durable` only after all writes covered by the destination's ordering state are durable. A destination may use a stronger barrier scope than the originating apply-loop stream. * `write_table_rows()` and `write_events()` must tolerate **duplicate delivery** because ETL may retry or replay after failure. * Handle **concurrent calls** safely, especially from parallel table sync workers. * Preserve **per-table event order**. During initial sync and catch-up, transaction markers are not a reliable all-tables transaction boundary. * Treat `Event::Relation` as an ordered schema transition, not a `write_events()` batch boundary. ETL batches ongoing replication events by size and time, so one call can contain multiple schema changes, including multiple relation events for the same table. * Always complete the supplied async result handle. Dropping it reports a destination error to ETL. * `startup()` runs after ETL has loaded destination metadata and table schemas from the store and purged tables removed from the publication, so destinations can compare active persisted ETL state with their physical objects before replication work starts. * All three write-like methods use async results, but ETL waits differently. `drop_table_for_copy()` waits immediately before copy-scoped store cleanup. `write_table_rows()` also waits immediately, requesting the next batch only after the current one reports `Accepted` or `Durable` for that copy partition. `write_events()` is the method where ETL can keep processing other work while the destination finishes the current batch; ETL still waits for that result before handing over the next ongoing-replication batch. See [Events](https://supabase.github.io/etl/explanation/events.md) for details on the events received by `write_events()`. `PipelineDestination` is a blanket-implemented facade for destinations that also satisfy the pipeline runtime clone and thread-safety bounds. Pipeline runtime code uses this facade when it needs to move destinations across worker tasks, but custom destinations only implement `Destination` directly. ## SchemaStore [#schemastore] Stores **versioned table schema information** (column names, types, primary keys, and snapshot IDs). A `SnapshotId` compares its commit LSN first and its message LSN second; store implementations should compare the type directly rather than its variable-width decimal display string. ```rust pub trait SchemaStore { fn get_table_schema(&self, table_id: &TableId, snapshot_id: SnapshotId) -> impl Future>>> + Send; fn get_table_schemas(&self) -> impl Future>>> + Send; fn load_table_schemas(&self) -> impl Future> + Send; fn store_table_schema(&self, table_schema: TableSchema) -> impl Future>> + Send; fn prune_table_schemas(&self, retention_snapshot_ids: BTreeMap) -> impl Future> + Send; } ``` ### Methods [#methods-1] | Method | Purpose | | ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `get_table_schema()` | Returns the schema version with the largest `snapshot_id <= requested_snapshot_id`. If it misses cache, it may load from persistent storage | | `get_table_schemas()` | Returns all cached schemas without reading persistent storage | | `load_table_schemas()` | Loads schemas from persistent storage into cache. Call once at startup. Returns the number of schemas loaded | | `store_table_schema()` | Saves a schema version to both cache and persistent storage and returns the cached `Arc` | | `prune_table_schemas()` | For the supplied per-table snapshot boundaries, preserves the newest schema version at or before each boundary, preserves newer versions, and removes older versions. The `BTreeMap` provides deterministic table-ID iteration. Implementations with both cache and persistent storage must prune both | ## StateStore [#statestore] Tracks **table states**, **persisted replication checkpoints**, and **destination table metadata**. ```rust pub trait StateStore { // Table state fn get_table_state(&self, table_id: TableId) -> impl Future>> + Send; fn get_table_states(&self) -> impl Future> + Send; fn load_table_states(&self) -> impl Future> + Send; fn update_table_states(&self, updates: Vec<(TableId, TableState)>) -> impl Future> + Send; fn update_table_state(&self, table_id: TableId, state: TableState) -> impl Future> + Send; fn rollback_table_state(&self, table_id: TableId) -> impl Future> + Send; // Persisted replication checkpoints fn get_replication_checkpoint(&self, worker_type: WorkerType) -> impl Future>> + Send; fn upsert_replication_checkpoint(&self, worker_type: WorkerType, checkpoint_lsn: PgLsn) -> impl Future> + Send; fn delete_replication_checkpoint(&self, worker_type: WorkerType) -> impl Future> + Send; // Destination table metadata fn get_destination_table_metadata(&self, table_id: TableId) -> impl Future>> + Send; fn get_applied_destination_table_metadata(&self, table_id: TableId) -> impl Future>> + Send; fn load_destination_tables_metadata(&self) -> impl Future> + Send; fn store_destination_table_metadata(&self, table_id: TableId, metadata: DestinationTableMetadata) -> impl Future> + Send; } ``` ### Table State Methods [#table-state-methods] | Method | Purpose | | ------------------------ | ---------------------------------------------------------------------------------------------------------- | | `get_table_state()` | Returns current state for a table from cache | | `get_table_states()` | Returns states for all tables from cache as \[`TableStates`] | | `load_table_states()` | Loads states from persistent storage into cache. Call once at startup. Returns the number of states loaded | | `update_table_states()` | Atomically updates multiple table states in both cache and persistent storage | | `update_table_state()` | Updates state in both cache and persistent storage | | `rollback_table_state()` | Reverts table to previous state. Returns the state after rollback | ### Replication Checkpoint Methods [#replication-checkpoint-methods] A persisted replication checkpoint records a safe replay frontier for the apply worker or a table-sync worker. ETL can select a durably flushed commit boundary or, when the apply loop is fully idle, its last received LSN. The checkpoint lets the worker resume safely after a restart. | Method | Purpose | | --------------------------------- | ------------------------------------------------------------------------------------------------------- | | `get_replication_checkpoint()` | Returns the persisted checkpoint for a worker, if present | | `upsert_replication_checkpoint()` | Monotonically stores a checkpoint and returns the stored LSN. Implementations must not move it backward | | `delete_replication_checkpoint()` | Deletes the checkpoint when a worker slot lineage is intentionally reset | ### Destination Metadata Methods [#destination-metadata-methods] Destination table metadata connects source table IDs to destination state, including the **destination table identifier**, the **schema snapshot under management**, the **schema status** (`Applying` or `Applied`), and the **replication mask**. Only `AppliedDestinationTableMetadata` guarantees the destination schema is ready for normal reads and writes. | Method | Purpose | | ------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------- | | `get_destination_table_metadata()` | Returns destination table metadata for a source table from cache | | `get_applied_destination_table_metadata()` | Returns destination table metadata only when the destination schema is fully applied. If metadata exists but is still `Applying`, this returns an error | | `load_destination_tables_metadata()` | Loads destination table metadata from persistent storage into cache. Call once during startup | | `store_destination_table_metadata()` | Saves destination table metadata to both cache and persistent storage | ### Table States [#table-states] Tables progress through these states: | State | Persisted | Description | | -------------------------------------------- | --------- | ------------------------------------------------------------------------------- | | `Init` | Yes | Table discovered, ready to start | | `DataSync` | Yes | Initial data being copied | | `FinishedCopy` | Yes | Copy complete, waiting for coordination | | `SyncWait` | No | Table sync worker signaling apply worker to pause | | `Catchup { lsn }` | No | Apply worker paused, table sync worker catching up to LSN | | `SyncDone { lsn }` | Yes | Caught up to LSN; durable decoder retained until Apply materializes local state | | `Ready` | Yes | Changes via apply worker | | `Errored { reason, solution, retry_policy }` | Yes | Error occurred, excluded until rollback | ## TableStateLifecycleStore [#tablestatelifecyclestore] Coordinates ETL table-state lifecycle operations across state, schema, destination metadata, persisted checkpoints, and any store caches. ```rust pub trait TableStateLifecycleStore { fn apply_table_state_operation( &self, operation: TableStateOperation, ) -> impl Future> + Send; fn prepare_table_state_for_copy( &self, table_id: TableId, ) -> impl Future> + Send; fn reset_table_states_for_resync( &self, ) -> impl Future> + Send; fn delete_table_state( &self, table_id: TableId, ) -> impl Future> + Send; } ``` | Method | Purpose | | --------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `apply_table_state_operation()` | Single implementation point for \[`TableStateOperation`]. Custom stores implement the prepare, reset, and delete semantics here | | `prepare_table_state_for_copy()` | Deletes destination metadata, schema versions, and the table-sync checkpoint while preserving the table state. This is called only after the destination object was dropped for a fresh copy | | `reset_table_states_for_resync()` | Resets all current table states to `Init` and deletes the apply-worker checkpoint while preserving destination metadata, schema versions, and table-sync checkpoints | | `delete_table_state()` | Deletes all stored ETL-owned state for a table removed from the publication. Does not modify destination tables | ## Combining Traits [#combining-traits] A single type typically implements **all store traits**: ```rust pub struct MyStore { /* ... */ } impl SchemaStore for MyStore { /* ... */ } impl StateStore for MyStore { /* ... */ } impl TableStateLifecycleStore for MyStore { /* ... */ } ``` `PipelineStore` is a blanket-implemented facade for stores that satisfy the full pipeline runtime store bounds. Pipeline runtime code uses this facade, while code that only needs one capability should depend on the narrower trait directly. `DestinationStore` is a blanket-implemented facade for stores that satisfy the destination runtime store bounds. Destination implementations use this when they need schema and state metadata but do not need lifecycle reset/removal operations. `SharedStateStore` covers state-only users with the corresponding worker-safe bounds. ETL provides two built-in implementations: * `MemoryStore`: In-memory storage, not persistent across restarts * `PostgresStore`: Persistent storage backed by PostgreSQL `PostgresStore::new()` runs only the Postgres-backed state-store migrations. `Pipeline::start()` runs the source migrations required by ETL itself, including the schema helper functions and DDL event trigger, regardless of which store implementation you use. ## Thread Safety [#thread-safety] All trait implementations must be **thread-safe**. ETL calls these methods concurrently from: * Multiple table sync workers (parallel initial sync) * Apply worker (ongoing replication) * Pipeline coordination Use `Arc>`, `RwLock`, or similar synchronization primitives for shared state. --- # Configure Postgres > Set up Postgres with the permissions and settings required by Supabase ETL. - Canonical HTML: https://supabase.github.io/etl/guides/configure-postgres/ - Agent-readable Markdown: https://supabase.github.io/etl/guides/configure-postgres.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/guides/configure-postgres.mdx Configure the Postgres settings, slots, publications, and retention needed by the open-source replication engine. Using the managed product in the Supabase Dashboard? Follow the canonical [Supabase Pipelines documentation](https://supabase.com/docs/guides/database/replication/pipelines) instead. This guide covers PostgreSQL configuration for developers who run or embed Supabase ETL themselves. ## Prerequisites [#prerequisites] * **PostgreSQL 14 through 18.** PostgreSQL 15 adds publication filtering; PostgreSQL 16 adds logical decoding on physical read replicas. * Superuser access to the Postgres server * Ability to restart Postgres (required for `wal_level` changes) ## Enable Logical WAL [#enable-logical-wal] Set `wal_level = logical` to enable Postgres to record **logical change data** in the WAL, which external tools can then decode and replicate. ```ini title="postgresql.conf" wal_level = logical ``` **Restart Postgres** after changing this setting. ## Replication Slots [#replication-slots] A replication slot records a consumer's progress and retains the WAL it still needs. This lets ETL catch up after a temporary disconnect. ### Creating Replication Slots [#creating-replication-slots] ```sql -- Create a logical replication slot SELECT pg_create_logical_replication_slot('my_slot', 'pgoutput'); ``` ### Viewing Replication Slots [#viewing-replication-slots] ```sql -- See all replication slots SELECT slot_name, slot_type, active, restart_lsn FROM pg_replication_slots; ``` ### Deleting Replication Slots [#deleting-replication-slots] ```sql -- Drop a replication slot when no longer needed SELECT pg_drop_replication_slot('my_slot'); ``` > **Warning** > > Only delete slots when you are sure they are not in use. Deleting an active > slot will break replication. ## Max Replication Slots [#max-replication-slots] Controls how many replication slots Postgres can maintain. ```ini title="postgresql.conf" # Default: 10 max_replication_slots = 20 ``` ETL uses a **single replication slot** for its main apply worker. Additional slots are created for parallel table copies during initial sync or when new tables are added to the publication. The `max_table_sync_workers` pipeline parameter controls parallel copies, so total slots used by ETL never exceed `max_table_sync_workers + 1`. Increase this setting when: * Running multiple ETL pipelines against the same database * Development/testing environments with frequent slot creation ## Max WAL Senders [#max-wal-senders] Controls concurrent streaming connections. Each active replication slot uses one WAL sender. ```ini title="postgresql.conf" # Default: 10 max_wal_senders = 20 ``` Set this to at least `max_replication_slots` to ensure all slots can connect. ## WAL Keep Size [#wal-keep-size] Provides a disk-backed safety buffer when replication consumers fall behind. ```ini title="postgresql.conf" wal_keep_size = 1GB ``` ## WAL Buildup and Disk Usage [#wal-buildup-and-disk-usage] Replication slots prevent Postgres from deleting WAL files until all consumers have processed them. This can cause **significant disk usage** if the pipeline falls behind or encounters errors. ### Common Causes of WAL Buildup [#common-causes-of-wal-buildup] * **Errored tables:** ETL retains their slots to preserve consistency. Fix the error or remove tables you no longer replicate. * **Slow destinations:** High latency, large transactions, or outages can make ingestion fall behind source writes. * **Long initial copies:** Each active table copy retains WAL from its consistent snapshot point. > **Warning** > > If WAL grows beyond the configured limit, Postgres will terminate the > replication slot. Control this with `max_slot_wal_keep_size`: ```ini title="postgresql.conf" # -1 = unlimited (dangerous for disk space) max_slot_wal_keep_size = 10GB ``` If the main replication slot is invalidated, ETL fails pipeline startup by default and requires operator intervention. Set `invalidated_slot_behavior` to `recreate` to have ETL delete and recreate the slot, reset table states, and resynchronize eligible tables from scratch. Automatic recreation can repeat a large initial copy, so choose it deliberately and monitor destination capacity. ### Monitoring WAL Usage [#monitoring-wal-usage] ```sql -- Check replication slot WAL usage. SELECT slot_name, active, wal_status, pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS retained_wal_bytes, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS confirmed_flush_lag_bytes, safe_wal_size FROM pg_replication_slots; -- Check total WAL directory size SELECT pg_size_pretty(sum(size)) AS wal_size FROM pg_ls_waldir(); ``` `wal_status` reports whether the WAL required by a replication slot is still available: * `reserved`: required WAL is within normal retention. * `extended`: required WAL exceeds `max_wal_size`, but Postgres is still retaining it. * `unreserved`: required WAL is no longer fully reserved and may be removed at the next checkpoint. * `lost`: required WAL was removed and the slot is no longer usable. * `NULL`: the slot has not reserved WAL yet, usually because `restart_lsn` is `NULL`. ETL API responses return unknown future Postgres `wal_status` values as `unknown`. ### Recommendations [#recommendations] * Bound retention with `max_slot_wal_keep_size` based on available disk. * Alert on `confirmed_flush_lag_bytes`, `retained_wal_bytes`, `safe_wal_size`, and `wal_status`. * Treat `safe_wal_size IS NULL` as unlimited retention and `0` as no remaining headroom. * Resolve errored tables promptly and size `max_table_sync_workers` to balance copy speed with resource use. ## Read Replicas [#read-replicas] ETL can read logical replication from a physical read replica when the replica runs **PostgreSQL 16 or newer**. PostgreSQL 14 and 15 can still be used with ETL, but logical decoding must run on the primary. See the PostgreSQL documentation on [logical slots on hot standby](https://www.postgresql.org/docs/16/logicaldecoding-explanation.html#LOGICALDECODING-REPLICATION-SLOTS). When using a read replica: * Configure `pg_connection` to point at the replica. ETL uses this connection for logical replication, table copy, schema reads, publications, slots, keepalives, and status updates. * Configure `store_pg_connection` when using `PostgresStore` and `pg_connection` points at a read-only replica. The store connection must be writable because it runs store migrations and persists pipeline state. * Apply ETL source migrations on the primary before starting the pipeline. Standby connections are read-only, so ETL skips source migration execution when the configured source is in recovery. * Let ETL create its logical replication slots on the read replica. Do not pre-create ETL logical slots on the primary for this mode. Publication, table, and ETL source-migration changes are ordinary WAL records. When you create them on the primary, the read replica can only see them after replay reaches that WAL position. Do not wait a fixed number of seconds; wait for a concrete replay LSN: ```sql -- Run on the primary after creating tables, source migrations, and publications. SELECT pg_current_wal_flush_lsn(); ``` Then wait on the read replica until replay reaches that LSN. If you just created a new database on the primary, connect to an existing maintenance database such as `postgres` for this check; the new database might not exist on the replica until replay catches up. ```sql -- Run on the read replica before starting ETL. SELECT pg_last_wal_replay_lsn() >= '0/16B6C50'::pg_lsn AS ready; ``` For extra confidence, also check that the expected publication is visible on the replica: ```sql SELECT 1 FROM pg_publication WHERE pubname = 'my_publication'; ``` If ETL only receives the replica `pg_connection`, it cannot derive the primary's setup LSN itself. The orchestrator that creates or updates the source-side schema and publication should perform this LSN barrier before starting the pipeline, or should retry pipeline startup until the replica catches up. Once the logical slot exists on the replica, ongoing primary writes can lag normally; ETL will decode them as the replica replays WAL and the slot keeps the replica-side restart position. The primary must generate logical WAL, and each server needs enough sender and slot capacity for the role it plays. On the primary, count the physical slots used by read replicas. On the read replica, count the logical slots ETL creates for its apply worker and table sync workers: ```ini title="postgresql.conf" wal_level = logical max_replication_slots = 20 max_wal_senders = 20 ``` For the physical replication link between the primary and the read replica, use a [physical replication slot](https://www.postgresql.org/docs/16/warm-standby.html#STREAMING-REPLICATION-SLOTS) and enable standby feedback: ```ini title="postgresql.conf (standby)" primary_conninfo = 'host=primary.example.com port=5432 dbname=postgres user=replicator password=...' primary_slot_name = 'etl_read_replica' hot_standby = on hot_standby_feedback = on wal_receiver_status_interval = '1s' ``` `hot_standby_feedback` helps prevent required catalog rows from being vacuumed away on the primary while standby logical slots need them. A physical slot between the primary and the standby keeps that protection across standby reconnects and restarts. If initial copies can run for longer than your standby conflict delay, tune `max_standby_streaming_delay` for that replica. A larger value reduces copy cancellations at the cost of allowing more replay lag while conflicting standby queries finish. Logical slot creation on a standby needs information about transactions running on the primary. If the primary is idle, creating a logical slot on the standby can wait until the primary emits that snapshot information. To speed this up during setup or tests, run this on the primary: ```sql SELECT pg_log_standby_snapshot(); ``` This is only a setup-time nudge for slot creation. ETL uses the regular PostgreSQL logical replication protocol for keepalives and status updates after ongoing replication starts. PostgreSQL 17+ also has logical failover slot synchronization, where failover-enabled logical slots on the primary are synchronized to standbys. That is a separate high-availability feature for resuming logical replication after promoting a standby. It is not required for ETL to read from a current read replica, and synchronized standby slots cannot be consumed on the standby while they are marked as synced. ## Publications [#publications] Publications define **which tables and operations** to replicate. ### Creating Publications [#creating-publications] ```sql -- Create publication for specific tables CREATE PUBLICATION my_publication FOR TABLE users, orders; -- Create publication for all tables (use with caution) CREATE PUBLICATION all_tables FOR ALL TABLES; -- Create publication for all tables in selected schemas CREATE PUBLICATION schema_tables FOR TABLES IN SCHEMA public, analytics; -- Include only specific operations CREATE PUBLICATION inserts_only FOR TABLE users WITH (publish = 'insert'); ``` Avoid publishing ETL-owned tables. If the source database is also used as the ETL state store, the `etl` schema contains ETL internal tables. Do not include that schema in the publication. In that setup, `FOR ALL TABLES` also includes ETL-owned tables, so use explicit table lists or `FOR TABLES IN SCHEMA ...` for customer-owned schemas instead. #### Partitioned Tables [#partitioned-tables] ETL supports PostgreSQL partition publications with either `publish_via_partition_root = true` or `publish_via_partition_root = false`. At startup, ETL loads the effective relation OIDs through `pg_get_publication_tables()`, so it tracks the same relation identities and schemas PostgreSQL uses for logical replication messages. | Publication shape | `publish_via_partition_root` | PostgreSQL publishes as | ETL tracks | | ----------------------------------------------------------------------------------------------------------------------------- | ---------------------------- | ----------------------------------- | --------------------------------------- | | `FOR TABLE orders` where `orders` is the top partitioned table | `true` | `orders` | `orders` | | `FOR TABLE orders` where `orders` is the top partitioned table | `false` | Leaf partitions under `orders` | The leaf partitions | | `FOR TABLE orders_2026` where `orders_2026` is a partitioned subtree | `true` | `orders_2026` | `orders_2026` | | `FOR TABLE orders_2026` where `orders_2026` is a partitioned subtree | `false` | Leaf partitions under `orders_2026` | The leaf partitions under `orders_2026` | | `FOR TABLE orders_2026_01` where `orders_2026_01` is a leaf partition | Either | `orders_2026_01` | `orders_2026_01` | | `FOR ALL TABLES` for the whole database, or `FOR TABLES IN SCHEMA ...` for selected schemas. Do not include ETL-owned tables. | `true` | Partition roots plus regular tables | Partition roots plus regular tables | | `FOR ALL TABLES` for the whole database, or `FOR TABLES IN SCHEMA ...` for selected schemas. Do not include ETL-owned tables. | `false` | Leaf partitions plus regular tables | Leaf partitions plus regular tables | For example, with this hierarchy: ```text orders ├── orders_2025 │ ├── orders_2025_01 │ └── orders_2025_02 └── orders_2026 ├── orders_2026_01 └── orders_2026_02 ``` `FOR TABLE orders_2026 WITH (publish_via_partition_root = true)` replicates the 2026 subtree as `orders_2026`. `FOR TABLE orders_2026 WITH (publish_via_partition_root = false)` replicates `orders_2026_01` and `orders_2026_02` as separate leaf tables. On PostgreSQL 15+, row filters on partition publications are applied during both the initial sync and ongoing replication. ETL uses the row filter attached to the effective publication table entry: the published root or subtree when `publish_via_partition_root = true`, and the published leaf relation when `publish_via_partition_root = false`. **Limitation:** With `publish_via_partition_root = true`, `TRUNCATE` operations on individual partitions are not replicated. Execute truncates on the published partition table instead. For a top-level publication, that is the top root; for a subtree publication, that is the published subtree root. ```sql -- This will NOT be replicated TRUNCATE TABLE orders_2026_01; -- This WILL be replicated TRUNCATE TABLE orders_2026; ``` ### Managing Publications [#managing-publications] Create the intended publication shape before starting a pipeline whenever possible. For a running pipeline, ETL currently supports changing the column list of a table it already tracks. Other `ALTER PUBLICATION` changes are not live publication-reconciliation operations. In particular, adding or removing tables, adding or removing schemas, and changing `publish_via_partition_root` can change the effective relation OIDs. ETL discovers those new and removed identities during pipeline startup, not from schema messages emitted by the DDL trigger. Restarting reconciles ETL's table state, but it does not rename, merge, or remove destination tables created under the previous relation identities. A root-to-leaf or leaf-to-root change may therefore require a table resynchronization or a new pipeline. > **Warning** > > Do not change publication membership, row filters, publication operation > settings, or `publish_via_partition_root` while a pipeline is running. These > changes are outside the supported live-update contract. ETL may skip events > for newly effective relation OIDs, and partition-identity changes can leave > stale destination tables or cause missing or duplicated data. Behavior in > this state is unsupported and must not be relied upon. ```sql -- View existing publications SELECT * FROM pg_publication; -- See which tables are in a publication SELECT * FROM pg_publication_tables WHERE pubname = 'my_publication'; -- Add tables to existing publication ALTER PUBLICATION my_publication ADD TABLE products; -- Remove tables from publication ALTER PUBLICATION my_publication DROP TABLE products; -- Drop publication DROP PUBLICATION my_publication; ``` ## Version Compatibility [#version-compatibility] ETL supports PostgreSQL 14 through 18. PostgreSQL 14 is limited to table-level publication filtering; use the newer versions for the features below. | Feature | PostgreSQL 14 | PostgreSQL 15 | PostgreSQL 16+ | | ------------------------------------------ | ------------- | ------------- | -------------- | | Table-level publication | Yes | Yes | Yes | | Column-level filtering | No | Yes | Yes | | Row-level filtering | No | Yes | Yes | | `FOR TABLES IN SCHEMA` | No | Yes | Yes | | Partitioned table support | Yes | Yes | Yes | | Logical decoding on physical read replicas | No | No | Yes | ## Complete Configuration Example [#complete-configuration-example] Minimal `postgresql.conf` setup: ```ini title="postgresql.conf" # Enable logical replication wal_level = logical # Replication capacity max_replication_slots = 20 max_wal_senders = 20 # WAL retention wal_keep_size = 1GB # Limit WAL retention per slot (optional but recommended) max_slot_wal_keep_size = 10GB ``` After editing the configuration: 1. Restart Postgres 2. Create your publication: ```sql CREATE PUBLICATION etl_publication FOR TABLE your_table; ``` 3. Verify the setup: ```sql SHOW wal_level; SHOW max_replication_slots; SELECT * FROM pg_publication WHERE pubname = 'etl_publication'; ``` ## Next Steps [#next-steps] * [First Pipeline](https://supabase.github.io/etl/guides/first-pipeline.md): Hands-on tutorial using these settings * [Custom Implementations](https://supabase.github.io/etl/guides/custom-implementations.md): Build your own components * [Architecture](https://supabase.github.io/etl/explanation/architecture.md): How ETL uses these settings --- # Custom Implementations > Implement your own stores and destinations. - Canonical HTML: https://supabase.github.io/etl/guides/custom-implementations/ - Agent-readable Markdown: https://supabase.github.io/etl/guides/custom-implementations.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/guides/custom-implementations.mdx Implement a store and an HTTP destination. Start with [First Pipeline](https://supabase.github.io/etl/guides/first-pipeline.md) 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](https://supabase.github.io/etl/reference/destinations.md#custom-destinations-and-upstream-support) for the distinction and upstream maintenance expectations. ## Understanding the Destination Trait [#understanding-the-destination-trait] ETL delivers data to destinations in **two phases**: | Phase | Method | When | Data Type | | ------------------- | -------------------- | --------------------------------------- | ----------------------------------------------- | | Initial sync | `write_table_rows()` | Startup | `Vec` | | Ongoing replication | `write_events()` | During catch-up and ongoing replication | `Vec` 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](https://supabase.github.io/etl/explanation/schema-changes.md) for the current semantics and limitations. ## Create the project [#create-the-project] ```bash title="Terminal" cargo new etl-custom --lib cd etl-custom rustup override set 1.95.0 ``` Update `Cargo.toml`: ```toml title="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 [#implement-a-custom-store] Create `src/custom_store.rs`. A store must implement **three traits** (see [Extension Points](https://supabase.github.io/etl/explanation/traits.md) 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. ```rust title="src/custom_store.rs" lineNumbers 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>, state: Option, state_history: Vec, destination_metadata: Option, } #[derive(Debug, Clone)] pub struct CustomStore { tables: Arc>>, checkpoints: Arc>>, } 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>> { 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>> { let tables = self.tables.lock().await; Ok(tables .values() .flat_map(|entry| entry.schemas.values().cloned()) .collect()) } async fn load_table_schemas(&self) -> EtlResult { Ok(0) } async fn store_table_schema(&self, schema: TableSchema) -> EtlResult> { 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, ) -> EtlResult { 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> { let tables = self.tables.lock().await; Ok(tables.get(&table_id).and_then(|e| e.state.clone())) } async fn get_table_states(&self) -> EtlResult { let tables = self.tables.lock().await; Ok(Arc::new( tables .iter() .filter_map(|(id, e)| e.state.clone().map(|s| (*id, s))) .collect::>(), )) } async fn load_table_states(&self) -> EtlResult { 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 { 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> { 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 { 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> { 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> { self.get_destination_table_metadata(table_id) .await? .map(|metadata| metadata.into_applied()) .transpose() } async fn load_destination_tables_metadata(&self) -> EtlResult { 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 { 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 [#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. ```rust title="src/http_destination.rs" lineNumbers 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 { 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, 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::>() }) }).collect::>() }); let result = self .post("rows", payload) .await .map(|_| DestinationWriteStatus::Durable); async_result.send(result); Ok(()) } async fn write_events( &self, events: Vec, _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::>() }), Event::Unsupported => json!({"type": "unsupported"}), } }).collect::>() }); let result = self.post("events", payload).await; async_result.send(result.map(|_| DestinationWriteStatus::Durable)); Ok(()) } } ``` **Check:** `cargo check` succeeds. ## Wire it together [#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. ```rust title="src/main.rs" lineNumbers 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> { 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 [#test-the-pipeline] ```bash title="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 [#next-steps] * [Extension Points](https://supabase.github.io/etl/explanation/traits.md) - Full trait API documentation * [Events](https://supabase.github.io/etl/explanation/events.md) - Details on all events your destination receives * [Configure Postgres](https://supabase.github.io/etl/guides/configure-postgres.md) - Production database setup * [Architecture](https://supabase.github.io/etl/explanation/architecture.md) - How ETL works internally --- # First Pipeline > Learn Supabase ETL by building a working Postgres replication pipeline. - Canonical HTML: https://supabase.github.io/etl/guides/first-pipeline/ - Agent-readable Markdown: https://supabase.github.io/etl/guides/first-pipeline.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/guides/first-pipeline.mdx Build a pipeline that performs an initial sync of a Postgres table, then replicates row changes through a small custom `Destination`. > **Active development** > > Supabase ETL is under active development. APIs and setup steps may change > before the first stable release. ## Prerequisites [#prerequisites] * Rust toolchain 1.95.0, matching `rust-toolchain.toml` * Postgres 14+ with logical replication enabled (`wal_level = logical` in `postgresql.conf`) * Basic familiarity with Rust and SQL New to Postgres logical replication? Read [Logical Replication](https://supabase.github.io/etl/explanation/concepts.md) first. ## Create the project [#create-the-project] ```bash title="Terminal" cargo new etl-tutorial cd etl-tutorial rustup override set 1.95.0 ``` Add dependencies to `Cargo.toml`: ```toml title="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 [#set-up-postgres] Connect to Postgres and create a **test database, table, seed rows, and publication**: ```sql title="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:** `SELECT * FROM pg_publication WHERE pubname = 'my_publication';` returns one row. ## Write the pipeline [#write-the-pipeline] Replace `src/main.rs`: ```rust title="src/main.rs" lineNumbers use etl::{ config::{ BatchConfig, InvalidatedSlotBehavior, PgConnectionConfig, PipelineConfig, TableSyncCopyConfig, TcpKeepaliveConfig, TlsConfig, }, data::TableRow, destination::{ Destination, DestinationWriteStatus, DropTableForCopyResult, WriteEventsDurability, 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, rows: Vec, async_result: WriteTableRowsResult, ) -> EtlResult<()> { println!("copied {} rows", rows.len()); async_result.send(Ok(DestinationWriteStatus::Durable)); Ok(()) } async fn write_events( &self, events: Vec, _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> { 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, 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: 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`. ## Run the pipeline [#run-the-pipeline] ```bash title="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 [#test-ongoing-replication] In another terminal, make changes to the database: ```sql title="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 [#cleanup] Stop the pipeline with `Ctrl+C`, then clean up the database: ```sql title="psql" -- Connect to a different database first (e.g., postgres) \c postgres DROP DATABASE etl_tutorial; ``` ## Next Steps [#next-steps] * [Custom Implementations](https://supabase.github.io/etl/guides/custom-implementations.md): Build your own components * [Configure Postgres](https://supabase.github.io/etl/guides/configure-postgres.md): Production Postgres setup * [Architecture](https://supabase.github.io/etl/explanation/architecture.md): How ETL works internally --- # Standalone Replicator > Build, configure, and run Supabase ETL as a standalone process. - Canonical HTML: https://supabase.github.io/etl/guides/standalone-replicator/ - Agent-readable Markdown: https://supabase.github.io/etl/guides/standalone-replicator.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/guides/standalone-replicator.mdx The standalone replicator is the ready-made `etl-replicator` application for running Supabase ETL without embedding the library in another Rust program. > **Active development** > > Supabase ETL is under active development. APIs and setup steps may change > before the first stable release. ## What the replicator is [#what-the-replicator-is] The replicator packages one ETL pipeline, a Postgres-backed state store, and one configured built-in destination into a long-lived process. It loads the source, destination, batching, retry, and operational settings from files and environment variables, starts replication, and handles graceful shutdown. It does not require Kubernetes or the managed Supabase product. Use the standalone replicator when a built-in destination and the standard Postgres-backed store fit your deployment. Embed the `etl` crate instead when you need a custom destination, a custom store, or application-specific runtime orchestration. See [First Pipeline](https://supabase.github.io/etl/guides/first-pipeline.md) for the library path. ## Prerequisites [#prerequisites] * Rust 1.95.0, as pinned by `rust-toolchain.toml`. * PostgreSQL 14 through 18 configured for logical replication. * A publication containing the tables and operations you want to replicate. * A Google Cloud project with the BigQuery API enabled and a dataset for the replicated tables. * A service account with the BigQuery Data Editor and BigQuery Job User roles. Complete [Configure Postgres](https://supabase.github.io/etl/guides/configure-postgres.md) before starting the replicator. PostgreSQL 15 or newer is recommended for column and row publication filters, and PostgreSQL 16 or newer is required when replication connects to a physical read replica. ## Build the binary [#build-the-binary] Clone the repository and build only the destination features you need. BigQuery is the stable, recommended default: ```bash title="Terminal" git clone https://github.com/supabase/etl.git cd etl cargo build --release -p etl-replicator --no-default-features --features bigquery ``` The executable is written to `target/release/etl-replicator`. Available destination features are `bigquery`, `clickhouse`, `ducklake`, `snowflake`, and the currently deprecated `iceberg` implementation. BigQuery is the stable, recommended default; review [Destinations](https://supabase.github.io/etl/reference/destinations.md) before choosing a module. ## Create the configuration [#create-the-configuration] The replicator reads a required `base.yaml` plus an environment-specific file from a configuration directory. Keep that directory outside the repository so credentials cannot be committed accidentally. Create `/absolute/path/to/etl-config/base.yaml` with the BigQuery destination and Postgres source settings: ```yaml title="base.yaml" destination: big_query: project_id: example-gcp-project dataset_id: etl_dataset service_account_key: "{}" max_staleness_mins: null pipeline: id: 1 publication_name: etl_publication pg_connection: host: 127.0.0.1 hostaddr: null port: 5432 name: example_database username: etl_user password: null tls: enabled: false trusted_root_certs: "" ``` Create `/absolute/path/to/etl-config/prod.yaml` for production overrides. It may be empty when `base.yaml` contains the complete configuration: ```yaml title="prod.yaml" {} ``` The default environment is `prod`. Set `APP_ENVIRONMENT` to `dev`, `staging`, or `prod` to load the corresponding file instead. Configuration values can be overridden with `APP_`-prefixed environment variables; use `__` between nested keys. For example: ```bash title="Terminal" export APP_PIPELINE__PG_CONNECTION__PASSWORD='placeholder-password' export APP_DESTINATION__BIG_QUERY__SERVICE_ACCOUNT_KEY='placeholder-service-account-json' ``` The credential values above and the `{}` service-account key in `base.yaml` are intentionally invalid placeholders. Supply the complete service-account JSON through your runtime's secret manager or protected environment rather than tracked files, shell history, logs, or process arguments. Enable TLS and provide trusted root certificates for networked production connections. ## Run the replicator [#run-the-replicator] `APP_CONFIG_DIR` must be an absolute path: ```bash title="Terminal" APP_CONFIG_DIR=/absolute/path/to/etl-config \ APP_ENVIRONMENT=prod \ ./target/release/etl-replicator ``` The process runs until it receives a shutdown signal or encounters a terminal pipeline error. Run it under a service manager that restarts failed processes, preserves logs, and provides resource limits. ## Recovery and operations [#recovery-and-operations] Validate destination behavior, resource limits, and recovery procedures in a non-production environment before operating the replicator on production data. * ETL provides at-least-once delivery. Destination writes must tolerate retries without producing incorrect duplicate state. * Pipeline state and checkpoints are stored in Postgres. When replication uses a read-only physical replica, configure a separate writable `store_pg_connection`. * The default `invalidated_slot_behavior` is `error`, which stops startup and requires operator intervention. Set it to `recreate` only when automatically resetting table states and repeating initial copies is acceptable. * Monitor PostgreSQL replication-slot WAL retention, destination capacity, process memory, replication lag, and terminal errors. * Back up configuration metadata, but never include credentials in diagnostic bundles or public issues. See [Architecture](https://supabase.github.io/etl/explanation/architecture.md) for the runtime model and [Extension Points](https://supabase.github.io/etl/explanation/traits.md) if you need a custom destination or state store. --- # Destinations > Official built-in destinations, maturity, requirements, and limitations. - Canonical HTML: https://supabase.github.io/etl/reference/destinations/ - Agent-readable Markdown: https://supabase.github.io/etl/reference/destinations.md - Source: https://github.com/supabase/etl/blob/main/site/content/docs/reference/destinations.mdx Supabase ETL ships official destination implementations in the `etl-destinations` crate. Enable only the destination feature you need when embedding ETL or building the standalone replicator. **BigQuery is the stable, recommended default.** | Feature | Destination | Status | Guidance | | ------------ | --------------- | ----------- | ------------------------------ | | `bigquery` | Google BigQuery | Stable | Recommended default | | `clickhouse` | ClickHouse | In progress | Evaluate for your workload | | `ducklake` | DuckLake | In progress | Evaluate for your workload | | `snowflake` | Snowflake | In progress | Evaluate for your workload | | `iceberg` | Apache Iceberg | Deprecated | Do not use for new deployments | ## Status definitions [#status-definitions] * **Stable:** The most mature built-in destination and the recommended default. Review its limitations and validate it against your production workload. * **In progress:** Functional, but behavior, configuration, schema support, and operational requirements may still change. Test recovery and schema changes before production use. * **Deprecated:** Retained for compatibility but no longer recommended for new deployments. Plan to move away from it. ## BigQuery [#bigquery] **Status: Stable** BigQuery uses the BigQuery Storage Write API and BigQuery change data capture to maintain destination tables. The standalone replicator guide uses BigQuery as its default example. ### Limitations [#limitations] * Every replicated source table needs a primary key, and all primary-key columns must be included in the publication. * PostgreSQL arrays containing `NULL` elements are not supported. * Supported schema changes are applied automatically, but PostgreSQL default expressions and backfill behavior are only supported where they map safely to BigQuery. See [Schema Changes](https://supabase.github.io/etl/explanation/schema-changes.md). ## ClickHouse [#clickhouse] **Status: In progress** ClickHouse supports a current-state layout with `ReplacingMergeTree` and an append-only event-log layout with `MergeTree`. ### Limitations [#limitations-1] * The default `ReplacingMergeTree` layout requires a source primary key with every primary-key column included in the publication, and requires ClickHouse 23.5 or newer. * `MergeTree` preserves an append-only event log; it does not expose a current-state replica by itself. * Tombstone cleanup and `OPTIMIZE ... FINAL CLEANUP` are operator-managed. ## DuckLake [#ducklake] **Status: In progress** DuckLake writes through DuckDB to a file or PostgreSQL catalog and local or object storage. ### Limitations [#limitations-2] * Data storage URLs currently support `file`, `s3`, and `gs` schemes. * Deployments must account for the required DuckDB extensions and the catalog, storage, and maintenance services they configure. * Primary-key sorting requires source tables to have a primary key; other sorting modes do not add that requirement. ## Snowflake [#snowflake] **Status: In progress** Snowflake uses direct Snowpipe Streaming and key-pair authentication. ### Limitations [#limitations-3] * Setup requires a Snowflake user and role with the documented warehouse, database, and schema privileges. * Schema evolution support is still in progress. Only defaults that ETL can translate safely are applied; unsupported expressions are skipped with a warning. * Validate channel recovery, committed offsets, and account-specific resource limits before production use. ## Iceberg [#iceberg] **Status: Deprecated** The Apache Iceberg implementation is retained for compatibility and is not recommended for new deployments. ### Limitations [#limitations-4] * Schema-change DDL is not supported. A newer relation schema is rejected. * The destination is deprecated and is not receiving the same product investment as the active destination implementations. ## Custom destinations and upstream support [#custom-destinations-and-upstream-support] Adding an official destination is a long-term maintenance commitment. It requires durable and idempotent writes, schema-evolution behavior, restart and recovery coverage, integration infrastructure, credential handling, ongoing dependency updates, and operational support. For that reason, maintainers are careful about accepting new destination implementations upstream. Open an issue or discussion before investing in an upstream implementation; acceptance is not guaranteed. You do not need to contribute a destination upstream to use ETL. Implement the [`Destination`](https://supabase.github.io/etl/explanation/traits.md#destination) trait in your own Rust project, wire it into `Pipeline::new`, and maintain it alongside your application. See [Custom Implementations](https://supabase.github.io/etl/guides/custom-implementations.md) for a complete example.