risingwavelabs/risingwave · error · SinkError::SqlServer

primary key does not match between RisingWave sink ({}: [{}]

Error message

primary key does not match between RisingWave sink ({}: [{}]) and SQL Server table {} ({}: [{}])

What it means

This is the fallback/mismatch report when the per-column PK checks cannot simply pinpoint the difference (e.g. multi-column PKs): it compares the count and ordered names of the RisingWave sink PK against the SQL Server table PK and fails with a full listing of both sides.

Source

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

                    }
                }
            }
        }

        if !self.is_append_only && sql_server_pk_count != self.pk_indices.len() {
            let sql_server_pk_columns = sql_server_table_metadata
                .values()
                .filter(|metadata| metadata.is_pk)
                .map(|metadata| metadata.name.as_str())
                .collect::<Vec<_>>()
                .join(",");
            let rw_pk_columns = self
                .pk_indices
                .iter()
                .map(|idx| self.schema[*idx].name.as_str())
                .collect::<Vec<_>>()
                .join(",");
            return Err(SinkError::SqlServer(anyhow!(format!(
                "primary key does not match between RisingWave sink ({}: [{}]) and SQL Server table {} ({}: [{}])",
                self.pk_indices.len(),
                rw_pk_columns,
                self.config.full_object_path(),
                sql_server_pk_count,
                sql_server_pk_columns,
            ))));
        }

        Ok(())
    }

    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        Ok(SqlServerSinkWriter::new(
            self.config.clone(),
            self.schema.clone(),
            self.pk_indices.clone(),
            self.is_append_only,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read both column lists in the message and align the sink's PRIMARY KEY to exactly the SQL Server table's PK columns (same set and order)
2. Recreate the SQL Server table with the PK matching the sink definition
3. Use an append_only sink if PK-based upsert matching is not required

Example fix

// before
CREATE SINK ... PRIMARY KEY (a, b) ...; -- table PK is (a)
// after
CREATE SINK ... PRIMARY KEY (a) ...;

fn check_pk_match(rw_pk: &[&str], sqlserver_pk: &[&str]) -> Result<(), String> {
    if rw_pk != sqlserver_pk {
        Err(format!("PK mismatch: rw={:?} sqlserver={:?}", rw_pk, sqlserver_pk))
    } else { Ok(()) }
}

match sink.validate().await {
    Err(e) if e.to_string().contains("primary key does not match between RisingWave sink") => {
        // message lists both sides; rebuild sink PK from the SQL Server side
        rebuild_sink_pk_from_error(&e)
    }
    other => other,
}

When it happens

Trigger: validate() detects a PK mismatch (count or composition) between the sink's pk_indices-derived columns and the downstream table's PK, and formats both column lists into this message.

Common situations: Composite primary keys defined differently on each side; reordered or partially overlapping PK columns; sink created against a table whose PK was later changed.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• column {} specified in primary_key mismatches with the downs
• column {} unspecified in primary_key mismatches with the dow
• DynamoDB table {} primary key {:?} must match RisingWave pri
• Primary key not defined for upsert SQL Server sink (please d
• column {} not found in the downstream SQL Server table {}

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