Set up the Openflow Connector for SQL Server¶

This topic describes how to set up the Openflow Connector for SQL Server.

For information on the incremental load process, see Incremental replication.

Prerequisites¶

Before setting up the connector, ensure that you have completed the following prerequisites:

1. Ensure that you have reviewed About Openflow Connector for SQL Server.

2. Ensure that you have reviewed Supported SQL Server versions.

3. Ensure that you have set up your runtime deployment. For more information, see the following topics:

   • Set up Openflow - BYOC
   • Set up Openflow - Snowflake Deployments.

4. If you use Openflow - Snowflake Deployments, ensure that you have reviewed configuring required domains and have granted access to the required domains for the SQL Server connector.

Set up your SQL Server instance¶

Before setting up the connector, perform the following tasks in your SQL Server environment:

1. Enable change tracking on the databases and tables that you plan to replicate, as shown in the following SQL Server example:

   Copy code

   ALTER DATABASE <database>
     SET CHANGE_TRACKING = ON
     (CHANGE_RETENTION = 5 DAYS, AUTO_CLEANUP = ON);
   
   ALTER TABLE <schema>.<table>
     ENABLE CHANGE_TRACKING
     WITH (TRACK_COLUMNS_UPDATED = OFF);
   

   Note

   Run the ALTER DATABASE command only once per database to enable change tracking on that database. Run the ALTER TABLE command for every table that you plan to replicate in that database.

   Important

   Keep the column tracking feature off by setting TRACK_COLUMNS_UPDATED = OFF (the default) when you enable change tracking on a table. The connector doesn’t use column-level change information, so enabling TRACK_COLUMNS_UPDATED = ON adds unnecessary storage and per-DML overhead on the source without providing any benefit to replication.

   Each table must have change tracking enabled before the connector can replicate it. Enable change tracking on every table you want to replicate before you start replication.

   Important

   Set CHANGE_RETENTION to a window that’s long enough to keep change data available for data continuity. SQL Server removes tracked changes once they’re older than the retention period, so if the connector can’t read changes before they expire, it can’t replicate them incrementally and the affected tables require a full reload.

   A longer retention window, such as 5 days, gives you time to detect and react to problems, such as a paused connector, a network outage, or a failed replication cycle, and to mitigate them without reloading entire tables. Choose a value that covers the longest interruption you expect to recover from, balanced against the additional storage that retained change data consumes in SQL Server.

2. Create a login for the SQL Server instance:

   Copy code

   CREATE LOGIN <user_name> WITH PASSWORD = '<password>';
   

   This login is used to create users for the databases you plan to replicate.

3. Create a user for each database you are replicating by running the following SQL Server command in each database:

   Copy code

   USE <source_database>;
   CREATE USER <user_name> FOR LOGIN <user_name>;
   

4. Grant the SELECT and VIEW CHANGE TRACKING permissions to the user for each database that you are replicating:

   Copy code

   GRANT SELECT ON <database>.<schema>.<table> TO <user_name>;
   GRANT VIEW CHANGE TRACKING ON <database>.<schema>.<table> TO <user_name>;
   

   Run these commands in each database for every table that you plan to replicate. These permissions must be granted to the user of each database that you created in a previous step.

5. (Optional) Grant the VIEW DEFINITION privilege on the User Defined Data Types (UDDT).

   If your tables contain columns that use User Defined Data Types (UDDT), and the UDDT is owned by a different user than the connector user, you must grant the VIEW DEFINITION permission to the connector user as shown in the following SQL Server example:

   Copy code

   GRANT VIEW DEFINITION TO <user_name>;
   

   Without this permission, columns using UDDT are silently excluded from replication.

6. (Optional) Configure SSL connection.

   If you use an SSL connection to connect to SQL Server, create the root certificate for your database server. This is required when configuring the connector.

Set up your Snowflake environment¶

For additional setup when you use key-pair authentication with BYOC:

Install the connector¶

Runtime sizing¶

