Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Risingwavelabs Risingwave RedShiftDialect

From Leeroopedia
Revision as of 16:32, 16 February 2026 by Admin (talk | contribs) (Auto-imported from implementations/Risingwavelabs_Risingwave_RedShiftDialect.md)
(diff) ← Older revision | Latest revision (diff) | Newer revision → (diff)


Property Value
Component risingwave-sink-jdbc
Language Java
Package com.risingwave.connector.jdbc
Extends PostgresDialect
Lines 66
Source RedShiftDialect.java

Overview

RedShiftDialect extends PostgresDialect to provide Amazon Redshift-specific behavior for the RisingWave JDBC sink connector. Since Redshift is based on PostgreSQL, this dialect inherits most of its SQL generation and type binding from the parent class, but explicitly disables operations that Redshift does not support in this context:

  • Upsert: Both getUpsertStatement and bindUpsertStatement throw UnsupportedOperationException. Redshift sinks are intended to be used in append-only mode via BatchAppendOnlyJDBCSink.
  • Delete: bindDeleteStatement throws UnsupportedOperationException, as the append-only sink does not perform row-level deletes.
  • JSONB handling: Overrides the parent's set_jsonb method to use setString instead of PGobject, since Redshift does not support the PostgreSQL-native PGobject type for JSONB.

Code Reference

Source Location

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

Signature

public class RedShiftDialect extends PostgresDialect {
    public RedShiftDialect(List<Integer> columnSqlTypes, List<Integer> pkIndices);

    @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 bindDeleteStatement(
            PreparedStatement stmt, TableSchema tableSchema, SinkRow row) throws SQLException;
    @Override public void set_jsonb(
            int placeholderIdx, int columnIdx, PreparedStatement stmt, 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;

I/O Contract

Constructor Input

Parameter Type Description
columnSqlTypes List<Integer> JDBC SQL type codes for all columns (passed to parent constructor)
pkIndices List<Integer> Indices of primary key columns (passed to parent constructor)

Overridden Methods

Method Behavior
getUpsertStatement Throws UnsupportedOperationException -- Redshift does not support upsert
bindUpsertStatement Throws UnsupportedOperationException -- Redshift does not support upsert
bindDeleteStatement Throws UnsupportedOperationException -- Redshift does not support DELETE in this context
set_jsonb Uses stmt.setString(placeholderIdx, (String) row.get(columnIdx)) instead of PGobject

Inherited Methods

All other methods are inherited from PostgresDialect, including:

  • createSchemaTableName
  • getNormalizedTableName (with optional schema qualification)
  • quoteIdentifier (double-quote style)
  • getInsertIntoStatement
  • getDeleteStatement
  • bindInsertIntoStatement (with Redshift-specific JSONB handling via overridden set_jsonb)

Usage Examples

// Create Redshift dialect
RedShiftDialect dialect = new RedShiftDialect(columnSqlTypes, pkIndices);

// Generate INSERT (inherited from PostgresDialect)
SchemaTableName stn = dialect.createSchemaTableName("public", "events");
String insertSql = dialect.getInsertIntoStatement(stn, fieldNames);

// Bind and execute
PreparedStatement stmt = conn.prepareStatement(insertSql);
dialect.bindInsertIntoStatement(stmt, conn, tableSchema, row);
stmt.executeUpdate();

// Upsert is NOT supported
// dialect.getUpsertStatement(stn, tableSchema, pks);  // throws UnsupportedOperationException

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment