Set up the Openflow Connector for PostgreSQL¶

This topic describes the steps to set up the Openflow Connector for PostgreSQL.

For details on the incremental load process, see Incremental replication. For information about restarting replication for failed tables, see Restart table replication.

Prerequisites¶

Before you begin¶

1. Ensure that you have reviewed About Openflow Connector for PostgreSQL.

2. Ensure that you have reviewed the supported PostgreSQL versions.

3. Make sure you have an Openflow deployment and runtime for this connector. If you don’t, see Set up Openflow - Snowflake Deployments or Set up Openflow - BYOC.

   A runtime’s size is fixed when you create it, so decide on a size before you create the runtime. See Runtime sizing and packing for CDC connectors.

4. If using Snowflake deployments, ensure that you’ve reviewed configuring required domains and have granted access to the required domains for the PostgreSQL connector.

Source database setup¶

As a database administrator, perform the following tasks:

1. Configure wal_level
2. Review other server settings
3. Create a publication
4. Ensure that there is enough disk space on your PostgreSQL server for the WAL. This is because once created, a replication slot causes PostgreSQL to retain the WAL data from the position held by the replication slot, until the connector confirms and advances that position.
5. Ensure that every table enabled for replication has one of the following identity key configurations:
   • Primary key: The connector uses primary key columns as the identity key and requires the table’s REPLICA IDENTITY to be set to DEFAULT.
   • Unique index (for tables without a primary key): Use a unique index that meets the requirements in Configure replica identity for tables without a primary key as the identity key. You must run ALTER TABLE <table_name> REPLICA IDENTITY USING INDEX <index_name> before enabling replication.
6. For tables without a primary key, see Configure replica identity for tables without a primary key.
7. Create a user for the connector. The connector requires a user with the REPLICATION attribute and permissions to SELECT from every replicated table. Create that user with a password to enter into the connector’s configuration. For more information on replication security, see Security.

Configure wal_level¶

Openflow Connector for PostgreSQL requires wal_level to be set to logical.

Depending on where your PostgreSQL server is hosted, you can configure the wal_level as follows:

Review server settings¶

Aside from wal_level, review and adjust the following PostgreSQL server settings as required. Set each high enough to cover all Openflow Connector for PostgreSQL instances on the server, plus any other replication traffic on the instance.

Create a publication¶

Openflow Connector for PostgreSQL requires a publication to be created and configured in PostgreSQL before replication starts. You can create it for all, or a subset of tables, as well as for specific tables with specified columns only. Make sure that every table and column that you plan to have replicated is included in the publication. You can also modify the publication later, while the connector is running. To create and configure a publication, do the following:

1. Log in as a user with the CREATE privilege on the database and run the following query:

   • For PostgreSQL 13 and later:

     Copy code

     CREATE PUBLICATION <publication name> WITH (publish_via_partition_root = true);
     

     The additional publish_via_partition_root is needed for correct replication of partitioned tables. To learn more about ingestion of partitioned tables, see Replicate a partitioned table.

   • For PostgreSQL versions earlier than 13:

     Copy code

     CREATE PUBLICATION <publication name>;
     

2. Define tables that the database agent will be able to see using:

Copy code

ALTER PUBLICATION <publication name> ADD TABLE <table name>;

For partitioned tables, it’s enough to just add the root partition table to the publication. See Replicate a partitioned table for more details.

Important

Important for PostgreSQL 15 and later: If your PostgreSQL publication is configured to exclude any table columns, exclude those same columns in the connector replication configuration. Use the column filtering settings so you don’t ingest columns that the publication omits. Snapshot replication can still read columns that match the filter, but incremental ingestion only receives data for columns that are in the publication, so including columns in the filter that are excluded from the publication can cause those columns to be suffixed with __DELETED, or cause the table to permanently fail.

For more information, see ALTER PUBLICATION.

Configure replica identity for tables without a primary key¶

For tables without a primary key, you can use a unique index as the identity key by setting REPLICA IDENTITY USING INDEX:

ALTER TABLE <table_name> REPLICA IDENTITY USING INDEX <unique_index_name>;

The connector automatically detects this setting during schema discovery and uses the unique index columns for UPDATE and DELETE operations.

The unique index must meet all of the following requirements. PostgreSQL validates these when you run the ALTER TABLE command and rejects any index that doesn’t qualify:

To check the current REPLICA IDENTITY setting for a table, run:

