{"record":{"id":"5c157218738b39a5","repo":"risingwavelabs/risingwave","slug":"cannot-find-5c1572","errorCode":null,"errorMessage":"Cannot find {}","messagePattern":"Cannot find (.+?)","errorType":"validation","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs","lineNumber":297,"sourceCode":"                return Err(SinkError::Config(anyhow!(\n                    \"please ensure the data type of {} is varchar.\",\n                    routing_column\n                )));\n            }\n        }\n        Ok(())\n    }\n\n    pub fn get_index_column_index(&self, schema: &Schema) -> Result<Option<usize>> {\n        let index_column_idx = self\n            .index_column\n            .as_ref()\n            .map(|n| {\n                schema\n                    .fields()\n                    .iter()\n                    .position(|s| &s.name == n)\n                    .ok_or_else(|| anyhow!(\"Cannot find {}\", ES_OPTION_INDEX_COLUMN))\n            })\n            .transpose()?;\n        Ok(index_column_idx)\n    }\n\n    pub fn get_routing_column_index(&self, schema: &Schema) -> Result<Option<usize>> {\n        let routing_column_idx = self\n            .routing_column\n            .as_ref()\n            .map(|n| {\n                schema\n                    .fields()\n                    .iter()\n                    .position(|s| &s.name == n)\n                    .ok_or_else(|| anyhow!(\"Cannot find {}\", ES_OPTION_ROUTING_COLUMN))\n            })\n            .transpose()?;\n        Ok(routing_column_idx)","sourceCodeStart":279,"sourceCodeEnd":315,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs#L279-L315","documentation":"`get_index_column_index` resolves the `index_column` name to its position in the sink schema. If no schema field matches the configured name, it returns 'Cannot find index_column' (the option's constant name). Note the column is normally also checked in `validate_config` with an unwrap, so this usually surfaces when the two checks are bypassed or the lookup happens against a different schema.","triggerScenarios":"Sink creation path `new` -> `get_index_column_index` with `index_column` set to a name that does not exist verbatim in the schema fields (case mismatch, extra whitespace, renamed column).","commonSituations":"Case-sensitive name mismatch ('UserID' vs 'user_id'), column renamed upstream after config was written, quoting artifacts in the WITH option value.","solutions":["Verify the exact column name in the sink schema and use it verbatim in `index_column`","Check for case differences or accidental whitespace in the WITH option","Query the schema (e.g. DESC the materialized view) to list available columns","If the column was renamed upstream, update or recreate the sink"],"exampleFix":"// before\nWITH (connector='elasticsearch', index_column='Topic')\n// after\nWITH (connector='elasticsearch', index_column='topic')","handlingStrategy":"validation","validationCode":"fn check_column_exists(schema: &Schema, name: &str) -> Result<(), String> {\n    if schema.fields().iter().any(|f| f.name == name) {\n        Ok(())\n    } else {\n        Err(format!(\"column '{}' not in schema; available: {:?}\",\n            name, schema.fields().iter().map(|f| f.name.clone()).collect::<Vec<_>>()))\n    }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Copy column names from DESC output rather than typing them","Remember names are case-sensitive and whitespace-sensitive","Re-check configs after upstream schema renames"],"tags":["rust","sink","config","schema","elasticsearch"],"backgroundTag":"entity-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"}