{"record":{"id":"480e507ec8365559","repo":"pathwaycom/pathway","slug":"header-name-r-is-reserved-for-the-pathway-inject","errorCode":null,"errorMessage":"{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>=...)`.","messagePattern":"(.+?) 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>=\\.\\.\\.\\)`\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/_utils.py","lineNumber":518,"sourceCode":"            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 \"\n                        \"produce a header with the same name. Alias one of \"\n                        \"them to a different name (e.g. via `table.select(...)`) \"\n                        \"to keep both as separate Kafka headers.\"\n                    )\n                header_fields[header.name] = cls.add_column_reference_to_extract(\n                    header, columns_to_extract, extracted_field_indices\n                )\n\n        # Format-dependent parts: handle json and dsv separately\n        if format == \"json\" or format == \"dsv\":","sourceCodeStart":500,"sourceCodeEnd":536,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/_utils.py#L500-L536","documentation":"Raised by MessageQueueOutputFormat.build when a column passed in 'headers' is named 'pathway_time' or 'pathway_diff'. The connector injects commit-time and diff headers under those exact names, so a user header with the same name would be overwritten (or collide) at write time; Pathway rejects it and suggests aliasing the column.","triggerScenarios":"pw.io.kafka.write(t, ..., headers=[pw.this.pathway_time, ...]) where the table happens to have a column with that name; tables produced by pw.io.debezium.read or replay modes that carry pathway_time/pathway_diff columns passed wholesale into headers.","commonSituations":"Feeding connector-internal or framework-reserved column names into Kafka headers; generic code that forwards all table columns as headers without filtering.","solutions":["Alias the column to a different name before passing it: headers=[pw.this.select(header_time=pw.this.pathway_time).header_time] or table.select(emit_time=pw.this.pathway_time).","Or drop that column from the headers list if you did not intend to emit it.","Reserve the names pathway_time/pathway_diff in your naming conventions."],"exampleFix":"# before\npw.io.kafka.write(t, ..., headers=[pw.this.pathway_time])\n\n# after\nt = t.select(*[c for c in t.columns], emit_time=pw.this.pathway_time)\npw.io.kafka.write(t, ..., headers=[pw.this.emit_time])","handlingStrategy":"validation","validationCode":"RESERVED = {'pathway_time', 'pathway_diff'}\nbad = [h.name for h in (headers or []) if h.name in RESERVED]\nassert not bad, f\"reserved header names: {bad}; alias them first\"","typeGuard":"def headers_avoid_reserved(headers) -> bool:\n    return all(h.name not in {'pathway_time', 'pathway_diff'} for h in headers or [])","tryCatchPattern":null,"preventionTips":["Filter or alias pathway_time/pathway_diff before passing columns as headers.","Treat pathway_* as a reserved prefix in your column naming conventions."],"tags":["pathway","kafka","headers","reserved-names","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}