SELECT n.nspname AS schema_name,
       c.relname AS table_name,
       CASE c.relreplident
           WHEN 'd' THEN 'DEFAULT'
           WHEN 'n' THEN 'NOTHING'
           WHEN 'f' THEN 'FULL'
           WHEN 'i' THEN 'USING INDEX: ' || i.relname
       END AS replica_identity
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
LEFT JOIN pg_index ix ON c.oid = ix.indrelid AND ix.indisreplident
LEFT JOIN pg_class i ON ix.indexrelid = i.oid
WHERE n.nspname = '<schema_name>'
  AND c.relname = '<table_name>'
  AND c.relkind IN ('r','p');

Replicate a partitioned table¶

The connector supports replication of partitioned tables for PostgreSQL servers with version >= 15. A PostgreSQL partitioned table will be replicated into Snowflake as a single destination table.

For example, if you have a partitioned table orders, with sub-partitions orders_2023, orders_2024, and configured the connector to ingest all tables matching orders.* pattern, then only the orders table will be replicated to Snowflake, and it will include data from all sub-partitions.

To support replication of partitioned tables, ensure that the publication created in PostgreSQL has the publish_via_partition_root option set to true.

Ingestion of partitioned tables currently has the following limitations:

• When a table is attached as a partition to a partitioned table after ingestion was started, the connector won’t fetch data that existed in the partition table before attaching.
• When a sub-partition table is detached from the partitioned table after ingestion was started, the connector won’t mark the data from this sub-partition as deleted in the root partition table.
• Truncate operation on sub-partitions won’t mark affected records as deleted.

Snowflake account setup¶

Additional setup for key-pair authentication (BYOC only)¶

Gather these before you install¶

You can stop and collect any of these later, but having them on hand first lets you install and configure the connector in one pass:

• The PostgreSQL JDBC driver .jar. You supply the driver file itself, so download it beforehand.
• A JDBC connection URL in the form jdbc:postgresql://<host>:<port>/<database>. SSL is configured in the URL itself rather than as a separate field, by appending an sslmode parameter.
• A source database user with the REPLICATION attribute and SELECT on every table you replicate, as described in Source database setup. On AWS RDS and Aurora, grant the rds_replication role instead, because those services don’t expose the native REPLICATION attribute.
• The source database user’s password. Both generations need it.
• A publication, created as described in Create a publication. You supply its name. Only tables in the publication are replicated: a table missing from it is skipped silently, even when it matches the tables you select for replication.
• Decisions on the settings that can’t be changed later. Destination schema naming, object identifier resolution (whether PostgreSQL names are stored case-sensitively or uppercased), and table storage format are fixed once the connector has applied its configuration and begun ingesting. Changing them afterwards requires a new connector instance and a fresh snapshot.

Install the connector¶

Choose your generation¶

This connector is available in both gen 1 and gen 2.

If you’re unsure, see Openflow gen 1 and gen 2 for a full comparison.

Configure the connector¶

Using gen 1? Skip to Configure on the canvas.

Configure with the setup wizard or SQL (gen 2)¶

CREATE SECRET <openflow_database>.<openflow_schema>.<secret_name>
  TYPE = GENERIC_STRING
  SECRET_STRING = '<source_db_password>';

GRANT READ ON SECRET <openflow_database>.<openflow_schema>.<secret_name> TO ROLE <execute_as_role>;

GRANT USAGE ON DATABASE <openflow_database> TO ROLE <execute_as_role>;
GRANT USAGE ON SCHEMA <openflow_database>.<openflow_schema> TO ROLE <execute_as_role>;

With the secret in place, configure the connector with the setup wizard or, for automation, by editing the connector’s config.json with SQL and stage file operations; see Configure a gen 2 connector with SQL.

Gen 2 parameters¶

The following table lists the gen 2 connector parameters, grouped by the wizard step where you configure them. The wizard’s Step Documentation panel describes each property in full; this table gives the parameter name and its purpose so you can prepare before entering the wizard.

The gen 2 parameters are listed below, grouped by wizard step. The gen 1 parameter reference follows in the next section.

Using gen 2? Skip to Run the flow.

Configure on the canvas (gen 1)¶

To configure the connector on the canvas, do the following as a data engineer:

1. Right-click on the imported process group and select Parameters.

2. Populate the required parameter values.

   For more information on the required parameter values, see the following sections:

   • PostgreSQL Source Parameters: Used to establish a connection with PostgreSQL.
   • PostgreSQL Destination Parameters: Used to establish a connection with Snowflake.
   • PostgreSQL Ingestion Parameters: Used to specify the tables to replicate.

