pathwaycom/pathway · error · ValueError

The topic name column must have a string type, however {topi

Error message

The topic name column must have a string type, however {topic_name._column.dtype.typehint} is used

What it means

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.

Source

Thrown at python/pathway/io/_utils.py:495

        if schema_registry_settings is not None and format != "json":
            raise ValueError(
                f"'schema_registry_settings' is only meaningful for the 'json' "
                f"format, but {format!r} was specified. The Confluent Schema "
                "Registry currently encodes JSON payloads only; remove "
                "'schema_registry_settings' or use format='json'."
            )

        key_field_index = None
        header_fields: dict[str, int] = {}
        extracted_field_indices: dict[str, int] = {}
        columns_to_extract: list[ColumnReference] = []

        if topic_name is not None:
            topic_name_index = cls.add_column_reference_to_extract(
                topic_name, columns_to_extract, extracted_field_indices
            )
            if topic_name._column.dtype not in (dt.STR, dt.ANY):
                raise ValueError(
                    "The topic name column must have a string type, however "
                    f"{topic_name._column.dtype.typehint} is used"
                )
        else:
            topic_name_index = None

        # Common part for all formats: obtain key field index and prepare header fields
        if key is not None:
            if (
                allowed_key_types is not None
                and table[key._name]._column.dtype not in allowed_key_types
            ):
                raise ValueError(
                    f"The key column must have one of the following types: {allowed_key_types}"
                )
            key_field_index = cls.add_column_reference_to_extract(
                key, columns_to_extract, extracted_field_indices
            )

View on GitHub (pinned to fa2f74a464)

Solutions

1. 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.
2. Or point topic_name at an existing string column.
3. If the column is genuinely string-valued but typed ANY, the check passes (ANY is allowed) — only concrete non-string dtypes fail.

Example fix

# before
pw.io.kafka.write(t, ..., topic_name=pw.this.topic_id)  # int column

# after
t = t.with_columns(topic_name_str=pw.this.topic_id.astype(str))
pw.io.kafka.write(t, ..., topic_name=pw.this.topic_name_str)

routing = table[routing_col]
assert routing._column.dtype in (dt.STR, dt.ANY), f"topic_name column must be str, got {routing._column.dtype}"

def is_string_or_any(dtype) -> bool:
    return dtype in (dt.STR, dt.ANY)

try:
    pw.io.kafka.write(t, ..., topic_name=pw.this[c])
except ValueError as e:
    if 'topic name column' in str(e):
        t = t.with_columns(**{c + '_str': pw.this[c].astype(str)})
        pw.io.kafka.write(t, ..., topic_name=pw.this[c + '_str'])
    else:
        raise

When it happens

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

Common situations: Topic-routing columns stored as integer IDs in the source table; upstream schema changes that retyped the routing column from str to int.

Related errors

• The key column must have one of the following types: {allowe
• The value column must have one of the following types: {allo
• Failed to install dependencies
• Column {pseudocolumn} has to contain integers only.
• Column {api.TIME_PSEUDOCOLUMN} cannot contain negative times

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/72fe0ff109258a6a. Report an issue: GitHub.