risingwavelabs/risingwave · error · SinkError::Config

topic field `{}` must be of type string but got {:?}

Error message

topic field `{}` must be of type string but got {:?}

What it means

`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.

Source

Thrown at src/connector/src/sink/mqtt.rs:469

            })
            .and_then(|dt| {
                // Iterate over the next fields to extract the fields from the nested structs
                iter.try_fold(dt, |dt, field| match dt {
                    DataType::Struct(st) => {
                        st.iter().enumerate().find(|(_, (s, _))| *s == field).map(
                            |(pos, (_, dt))| {
                                path.push(pos);
                                dt
                            },
                        )
                    }
                    _ => None,
                })
            });

    match dt {
        Some(DataType::Varchar) => Ok(path),
        Some(dt) => Err(SinkError::Config(anyhow!(
            "topic field `{}` must be of type string but got {:?}",
            topic_field,
            dt
        ))),
        None => Err(SinkError::Config(anyhow!(
            "topic field `{}`  not found",
            topic_field
        ))),
    }
}

#[cfg(test)]
mod test {
    use risingwave_common::array::{DataChunk, DataChunkTestExt, RowRef};
    use risingwave_common::catalog::{Field, Schema};
    use risingwave_common::types::{DataType, StructType};

    use super::{get_topic_field_index_path, get_topic_from_index_path};

View on GitHub (pinned to 6469eb736d)

Solutions

1. Change `topic_field` to point at a VARCHAR column
2. Cast the column in the sink's SELECT: `SELECT id::varchar AS id_str, ... FROM t` and set `topic_field='id_str'`
3. Re-create the sink after fixing the option

Example fix

// before
CREATE SINK s FROM t WITH (connector='mqtt', topic_field='device_id'); -- device_id: INT
// after
CREATE SINK s FROM (SELECT device_id::varchar AS device_id, * FROM t) WITH (connector='mqtt', topic_field='device_id');

// ensure topic_field points to a varchar column
let col = schema.fields().iter().find(|f| f.name.as_str() == topic_field)
    .unwrap_or_else(|| panic!("topic_field '{}' not in schema", topic_field));
assert_eq!(col.data_type, DataType::Varchar, "topic_field must be varchar");

fn is_string_column(f: &Field) -> bool { matches!(f.data_type, DataType::Varchar) }

When it happens

Trigger: 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`.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• topic field `{}` not found
• Turbopuffer namespace_column must be varchar, got {:?}
• sink type unsupported: {}
• sink format unsupported: {}
• sink encode unsupported: {}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/365d50c2913f41cb. Report an issue: GitHub.