Start with setting the parameters of the PostgreSQL Source Parameters context, then the PostgreSQL Destination Parameters context. Once this is done, you can enable the connector, and it should connect to both PostgreSQL and Snowflake and start running. However, it won’t replicate any data until tables are explicitly added to its configuration.

To configure specific tables for replication, edit the PostgreSQL Ingestion Parameters context. Shortly after you apply the changes to the Replication Parameters context, the configuration will be picked up by the connector, and the replication lifecycle will start for every table.

To run multiple CDC connector instances on one runtime, see .

PostgreSQL Source Parameters¶

PostgreSQL Destination Parameters¶

PostgreSQL Ingestion Parameters¶

Replicate tables from a PostgreSQL replica server¶

The connector can ingest data from a primary server, a hot standby replica, or a subscriber server using logical replication. Before configuring the connector to connect to a PostgreSQL replica, ensure that replication between primary and replica nodes works correctly. When investigating issues with missing data in the connector, first ensure that missing rows are present in the replica server used by the connector.

Additional considerations when connecting to a standby replica:

• The PostgreSQL version of the server must be >= 16. Amazon Aurora is not supported because it doesn’t offer logical decoding from read replicas.
• Only connecting to a hot standby replica is supported. Note that warm standby replicas can’t accept connections from clients until they are promoted to a primary instance.
• The publication needed by the connector must be created on the primary server, not the standby server. The standby server is read-only and doesn’t let you create a publication.

If you connect to a hot standby instance and see Trying to create the replication slot ‘<replication slot>’ timed out. If connecting to a standby instance, ensure there is some traffic on the primary PostgreSQL instance, otherwise the call to create a replication slot will never return. error in the Openflow bulletin, or the Read PostgreSQL CDC Stream processor isn’t starting, log in to the primary PostgreSQL instance and execute the following query:

SELECT pg_log_standby_snapshot();

The error occurs when there are no data changes in the primary server. As such, the connector can stall while creating a replication slot on the replica server. This results from the replica server requiring information about running transactions from the primary server to be able to create a replication slot. Primary servers won’t send the information while idle. The pg_log_standby_snapshot() function forces the primary server to send information about running transactions to the replica server.

On PostgreSQL 17 and later, if you want the replication slot to survive a primary failover, the connector must connect to the primary rather than a standby. See PostgreSQL 17+ failover slot support.

Replicate a subset of columns in a table¶

The following describes the gen 1 Column Filter JSON parameter. In gen 2, you select columns per table in the Replication columns step of the setup wizard; see Configure with the setup wizard or SQL (gen 2) for that step.

[
    {
        "schema": "<schema>" | "schemaPattern": "<regex>",
        "table": "<table>" | "tablePattern": "<regex>",
        "included": ["<column>", "<column>"],
        "excluded": ["<column>", "<column>"],
        "includedPattern": "<regex>",
        "excludedPattern": "<regex>"
    }
]

[
    {
        "schema": "public",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

[
    {
        "schema": "public",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

[
    {
        "schema": "public",
        "table": "contacts",
        "includedPattern": ".*_email",
        "excluded": ["admin_email"]
    }
]

[
    {
        "schemaPattern": "data_.*",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

[
    {"schema": "public", "table": "orders", "included": ["account_id", "status"]},
    {"schema": "public", "table": "customers", "excludedPattern": ".*_internal"}
]

Specify a logical key for a table¶

The connector requires a replication key for every table it replicates. By default, the connector picks the replication key automatically: first a primary key, then a qualifying unique index. A logical key is a user-declared replacement for the auto-detected key. Configure a logical key when:

• A table has no primary key and no qualifying unique index, but one or more columns are unique in the data.
• A specific column or set of columns should be used as the replication key, regardless of what the connector would auto-detect (for example, to override a synthetic primary key).

A logical key takes the highest priority. When the connector finds a logical key for a table, it uses that key and ignores any primary key or unique index on the table.

REPLICA IDENTITY requirement¶

When the logical key columns don’t match the table’s primary key columns exactly (or the table has no primary key), PostgreSQL DELETE events in the WAL don’t carry the logical-key column values. The connector can’t identify which destination row to delete.

Before you configure a logical key and enable replication for the table, set:

ALTER TABLE <schema_name>.<table_name> REPLICA IDENTITY FULL;

If the table already has REPLICA IDENTITY USING INDEX and the logical key columns exactly match that index, setting FULL isn’t required.

JSON syntax¶

The Table Key Configuration JSON value is a JSON array. Each entry maps one table to its logical key columns:

[
    {
        "schema": "<schema>",
        "table": "<table>",
        "logicalKey": ["<column>", "<column>"]
    }
]

The fields are:

The following rules apply:

• schema, table, and logicalKey column matching is case-sensitive. Use the exact names as reported by PostgreSQL.
• An entry whose schema and table don’t match any replicated table is silently ignored.

Logical key configuration examples¶

A single-column logical key on a table without a primary key:

[
    {
        "schema": "sales",
        "table": "audit_log",
        "logicalKey": ["event_id"]
    }
]

A composite logical key:

[
    {
        "schema": "sales",
        "table": "order_lines",
        "logicalKey": ["order_id", "line_item_id"]
    }
]

Logical keys for several tables in one JSON value:

[
    {
        "schema": "sales",
        "table": "audit_log",
        "logicalKey": ["event_id"]
    },
    {
        "schema": "sales",
        "table": "order_lines",
        "logicalKey": ["order_id", "line_item_id"]
    }
]

Restrictions¶

The connector rejects the configuration when any of the following is true:

• logicalKey is missing, empty, or not an array.
• logicalKey contains duplicate column names.
• logicalKey contains a nullable column. Logical key columns must be defined as NOT NULL to reliably identify rows.
• logicalKey contains the PostgreSQL system column ctid. ctid isn’t a reliable replication key because it’s a physical row pointer that can change when a row is vacuumed or updated.
• logicalKey contains a column name that doesn’t exist in the source table.

When the configuration is rejected, the connector either fails to enable the controller service (for structural issues detected at enablement time) or holds the table in the NEW state (for issues detected when the table is initialized). After you fix the configuration, replication for the table resumes without resetting state.

Warnings logged for risky configurations¶

The connector accepts the following configurations but logs a warning at table initialization.

When choosing logical-key columns, prefer columns with high cardinality and, where possible, monotonically increasing values. Low-cardinality or non-monotonic keys can degrade snapshot performance.

• A logical-key column is a large-object type (bytea). Using large objects as keys severely degrades MERGE performance.
• A logical-key column is a floating-point type (float4, float8, money). Floating-point comparisons can produce inconsistent results because of precision differences.
• A logical-key column is a semi-structured type (json, jsonb). Semi-structured values may produce non-deterministic equality comparisons.
• The composite logical key includes more than five columns. Long composite keys often indicate a design issue and might degrade MERGE performance.
• The logical key overrides an existing primary key on the table. Verify that the replacement key is intentional: the connector no longer uses the primary key for MERGE operations.

If you observe data divergence after any of these warnings, run a periodic full reload to reconcile the destination with the source.

Schema changes that affect a logical key¶

Logical keys reference column names. The connector doesn’t follow renames or drops of those columns:

• If a logical-key column is dropped on the source, replication for the affected table fails. The table is marked FAILED. For more information, see Restart table replication.
• If a logical-key column is renamed on the source, the configuration still references the old name and replication fails. Update the JSON to use the new name and restart table replication.

Run the flow¶

Gen 2¶

After you apply your configuration in the wizard, the connector’s status moves to Upgrading and then to Stopped once upgrading finishes. Start it from the Installed Connectors tab: open the connector’s menu and select Start.

If the connector is still in Draft when you reach this step, its configuration hasn’t been applied. Open the setup wizard and select Apply so your changes take effect before you start.

After starting, open the connector’s observability dashboard to confirm data is moving and there are no errors.

To start, stop, or otherwise manage a gen 2 connector programmatically, see Manage the gen 2 Openflow connector lifecycle.

Gen 1¶

1. Right-click on the canvas and select Enable all Controller Services.
2. Right-click on the imported process group and select Start. The connector starts the data ingestion.

Set up alerts¶

Snowflake recommends setting up alerts so that you’re notified of ingestion errors or stalled replication without having to check the connector manually. This applies to both gen 1 and gen 2 connectors.

Openflow writes telemetry, including logs and metrics, to an event table. Build an alert on that telemetry with a scheduled query: see Monitor Openflow using telemetry data for the available telemetry and example queries, and Setting up alerts based on data in Snowflake for how to create an alert from a query.