{"record":{"id":"8c5e35eb8e7742cc","repo":"risingwavelabs/risingwave","slug":"please-ensure-the-data-type-of-is-varchar","errorCode":null,"errorMessage":"please ensure the data type of {} is varchar.","messagePattern":"please ensure the data type of (.+?) is varchar\\.","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs","lineNumber":265,"sourceCode":"    }\n\n    pub fn validate_config(&self, schema: &Schema) -> Result<()> {\n        if self.index_column.is_some() && self.index.is_some()\n            || self.index_column.is_none() && self.index.is_none()\n        {\n            return Err(SinkError::Config(anyhow!(\n                \"please set only one of the 'index_column' or 'index' properties.\"\n            )));\n        }\n\n        if let Some(index_column) = &self.index_column {\n            let filed = schema\n                .fields()\n                .iter()\n                .find(|f| &f.name == index_column)\n                .unwrap();\n            if filed.data_type() != DataType::Varchar {\n                return Err(SinkError::Config(anyhow!(\n                    \"please ensure the data type of {} is varchar.\",\n                    index_column\n                )));\n            }\n        }\n\n        if let Some(routing_column) = &self.routing_column {\n            let filed = schema\n                .fields()\n                .iter()\n                .find(|f| &f.name == routing_column)\n                .unwrap();\n            if filed.data_type() != DataType::Varchar {\n                return Err(SinkError::Config(anyhow!(\n                    \"please ensure the data type of {} is varchar.\",\n                    routing_column\n                )));\n            }","sourceCodeStart":247,"sourceCodeEnd":283,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs#L247-L283","documentation":"When `index_column` is configured, the sink reads that column's value per row to decide the destination index. `validate_config` requires the column's data type to be VARCHAR, because index names must be strings. If the referenced column has any other type (int, timestamp, etc.) the sink creation fails with this error naming the offending column.","triggerScenarios":"CREATE SINK with `index_column = '<col>'` where `<col>` exists in the sink schema but its declared data type is not VARCHAR. The column lookup itself uses `.unwrap()` so the column must exist; this error fires only on type mismatch.","commonSituations":"Pointing index_column at an integer id or timestamp column assuming automatic casting; schema evolution changed the column type after the sink was defined; copying a config where the column type differs across environments.","solutions":["Cast or add a varchar column in the source/materialized view, e.g. `CAST(ts AS varchar) AS index_key`, and point index_column at it","Choose a different varchar column for index_column","Change the upstream schema so the column is varchar before creating the sink"],"exampleFix":"// before\nCREATE MATERIALIZED VIEW mv AS SELECT id, ts FROM src;\nWITH (connector='elasticsearch', index_column='ts')  -- ts is TIMESTAMP\n// after\nCREATE MATERIALIZED VIEW mv AS SELECT id, CAST(ts AS varchar) AS ts_str FROM src;\nWITH (connector='elasticsearch', index_column='ts_str')","handlingStrategy":"validation","validationCode":"fn check_index_column_type(schema: &Schema, index_column: &str) -> Result<(), String> {\n    match schema.fields().iter().find(|f| f.name == index_column) {\n        Some(f) if f.data_type() == DataType::Varchar => Ok(()),\n        Some(f) => Err(format!(\"column {} is {:?}, must be varchar\", index_column, f.data_type())),\n        None => Err(format!(\"column {} not found\", index_column)),\n    }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Add explicit CAST(... AS varchar) columns for dynamic index keys","Check column types with DESC before wiring index_column","Keep index-key columns as varchar in upstream views by convention"],"tags":["rust","sink","config","data-type","elasticsearch"],"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"}