risingwavelabs/risingwave · error · SinkError::Config

Primary key not defined for upsert SQL Server sink (please d

Error message

Primary key not defined for upsert SQL Server sink (please define in `primary_key` field)

What it means

An upsert SQL Server sink needs primary key indices to know which columns to match on for UPDATE/MERGE semantics. If the sink is not append-only and pk_indices is empty, validate() rejects the sink definition.

Source

Thrown at src/connector/src/sink/sqlserver.rs:187

        let config = SqlServerConfig::from_btreemap(param.properties)?;
        SqlServerSink::new(config, schema, pk_indices, param.sink_type.is_append_only())
    }
}

impl Sink for SqlServerSink {
    type LogSinker = LogSinkerOf<SqlServerSinkWriter>;

    const SINK_NAME: &'static str = SQLSERVER_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn validate(&self) -> Result<()> {
        risingwave_common::license::Feature::SqlServerSink
            .check_available()
            .map_err(|e| anyhow::anyhow!(e))?;

        if !self.is_append_only && self.pk_indices.is_empty() {
            return Err(SinkError::Config(anyhow!(
                "Primary key not defined for upsert SQL Server sink (please define in `primary_key` field)"
            )));
        }

        for f in self.schema.fields() {
            check_data_type_compatibility(&f.data_type)?;
        }

        let mut sql_client = SqlServerClient::new(&self.config).await?;
        validate_sql_server_write_permission(&mut sql_client, &self.config, self.is_append_only)
            .await?;
        let sql_server_table_metadata =
            query_sql_server_table_metadata(&mut sql_client, &self.config).await?;
        let sql_server_pk_count = sql_server_table_metadata
            .iter()
            .filter(|metadata| metadata.is_pk)
            .count();
        let sql_server_table_metadata = sql_server_table_metadata

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add `PRIMARY KEY (...)` (or the `primary_key` field) to the CREATE SINK statement naming the key columns
2. Change the sink type to 'append_only' if the data stream is truly append-only and no upsert semantics are needed
3. Ensure the upstream materialized view/table has a primary key so it is inherited by the sink

Example fix

// before
CREATE SINK s FROM mv INTO sqlserver ... WITH ( 'type' = 'upsert' );
// after
CREATE SINK s FROM mv INTO sqlserver ... WITH ( 'type' = 'upsert' ) AS SELECT ... PRIMARY KEY (id);

fn check_upsert_pk(sink_type: &str, pk_columns: &[&str]) -> Result<(), String> {
    if sink_type == "upsert" && pk_columns.is_empty() {
        Err("upsert SQL Server sink requires a primary_key".into())
    } else { Ok(()) }
}

match create_sink(cfg).await {
    Err(e) if e.to_string().contains("Primary key not defined for upsert SQL Server sink") => return Err("add PRIMARY KEY to sink definition".into()),
    other => other,
}

When it happens

Trigger: Creating a SQL Server sink with `type = 'upsert'` (or non-append-only) but without a `primary_key` field defined in the sink definition.

Common situations: Upsert sink over a source without a primary key and forgetting to specify `primary_key` in the CREATE SINK statement; append-only data being sunk as upsert.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Primary key not defined for upsert mongodb sink (please defi
• Primary key not defined for upsert clickhouse sink (please d
• Primary key not defined for upsert doris sink (please define
• please set the separator in the with option, when there are
• primary key must be specified for upsert iceberg sink

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/d4b80cd2eaed3e36. Report an issue: GitHub.