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 table replication for failed tables, see Restart table replication.

Prerequisites¶

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

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

3. Ensure that you have Set up Openflow - BYOC or Set up Openflow - Snowflake Deployments.

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

5. 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 (https://www.postgresql.org/docs/current/logical-replication-security.html).

6. As a Snowflake account administrator, perform the following tasks:

   1. Create a Snowflake user with the type as SERVICE. Create a database to store the replicated data, and set up privileges for the Snowflake user to create objects in that database by granting the USAGE and CREATE SCHEMA privileges.

      Copy code

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
        WITH
          WAREHOUSE_SIZE = 'XSMALL'
          AUTO_SUSPEND = 300
          AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

   2. Create a pair of secure keys (public and private). Store the private key for the user in a file to use while configuring the connector. Assign the public key to the Snowflake service user:

      Copy code

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      For more information, see key-pair authentication.

   3. Designate a warehouse for the connector to use. Start with the XSMALL warehouse size, then experiment with size depending on the amount of tables being replicated, and the amount of data transferred. Large numbers of tables typically scale better with multi-cluster warehouses, rather than the warehouse size.

Configure wal_level¶

Openflow Connector for PostgreSQL requires wal_level (https://www.postgresql.org/docs/current/runtime-config-wal.html#GUC-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 (https://www.postgresql.org/docs/current/logical-replication-publication.html#LOGICAL-REPLICATION-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

PostgreSQL 15 and later support configuring publications for a specified subset of table columns. For the connector to support this correctly, you must use the column filtering settings to include the same columns as set on the publication.

Without this setting, the connector will behave as follows:

• In the destination table, columns that aren’t included in the filter will be suffixed with __DELETED. All data replicated during the snapshot phase will be retained.
• After you add new columns to the publication, the table will be permanently failed, and you will need to restart its replication.

For more information, see ALTER PUBLICATION (https://www.postgresql.org/docs/current/sql-alterpublication.html).

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');

Install the connector¶

Runtime sizing¶

For migration instructions, see Reinstall the connector.

Configure the connector¶

To configure the connector, 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 both to PostgreSQL and Snowflake and start running. However, it won’t replicate any data until any 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.

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 (https://www.postgresql.org/docs/current/hot-standby.html), or subscriber server using logical replication (https://www.postgresql.org/docs/current/logical-replication.html). 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 replica server used by the connector.

Additional considerations when connecting to a standby replica:

• PostgreSQL version of the server must be >= 16. Amazon Aurora is not supported because it doesn’t offer logical decoding from read replicas (https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/AuroraPostgreSQL.Replication.Logical.html).
• Only connecting to 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.

Replicate a subset of columns in a table¶

[
    {
        "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"}
]

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 has currently 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 subpartitions won’t mark affected records as deleted.

Track data changes in tables¶

The connector replicates not only the current state of data from the source tables, but also every state of every row from every changeset. This data is stored in journal tables created in the same schema as the destination table.

The journal table names are formatted as: <source_table_name>_JOURNAL_<timestamp>_<schema_generation> where <timestamp> is the value of epoch seconds when the source table was added to replication, and <schema_generation> is an integer increasing with every schema change on the source table. As a result, source tables that undergo schema changes will have multiple journal tables.

When a table is removed from replication, then added back, the <timestamp> value will change, and <schema_generation> will start again from 1.

The connector never drops journal tables, but does make use of the latest journal for every replicated source table, only reading append-only streams on top of journals. To reclaim the storage, you can:

• Truncate all journal tables at any time.
• Drop the journal tables related to source tables that were removed from replication.
• Drop all but the latest generation journal tables for actively replicated tables.

For example, if your connector is set to actively replicate source table orders, and you have earlier removed table customers from replication, you may have the following journal tables. In this case you can drop all of them except orders_5678_2.

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

Configure scheduling of merge tasks¶

The connector uses a warehouse to merge change data capture (CDC) data into destination tables. This operation is triggered by the MergeSnowflakeJournalTable processor. If there are no new changes or if no new flow files are waiting in the MergeSnowflakeJournalTable queue, no merge is triggered and the warehouse auto-suspends.

To limit the warehouse cost and limit merges to only scheduled time, use the CRON expression in the Merge task Schedule CRON parameter. It throttles the flow files coming to the MergeSnowflakeJournalTable processor and merges are triggered only in a dedicated period of time. For more information about scheduling, see Scheduling strategy (https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#scheduling-strategy).

Stop or delete the connector¶

When stopping or removing the connector, you have to consider the replication slot (https://www.postgresql.org/docs/current/warm-standby.html#STREAMING-REPLICATION-SLOTS) that the connector uses.

The connector creates its own replication slot with a name starting with snowflake_connector_ followed by a random suffix. As the connector reads the replication stream, it advances the slot, so that PostgreSQL can trim its WAL log and free up disk space.

When the connector is paused, the slot isn’t advanced, and changes to the source database keep increasing the WAL log size. You should not keep the connector paused for extended periods of time, especially on high-traffic databases.

When the connector is removed, whether by deleting it from the Openflow canvas, or any other means, such as deleting the whole Openflow instance, the replication slot remains in place, and must be dropped manually.

If you have multiple connector instances replicating from the same PostgreSQL database, each instance will create its own uniquely-named replication slot. When dropping a replication slot manually, make sure it’s the right one. You can see which replication slot is used by a given connector instance by checking the state of the CaptureChangePostgreSQL processor.

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¶

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