risingwavelabs/risingwave · error · SinkError::Config

Primary key mismatch: Postgres table has primary key on colu

Error message

Primary key mismatch: Postgres table has primary key on column `{}`, but sink schema does not define it as a primary key

What it means

After the PK count check passes, validation iterates the Postgres table's PK column names and requires each to appear in the sink's declared PK set. Any PG PK column missing from the sink's primary_key definition aborts CREATE SINK for that single column.

Source

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

            // check that pk matches
            {
                let pg_pk_names = pg_table.pk_names();
                let sink_pk_names = self
                    .pk_indices
                    .iter()
                    .map(|i| &self.schema.fields()[*i].name)
                    .collect::<HashSet<_>>();
                if pg_pk_names.len() != sink_pk_names.len() {
                    return Err(SinkError::Config(anyhow!(
                        "Primary key mismatch: Postgres table has primary key on columns {:?}, but sink schema defines primary key on columns {:?}",
                        pg_pk_names,
                        sink_pk_names
                    )));
                }
                for name in pg_pk_names {
                    if !sink_pk_names.contains(name) {
                        return Err(SinkError::Config(anyhow!(
                            "Primary key mismatch: Postgres table has primary key on column `{}`, but sink schema does not define it as a primary key",
                            name
                        )));
                    }
                }
            }
        }

        Ok(())
    }

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

View on GitHub (pinned to 6469eb736d)

Solutions

1. Rename the column in the sink query (AS <pg_pk_name>) so its name matches the Postgres primary key column.
2. Update the sink's `primary_key` option to use the exact PG PK column names, then recreate the sink.
3. ALTER the Postgres table PK to match the names used in the sink schema.

Example fix

-- before (pg PK on 'user_id')
CREATE SINK s AS SELECT user_id AS key FROM mv WITH (connector='postgres', table='t', primary_key='key');
-- after
CREATE SINK s AS SELECT user_id FROM mv WITH (connector='postgres', table='t', primary_key='user_id');

-- every PG PK column name must equal a sink schema column name and be in primary_key
SELECT conkey, conname FROM pg_constraint WHERE conrelid='t'::regclass AND contype='p';

When it happens

Trigger: PG table PK is composite (a,b) and the sink declares primary_key='a,b' but with a renamed column, or a name differs by case/alias so sink_pk_names lacks the PG name.

Common situations: Aliasing a PK column in the sink query so its name no longer matches the PG PK column; case-sensitivity mismatch; PG PK changed since the sink config was written.

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

• Primary key not defined for upsert Postgres sink (please def
• 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/cfe4762c633aba27. Report an issue: GitHub.