Implementation:Risingwavelabs Risingwave MySqlDialect

Property	Value
Component	risingwave-sink-jdbc
Language	Java
Package	com.risingwave.connector.jdbc
Implements	JdbcDialect
Lines	132
Source	MySqlDialect.java

Overview

MySqlDialect implements the JdbcDialect interface to provide MySQL-compatible SQL generation for the RisingWave JDBC sink connector. It supports both MySQL and MariaDB databases through the same dialect.

Key characteristics of this dialect:

• Identifier quoting: Uses backticks (`identifier`), which is the MySQL convention.
• Upsert strategy: Uses MySQL's INSERT ... ON DUPLICATE KEY UPDATE syntax, which updates all columns when a duplicate key is detected.
• Table name normalization: Uses only the table name without schema qualification (MySQL uses databases rather than schemas).
• Type binding: Handles DECIMAL, INTERVAL, JSONB, BYTEA, LIST (converted to comma-separated string), and general types. STRUCT type is explicitly rejected.

Code Reference

Source Location

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

Signature

public class MySqlDialect implements JdbcDialect {
    public MySqlDialect(List<Integer> columnSqlTypes, List<Integer> pkIndices);

    @Override public SchemaTableName createSchemaTableName(String schemaName, String tableName);
    @Override public String getNormalizedTableName(SchemaTableName schemaTableName);
    @Override public String quoteIdentifier(String identifier);
    @Override public Optional<String> getUpsertStatement(
            SchemaTableName schemaTableName, TableSchema tableSchema, List<String> uniqueKeyFields);
    @Override public void bindUpsertStatement(
            PreparedStatement stmt, Connection conn, TableSchema tableSchema, SinkRow row) throws SQLException;
    @Override public void bindInsertIntoStatement(
            PreparedStatement stmt, Connection conn, TableSchema tableSchema, SinkRow row) throws SQLException;
    @Override public void bindDeleteStatement(
            PreparedStatement stmt, TableSchema tableSchema, SinkRow row) throws SQLException;
}

Imports

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;
import org.apache.commons.lang3.StringUtils;

I/O Contract

Constructor Input

Parameter	Type	Description
columnSqlTypes	List<Integer>	JDBC SQL type codes for all columns
pkIndices	List<Integer>	Indices of primary key columns within the column list

The constructor derives pkColumnSqlTypes by mapping pkIndices to their corresponding SQL types, which are later used for binding DELETE statement parameters.

SQL Generation

Upsert statement produces:

INSERT INTO `table_name`(`col1`, `col2`, `col3`) VALUES (?, ?, ?)
ON DUPLICATE KEY UPDATE `col1`=VALUES(`col1`), `col2`=VALUES(`col2`), `col3`=VALUES(`col3`)

Insert statement (inherited default with backtick quoting):

INSERT INTO `table_name`(`col1`, `col2`) VALUES (?, ?)

Delete statement (inherited default with backtick quoting):

DELETE FROM `table_name` WHERE `pk1` = ? AND `pk2` = ?

Type Binding Behavior

RisingWave Type	Binding Method	Notes
DECIMAL	setBigDecimal	Direct BigDecimal binding
INTERVAL, JSONB	setObject	Generic object binding
BYTEA	setBytes	Raw byte array
LIST	setString	Converts array to comma-separated string via StringUtils.join
STRUCT	throws RuntimeException	Not supported
All others	setObject	Generic fallback

Usage Examples

// Create dialect with column types and primary key indices
MySqlDialect dialect = new MySqlDialect(columnSqlTypes, pkIndices);

// Generate upsert SQL
SchemaTableName stn = dialect.createSchemaTableName(null, "orders");
Optional<String> upsertSql = dialect.getUpsertStatement(stn, tableSchema, primaryKeys);

// Bind and execute
PreparedStatement stmt = conn.prepareStatement(upsertSql.get());
dialect.bindUpsertStatement(stmt, conn, tableSchema, row);
stmt.executeUpdate();

Related Pages

• Risingwavelabs_Risingwave_JdbcDialect_Interface -- Interface implemented by this class
• Risingwavelabs_Risingwave_JdbcUtils -- Resolves this dialect for jdbc:mysql and jdbc:mariadb URLs
• Risingwavelabs_Risingwave_PostgresDialect -- PostgreSQL dialect (uses ON CONFLICT instead of ON DUPLICATE KEY)
• Risingwavelabs_Risingwave_SqlServerDialect -- SQL Server dialect with similar type binding but different upsert syntax
• Risingwavelabs_Risingwave_JDBCSinkConfig -- Configuration class providing connection parameters