{"record":{"id":"bf4e52103b2391ba","repo":"pathwaycom/pathway","slug":"the-key-column-must-have-one-of-the-following-type","errorCode":null,"errorMessage":"The key column must have one of the following types: {allowed_key_types}","messagePattern":"The key column must have one of the following types: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/_utils.py","lineNumber":508,"sourceCode":"        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            )\n        if headers is not None:\n            reserved_header_names = {\"pathway_time\", \"pathway_diff\"}\n            for header in headers:\n                if header.name in reserved_header_names:\n                    raise ValueError(\n                        f\"{header.name!r} is reserved for the Pathway-injected \"\n                        \"headers (pathway_time / pathway_diff) and cannot be \"\n                        \"used as a user header name. Alias the column to \"\n                        \"another name with `table.select(<new_name>=...)`.\"\n                    )\n                if header.name in header_fields:\n                    raise ValueError(\n                        f\"Duplicate header name {header.name!r}: two columns \"","sourceCodeStart":490,"sourceCodeEnd":526,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/_utils.py#L490-L526","documentation":"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.","triggerScenarios":"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.","commonSituations":"Auto-increment integer primary keys used as Kafka keys; upstream re-typing of the key column during schema evolution.","solutions":["Cast the key column: key on pw.this.user_id.astype(str) (e.g. t.with_columns(key=pw.this.user_id.astype(str))).","Or pick a column that is already BYTES/STR.","As a last resort, pass allowed_key_types=(dt.INT_64, ...) if the underlying serializer genuinely supports that type."],"exampleFix":"# before\npw.io.kafka.write(t, ..., key=pw.this.user_id)  # int\n\n# after\nt = t.with_columns(user_id_str=pw.this.user_id.astype(str))\npw.io.kafka.write(t, ..., key=pw.this.user_id_str)","handlingStrategy":"type-guard","validationCode":"key_dtype = table[key_name]._column.dtype\nassert allowed_key_types is None or key_dtype in allowed_key_types, f\"key dtype {key_dtype} not in {allowed_key_types}\"","typeGuard":"def key_dtype_allowed(dtype, allowed=(dt.BYTES, dt.STR, dt.ANY)) -> bool:\n    return dtype in allowed","tryCatchPattern":"try:\n    pw.io.kafka.write(t, ..., key=pw.this[k])\nexcept ValueError as e:\n    if 'key column' in str(e):\n        t = t.with_columns(**{k + '_key': pw.this[k].astype(str)})\n        pw.io.kafka.write(t, ..., key=pw.this[k + '_key'])\n    else:\n        raise","preventionTips":["Normalize Kafka keys to strings in your table-preparation step: .astype(str) on id-like columns.","Keep the allowed_key_types tuple in one place if you customize it."],"tags":["pathway","kafka","dtype","key","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}