SQL Server Connector

SQL Server connector allows reading snapshot data and incremental data from SQL Server databases and provides end-to-end table synchronization capabilities. This document describes how to set up the SQL Server connector.

Prerequisites

• SQL Server Agent is running.
• Change Data Capture is enabled for the database and captured tables.
• The configured SQL Server user can connect to the server and read the captured tables.
• In tables, the database must be a single fixed database name; schema and table support regular expressions to match multiple objects.

Enable CDC on a database and table:

USE MyDB;
GO
EXEC sys.sp_cdc_enable_db;
GO

EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',
@source_name   = N'MyTable',
@role_name     = NULL,
@supports_net_changes = 0;
GO

Verify that the table is enabled for CDC:

USE MyDB;
GO
EXEC sys.sp_cdc_help_change_data_capture;
GO

Example

An example of the pipeline for reading data from SQL Server and sink to Doris can be defined as follows:

source:
   type: sqlserver
   name: SQL Server Source
   hostname: 127.0.0.1
   port: 1433
   username: root
   password: 123456
   # The database must be a single fixed name; schema and table support regex to match multiple tables.
   tables: inventory.dbo.\.*
   schema-change.enabled: true

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: 123456

pipeline:
   name: SQL Server to Doris Pipeline
   parallelism: 4

Connector Options

Option	Required	Default	Type	Description
hostname	required	(none)	String	IP address or hostname of the SQL Server database server.
port	optional	1433	Integer	Integer port number of the SQL Server database server.
username	required	(none)	String	Name of the SQL Server user to use when connecting to the SQL Server database server.
password	required	(none)	String	Password to use when connecting to the SQL Server database server.
tables	required	(none)	String	Table names of the SQL Server tables to monitor. The database must be a single fixed database name, while schema and table support regular expressions to match multiple objects. Database-level regular expressions and cross-database patterns are not supported. The dot (.) is treated as a delimiter for database, schema, and table names. If a dot (.) is needed in a regular expression to match any character, escape it with a backslash. Examples: inventory.dbo.\.*, inventory.dbo.user_table_[0-9]+, inventory.dbo.(app|web)_order_\.*
tables.exclude	optional	(none)	String	Table names of the SQL Server tables to exclude after applying `tables`. Schema and table support regular expressions to match multiple objects. All exclude patterns must use the same fixed database name as the `tables` option. Examples: inventory.dbo.audit_\.*, inventory.dbo.tmp_[0-9]+
schema-change.enabled	optional	true	Boolean	Whether to send schema change events so downstream sinks can synchronize table structure changes.
server-time-zone	optional	(none)	String	The session time zone in database server. If not set, ZoneId.systemDefault() is used to determine the server time zone.
scan.incremental.snapshot.chunk.key-column	optional	(none)	String	The chunk key of table snapshot. By default, the chunk key is the first column of the primary key. This column must be a primary key column.
scan.incremental.snapshot.chunk.size	optional	8096	Integer	The chunk size, in rows, of table snapshots.
scan.snapshot.fetch.size	optional	1024	Integer	The maximum fetch size per poll when reading table snapshots.
scan.startup.mode	optional	initial	String	Optional startup mode for SQL Server CDC consumer. Valid values are "initial", "latest-offset", "snapshot", and "timestamp".
scan.startup.timestamp-millis	optional	(none)	Long	Optional timestamp used when `scan.startup.mode` is "timestamp".
scan.incremental.snapshot.backfill.skip	optional	false	Boolean	Whether to skip backfill in snapshot reading phase. Skipping backfill may lead to replayed change log events with at-least-once semantics.
scan.incremental.snapshot.unbounded-chunk-first.enabled	optional	true	Boolean	Whether to assign the unbounded chunks first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.
connect.timeout	optional	30s	Duration	The maximum time that the connector should wait after trying to connect to the SQL Server database server before timing out.
connect.max-retries	optional	3	Integer	The maximum retry times for building SQL Server database server connections.
connection.pool.size	optional	20	Integer	The connection pool size.
metadata.list	optional	(none)	String	List of readable metadata from SourceRecord to be passed to downstream, split by commas. Available metadata keys are: database_name, schema_name, table_name, op_ts.
scan.incremental.close-idle-reader.enabled	optional	false	Boolean	Whether to close idle readers at the end of the snapshot phase. This feature depends on FLIP-147.
scan.newly-added-table.enabled	optional	false	Boolean	Whether to scan newly added tables. This option is only useful when starting the job from a savepoint or checkpoint.
debezium.*	optional	(none)	String	Pass-through Debezium properties to Debezium Embedded Engine.
jdbc.properties.*	optional	(none)	String	Pass custom JDBC URL properties. For example: jdbc.properties.encrypt=false.

Compatibility note: To keep the behavior of older Pipeline versions, explicitly configure scan.incremental.snapshot.backfill.skip: true and scan.incremental.snapshot.unbounded-chunk-first.enabled: false.

Available Metadata

The following metadata can be passed downstream when configured in metadata.list.

Key	DataType	Description
database_name	STRING NOT NULL	Name of the database that contains the row.
schema_name	STRING NOT NULL	Name of the schema that contains the row.
table_name	STRING NOT NULL	Name of the table that contains the row.
op_ts	TIMESTAMP_LTZ(3) NOT NULL	Time when the change was made in the database. For snapshot records, the value is always 0.

Startup Reading Position

The config option scan.startup.mode specifies the startup mode for SQL Server CDC consumer. The valid values are:

• initial: Performs an initial snapshot on the monitored tables and continues to read the latest changes.
• latest-offset: Starts from the latest change log offset.
• snapshot: Reads the snapshot only.
• timestamp: Starts from the specified timestamp in scan.startup.timestamp-millis.

Data Type Mapping

SQL Server type	CDC type	NOTE
BIT	BOOLEAN
TINYINT	SMALLINT	SQL Server TINYINT is unsigned (0-255), so it is widened to SMALLINT.
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
REAL	FLOAT
FLOAT	DOUBLE
DECIMAL(p, s) NUMERIC(p, s)	DECIMAL(p, s)	When precision is greater than 38, it falls back to DECIMAL(38, 0).
MONEY	DECIMAL(19, 4)
SMALLMONEY	DECIMAL(10, 4)
CHAR(n) NCHAR(n)	CHAR(n)	Mapped to STRING when the length is not available.
VARCHAR(n) NVARCHAR(n)	VARCHAR(n)	Mapped to STRING when the length is not available.
TEXT NTEXT	STRING
BINARY VARBINARY IMAGE	BYTES
TIMESTAMP ROWVERSION	BYTES	SQL Server TIMESTAMP/ROWVERSION is a row-version binary value, not a datetime.
DATE	DATE
TIME(p)	TIME(p)
SMALLDATETIME	TIMESTAMP(0)
DATETIME	TIMESTAMP(3)
DATETIME2(p)	TIMESTAMP(p)	Precision defaults to 7 when not specified.
DATETIMEOFFSET(p)	TIMESTAMP_LTZ(p)	Precision defaults to 7 when not specified.
UNIQUEIDENTIFIER XML SQL_VARIANT HIERARCHYID GEOMETRY GEOGRAPHY	STRING

Limitations

Single Database

All entries in tables must belong to the same database. The database must be a single fixed name, while schema and table support regular expressions to match multiple objects.