{"record":{"id":"975f0fb97ef057bc","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-mongodb-sink-p","errorCode":null,"errorMessage":"Primary key not defined for upsert mongodb sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert mongodb sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":274,"sourceCode":"impl TryFrom<SinkParam> for MongodbSink {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        MongodbSink::new(param)\n    }\n}\n\nimpl Sink for MongodbSink {\n    type LogSinker = AsyncTruncateLogSinkerOf<MongodbSinkWriter>;\n\n    const SINK_NAME: &'static str = MONGODB_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        if !self.is_append_only {\n            if self.pk_indices.is_empty() {\n                return Err(SinkError::Config(anyhow!(\n                    \"Primary key not defined for upsert mongodb sink (please define in `primary_key` field)\"\n                )));\n            }\n\n            // checking if there is a non-pk field's name is `_id`\n            if self\n                .schema\n                .fields\n                .iter()\n                .enumerate()\n                .any(|(i, field)| !self.pk_indices.contains(&i) && field.name == MONGODB_PK_NAME)\n            {\n                return Err(SinkError::Config(anyhow!(\n                    \"_id field must be the sink's primary key, but a non primary key field name is _id\",\n                )));\n            }\n\n            // assume the sink's pk is (a, b) and then the data written to mongodb will be","sourceCodeStart":256,"sourceCodeEnd":292,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L256-L292","documentation":"When a MongoDB sink is created in upsert mode (the stream is not append-only), RisingWave requires an explicit primary key so it can build the MongoDB `_id` field for upserts. `validate` throws this error at sink creation when `pk_indices` is empty, meaning no `primary_key` was defined in the sink's WITH options. Without a PK, upsert semantics cannot be expressed in MongoDB.","triggerScenarios":"Creating a `CREATE SINK ... INTO mongodb ...` sink without a `primary_key` option while the source/materialized view is not append-only (i.e., it can emit UPDATE/DELETE events).","commonSituations":"Developers forgetting to add `primary_key='col'` to the WITH clause when the upstream is a materialized view with updates; assuming a PK on the source table is automatically inherited by the sink.","solutions":["Add `primary_key='your_column'` (or a comma-separated list) to the WITH options of the CREATE SINK statement.","If the sink should only receive inserts, ensure the upstream stream is append-only (e.g., sink from an append-only source) or use append-only mode so PK is not required.","If upserting on the full row is intended, specify all columns as the primary key."],"exampleFix":"-- before\nCREATE SINK s FROM mv INTO mongodb WITH (\n  connector='mongodb', url='mongodb://localhost:27017', collection='c'\n);\n-- after\nCREATE SINK s FROM mv INTO mongodb WITH (\n  connector='mongodb', url='mongodb://localhost:27017', collection='c',\n  primary_key='id'\n);","handlingStrategy":"validation","validationCode":"// Before creating the sink, ensure upsert sinks declare a primary key\nconst isAppendOnly = /* determined from source */ false;\nconst hasPk = sinkOptions.primary_key != null && sinkOptions.primary_key.length > 0;\nif (!isAppendOnly && !hasPk) {\n  throw new Error(\"MongoDB upsert sink requires `primary_key` in WITH options\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always set `primary_key` when the upstream can emit updates/deletes","Only omit primary_key for genuinely append-only streams","Review sink WITH options in code review"],"tags":["mongodb","sink","upsert","primary-key","config"],"backgroundTag":"missing-required-config-field","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"}