Implementation:Risingwavelabs Risingwave JDBCSinkConfig

Property	Value
Component	risingwave-sink-jdbc
Language	Java
Package	com.risingwave.connector
Extends	CommonSinkConfig
Lines	122
Source	JDBCSinkConfig.java

Overview

JDBCSinkConfig is the configuration class for the JDBC sink connector in RisingWave. It extends CommonSinkConfig and holds all connection and behavioral parameters needed to establish a JDBC connection to an external database and control sink behavior. The class uses Jackson annotations for JSON deserialization from the sink's WITH clause properties.

The configuration supports multiple database backends (MySQL, PostgreSQL, Redshift, Snowflake, SQL Server) through a common set of JDBC properties. It determines whether the sink operates in upsert or append-only mode based on the type property, and provides a factory method for creating JDBC connections with sensible defaults.

Code Reference

Source Location

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCSinkConfig.java

Signature

@JsonIgnoreProperties(ignoreUnknown = true)
public class JDBCSinkConfig extends CommonSinkConfig {
    @JsonCreator
    public JDBCSinkConfig(
            @JsonProperty(value = "jdbc.url") String jdbcUrl,
            @JsonProperty(value = "table.name") String tableName,
            @JsonProperty(value = "type") String sinkType)
}

Imports

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.risingwave.connector.api.sink.CommonSinkConfig;
import java.sql.Connection;
import java.sql.SQLException;

I/O Contract

Input

The class is deserialized from a Map<String, String> of table properties provided via the SQL WITH clause. The following properties are recognized:

Property Key	Field	Type	Default	Description
jdbc.url	jdbcUrl	String	(required)	JDBC connection URL (e.g., jdbc:postgresql://host:5432/db)
table.name	tableName	String	(required)	Target table name in the downstream database
type	sinkType	String	(required)	Sink type: "upsert" or "append-only"
user	user	String	null	Database authentication username
password	password	String	null	Database authentication password
schema.name	schemaName	String	null	Target schema name (for schema-qualified table references)
jdbc.query.timeout	queryTimeoutSeconds	int	60	Query timeout in seconds
jdbc.auto.commit	autoCommit	boolean	false	Whether to enable auto-commit on the JDBC connection
database.name	databaseName	String	null	Target database name
batch.insert.rows	batchInsertRows	int	0	Batch insert row count (only applicable for Redshift BatchAppendOnlyJDBCSink)

Output

• getConnection(): Returns a java.sql.Connection configured with auto-commit disabled by default. Subclasses (e.g., SnowflakeJDBCSinkConfig) can override this method for specialized connection logic.
• isUpsertSink(): Returns true if the sink type is "upsert" (case-insensitive comparison).

Usage Examples

SQL CREATE SINK Statement

CREATE SINK my_sink FROM my_mv WITH (
    connector = 'jdbc',
    jdbc.url = 'jdbc:postgresql://localhost:5432/mydb',
    user = 'admin',
    password = 'secret',
    table.name = 'target_table',
    schema.name = 'public',
    type = 'upsert'
);

Programmatic Usage

ObjectMapper mapper = new ObjectMapper();
JDBCSinkConfig config = mapper.convertValue(tableProperties, JDBCSinkConfig.class);

// Obtain a JDBC connection
Connection conn = config.getConnection();

// Check sink mode
if (config.isUpsertSink()) {
    // upsert logic
}

Related Pages

• Risingwavelabs_Risingwave_JDBCSinkFactory -- Factory that creates JDBC sink writers using this configuration
• Risingwavelabs_Risingwave_JdbcUtils -- Utility class that provides the default connection logic called by getConnection()
• Risingwavelabs_Risingwave_SnowflakeJDBCSinkConfig -- Subclass providing Snowflake-specific authentication (key-pair)
• Risingwavelabs_Risingwave_JdbcDialect_Interface -- Interface for database-specific SQL dialect generation
• Risingwavelabs_Risingwave_BatchAppendOnlyJDBCSink -- Batch append-only sink used for Redshift and Snowflake