Implementation:Risingwavelabs Risingwave PostgresDialect

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

Overview

PostgresDialect implements the JdbcDialect interface to provide PostgreSQL-compatible SQL generation for the RisingWave JDBC sink connector. It is the most feature-rich dialect implementation, serving as the base class for the RedShiftDialect.

Key characteristics of this dialect:

• Identifier quoting: Uses double-quotes ("identifier"), making identifiers case-sensitive per PostgreSQL conventions.
• Schema-qualified table names: Supports "schema"."table" notation when a schema name is provided.
• Upsert strategy: Uses PostgreSQL's INSERT ... ON CONFLICT (...) DO UPDATE SET syntax with EXCLUDED references.
• Type mapping: Maintains a static map (RW_TYPE_TO_JDBC_TYPE_NAME) from RisingWave type names to PostgreSQL JDBC type names for array element type resolution.
• Special type handling: Supports JSONB via PGobject, INTERVAL via PGInterval, native PostgreSQL arrays via createArrayOf, and VARCHAR-to-UUID mapping.

Code Reference

Source Location

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

Signature

public class PostgresDialect implements JdbcDialect {
    static final HashMap<TypeName, String> RW_TYPE_TO_JDBC_TYPE_NAME;

    public PostgresDialect(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> primaryKeyFields);
    @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;
    public void set_jsonb(int placeholderIdx, int columnIdx, PreparedStatement stmt, 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 com.risingwave.proto.Data.DataType.TypeName;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import org.postgresql.util.PGInterval;
import org.postgresql.util.PGobject;

I/O Contract

Constructor Input

Parameter	Type	Description
columnSqlTypes	List<Integer>	JDBC SQL type codes for all columns (used for VARCHAR-to-UUID and DELETE binding)
pkIndices	List<Integer>	Indices of primary key columns within the column list

RisingWave-to-PostgreSQL Type Mapping

RisingWave TypeName	PostgreSQL Type Name
INT16	int2
INT32	int4
INT64	int8
FLOAT	float4
DOUBLE	float8
BOOLEAN	bool
VARCHAR	varchar
DECIMAL	numeric
TIME	time
TIMESTAMP	timestamp
INTERVAL	varchar
DATE	date
TIMESTAMPTZ	timestamptz
JSONB	varchar

SQL Generation

Upsert statement produces:

INSERT INTO "schema"."table"("col1", "col2", "col3") VALUES (?, ?, ?)
ON CONFLICT ("pk1", "pk2") DO UPDATE SET "col1"=EXCLUDED."col1", "col2"=EXCLUDED."col2", "col3"=EXCLUDED."col3"

Table name normalization:

• With schema: "public"."users"
• Without schema: "users"

Type Binding Behavior

RisingWave Type	Binding Method	Notes
DECIMAL	setBigDecimal	Direct BigDecimal binding
INTERVAL	setObject(new PGInterval(...))	Uses PostgreSQL-specific PGInterval type
JSONB	setObject(PGobject)	Creates PGobject with type "jsonb" via set_jsonb method
BYTEA	setBytes	Raw byte array
LIST	setArray(conn.createArrayOf(...))	Uses native PostgreSQL arrays with proper element type mapping
VARCHAR	setObject(value, sqlType)	Uses column SQL type to support VARCHAR-to-UUID mapping
All others	setObject	Generic fallback

Usage Examples

// Create dialect
PostgresDialect dialect = new PostgresDialect(columnSqlTypes, pkIndices);

// Generate schema-qualified upsert
SchemaTableName stn = dialect.createSchemaTableName("public", "users");
Optional<String> upsertSql = dialect.getUpsertStatement(stn, tableSchema, List.of("id"));
// Produces: INSERT INTO "public"."users"("id", "name") VALUES (?, ?)
//           ON CONFLICT ("id") DO UPDATE SET "id"=EXCLUDED."id", "name"=EXCLUDED."name"

// 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_RedShiftDialect -- Subclass that extends this dialect for Amazon Redshift
• Risingwavelabs_Risingwave_JdbcUtils -- Resolves this dialect for jdbc:postgresql URLs
• Risingwavelabs_Risingwave_MySqlDialect -- MySQL dialect alternative (uses ON DUPLICATE KEY)
• Risingwavelabs_Risingwave_SqlServerDialect -- SQL Server dialect alternative (uses MERGE)
• Risingwavelabs_Risingwave_JDBCSinkConfig -- Configuration providing connection parameters