Set up the Openflow Connector for SQL Server (CDC)¶

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

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 (CDC).

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 Data Capture on the databases and tables that you plan to replicate:

   Copy code

   USE <database>;
   EXEC sys.sp_cdc_enable_db;
   
   EXEC sys.sp_cdc_enable_table
     @source_schema = N'<schema>',
     @source_name = N'<table>',
     @role_name = NULL;
   

   Note

   Run the sp_cdc_enable_table procedure for every table that you plan to replicate. Run sp_cdc_enable_db once per database.

   The connector requires that CDC is enabled on the databases and tables before replication starts. You can also enable CDC on additional tables while the connector is running.

   Note

   Platform-specific variants for enabling CDC at the database level. The sp_cdc_enable_table call shown above is the same on every platform; only the database-level enable procedure differs.

   • AWS RDS for SQL Server. You can’t call sys.sp_cdc_enable_db directly on RDS because RDS doesn’t expose the sysadmin server role. Use the RDS-provided wrapper instead:

     Copy code

     EXEC msdb.dbo.rds_cdc_enable_db '<database>';
     

     See Using change data capture for Amazon RDS for SQL Server. CDC isn’t supported on the Web edition of RDS for SQL Server.

   • Google Cloud SQL for SQL Server. You can’t call sys.sp_cdc_enable_db directly. Use the Cloud SQL-provided wrapper instead:

     Copy code

     EXEC msdb.dbo.gcloudsql_cdc_enable_db '<database>';
     

     See Enable change data capture (CDC) on Cloud SQL for SQL Server. Cloud SQL for SQL Server currently offers SQL Server 2017, 2019, and 2022 only.

   • Azure SQL Database (single database). Use the standard sys.sp_cdc_enable_db procedure. On the DTU-based purchasing model, CDC requires the S3 service tier or higher (CDC isn’t supported on Basic, S0, S1, or S2). On the vCore-based purchasing model, CDC is supported on any tier. See Change data capture with Azure SQL Database.

   • Azure SQL Managed Instance. Use the standard sys.sp_cdc_enable_db procedure. Enabling CDC requires membership in the sysadmin server role.

   Raise max text repl size for large LOB columns

   If replicated tables contain LOB columns (such as VARCHAR(MAX), NVARCHAR(MAX), or VARBINARY(MAX)) with values larger than 64 KB, raise the SQL Server max text repl size setting on the source instance. SQL Server CDC defaults this setting to 65536 bytes (64 KB), which is lower than the connector’s 16 MB per-value limit. Without raising it, replication can fail with an error such as the following:

   Length of LOB data (N) to be replicated exceeds configured maximum 65536. Use the stored procedure sp_configure to increase the configured maximum value for max text repl size option.

   Set the value based on the largest LOB sizes in your source data. It must be at least as large as the biggest value SQL Server CDC needs to replicate, including when Oversized Value Strategy is set to Set Null (the connector still reads the full value before replacing it with NULL).

   Warning

   Raising max text repl size allows SQL Server CDC to capture larger LOB values, but those values are written to the transaction log and copied into CDC change tables. Capturing very large values can increase transaction log generation, storage consumption, and latency in CDC capture and cleanup jobs. Set the limit to match the LOB sizes you actually need rather than the maximum, unless you require that headroom.

   How you change this setting depends on your platform:

   • On-premises SQL Server, Azure SQL Managed Instance, and Google Cloud SQL for SQL Server. Use sp_configure. The example below sets the maximum allowed value of 2147483647 (~2 GB):

     Copy code

     EXEC sp_configure 'show advanced options', 1;
     RECONFIGURE;
     EXEC sp_configure 'max text repl size', 2147483647;
     RECONFIGURE;
     

     For more information, see Configure the max text repl size server configuration option.

   • AWS RDS for SQL Server. You can’t change this setting with sp_configure or msdb.dbo.rds_set_configuration. Configure it through an RDS DB parameter group instead:

     1. Create a custom DB parameter group for your SQL Server version family (for example, sqlserver-se-16.0):

        Copy code

        aws rds create-db-parameter-group \
          --db-parameter-group-name <name> \
          --db-parameter-group-family sqlserver-se-16.0 \
          --description "Custom SQL Server params"
        

     2. Set max text repl size (b) (note the exact parameter name, including (b) in lowercase). The example below uses 2147483647 (~2 GB), the maximum allowed value:

        Copy code

        aws rds modify-db-parameter-group \
          --db-parameter-group-name <name> \
          --parameters "ParameterName='max text repl size (b)',ParameterValue=2147483647,ApplyMethod=immediate"
        

     3. Attach the parameter group to the RDS instance:

        Copy code

        aws rds modify-db-instance \
          --db-instance-identifier <instance-id> \
          --db-parameter-group-name <name> \
          --apply-immediately
        

     4. Reboot the RDS instance. This parameter requires a reboot to take effect on RDS for SQL Server.

   • Azure SQL Database (single database). Open a query window connected to the specific database and run:

     Copy code

     EXEC sp_configure 'max text repl size', 2147483647;
     RECONFIGURE;
     

     A value of -1 is also supported and removes the size limit other than the limit imposed by the column data type.

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 required permissions to the user for each database that you are replicating.

   Add the user to the db_datareader role and grant SELECT on the cdc schema so the connector can read both the source tables and the CDC change tables:

   Copy code

   ALTER ROLE db_datareader ADD MEMBER <user_name>;
   GRANT SELECT ON SCHEMA::cdc TO <user_name>;
   

   Run these commands in each database that you plan to replicate.

   Note

   These permissions give the connector read access to every user table in the database. To scope access more tightly, grant SELECT only on the specific tables being replicated and on SCHEMA::cdc instead of adding the user to the db_datareader role.

   Note

   Azure SQL Database (single database) only — database owner before deploying wrapper scripts. The wrapper procedures use EXECUTE AS OWNER. If the database owner is a Microsoft Entra ID principal (common after importing a database from a .bacpac file) and the connector authenticates with SQL Server authentication, calls to dbo.sf_openflow_cdc_enable_table and dbo.sf_openflow_cdc_disable_table fail with an error such as Only active directory users can impersonate other active directory users (error 33171). The connector does not receive extra privileges from this step; it only changes who owns the database.

   Before deploying the Openflow CDC wrapper scripts in step 5, connect to each replicated database as the server administrator and run:

   Copy code

   ALTER AUTHORIZATION ON DATABASE::[<database>] TO [<sql_server_admin_login>];
   

   Use the SQL Server authentication login that administers the logical server (for example the login you specified when you created the server), not the connector login.

