pathwaycom/pathway · error · ValueError

The key column must have one of the following types: {allowe

Error message

The key column must have one of the following types: {allowed_key_types}

What it means

Raised by MessageQueueOutputFormat.build when the column passed as 'key' has a dtype outside allowed_key_types (by default (BYTES, STR, ANY)). Kafka message keys are serialized as bytes, so only byte-like or string columns are accepted by default; the error lists the tuple of permitted dtypes so you know exactly what is allowed. Connectors may narrow or widen allowed_key_types.

Source

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

        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
            )
        if headers is not None:
            reserved_header_names = {"pathway_time", "pathway_diff"}
            for header in headers:
                if header.name in reserved_header_names:
                    raise ValueError(
                        f"{header.name!r} is reserved for the Pathway-injected "
                        "headers (pathway_time / pathway_diff) and cannot be "
                        "used as a user header name. Alias the column to "
                        "another name with `table.select(<new_name>=...)`."
                    )
                if header.name in header_fields:
                    raise ValueError(
                        f"Duplicate header name {header.name!r}: two columns "

View on GitHub (pinned to fa2f74a464)

Solutions

1. Cast the key column: key on pw.this.user_id.astype(str) (e.g. t.with_columns(key=pw.this.user_id.astype(str))).
2. Or pick a column that is already BYTES/STR.
3. As a last resort, pass allowed_key_types=(dt.INT_64, ...) if the underlying serializer genuinely supports that type.

Example fix

# before
pw.io.kafka.write(t, ..., key=pw.this.user_id)  # int

# after
t = t.with_columns(user_id_str=pw.this.user_id.astype(str))
pw.io.kafka.write(t, ..., key=pw.this.user_id_str)

key_dtype = table[key_name]._column.dtype
assert allowed_key_types is None or key_dtype in allowed_key_types, f"key dtype {key_dtype} not in {allowed_key_types}"

def key_dtype_allowed(dtype, allowed=(dt.BYTES, dt.STR, dt.ANY)) -> bool:
    return dtype in allowed

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

When it happens

Trigger: pw.io.kafka.write(t, ..., key=pw.this.user_id) where user_id is int or a list column; using a struct column as key; a connector that passes a stricter allowed_key_types tuple.

Common situations: Auto-increment integer primary keys used as Kafka keys; upstream re-typing of the key column during schema evolution.

Related errors

• The topic name column must have a string type, however {topi
• 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/bf4e52103b2391ba. Report an issue: GitHub.