{"record":{"id":"365d50c2913f41cb","repo":"risingwavelabs/risingwave","slug":"topic-field-must-be-of-type-string-but-got","errorCode":null,"errorMessage":"topic field `{}` must be of type string but got {:?}","messagePattern":"topic field `(.+?)` must be of type string but got (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mqtt.rs","lineNumber":469,"sourceCode":"            })\n            .and_then(|dt| {\n                // Iterate over the next fields to extract the fields from the nested structs\n                iter.try_fold(dt, |dt, field| match dt {\n                    DataType::Struct(st) => {\n                        st.iter().enumerate().find(|(_, (s, _))| *s == field).map(\n                            |(pos, (_, dt))| {\n                                path.push(pos);\n                                dt\n                            },\n                        )\n                    }\n                    _ => None,\n                })\n            });\n\n    match dt {\n        Some(DataType::Varchar) => Ok(path),\n        Some(dt) => Err(SinkError::Config(anyhow!(\n            \"topic field `{}` must be of type string but got {:?}\",\n            topic_field,\n            dt\n        ))),\n        None => Err(SinkError::Config(anyhow!(\n            \"topic field `{}`  not found\",\n            topic_field\n        ))),\n    }\n}\n\n#[cfg(test)]\nmod test {\n    use risingwave_common::array::{DataChunk, DataChunkTestExt, RowRef};\n    use risingwave_common::catalog::{Field, Schema};\n    use risingwave_common::types::{DataType, StructType};\n\n    use super::{get_topic_field_index_path, get_topic_from_index_path};","sourceCodeStart":451,"sourceCodeEnd":487,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mqtt.rs#L451-L487","documentation":"`get_topic_field_index_path` validates that the column referenced by the `topic_field` option has type VARCHAR, so it can be used to build MQTT topic names. This error is thrown when the configured topic field exists but is of some other data type.","triggerScenarios":"Creating/validating an MQTT sink whose `topic_field` option points to a column of a non-string type (e.g. INT, TIMESTAMP); raised from `validate` and `new` via `get_topic_field_index_path`.","commonSituations":"User points `topic_field` at an id column (integer) or a timestamp column instead of a string column; schema changed upstream so the column type drifted from varchar.","solutions":["Change `topic_field` to point at a VARCHAR column","Cast the column in the sink's SELECT: `SELECT id::varchar AS id_str, ... FROM t` and set `topic_field='id_str'`","Re-create the sink after fixing the option"],"exampleFix":"// before\nCREATE SINK s FROM t WITH (connector='mqtt', topic_field='device_id'); -- device_id: INT\n// after\nCREATE SINK s FROM (SELECT device_id::varchar AS device_id, * FROM t) WITH (connector='mqtt', topic_field='device_id');","handlingStrategy":"validation","validationCode":"// ensure topic_field points to a varchar column\nlet col = schema.fields().iter().find(|f| f.name.as_str() == topic_field)\n    .unwrap_or_else(|| panic!(\"topic_field '{}' not in schema\", topic_field));\nassert_eq!(col.data_type, DataType::Varchar, \"topic_field must be varchar\");","typeGuard":"fn is_string_column(f: &Field) -> bool { matches!(f.data_type, DataType::Varchar) }","tryCatchPattern":null,"preventionTips":["Verify column types in the sinked stream before setting topic_field","Cast non-string columns with ::varchar in the sink's SELECT","Re-check after upstream schema changes"],"tags":["mqtt","sink","config","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"}