SQL Server CDC (Public Preview) - Realtime Compute for Apache Flink

The SQL Server CDC connector reads snapshot and incremental data from a SQL Server database.

Background information

The connector supports the following features.

Category	Details
Supported type	Source table
Run mode	Only stream mode is supported.
Data format	Not applicable
Connector-specific monitoring metrics	Connector-specific monitoring metrics currentFetchEventTimeLag: The delay between event generation and when the source operator fetches it. currentEmitEventTimeLag: The delay between event generation and when the source operator emits it. sourceIdleTime: The duration the source has been idle without producing data. Note These metrics are 0 during the full phase. For more information about these metrics, see Monitoring Metrics.
API type	SQL
Supports updates or deletions in the result table	Not applicable

Prerequisites

Before using the SQL Server CDC connector, you must enable the Change Data Capture (CDC) feature for the target database and tables.

To enable CDC for a table, run the following SQL statements:

RDS SQL Server

EXEC sp_rds_cdc_enable_db;       -- Enable CDC for the database
EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',         -- Specify the schema of the source table
@source_name = N'MyTable',       -- Specify the name of the table to capture
@role_name = N'MyRole',          -- Specify the role MyRole
@supports_net_changes = 1;

Self-managed SQL Server

EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',        -- Specify the schema of the source table
@source_name   = N'MyTable',    -- Specify the name of the table to capture
@role_name     = N'MyRole',     -- Specify the role MyRole. You can add users who require SELECT 
                                -- permissions on the captured columns to this role.
                                -- Members of the sysadmin or db_owner roles can also
                                -- access the specified change tables. If this parameter is NULL, 
                                -- only members of these roles can access the capture data.
@filegroup_name = N'MyDB_CT',   -- Specify the filegroup where SQL Server will place the change
                                -- tables. This filegroup must already exist.
                                -- We recommend placing change tables in a different filegroup from the source tables.
@supports_net_changes = 0;

To verify access permissions for the CDC tables, run the following SQL statement:

EXEC sys.sp_cdc_help_change_data_capture;

This query returns the configuration for each CDC-enabled table in the database. If a table is not listed, verify that CDC is enabled for it and that the user has the required access permissions.

Limitations

This connector is available only for VVR 11.7 and later.
The source database must have CDC enabled.

Because the connector uses the Change Data Capture feature, it has the following SQL Server version requirements:

For Standard Edition, SQL Server 2016 SP1 or later is required.
For Enterprise Edition, SQL Server 2012 or later is required.

SQL

Syntax

CREATE TABLE sqlserver_cdc_source (
    id INT,
    order_date DATE,
    purchaser INT,
    quantity INT,
    product_id INT,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname'  = '<hostname>',
  'port'      = '<port>',
  'username'  = '<username>',
  'password'  = '<password>',
  'database-name' = '<database name>',
  'table-name'    = 'dbo.orders'
);

WITH parameters

Parameter

Description

Type

Required

Default

Remarks

connector

The connector type.

STRING

Yes

None

The value is fixed at sqlserver-cdc.

hostname

The IP address or hostname of the SQL Server database.

STRING

Yes

None

None.

username

The username to connect to the SQL Server database.

STRING

Yes

None

None.

password

The password for the specified username.

STRING

Yes

None

None.

database-name

The name of the database to monitor.

STRING

Yes

None

None.

table-name

The names of the tables to capture.

STRING

Yes

None

To match multiple tables, you can use , to separate multiple <schemaName>.<tableName> entries, and regular expressions are also supported. For example:

Value	Description
'table-name'='s1.t1,s2.t2'	Includes tables s1.t1 and s2.t2.
'table-name'='s1\..,s2\..'	Includes all tables under schemas s1 and s2.

scan.startup.mode

The startup mode for data consumption.

STRING

initial

Valid values:

initial (default): Performs a full snapshot of existing data, then reads the latest change logs.
latest-offset: Skips the snapshot and reads only changes that occurred after the connector started.

Important

The scan.startup.mode option depends on the Debezium snapshot.mode configuration. Do not use both options at the same time. If both are specified in the table DDL, scan.startup.mode may not take effect.

port

The port number of the SQL Server database service.

INTEGER

1433

None.

server-time-zone

The session time zone of the database server, such as "Asia/Shanghai"

STRING

None

None.

scan.incremental.snapshot.enabled

Enables or disables the incremental snapshot.

BOOLEAN

true

Valid values:

false: Disables the incremental snapshot.
true (default): Enables the incremental snapshot.

Note