For sizing guidance, including node type tiers and how to resize after creation, see Runtime sizing and packing for CDC connectors.

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:

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

Start by setting the parameters of the SQLServer Source Parameters context, then the SQLServer Destination Parameters context. After you complete this, enable the connector. The connector connects to both SQLServer and Snowflake and starts running. However, the connector doesn’t replicate any data until any tables to be replicated are explicitly added to its configuration.

To configure specific tables for replication, edit the SQLServer Ingestion Parameters context. After you apply the changes to the SQLServer Ingestion Parameters context, the configuration is picked up by the connector, and the replication lifecycle starts for every table.

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

SQLServer Source Parameters¶

For failover behavior, see Always On Availability Groups and source failover.

SQLServer Destination Parameters¶

The following destination parameter controls how the connector handles invalid rows:

SQLServer Ingestion Parameters¶

Read the source under SNAPSHOT isolation¶

The connector reads the source tables during both snapshot and incremental replication. Under SQL Server’s default READ COMMITTED isolation level, these reads acquire shared locks that can deadlock with concurrent writes from other database clients. To avoid these deadlocks without affecting the isolation level that other applications use, configure the connector to read under SNAPSHOT isolation. For background, see Source database locking behavior.

Enable SNAPSHOT isolation for the connector in two steps:

1. On each source database, allow snapshot isolation:

   Copy code

   ALTER DATABASE <database> SET ALLOW_SNAPSHOT_ISOLATION ON;
   

2. Add a dynamic property named Use Snapshot Isolation with the value true to the connector’s processors. Because the connector reads the source tables during both snapshot and incremental replication, add this dynamic property to both the MultiDatabaseFetchTableSnapshot processor and the MultiDatabaseCaptureChangeSqlServer processor.

The connector checks each source database when it starts and uses SNAPSHOT isolation only for databases that have ALLOW_SNAPSHOT_ISOLATION enabled. For a database that doesn’t have it enabled, the connector falls back to the default isolation level. Because this check runs at startup, restart the processors after you change ALLOW_SNAPSHOT_ISOLATION.

Replicate tables from a SQL Server replica server¶

The connector can ingest data from a primary server or from a subscriber server using transactional replication. Before configuring the connector to connect to a SQL Server replica, ensure that replication between the primary and replica nodes works correctly. For instructions on setting up transactional replication, see Tutorial: Configure transactional replication. When investigating issues with missing data in the connector, first ensure that missing rows and change tracking events are present in the replica server used by the connector.

To configure the connector to read from a subscriber server instead of the publisher, specify the subscriber server URL in the SQLServer Connection URL parameter.

Replicate a subset of columns in a table¶

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

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

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

[
    {
        "databasePattern": "prod_.*",
        "schema": "dbo",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

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

Replicate a partitioned table¶

The connector supports replication of partitioned tables. A SQL Server partitioned table is replicated into Snowflake as a single destination table, containing data from all partitions.

To replicate a partitioned table, ensure that change tracking is enabled on the partitioned table, as described in Set up your SQL Server instance.

For more information about how the connector handles snapshots of large partitioned tables, see Snapshot of partitioned tables.

Track data changes in tables¶

The connector replicates the current state of data from the source tables, as well as detected changes from each polling interval. 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 you remove a table from replication, then add it back, the <timestamp> value changes, and <schema_generation> starts again from 1.

The connector never drops journal tables, but uses 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. The processor named Merge Journal to Destination triggers this operation. When there are no new changes, or when no new FlowFiles are waiting in the Merge Journal to Destination queue, no merge is triggered and the warehouse is available for auto-suspension.

To limit warehouse cost and restrict merges to scheduled times, use the CRON expression in the Merge Task Schedule CRON parameter. It throttles the FlowFiles that reach the Merge Journal to Destination processor, so merges are triggered only during the specified period. The connector evaluates the schedule in the UTC time zone.

For additional information and examples, see the cron triggers tutorial in the Quartz Documentation.

Run the flow¶

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.