{"record":{"id":"30aa75062949600d","repo":"risingwavelabs/risingwave","slug":"the-type-of-collection-name-field-must-be-varch","errorCode":null,"errorMessage":"the type of collection.name.field {} must be varchar","messagePattern":"the type of collection\\.name\\.field (.+?) must be varchar","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":357,"sourceCode":"            let fields = self.schema.fields();\n\n            let coll_field_index = fields\n                .iter()\n                .enumerate()\n                .find_map(|(index, field)| {\n                    if &field.name == coll_field {\n                        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(","sourceCodeStart":339,"sourceCodeEnd":375,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L339-L375","documentation":"The dynamic collection-name column's value becomes a MongoDB collection name, so it must be a string (VARCHAR). `validate` fails if the column referenced by `collection.name.field` has any other data type. Non-string types could not produce valid collection names without arbitrary casting rules.","triggerScenarios":"CREATE SINK into mongodb with `collection.name.field` pointing to an INT/DATE/etc. column.","commonSituations":"Routing by a numeric tenant ID or a date column without casting it to a string first.","solutions":["Cast the column to VARCHAR in the sink's source query (e.g., `SELECT tenant_id::varchar AS tenant, ...`) and point `collection.name.field` at the new string column.","Or change the upstream column type to VARCHAR.","Keep collection names valid MongoDB identifiers (lowercase, no spaces/special chars)."],"exampleFix":"-- before\nCREATE SINK s FROM (SELECT tenant_id, ... FROM mv) INTO mongodb WITH (\n  connector='mongodb', collection='db.c', collection.name.field='tenant_id'\n); -- tenant_id is INT\n-- after\nCREATE SINK s FROM (SELECT tenant_id::varchar AS tenant, ... FROM mv) INTO mongodb WITH (\n  connector='mongodb', collection='db.c', collection.name.field='tenant'\n);","handlingStrategy":"validation","validationCode":"const f = schema.fields.find(f => f.name === options['collection.name.field']);\nif (f && f.type !== 'varchar') {\n  throw new Error(`collection.name.field '${f.name}' must be VARCHAR, got ${f.type}`);\n}","typeGuard":"function isVarcharField(f) { return f != null && f.type === 'varchar'; }","tryCatchPattern":null,"preventionTips":["Cast routing columns to VARCHAR in the sink source query","Prefer string routing keys (tenant names, slugs) over numeric ids"],"tags":["mongodb","sink","dynamic-collection","type-mismatch"],"backgroundTag":"type-mismatch","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"}