Implementation:Risingwavelabs Risingwave SqlServerDialect

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

Overview

SqlServerDialect implements the JdbcDialect interface to provide Microsoft SQL Server-compatible (T-SQL) SQL generation for the RisingWave JDBC sink connector.

Key characteristics of this dialect:

• Identifier quoting: Uses square brackets ([identifier]), which is the SQL Server convention.
• Schema-qualified table names: Always uses [schema].[table] notation, as SQL Server requires explicit schema qualification.
• Upsert strategy: Uses SQL Server's MERGE ... USING ... ON ... WHEN MATCHED THEN UPDATE ... WHEN NOT MATCHED THEN INSERT syntax, which is a single atomic statement for upsert operations.
• 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/SqlServerDialect.java

Signature

public class SqlServerDialect implements JdbcDialect {
    public SqlServerDialect(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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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, used for DELETE statement parameter binding.

SQL Generation

Upsert statement produces (using MERGE syntax):

MERGE [schema].[table] AS [TARGET]
USING (VALUES (?, ?, ?)) AS [SOURCE] ([col1], [col2], [col3])
ON [SOURCE].[pk1]=[TARGET].[pk1]
WHEN MATCHED THEN UPDATE SET [col1]=[SOURCE].[col1], [col2]=[SOURCE].[col2], [col3]=[SOURCE].[col3]
WHEN NOT MATCHED THEN INSERT ([col1], [col2], [col3]) VALUES ([SOURCE].[col1], [SOURCE].[col2], [SOURCE].[col3]);

Table name normalization always produces:

[schema_name].[table_name]

Insert statement (inherited default with bracket quoting):

INSERT INTO [schema].[table]([col1], [col2]) VALUES (?, ?)

Delete statement (inherited default with bracket quoting):

DELETE FROM [schema].[table] 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 SQL Server dialect
SqlServerDialect dialect = new SqlServerDialect(columnSqlTypes, pkIndices);

// Generate MERGE upsert
SchemaTableName stn = dialect.createSchemaTableName("dbo", "orders");
Optional<String> upsertSql = dialect.getUpsertStatement(stn, tableSchema, List.of("order_id"));

// 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:sqlserver URLs
• Risingwavelabs_Risingwave_MySqlDialect -- MySQL dialect with similar type binding but different upsert syntax
• Risingwavelabs_Risingwave_PostgresDialect -- PostgreSQL dialect using ON CONFLICT upsert
• Risingwavelabs_Risingwave_JDBCSinkConfig -- Configuration class providing connection parameters