risingwavelabs/risingwave · error · SinkError::Config

Primary key not defined for upsert Postgres sink (please def

Error message

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

What it means

An upsert (non-append-only) PostgreSQL sink requires a primary key so rows can be keyed for upsert/delete operations. If the streaming fragment is not append-only and `pk_indices` is empty, validate() rejects the sink and asks the user to define a primary key via the `primary_key` field.

Source

Thrown at src/connector/src/sink/postgres.rs:242

impl Sink for PostgresSink {
    type LogSinker = BatchingLogSinker<PostgresSinkWriter>;

    const SINK_NAME: &'static str = POSTGRES_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn validate(&self) -> Result<()> {
        if !(1..=MAX_BATCH_ROWS_LIMIT).contains(&self.config.max_batch_rows) {
            return Err(SinkError::Config(anyhow!(
                "`max_batch_rows` must be between 1 and {}, got {}",
                MAX_BATCH_ROWS_LIMIT,
                self.config.max_batch_rows
            )));
        }

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

        ensure_no_foreign_key(&self.config).await?;

        // Verify our sink schema is compatible with Postgres
        {
            let pg_conn = self.config.pg_connection_config();
            let pg_table = PostgresExternalTable::connect(
                &pg_conn,
                &self.config.schema,
                &self.config.table,
                self.is_append_only,
                None,
            )
            .await
            .context(format!(

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add `primary_key = '<column>'` (or a comma-separated list) to the WITH options of CREATE SINK.
2. Ensure the queried relation has a primary key so pk_indices is populated automatically.
3. If the data is truly append-only, adjust the query/sink so it is treated as append-only.

Example fix

// before
CREATE SINK s FROM mv WITH (connector = 'postgres');
// after
CREATE SINK s FROM mv WITH (connector = 'postgres', primary_key = 'id');

-- ensure the relation is append-only or a PK is supplied
-- if relation is not append-only: WITH (... , primary_key = 'id')

match err {
    SinkError::Config(e) if e.to_string().contains("Primary key not defined") => {
        // recreate sink with primary_key option
    }
    _ => {}
}

When it happens

Trigger: CREATE SINK from a non-append-only source/MV without specifying `primary_key` in the WITH options, so the sink has empty pk_indices.

Common situations: Sinking an append-only=false materialized view (or one containing updates/deletes) to Postgres; forgetting that the upstream stream's PK is not automatically propagated to the sink.

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 mismatch: Postgres table has primary key on colu
• Primary key mismatch: Postgres table has primary key on colu
• Primary key not defined for upsert doris sink (please define
• no value find in sink schema, index is {:?}
• please set the separator in the with option, when there are

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