{"record":{"id":"7d9625a843230c75","repo":"risingwavelabs/risingwave","slug":"postgres-table-should-define-the-primary-key-for-n","errorCode":null,"errorMessage":"Postgres table should define the primary key for non-append-only tables","messagePattern":"Postgres table should define the primary key for non-append-only tables","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/connector_common/postgres.rs","lineNumber":453,"sourceCode":"                        Some(scalar),\n                    ),\n                    Err(err) => {\n                        tracing::warn!(\n                            error=%err.as_report(),\n                            \"failed to parse the PostgreSQL default value expression; only constants are supported\",\n                        );\n                        ColumnDesc::named(col.name.clone(), ColumnId::placeholder(), rw_data_type)\n                    }\n                }\n            } else {\n                ColumnDesc::named(col.name.clone(), ColumnId::placeholder(), rw_data_type)\n            };\n            column_descs.push(column_desc);\n        }\n\n        // Check primary key existence using the directly discovered pk_names\n        if !is_append_only && pk_names.is_empty() {\n            return Err(anyhow!(\n                \"Postgres table should define the primary key for non-append-only tables\"\n            )\n            .into());\n        }\n\n        Ok(Self {\n            column_descs,\n            pk_names,\n        })\n    }\n\n    // return the mapping from column name to pg type, the pg type is used for writing data to postgres\n    pub async fn type_mapping(\n        config: &PgConnectionConfig,\n        schema: &str,\n        table: &str,\n        is_append_only: bool,\n    ) -> ConnectorResult<HashMap<String, tokio_postgres::types::Type>> {","sourceCodeStart":435,"sourceCodeEnd":471,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/connector_common/postgres.rs#L435-L471","documentation":"RisingWave's PostgreSQL connector requires a primary key for any non-append-only source table, because CDC row updates/deletes must be keyed. `connect` throws this error when the discovered table schema (`pk_names`) is empty while `is_append_only` is false.","triggerScenarios":"Creating a CDC table/source via `PgCdcConnector::connect` on a Postgres table with no PRIMARY KEY or UNIQUE constraints recognized as pk_names, with append-only mode not enabled.","commonSituations":"Upstream table was created without a primary key; the key is enforced only at application level; the user assumed RisingWave tolerates keyless tables for CDC.","solutions":["Add a primary key to the upstream table: `ALTER TABLE <table> ADD PRIMARY KEY (<cols>);`","If the table truly receives only inserts, declare the RisingWave source as append-only so the PK requirement is skipped","Use a UNIQUE NOT NULL column set as the table primary key"],"exampleFix":"-- before\nCREATE TABLE orders (id BIGINT, amount NUMERIC);\n-- after\nCREATE TABLE orders (id BIGINT PRIMARY KEY, amount NUMERIC);","handlingStrategy":"validation","validationCode":"-- before creating the RW source\nSELECT COUNT(*) FROM pg_constraint\nWHERE contype = 'p' AND conrelid = 'public.orders'::regclass; -- must be > 0","typeGuard":null,"tryCatchPattern":"if let Err(e) = connect_result {\n    if e.to_string().contains(\"should define the primary key\") {\n        // either add a PK upstream or declare the source append-only\n        return Err(PlanError::MissingPrimaryKey(e));\n    }\n    return Err(e.into());\n}","preventionTips":["Always define PRIMARY KEY on tables used for CDC","If a table is insert-only, configure it as append-only explicitly","Include PK existence checks in DBA runbooks"],"tags":["postgres","cdc","primary-key","schema"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}