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 columns {:?}, but sink schema defines primary key on columns {:?}

What it means

For non-append-only sinks, the sink's declared primary key column-name set must exactly match the Postgres table's primary key column-name set, starting with a count check. If the number of PK columns differs (or the sets otherwise diverge), CREATE SINK fails with the full sets printed via {:?}.

Source

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

                                    pg_column,
                                    sink_column.data_type()
                                )));
                            }
                        }
                    }
                }
            }

            // 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(())
    }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Make the sink `primary_key` list exactly the PG table's primary key columns (ALTER the WITH options and recreate the sink).
2. ALTER the Postgres table's primary key to match the sink's declared PK.
3. If the stream is append-only, drop the primary_key requirement by restructuring the sink as append-only.

Example fix

-- before (pg t has PK (a,b))
CREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='a');
-- after
CREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='a,b');

SELECT a.attname FROM pg_index i
JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
WHERE i.indrelid = 't'::regclass AND i.indisprimary;
-- sink primary_key must list exactly these columns

When it happens

Trigger: `primary_key` in WITH options names a different number of columns than the PG table's actual PRIMARY KEY constraint; duplicate names collapsing in the HashSet comparison.

Common situations: Defining a single-column sink PK while the PG table uses a composite PK (or vice versa); PG table PK changed after sink creation 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/296e610794a900a9. Report an issue: GitHub.