When the incremental snapshot framework is disabled (scan.incremental.snapshot.enabled = false), checkpoints cannot be performed during the full phase. If a table contains a large amount of data, a checkpoint timeout causes the job to fail and restart. We recommend that you enable the incremental snapshot framework for large tables.
For more information about the benefits, prerequisites, and limitations of incremental snapshots, see Features, Prerequisites, and Limitations.

chunk-meta.group.size

The group size for chunk metadata. If the size is exceeded, the metadata is split into multiple groups.

INTEGER

1000

None.

chunk-key.even-distribution.factor.lower-bound

The lower bound of the distribution factor for the chunk key.

DOUBLE

0.05

The distribution factor is used to determine whether table data is uniformly distributed. A uniform computation optimization is used for a uniform distribution, while a non-uniform distribution triggers a split query. The formula is: (MAX(id) - MIN(id) + 1) / rowCount.

chunk-key.even-distribution.factor.upper-bound

The upper bound of the distribution factor for the chunk key.

DOUBLE

1000.0

The distribution factor is used to determine whether table data is evenly distributed. If the data is evenly distributed, uniform calculation optimization is used. Otherwise, a split query is triggered. The calculation formula is: (MAX(id) - MIN(id) + 1) / rowCount.

scan.incremental.close-idle-reader.enabled

Specifies whether to close idle readers after the snapshot is complete.

BOOLEAN

false

For this configuration to take effect, you must set execution.checkpointing.checkpoints-after-tasks-finish.enabled to true.

scan.incremental.snapshot.chunk.key-column

The column to use as a key for splitting chunks during the snapshot phase.

STRING

None

The column used to split the table into chunks for the snapshot read. This column must be part of the primary key. Defaults to the first primary key column.

scan.incremental.snapshot.unbounded-chunk-first.enabled

Prioritizes processing the largest, unbounded chunk first. This reduces the risk of TaskManager OutOfMemory (OOM) errors.

BOOLEAN

true

None.

scan.incremental.snapshot.backfill.skip

Specifies whether to skip reading the transaction log during the full phase.

Boolean

false

Valid values:

true: Skips the log reading.

Recommended if downstream operators are idempotent.

Be aware that this only guarantees at-least-once semantics.
false (default): Does not skip the log reading.

During the full phase, when reading chunks, the transaction log between the low watermark and high watermark is read to ensure consistency.

If your SQL job involves aggregations or joins, we recommend not skipping the log reading during the full phase.

debezium.*

Debezium property parameters.

STRING

None

Use these options for finer-grained control over the Debezium client's behavior. For more information, see Configuration Properties.

Use with caution. Incorrect settings can prevent the connector from capturing data correctly.

Type mapping

The following table lists the data type mappings between SQL Server and Flink.

SQL Server type	Flink type
bit	BOOLEAN
tinyint	SMALLINT
smallint	SMALLINT
int	INT
bigint	BIGINT
float	DOUBLE
real	DOUBLE
numeric	DECIMAL(p,s)
decimal(p,s)
money
smallmoney
date	DATE
time(n)	TIME(n)
datetime2	TIMESTAMP(n)
datetime
smalldatetime
datetimeoffset	TIMESTAMP_LTZ(3)
char(n)	CHAR(n)
varchar(n)	VARCHAR(n)
nvarchar(n)
nchar(n)
text	STRING
ntext
xml

Additional information

Metadata

Metadata in the following formats can be exposed as read-only (VIRTUAL) columns in a table definition:

Key	Type	Description
database_name	STRING NOT NULL	The name of the database where the row originated.
schema_name	STRING NOT NULL	The name of the schema where the row originated.
table_name	STRING NOT NULL	The name of the table where the row originated.
op_ts	TIMESTAMP_LTZ(3) NOT NULL	The time the change occurred in the database. This value is 0 for records from a table snapshot instead of the transaction log.

The following example creates a table that uses metadata fields:

CREATE TABLE products (
    table_name    STRING METADATA FROM 'table_name' VIRTUAL,
    schema_name   STRING METADATA FROM 'schema_name' VIRTUAL,
    db_name       STRING METADATA FROM 'database_name' VIRTUAL,
    operation_ts  TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    id            INT NOT NULL,
    name          STRING,
    description   STRING,
    weight        DECIMAL(10, 3)
) WITH (
    'connector'     = 'sqlserver-cdc',
    'hostname'      = 'localhost',
    'port'          = '1433',
    'username'      = 'sa',
    'password'      = 'Password!',
    'database-name' = 'inventory',
    'table-name'    = 'dbo.products'
);