{"record":{"id":"72fe0ff109258a6a","repo":"pathwaycom/pathway","slug":"the-topic-name-column-must-have-a-string-type-how","errorCode":null,"errorMessage":"The topic name column must have a string type, however {topic_name._column.dtype.typehint} is used","messagePattern":"The topic name column must have a string type, however (.+?) is used","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/_utils.py","lineNumber":495,"sourceCode":"        if schema_registry_settings is not None and format != \"json\":\n            raise ValueError(\n                f\"'schema_registry_settings' is only meaningful for the 'json' \"\n                f\"format, but {format!r} was specified. The Confluent Schema \"\n                \"Registry currently encodes JSON payloads only; remove \"\n                \"'schema_registry_settings' or use format='json'.\"\n            )\n\n        key_field_index = None\n        header_fields: dict[str, int] = {}\n        extracted_field_indices: dict[str, int] = {}\n        columns_to_extract: list[ColumnReference] = []\n\n        if topic_name is not None:\n            topic_name_index = cls.add_column_reference_to_extract(\n                topic_name, columns_to_extract, extracted_field_indices\n            )\n            if topic_name._column.dtype not in (dt.STR, dt.ANY):\n                raise ValueError(\n                    \"The topic name column must have a string type, however \"\n                    f\"{topic_name._column.dtype.typehint} is used\"\n                )\n        else:\n            topic_name_index = None\n\n        # Common part for all formats: obtain key field index and prepare header fields\n        if key is not None:\n            if (\n                allowed_key_types is not None\n                and table[key._name]._column.dtype not in allowed_key_types\n            ):\n                raise ValueError(\n                    f\"The key column must have one of the following types: {allowed_key_types}\"\n                )\n            key_field_index = cls.add_column_reference_to_extract(\n                key, columns_to_extract, extracted_field_indices\n            )","sourceCodeStart":477,"sourceCodeEnd":513,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/_utils.py#L477-L513","documentation":"Raised by MessageQueueOutputFormat.build when the column passed as topic_name does not have dtype STR (or ANY, which is not yet constrained). Because the connector routes each row to a Kafka topic whose name comes from that column's runtime value, the value must be a string; passing a typed non-string column is rejected before the pipeline runs. The message reports the offending dtype.","triggerScenarios":"pw.io.kafka.write(t, ..., topic_name=pw.this.topic_id) where topic_id is int or bytes; computing the topic column with an expression that yields a non-string type; using a column typed as dt.INT as the routing column.","commonSituations":"Topic-routing columns stored as integer IDs in the source table; upstream schema changes that retyped the routing column from str to int.","solutions":["Cast the routing column to string, e.g. topic_name=t.select(topic=pw.this.topic_id.astype(str)).topic or apply .astype(pw.Text) in a select.","Or point topic_name at an existing string column.","If the column is genuinely string-valued but typed ANY, the check passes (ANY is allowed) — only concrete non-string dtypes fail."],"exampleFix":"# before\npw.io.kafka.write(t, ..., topic_name=pw.this.topic_id)  # int column\n\n# after\nt = t.with_columns(topic_name_str=pw.this.topic_id.astype(str))\npw.io.kafka.write(t, ..., topic_name=pw.this.topic_name_str)","handlingStrategy":"type-guard","validationCode":"routing = table[routing_col]\nassert routing._column.dtype in (dt.STR, dt.ANY), f\"topic_name column must be str, got {routing._column.dtype}\"","typeGuard":"def is_string_or_any(dtype) -> bool:\n    return dtype in (dt.STR, dt.ANY)","tryCatchPattern":"try:\n    pw.io.kafka.write(t, ..., topic_name=pw.this[c])\nexcept ValueError as e:\n    if 'topic name column' in str(e):\n        t = t.with_columns(**{c + '_str': pw.this[c].astype(str)})\n        pw.io.kafka.write(t, ..., topic_name=pw.this[c + '_str'])\n    else:\n        raise","preventionTips":["Select routing columns explicitly as str: t.select(topic=pw.this.topic_id.astype(str)).","After schema changes, re-check dtypes of key/value/topic columns used by sinks."],"tags":["pathway","kafka","dtype","topic-routing","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}