{"record":{"id":"e9c8abadba2d651a","repo":"pathwaycom/pathway","slug":"duplicate-header-name-header-name-r-two-columns","errorCode":null,"errorMessage":"Duplicate header name {header.name!r}: two columns produce a header with the same name. Alias one of them to a different name (e.g. via `table.select(...)`) to keep both as separate Kafka headers.","messagePattern":"Duplicate header name (.+?): two columns produce a header with the same name\\. Alias one of them to a different name \\(e\\.g\\. via `table\\.select\\(\\.\\.\\.\\)`\\) to keep both as separate Kafka headers\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/_utils.py","lineNumber":525,"sourceCode":"            ):\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\":\n            if value is not None:\n                raise ValueError(\n                    f\"'value' and format='{format}' cannot be set at the same time\"\n                )\n            if format == \"json\":\n                reserved = {\"time\", \"diff\"}\n                conflicting = reserved.intersection(table._columns.keys())","sourceCodeStart":507,"sourceCodeEnd":543,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/_utils.py#L507-L543","documentation":"Raised by MessageQueueOutputFormat.build when two entries in the 'headers' list use the same column name. Each header becomes one Kafka header keyed by the column name, so two same-named headers would clobber each other; the duplicate is detected while building the header_fields mapping and rejected with a hint to alias.","triggerScenarios":"pw.io.kafka.write(t, ..., headers=[pw.this.tag, pw.this.tag]) — e.g. the same column listed twice, or two different columns that were both aliased to 'tag' in a select; programmatically building the headers list with a bug that appends a column twice.","commonSituations":"Header lists assembled from multiple sources (base columns + extra columns) that overlap; deduplication missing in generated configs.","solutions":["Alias one of the duplicates: t = t.rename(tag_source=pw.this.tag) and use distinct names in headers.","Deduplicate the headers list before the call: headers=list(dict.fromkeys(headers))."],"exampleFix":"# before\npw.io.kafka.write(t, ..., headers=[pw.this.tag, pw.this.tag])\n\n# after\nheaders = list(dict.fromkeys([pw.this.tag]))  # dedupe\npw.io.kafka.write(t, ..., headers=headers)","handlingStrategy":"validation","validationCode":"names = [h.name for h in headers]\nassert len(names) == len(set(names)), f\"duplicate header names: {sorted({n for n in names if names.count(n) > 1})}\"","typeGuard":"def no_duplicate_headers(headers) -> bool:\n    names = [h.name for h in headers]\n    return len(names) == len(set(names))","tryCatchPattern":null,"preventionTips":["Deduplicate header lists: headers = list(dict.fromkeys(headers)).","When merging header lists from multiple sources, merge by name and detect conflicts."],"tags":["pathway","kafka","headers","duplicate","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}