Sink data from RisingWave to SQL Server

This guide describes how to sink data from RisingWave to Microsoft SQL Server.

This guide covers two connector options for sinking data to SQL Server:

• JDBC connector (connector='jdbc') - Recommended for better performance
• SQL Server native connector (connector='sqlserver')

You can test out this process on your own device by using the sqlserver_sink.slt demo in the e2e_test directory of the RisingWave repository.

PREMIUM FEATURE

This is a premium feature. For a comprehensive overview of all premium features and their usage, please see RisingWave premium features.

Prerequisites

Before sinking data from RisingWave to SQL Server, please ensure the following:

• The SQL Server table you want to sink to is accessible from RisingWave.
• You have an upstream materialized view or table in RisingWave that you can sink data from.
• For the SQL Server native connector, the SQL Server user must be able to inspect the target table metadata. During sink creation, RisingWave validates the target columns, data types, and primary key definition against the downstream table.
• For the SQL Server native connector, append-only sinks require INSERT permission on the target table. Upsert sinks require INSERT, UPDATE, and DELETE permissions.
• If you are running RisingWave locally from binaries and intend to use the JDBC sink connector, make sure you have JDK 11 or later versions installed in your environment.

Create a sink

RisingWave supports two connector types for SQL Server sinks. The JDBC connector is recommended for better performance.

Syntax

CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector='jdbc' | 'sqlserver',
   connector_parameter = 'value', ...
);

Parameters (JDBC connector)

Parameter Names	Description
connector	Required. Sink connector type. Use jdbc for the JDBC connector.
jdbc.url	Required. The JDBC URL of the destination database necessary for the driver to recognize and connect to the database.
table.name	Required. The SQL Server table you want to sink to.
schema.name	Optional. The SQL Server schema name. If not specified, the default schema will be used.
type	Required. Allowed values: append-only and upsert.
force_append_only	Optional. If true, forces the sink to be append-only, even if it cannot be.
primary_key	Conditional. The primary keys of the sink. Use ’,’ to delimit the primary key columns. Primary keys are required for upsert sinks.
jdbc.query.timeout	Optional. Specifies the timeout for the operations to downstream. If not set, the default is 60s.
jdbc.auto.commit	Optional. Controls whether to automatically commit transactions for JDBC sink. If not set, the default is false.
tcp.keepalive.enable	Optional. Enables TCP keepalive for the connection to the downstream database. If not set, the default is false.
tcp.keepalive.idle	Optional. The time (in seconds) the connection must be idle before the first keepalive probe is sent. Takes effect only when tcp.keepalive.enable is true. If not set, the default is 600 (10 minutes).
tcp.keepalive.interval	Optional. The time (in seconds) between consecutive keepalive probes. Takes effect only when tcp.keepalive.enable is true. If not set, the default is 10.
tcp.keepalive.count	Optional. The number of unacknowledged keepalive probes before the connection is considered dead. Takes effect only when tcp.keepalive.enable is true. If not set, the default is 3.

Parameters (SQL Server native connector)

Parameter Names	Description
connector	Required. Sink connector type. Use sqlserver for the native SQL Server connector.
type	Required. Allowed values: append-only and upsert.
force_append_only	Optional. If true, forces the sink to be append-only, even if it cannot be.
primary_key	Conditional. The primary keys of the sink. Use ’,’ to delimit the primary key columns. Primary keys are required for upsert sinks.
sqlserver.host	Required. The SQL Server host.
sqlserver.port	Required. The SQL Server port.
sqlserver.user	Required. The user for SQL Server access.
sqlserver.password	Required. The password for SQL Server access.
sqlserver.database	Required. The SQL Server database you want to sink to.
sqlserver.schema	Optional. The SQL Server schema name. If not specified, the default schema dbo is used.
sqlserver.table	Required. The SQL Server table you want to sink to.

For native SQL Server sinks, RisingWave validates the downstream table when you create the sink:

• Sink columns must exist in the target SQL Server table and use compatible data types.
• For upsert sinks, the primary_key setting must match the target table primary key columns.
• timestamptz columns are written according to the target SQL Server column type:
  • datetimeoffset stores an offset-aware timestamp.
  • datetime, datetime2, and smalldatetime store a UTC timestamp without offset.
  • bigint, int, smallint, and tinyint store microsecond epoch values.

Examples

jdbc as connector

CREATE SINK s_many_data_type from t_many_data_type_rw WITH (
  connector = 'jdbc',
  type = 'upsert',
  jdbc.url='jdbc:sqlserver://sqlserver-server:1433;databaseName=SinkTest;user=SA;password=SomeTestOnly@SA;trustServerCertificate=true',
  primary_key = 'k1,k2',
  schema.name='test_schema',
  table.name = 't_many_data_type',
);

sqlserver as connector

CREATE SINK s_many_data_type from t_many_data_type_rw WITH (
  connector = 'sqlserver',
  type = 'upsert',
  sqlserver.host = 'sqlserver-server',
  sqlserver.port = 1433,
  sqlserver.user = 'SA',
  sqlserver.password = 'SomeTestOnly@SA',
  sqlserver.database = 'SinkTest',
  sqlserver.schema = 'test_schema',
  sqlserver.table = 't_many_data_type',
  primary_key = 'k1,k2',
);

Data type mapping

The following table shows the corresponding data types between RisingWave and SQL Server that should be specified when creating a sink. For details on native RisingWave data types, see Overview of data types.

SQL Server type	RisingWave type
bit	boolean
smallint	smallint
int	integer
bigint	bigint
float(24)	real
float(53)	double
decimal	decimal
date	date
nvarchar	varchar
time	time
datetime / datetime2 / smalldatetime	timestamp without time zone
datetimeoffset	timestamp with time zone
bigint / int / smallint / tinyint	timestamp with time zone
No support	interval
No support	struct
No support	array
varbinary	bytea
No support	jsonb
No support	serial