Implementation:Risingwavelabs Risingwave JdbcDialect Interface

Property	Value
Component	risingwave-sink-jdbc
Language	Java
Package	com.risingwave.connector.jdbc
Type	Interface
Lines	119
Source	JdbcDialect.java

Overview

JdbcDialect is the core interface that defines the contract for database-specific SQL dialect implementations in the RisingWave JDBC sink connector. Inspired by the Apache Flink JDBC connector, it provides methods for generating SQL statements (INSERT, UPSERT, DELETE) and binding parameters to PreparedStatement objects in a database-specific manner.

The interface provides default implementations for getInsertIntoStatement and getDeleteStatement, which generate standard SQL that works across most databases. Database-specific dialects override the abstract methods to provide specialized upsert syntax (e.g., MySQL's ON DUPLICATE KEY UPDATE, PostgreSQL's ON CONFLICT ... DO UPDATE SET, SQL Server's MERGE) and type-specific parameter binding.

Code Reference

Source Location

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

Signature

public interface JdbcDialect {
    SchemaTableName createSchemaTableName(String schemaName, String tableName);

    String getNormalizedTableName(SchemaTableName schemaTableName);

    String quoteIdentifier(String identifier);

    Optional<String> getUpsertStatement(
            SchemaTableName schemaTableName, TableSchema tableSchema, List<String> uniqueKeyFields);

    default String getInsertIntoStatement(
            SchemaTableName schemaTableName, List<String> fieldNames);

    default String getDeleteStatement(
            SchemaTableName schemaTableName, List<String> conditionFields);

    void bindUpsertStatement(
            PreparedStatement stmt, Connection conn, TableSchema tableSchema, SinkRow row)
            throws SQLException;

    void bindInsertIntoStatement(
            PreparedStatement stmt, Connection conn, TableSchema tableSchema, SinkRow row)
            throws SQLException;

    void bindDeleteStatement(PreparedStatement stmt, TableSchema tableSchema, SinkRow row)
            throws SQLException;
}

Imports

import static java.lang.String.format;
import com.risingwave.connector.api.TableSchema;
import com.risingwave.connector.api.sink.SinkRow;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

I/O Contract

Abstract Methods

Method	Input	Output	Description
createSchemaTableName	schemaName, tableName	SchemaTableName	Creates a schema-qualified table name object
getNormalizedTableName	SchemaTableName	String	Returns the fully qualified, quoted table name for SQL statements
quoteIdentifier	String identifier	String	Wraps an identifier with database-specific quote characters
getUpsertStatement	SchemaTableName, TableSchema, List<String> uniqueKeyFields	Optional<String>	Returns the database-specific upsert SQL, or empty if unsupported
bindUpsertStatement	PreparedStatement, Connection, TableSchema, SinkRow	void (modifies PreparedStatement)	Binds row values to an upsert PreparedStatement
bindInsertIntoStatement	PreparedStatement, Connection, TableSchema, SinkRow	void (modifies PreparedStatement)	Binds row values to an INSERT PreparedStatement
bindDeleteStatement	PreparedStatement, TableSchema, SinkRow	void (modifies PreparedStatement)	Binds primary key values to a DELETE PreparedStatement

Default Methods

getInsertIntoStatement generates a standard SQL INSERT:

// Produces: INSERT INTO "schema"."table"("col1", "col2") VALUES (?, ?)
default String getInsertIntoStatement(SchemaTableName schemaTableName, List<String> fieldNames)

getDeleteStatement generates a standard SQL DELETE with a WHERE clause on condition fields:

// Produces: DELETE FROM "schema"."table" WHERE "pk1" = ? AND "pk2" = ?
default String getDeleteStatement(SchemaTableName schemaTableName, List<String> conditionFields)

Usage Examples

Generating an Upsert Statement

JdbcDialect dialect = new PostgresDialect(columnSqlTypes, pkIndices);
SchemaTableName stn = dialect.createSchemaTableName("public", "users");

Optional<String> upsertSql = dialect.getUpsertStatement(stn, tableSchema, primaryKeys);
if (upsertSql.isPresent()) {
    PreparedStatement stmt = conn.prepareStatement(upsertSql.get());
    dialect.bindUpsertStatement(stmt, conn, tableSchema, row);
    stmt.executeUpdate();
}

Generating an Insert Statement

JdbcDialect dialect = new MySqlDialect(columnSqlTypes, pkIndices);
SchemaTableName stn = dialect.createSchemaTableName(null, "orders");

String insertSql = dialect.getInsertIntoStatement(stn, fieldNames);
PreparedStatement stmt = conn.prepareStatement(insertSql);
dialect.bindInsertIntoStatement(stmt, conn, tableSchema, row);
stmt.executeUpdate();

Implementations

Implementation	Database	Identifier Quoting	Upsert Strategy
MySqlDialect	MySQL / MariaDB	Backtick (`)	INSERT ... ON DUPLICATE KEY UPDATE
PostgresDialect	PostgreSQL	Double-quote (")	INSERT ... ON CONFLICT ... DO UPDATE SET
RedShiftDialect	Amazon Redshift	Double-quote (")	Unsupported (throws exception)
SnowflakeDialect	Snowflake	Double-quote (")	Unsupported (throws exception)
SqlServerDialect	SQL Server	Square bracket ([])	MERGE ... WHEN MATCHED ... WHEN NOT MATCHED

Related Pages

• Risingwavelabs_Risingwave_JdbcUtils -- Resolves the appropriate dialect factory based on JDBC URL
• Risingwavelabs_Risingwave_JDBCSinkFactory -- Factory that creates sink writers using dialects
• Risingwavelabs_Risingwave_JDBCSinkConfig -- Configuration that determines which dialect is used