About Openflow Connector for PostgreSQL¶

How tables are replicated¶

1. Schema introspection: The connector discovers the columns in the source table, their names, types, then validates them against Snowflake’s and the connector’s limitations. Validation failures cause this stage to fail, and the cycle completes. After successful completion of Schema Introspection, the connector creates an empty destination table.

2. Snapshot load: The connector copies all data available in the source table into the destination table. Failure of this stage finishes the cycle, and no more data is replicated. After successful completion, the whole set of data from the source table is available in the destination table.

3. Incremental load: The connector keeps tracking changes in the source table, and copying them into the destination table. This continues until the table is removed from replication. Failure at this stage permanently stops replication of the source table, until the issue is removed.

   Note

   This connector can be configured to immediately start replicating incremental changes for newly added tables, bypassing the snapshot load phase. This option is often useful when reinstalling the connector in an account where previously replicated data exists and you want to continue replication without having to re-snapshot tables.

   For details on the bypassing snapshot load and using the incremental load process, see Incremental replication.

TOASTed value support¶

The connector supports replicating tables with TOAST values (https://www.postgresql.org/docs/current/storage-toast.html) for columns of types: array, bytea, json, jsonb, text, varchar, xml.

Whenever the connector encounters a TOASTed value in the CDC stream, it substitutes a default placeholder of __previous_value_unchanged, formatted for the given column type, and stores it in the journal table. The MERGE query then accounts for placeholder values, so that the destination table always contains the last non-TOASTed value.

Out of range value support¶

The connector supports replicating tables with columns of types date, timestamp, and timestamptz that contain out-of-range values. If the connector encounters an out-of-range value in the CDC stream, it substitutes a default placeholder based on the type of the column.

Table replication status¶

Interim failures, such as connection errors, do not prevent table replication. However, permanent failures, such as unsupported data types, prevent table replication.

To troubleshoot replication issues or verify that a table has been successfully removed from the replication flow, check the Table State Store:

1. In the Openflow runtime canvas, right-click a processor group and choose Controller Services. A table listing controller services displays.

2. Locate the row labeled Table State Store, click the More button on the right side of the row, and then choose View State.

A list of tables and their current states displays. Type in the search box to filter the list by table name. The possible states are:

• NEW: The table is scheduled for replication but replication hasn’t started.

• SNAPSHOT_REPLICATION: The connector is copying existing data. This status displays until all records are stored in the destination table.

• INCREMENTAL_REPLICATION: The connector is actively replicating changes. This status displays after snapshot replication ends and continues to display indefinitely until a table is either removed from replication or replication fails.

• FAILED: Replication has permanently stopped due to an error.

Replication state for table <database_name>.<schema_name>.<table_name> changed from <old_state> to <new_state>

If a permanent failure prevents table replication, remove the table from replication. After you address the problem that caused the failure, you can add the table back to replication. For more information, see Restart table replication.

Destination table metadata columns¶

Each destination table includes the following metadata columns that track replication information:

Soft-deleted rows¶

When a row is deleted from the source table, the connector does not physically remove it from the destination table. Instead, the row is marked as deleted by setting the _SNOWFLAKE_DELETED metadata column to true.

This approach allows you to:

• Retain historical data for auditing or compliance purposes.

• Query deleted records when needed.

• Decide when and how to permanently remove data based on your requirements.

To query only active (non-deleted) rows, filter on the _SNOWFLAKE_DELETED column:

SELECT * FROM my_table WHERE _SNOWFLAKE_DELETED = FALSE;

To query deleted rows:

SELECT * FROM my_table WHERE _SNOWFLAKE_DELETED = TRUE;

Dropped columns¶

When a column is dropped from the source table, the connector does not drop the corresponding column from the destination table. Instead, the column is renamed by appending the __SNOWFLAKE_DELETED suffix to preserve historical values.

For example, if a column named EMAIL is dropped from the source table, it is renamed to EMAIL__SNOWFLAKE_DELETED in the destination table. Rows that existed before the column was dropped retain their original values, while rows added after the drop have NULL in this column.

You can still query historical values from the renamed column:

SELECT EMAIL__SNOWFLAKE_DELETED FROM my_table;

Renamed columns¶

Due to limitations in CDC (Change Data Capture) mechanisms, the connector cannot distinguish between a column being renamed and a column being dropped followed by a new column being added. As a result, when you rename a column in the source table, the connector treats this as two separate operations: dropping the original column and adding a new column with the new name.

For example, if you rename a column from A to B in the source table, the destination table will contain:

• A__SNOWFLAKE_DELETED: Contains values from before the rename. Rows added after the rename have NULL in this column.

• B: Contains values from after the rename. Rows that existed before the rename have NULL in this column.

SELECT
    COALESCE(B, A__SNOWFLAKE_DELETED) AS A_RENAMED_TO_B
FROM my_table;

SELECT
    CASE
        WHEN B IS NOT NULL THEN B
        ELSE A__SNOWFLAKE_DELETED
    END AS A_RENAMED_TO_B
FROM my_table;

CREATE VIEW my_table_unified AS
SELECT
    *,
    COALESCE(B, A__SNOWFLAKE_DELETED) AS A_RENAMED_TO_B
FROM my_table;

Journal tables¶

During incremental replication, changes from the source database are first written to journal tables before being merged into the destination tables. The connector does not automatically remove data from journal tables, as this data may be useful for auditing, debugging, or reprocessing purposes.

Journal tables are created in the same schema as their corresponding destination tables and follow this naming convention:

<TABLE_NAME>_JOURNAL_<timestamp>_<number>

Where:

• <TABLE_NAME> is the name of the destination table.

• <timestamp> is the creation timestamp in Unix epoch format (seconds since January 1, 1970), ensuring uniqueness.

• <number> starts at 1 and increments whenever the destination table schema changes, either due to schema changes in the source table or modifications to column filters.

For example, if your destination table is SALES.ORDERS, the journal table might be named SALES.ORDERS_JOURNAL_1705320000_1.