{"record":{"id":"11b399b7bf3bfed1","repo":"risingwavelabs/risingwave","slug":"collection-name-field-not-found","errorCode":null,"errorMessage":"collection.name.field {} not found","messagePattern":"collection\\.name\\.field (.+?) not found","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":351,"sourceCode":"                \"collection.name.field must be specified when collection.name.field.drop is enabled\"\n            )));\n        }\n\n        // checking dynamic collection name settings\n        if let Some(coll_field) = &self.config.collection_name_field {\n            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        }","sourceCodeStart":333,"sourceCodeEnd":369,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L333-L369","documentation":"For dynamic collection routing, `collection.name.field` must name an existing column of the sink's schema; `validate` searches the schema fields and fails if none matches. This ensures the sink can resolve the per-row collection name at write time.","triggerScenarios":"CREATE SINK into mongodb with `collection.name.field='X'` where no column named `X` exists in the sink's output schema (typo, renamed column, wrong case).","commonSituations":"Typos or case-sensitivity mismatches; upstream schema changed (column renamed/dropped) after the sink config was written; confusing the field name with the collection name.","solutions":["Set `collection.name.field` to an exact existing column name (case-sensitive).","If the routing key doesn't exist, add it to the upstream query (e.g., a computed column) and refresh the sink.","Re-check the current schema with `DESCRIBE` on the source/materialized view."],"exampleFix":"-- before\nCREATE SINK s FROM mv INTO mongodb WITH (\n  connector='mongodb', collection='db.c', collection.name.field='coll'\n); -- mv has column `coll_name`\n-- after\nCREATE SINK s FROM mv INTO mongodb WITH (\n  connector='mongodb', collection='db.c', collection.name.field='coll_name'\n);","handlingStrategy":"validation","validationCode":"const collField = options['collection.name.field'];\nif (collField != null && !schema.fields.some(f => f.name === collField)) {\n  throw new Error(`collection.name.field '${collField}' not found in sink schema`);\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Verify column names with DESCRIBE before writing the sink DDL","Watch for case sensitivity in column names","Re-check config after upstream schema changes"],"tags":["mongodb","sink","dynamic-collection","schema"],"backgroundTag":"resource-not-found","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"}