Missing Postgres migrations for PostgresStore

[kimtore]

Bug report

• I confirm this is a bug with Supabase, not with my own application.
• I confirm I have searched the Docs, GitHub Discussions, and Discord.

Describe the bug

Moving on from the Getting Started example, I tried using a Postgres backend for state persistence and get the following error message:

Error: EtlError { repr: Single(ErrorPayload { kind: SourceQueryFailed, description: "Table mappings loading failed", detail: Some("Failed to load table mappings from PostgreSQL: error returned from database: relation "etl.table_mappings" does not exist") ...

A quick grep shows that there are some migrations defined in etl-replicator/migrations/20250827000000_base.sql:

-- Replication state
create table etl.replication_state (
...
);

However it is unclear to me how to actually trigger the migrations to get this up and running.

To Reproduce

# Cargo.toml
[package]
name = "extracterator"
version = "0.1.0"
edition = "2024"

[dependencies]
etl = { git = "https://github.com/supabase/etl" }
etl-destinations = { git = "https://github.com/supabase/etl", features = ["iceberg"] }
tokio = "1"
tracing = { version = "0.1.41", features = ["log"] }
tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }

// main.rs
use etl::store::both::postgres::PostgresStore;
use etl::types::PipelineId;
use etl::{
    config::{BatchConfig, PgConnectionConfig, PipelineConfig, TlsConfig},
    pipeline::Pipeline,
};
use etl_destinations::iceberg::{DestinationNamespace, IcebergClient, IcebergDestination};
use std::collections::HashMap;
use tracing_subscriber::EnvFilter;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .init();

    const ICEBERG_CATALOG_URL: &str = "http://localhost:8181/catalog";
    const WAREHOUSE_NAME: &str = "demo";
    const NAMESPACE_NAME: &str = "whatever";
    const PUBLICATION_NAME: &str = "my_publication";
    const PIPELINE_ID: PipelineId = 1;
    const POSTGRES_SOURCE_DB: &str = "extract";
    const POSTGRES_STORE_DB: &str = "etl";

    let client = IcebergClient::new_with_rest_catalog(
        ICEBERG_CATALOG_URL.to_string(),
        WAREHOUSE_NAME.to_string(),
        HashMap::new(),
    )
    .await
    .unwrap();

    let pg_store = PostgresStore::new(PIPELINE_ID, postgres_config(POSTGRES_STORE_DB));
    let namespace = DestinationNamespace::Single(NAMESPACE_NAME.to_string());
    let destination = IcebergDestination::new(client, namespace, pg_store.clone());

    let config = PipelineConfig {
        id: PIPELINE_ID,
        publication_name: PUBLICATION_NAME.into(),
        pg_connection: postgres_config(POSTGRES_SOURCE_DB),
        batch: BatchConfig {
            max_size: 1000,
            max_fill_ms: 5000,
        },
        table_error_retry_delay_ms: 10_000,
        table_error_retry_max_attempts: 5,
        max_table_sync_workers: 4,
    };

    let mut pipeline = Pipeline::new(config, pg_store, destination);
    pipeline.start().await?;
    pipeline.wait().await?;

    Ok(())
}

fn postgres_config(db_name: impl ToString) -> PgConnectionConfig {
    PgConnectionConfig {
        host: "localhost".into(),
        port: 5432,
        name: db_name.to_string(),
        username: "postgres".into(),
        password: None,
        tls: TlsConfig {
            enabled: false,
            trusted_root_certs: String::new(),
        },
    }
}

Expected behavior

etl schema containing the needed tables should be created, and the program should sync my table data.

System information

• OS: MacOS 14.8.2
• Postgres 18 as source database
• Rust 1.91

Additional context

Memory store and Iceberg catalog integration seems to work fine? I'm pretty new to Iceberg so I'm just messing around for now.

[iambriccardo]

Thanks for reporting this issue! This is mostly a problem of the docs, right now to run the state store you indeed need to run the migrations that are declared in the file you mentioned.

How would you expect the migrations to be ran? Would you like a script to apply them for you? We can definitely improve that.

[kimtore]

Thanks for acknowledging! I don't have strong opinions on the matter, but I suppose the migrations could be run from within the compiled program as a separate library call. This way, the database schema is always coupled with the library version, and the caller may configure which credentials are used for the migrations.

A script might also be useful for those cases where more granular control is needed.

[iambriccardo]

The database schema is indeed initialized automatically when running the etl-replicator:

etl/etl-replicator/src/core.rs

Line 269 in 5c4669b

migrate_state_store(&pg_connection_config).await?;

[iambriccardo]

I have just merged better docs: https://github.com/supabase/etl/blob/main/DEVELOPMENT.md

[kimtore]

Thanks! These docs are much appreciated.

On a related note, from the changeset:

Apache Iceberg – append-only log of operations (updates coming soon)

Are you planning to support data sink capabilities to Iceberg for usage in analytics workloads, as you've done for BigQuery?

[amdjachenko]

hi, I've bypassed the same issue by just copying the migration content as a string and call sqlx::raw_sql, however the current migration script is not idempotent and I had to fix it. So my point is:

• I can provide my changes to the script to make it idempotent
• it would be nice to call migration unconditionally in PostgresStore::new alternatively provide an explicit pub function for migration like pub async fn PostgresStore::migrate(source_config: PgConnectionConfig) -> EtlResult<()>

[kimtore]

it would be nice to call migration unconditionally in PostgresStore::new

Personally, I would prefer if this could be configurable, in case it is favorable that the client has restricted privileges (no CREATE TABLE or ALTER TABLE permissions, for instance).

alternatively provide an explicit pub function for migration like pub async fn PostgresStore::migrate(source_config: PgConnectionConfig) -> EtlResult<()>

Makes more sense to me. This would allow the client better control over the process, costing only one line of code.

[iambriccardo]

Thanks! These docs are much appreciated.

On a related note, from the changeset:

Apache Iceberg – append-only log of operations (updates coming soon)

Are you planning to support data sink capabilities to Iceberg for usage in analytics workloads, as you've done for BigQuery?

We are yes, the problem is that it might require a third-piece of infra. We are currently investigating merging the changelog data (the append only log) into mirror tables in the background.

[iambriccardo]

hi, I've bypassed the same issue by just copying the migration content as a string and call sqlx::raw_sql, however the current migration script is not idempotent and I had to fix it. So my point is:

• I can provide my changes to the script to make it idempotent
• it would be nice to call migration unconditionally in PostgresStore::new alternatively provide an explicit pub function for migration like pub async fn PostgresStore::migrate(source_config: PgConnectionConfig) -> EtlResult<()>

For the idempotency of migrations, we rely on sqlx to track applied migrations through its own _sqlx_migrations table.

As for having the Postgres store run the migrations on its own, that could definitely be an option, especially for people who use ETL (the library) independently. If you want, feel free to open a PR to introduce these changes and remove the migrations from etl-replicator.

Quick question: to make your migrations idempotent, did you just add if not exists?

[kimtore]

Are you planning to support data sink capabilities to Iceberg for usage in analytics workloads, as you've done for BigQuery?

We are yes, the problem is that it might require a third-piece of infra. We are currently investigating merging the changelog data (the append only log) into mirror tables in the background.

If I understand correctly, the data layer would be stored on the same infrastructure as the metadata layer. So I'm curious about what this third piece of infrastructure would consist of, specifically?

[iambriccardo]

Are you planning to support data sink capabilities to Iceberg for usage in analytics workloads, as you've done for BigQuery?

We are yes, the problem is that it might require a third-piece of infra. We are currently investigating merging the changelog data (the append only log) into mirror tables in the background.

If I understand correctly, the data layer would be stored on the same infrastructure as the metadata layer. So I'm curious about what this third piece of infrastructure would consist of, specifically?

Could you define the data and metadata layers?

[kimtore]

Could you define the data and metadata layers?

I will refer to the diagram from the official Apache Iceberg documentation: https://iceberg.apache.org/assets/external/iceberg.apache.org/assets/images/iceberg-metadata.png

[iambriccardo]

We are using the Iceberg Catalog REST API, so we are not making any assumptions on how data is stored.

For the solution we are exploring, we might have to build a separate component that is aware of the storage part and can run this incremental writing of the mirror table.

[kimtore]

For the solution we are exploring, we might have to build a separate component that is aware of the storage part and can run this incremental writing of the mirror table.

Sounds exactly like what I'm after; very much looking forward to this component! 😊

FWIW, what I need is a minimal footprint component that can perform continuous data capture from Postgres to Iceberg, running alongside workloads in Kubernetes. Supabase ETL looks promising, so I'd be glad if I could track your progress and happy to help with testing.