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.Connectionconfigured with auto-commit disabled by default. Subclasses (e.g.,SnowflakeJDBCSinkConfig) can override this method for specialized connection logic. - isUpsertSink(): Returns
trueif 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