5. Deploy the Openflow CDC wrapper procedures so the connector can manage capture instances and apply source schema changes autonomously. For more information, see Deploy the Openflow CDC wrapper procedures.

6. (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.

7. (Optional) Configure SSL connection.

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

Deploy the Openflow CDC wrapper procedures¶

The connector applies supported source table schema changes (DDL) without stopping replication or requiring a manual re-snapshot. To do this, the connector manages SQL Server capture instances autonomously: when a tracked table’s schema changes, the connector creates a new capture instance for the updated schema and drops the old one after it finishes the transition. For an overview of the process, see Schema changes.

Creating and dropping capture instances normally requires db_owner. Rather than granting the connector that level of access, deploy a small set of wrapper procedures that perform these operations on the connector’s behalf and grant the connector permission to run only those two procedures.

This design has the following properties:

• The connector can run only the two wrapper procedures. For capture-instance management, the connector is granted EXECUTE on only dbo.sf_openflow_cdc_enable_table and dbo.sf_openflow_cdc_disable_table. It doesn’t hold db_owner and can’t call the underlying sys.sp_cdc_enable_table or sys.sp_cdc_disable_table procedures directly. The wrapper procedures run with EXECUTE AS OWNER, so they supply the elevated privileges only for the specific, audited operation.
• Every operation is recorded in an audit table. Each invocation of a wrapper procedure writes an attempt row to the dbo.openflow_cdc_audit table before it calls the engine, then a success or failure row (including the SQL Server error number and message on failure) after the call. The rows are append-only: the wrappers never update or delete audit rows.

Deploy the procedures as a database administrator. Run the following scripts, in order, in each CDC-enabled database being replicated. Run them as a principal that already holds db_owner.

1. openflow_cdc_audit_setup.sql: Creates the append-only dbo.openflow_cdc_audit table that the wrapper procedures write to.

   Copy code

   SET NOCOUNT ON;
   
   -- Audit log. Each wrapper invocation writes one 'attempt' row before delegating to
   -- sys.sp_cdc_*, then either a 'success' or 'failure' row sharing the same attempt_id.
   -- Rows are append-only.
   IF OBJECT_ID(N'dbo.openflow_cdc_audit', N'U') IS NULL
   BEGIN
       CREATE TABLE dbo.openflow_cdc_audit (
           audit_id         bigint           IDENTITY(1,1) NOT NULL CONSTRAINT pk_openflow_cdc_audit PRIMARY KEY,
           attempt_id       uniqueidentifier NOT NULL,
           event_time       datetime2(3)     NOT NULL,
           event_kind       varchar(16)      NOT NULL,
           action           varchar(16)      NOT NULL,
           source_schema    sysname          NOT NULL,
           source_name      sysname          NOT NULL,
           capture_instance sysname          NOT NULL,
           caller           sysname          NOT NULL,
           error_number     int              NULL,
           error_message    nvarchar(4000)   NULL,
           CONSTRAINT ck_openflow_cdc_audit_event_kind CHECK (event_kind IN ('attempt', 'success', 'failure')),
           CONSTRAINT ck_openflow_cdc_audit_action     CHECK (action     IN ('enable',  'disable'))
       );
   
       CREATE INDEX ix_openflow_cdc_audit_attempt    ON dbo.openflow_cdc_audit(attempt_id);
       CREATE INDEX ix_openflow_cdc_audit_event_time ON dbo.openflow_cdc_audit(event_time);
   END;
   GO
   

2. sf_openflow_cdc_enable_table.sql: Creates the wrapper procedure that the connector calls to add a new capture instance during a schema transition.

   Copy code

   CREATE OR ALTER PROCEDURE dbo.sf_openflow_cdc_enable_table
       @source_schema    sysname,
       @source_name      sysname,
       @capture_instance sysname
   WITH EXECUTE AS OWNER
   AS
   BEGIN
       SET NOCOUNT ON;
   
       -- 1. Table must exist and not be a system table.
       IF OBJECT_ID(QUOTENAME(@source_schema) + N'.' + QUOTENAME(@source_name), N'U') IS NULL
           THROW 50001, 'Source table does not exist or is not a user table.', 1;
   
       -- 2. The connector supplies the full capture instance name. Validate that
       --    it is present and fits within the 100-character limit imposed by CDC.
       IF @capture_instance IS NULL OR LEN(@capture_instance) = 0
           THROW 50002, 'Capture instance name must be provided.', 1;
   
       IF LEN(@capture_instance) > 100
           THROW 50003, 'Capture instance name exceeds the 100-character limit.', 1;
   
       -- 3. Record the attempt.
       DECLARE @attempt_id uniqueidentifier = NEWID();
       DECLARE @caller     sysname          = ORIGINAL_LOGIN();
   
       INSERT INTO dbo.openflow_cdc_audit
           (attempt_id, event_time, event_kind, action, source_schema, source_name, capture_instance, caller)
       VALUES
           (@attempt_id, SYSUTCDATETIME(), 'attempt', 'enable',
            @source_schema, @source_name, @capture_instance, @caller);
   
       -- 4. Delegate to the engine procedure and record the outcome.
       BEGIN TRY
           EXEC sys.sp_cdc_enable_table
               @source_schema    = @source_schema,
               @source_name      = @source_name,
               @capture_instance = @capture_instance,
               @role_name        = NULL;
   
           INSERT INTO dbo.openflow_cdc_audit
               (attempt_id, event_time, event_kind, action, source_schema, source_name, capture_instance, caller)
           VALUES
               (@attempt_id, SYSUTCDATETIME(), 'success', 'enable',
                @source_schema, @source_name, @capture_instance, @caller);
       END TRY
       BEGIN CATCH
           DECLARE @err_num int            = ERROR_NUMBER();
           DECLARE @err_msg nvarchar(4000) = ERROR_MESSAGE();
   
           INSERT INTO dbo.openflow_cdc_audit
               (attempt_id, event_time, event_kind, action, source_schema, source_name,
                capture_instance, caller, error_number, error_message)
           VALUES
               (@attempt_id, SYSUTCDATETIME(), 'failure', 'enable',
                @source_schema, @source_name, @capture_instance, @caller,
                @err_num, @err_msg);
   
           ;THROW;
       END CATCH
   END;
   GO
   

3. sf_openflow_cdc_disable_table.sql: Creates the wrapper procedure that the connector calls to drop the old capture instance after a schema transition completes.

   Copy code

   CREATE OR ALTER PROCEDURE dbo.sf_openflow_cdc_disable_table
       @source_schema    sysname,
       @source_name      sysname,
       @capture_instance sysname
   WITH EXECUTE AS OWNER
   AS
   BEGIN
       SET NOCOUNT ON;
   
       -- 1. The connector supplies the full capture instance name. Validate that it is present.
       IF @capture_instance IS NULL OR LEN(@capture_instance) = 0
           THROW 50002, 'Capture instance name must be provided.', 1;
   
       -- 2. Record the attempt.
       DECLARE @attempt_id uniqueidentifier = NEWID();
       DECLARE @caller     sysname          = ORIGINAL_LOGIN();
   
       INSERT INTO dbo.openflow_cdc_audit
           (attempt_id, event_time, event_kind, action, source_schema, source_name, capture_instance, caller)
       VALUES
           (@attempt_id, SYSUTCDATETIME(), 'attempt', 'disable',
            @source_schema, @source_name, @capture_instance, @caller);
   
       -- 3. Delegate to the engine procedure and record the outcome.
       BEGIN TRY
           EXEC sys.sp_cdc_disable_table
               @source_schema    = @source_schema,
               @source_name      = @source_name,
               @capture_instance = @capture_instance;
   
           INSERT INTO dbo.openflow_cdc_audit
               (attempt_id, event_time, event_kind, action, source_schema, source_name, capture_instance, caller)
           VALUES
               (@attempt_id, SYSUTCDATETIME(), 'success', 'disable',
                @source_schema, @source_name, @capture_instance, @caller);
       END TRY
       BEGIN CATCH
           DECLARE @err_num int            = ERROR_NUMBER();
           DECLARE @err_msg nvarchar(4000) = ERROR_MESSAGE();
   
           INSERT INTO dbo.openflow_cdc_audit
               (attempt_id, event_time, event_kind, action, source_schema, source_name,
                capture_instance, caller, error_number, error_message)
           VALUES
               (@attempt_id, SYSUTCDATETIME(), 'failure', 'disable',
                @source_schema, @source_name, @capture_instance, @caller,
                @err_num, @err_msg);
   
           ;THROW;
       END CATCH
   END;
   GO
   

4. openflow_cdc_grants.sql: Grants the connector’s database user permission to run the two wrapper procedures and to read the CDC metadata, change tables, and audit trail. Replace <user_name> with the connector’s database user created in Set up your SQL Server instance, then run the script.

   Copy code

   SET NOCOUNT ON;
   
   -- Replace <user_name> with the connector's SQL Server database user.
   DECLARE @connector sysname = N'<user_name>';
   
   -- The principal must already exist in this database.
   IF DATABASE_PRINCIPAL_ID(@connector) IS NULL
       THROW 50100,
             'Connector principal does not exist in this database. Create the user first, then re-run this script.',
             1;
   
   DECLARE @sql nvarchar(max);
   
   -- 1. Allow the connector to call the two wrapper procedures. All CDC management
   --    goes through these wrappers, so no elevated role is required.
   SET @sql = N'GRANT EXECUTE ON dbo.sf_openflow_cdc_enable_table  TO ' + QUOTENAME(@connector);
   EXEC sys.sp_executesql @sql;
   
   SET @sql = N'GRANT EXECUTE ON dbo.sf_openflow_cdc_disable_table TO ' + QUOTENAME(@connector);
   EXEC sys.sp_executesql @sql;
   
   -- 2. Allow the connector to read its own audit trail.
   SET @sql = N'GRANT SELECT ON dbo.openflow_cdc_audit TO ' + QUOTENAME(@connector);
   EXEC sys.sp_executesql @sql;
   
   -- 3. Allow the connector to enumerate capture instances and read CDC change tables.
   SET @sql = N'GRANT SELECT ON SCHEMA::cdc TO ' + QUOTENAME(@connector);
   EXEC sys.sp_executesql @sql;
   GO
   

5. Verify the deployment. Confirm that the two wrapper procedures exist and that the audit table is queryable:

   Copy code

   SELECT name FROM sys.procedures WHERE name LIKE 'sf_openflow%';
   SELECT TOP 1 1 FROM dbo.openflow_cdc_audit;
   

   The first query returns both sf_openflow_cdc_enable_table and sf_openflow_cdc_disable_table. The second query confirms that the audit table exists and is readable.

Set up your Snowflake environment¶

As a Snowflake administrator, perform the following tasks:

1. Create a destination database in Snowflake to store the replicated data:

   Copy code

   CREATE DATABASE <destination_database>;
   

2. Create a Snowflake service user:

   Copy code

   CREATE USER <openflow_user>
     TYPE = SERVICE
     COMMENT='Service user for automated access of Openflow';
   

3. Create a Snowflake role for the connector and grant the required privileges:

   Copy code

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

   Use this role to manage the connector’s access to the Snowflake database.

   To create objects in the destination database, you must grant the USAGE and CREATE SCHEMA privileges on the database to the role used to manage access.

4. Create a Snowflake warehouse for the connector and grant the required privileges:

   Copy code

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

   Snowflake recommends starting with an XSMALL warehouse size, then experimenting with size depending on the number of tables being replicated and the amount of data transferred. Large numbers of tables typically scale better with multi-cluster warehouses, rather than a larger warehouse size. For more information, see multi-cluster warehouses.

5. Set up the public and private keys for key pair authentication:

   1. Create a pair of secure keys (public and private).

   2. Store the private key for the user in a file to supply to the connector’s configuration.

   3. 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 and key-pair rotation.

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:

   • 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 SQL Server and Snowflake and starts running. However, the connector doesn’t replicate any data until 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.

SQLServer Source Parameters¶

For ApplicationIntent in the JDBC URL on Always On Availability Groups:

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

SQLServer Destination Parameters¶

SQLServer Ingestion Parameters¶

Read the source under SNAPSHOT isolation¶

During the snapshot phase, the connector reads directly from the source tables to perform the initial full copy. Under SQL Server’s default READ COMMITTED isolation level, these reads acquire shared locks that can deadlock with concurrent writes from other database clients. During incremental replication, the connector reads from dedicated CDC change tables instead of the source tables, so it doesn’t take these locks. To avoid deadlocks during the snapshot phase 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 MultiDatabaseFetchTableSnapshot processor. Only the snapshot phase takes shared locks on the source tables, so incremental replication doesn’t require the Use Snapshot Isolation property.

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 processor after you change ALLOW_SNAPSHOT_ISOLATION.

Restart table replication¶

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 CDC 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.

Specify a logical key for a table¶

The connector requires a replication key for every table it replicates. By default, the connector uses the table’s primary key, or falls back to a qualifying unique constraint or unique index if no primary key exists. For the full priority order the connector uses to choose a key, see How the connector chooses a replication key. A logical key is a user-declared replacement for the auto-detected key. Configure a logical key when:

• A table has no primary key, 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 on the table.

JSON syntax¶

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

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

The fields are:

The following rules apply:

• database, schema, and table matching is case-sensitive. Use the exact names as reported by SQL Server.
• logicalKey column names are matched case-insensitively: the connector lowercases both the configured names and the source column names before comparing them. On a case-sensitive SQL Server collation this matching is lenient. A key whose case differs from the actual column is still accepted, and two source columns that differ only by letter case are treated as the same key column. Use the exact column case to avoid ambiguity.
• An entry whose database, 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:

[
    {
        "database": "SalesDB",
        "schema": "dbo",
        "table": "audit_log",
        "logicalKey": ["event_id"]
    }
]

A composite logical key:

[
    {
        "database": "SalesDB",
        "schema": "dbo",
        "table": "order_lines",
        "logicalKey": ["order_id", "line_item_id"]
    }
]

Logical keys for several tables in one JSON value:

[
    {
        "database": "SalesDB",
        "schema": "dbo",
        "table": "audit_log",
        "logicalKey": ["event_id"]
    },
    {
        "database": "SalesDB",
        "schema": "dbo",
        "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 a column name that doesn’t exist in the source table.

When the configuration is rejected, verification surfaces a clear error and the table stays in the NEW state (never FAILED). 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 floating-point type (float, real). Floating-point comparisons can produce inconsistent results because of precision differences.
• A logical-key column is a large-object type (text, image, varbinary(max)). Using large objects as keys severely degrades MERGE performance.
• 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¶

The connector doesn’t track schema evolution of the unique or logical key columns after the CDC capture instance exists. Dropping or altering a logical-key column isn’t detected at runtime:

• If a logical-key column is dropped on the source, replication for the affected table fails. Restart table replication to recover. 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.

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.