关于 Openflow Connector for MySQL¶

如何复制表¶

表的复制分为以下几个阶段：

1. 架构自检：连接器会发现源表中的列、包括列名和数据类型，然后根据 Snowflake 和连接器的 限制 来验证它们。验证失败会导致此阶段失败，并且周期完成。如果成功完成该阶段，连接器将创建一个空目标表。

2. 快照加载：连接器将源表中的所有可用数据复制到目标表中。如果此阶段失败，则不会再复制任何数据。成功完成后，源表中的数据将在目标表中可用。

3. 增量加载连接器跟踪源表中的更改并将这些更改应用到目标表。此过程将一直持续，直到从复制中移除该表。如果该阶段失败，将永久停止源表的复制，直到问题解决。

   备注

   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.

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.