{"record":{"id":"a61d06a80dafe430","repo":"risingwavelabs/risingwave","slug":"collection-name-field-must-not-be-equal-to-the","errorCode":null,"errorMessage":"collection.name.field {} must not be equal to the primary key field","messagePattern":"collection\\.name\\.field (.+?) must not be equal to the primary key field","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":364,"sourceCode":"                        Some(index)\n                    } else {\n                        None\n                    }\n                })\n                .ok_or(SinkError::Config(anyhow!(\n                    \"collection.name.field {} not found\",\n                    coll_field\n                )))?;\n\n            if fields[coll_field_index].data_type() != risingwave_common::types::DataType::Varchar {\n                return Err(SinkError::Config(anyhow!(\n                    \"the type of collection.name.field {} must be varchar\",\n                    coll_field\n                )));\n            }\n\n            if !self.is_append_only && self.pk_indices.contains(&coll_field_index) {\n                return Err(SinkError::Config(anyhow!(\n                    \"collection.name.field {} must not be equal to the primary key field\",\n                    coll_field\n                )));\n            }\n        }\n\n        Ok(())\n    }\n\n    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        Ok(MongodbSinkWriter::new(\n            format!(\"{}-{}\", writer_param.executor_id, self.param.sink_name),\n            self.config.clone(),\n            self.schema.clone(),\n            self.pk_indices.clone(),\n            self.is_append_only,\n        )\n        .await?","sourceCodeStart":346,"sourceCodeEnd":382,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L346-L382","documentation":"In upsert mode, the primary key is written as the document's `_id`; the dynamic collection-name column must survive in the document per row, so it cannot be part of the primary key (otherwise it would be consumed into `_id`, breaking per-collection routing). `validate` rejects this combination for non-append-only sinks.","triggerScenarios":"CREATE SINK into mongodb (upsert mode) where `collection.name.field` references a column that is also listed in `primary_key`.","commonSituations":"Routing by tenant column while also using tenant as part of the upsert PK — a common but unsupported combo here.","solutions":["Remove the collection-name column from `primary_key` and pick a different PK (e.g., add a synthetic id column and use `primary_key='id'`).","If the collection name column must be the key, keep the sink append-only (no updates) so the constraint doesn't apply.","Duplicate the value: route on a plain string column and use a different column set as PK."],"exampleFix":"-- before\nCREATE SINK s FROM mv INTO mongodb WITH (\n  connector='mongodb', collection='db.c',\n  primary_key='tenant', collection.name.field='tenant'\n);\n-- after\nCREATE SINK s FROM mv INTO mongodb WITH (\n  connector='mongodb', collection='db.c',\n  primary_key='id', collection.name.field='tenant'\n);","handlingStrategy":"validation","validationCode":"const pk = new Set(sinkOptions.primary_key.split(',').map(s => s.trim()));\nconst cf = options['collection.name.field'];\nif (!isAppendOnly && cf != null && pk.has(cf)) {\n  throw new Error(\"collection.name.field must not be part of primary_key in upsert mongodb sink\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep the routing column out of primary_key","Use a separate id column for the PK","Only combine routing column + PK in append-only sinks"],"tags":["mongodb","sink","upsert","primary-key","dynamic-collection"],"backgroundTag":"conflicting-config-options","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"}