{"record":{"id":"6463d2fd88884ca7","repo":"pathwaycom/pathway","slug":"the-table-has-columns-sorted-conflicting-which","errorCode":null,"errorMessage":"The table has columns {sorted(conflicting)} which clash with the reserved JSON fields written by the connector ('time', 'diff'). Rename or drop the conflicting column(s) before writing in 'json' format, otherwise the output JSON would contain duplicate keys.","messagePattern":"The table has columns (.+?) which clash with the reserved JSON fields written by the connector \\('time', 'diff'\\)\\. Rename or drop the conflicting column\\(s\\) before writing in 'json' format, otherwise the output JSON would contain duplicate keys\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/_utils.py","lineNumber":545,"sourceCode":"                        \"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())\n                if conflicting:\n                    raise ValueError(\n                        f\"The table has columns {sorted(conflicting)} which \"\n                        f\"clash with the reserved JSON fields written by the \"\n                        f\"connector ('time', 'diff'). Rename or drop the \"\n                        f\"conflicting column(s) before writing in 'json' \"\n                        f\"format, otherwise the output JSON would contain \"\n                        f\"duplicate keys.\"\n                    )\n            for column_name in table._columns:\n                cls.add_column_reference_to_extract(\n                    table[column_name], columns_to_extract, extracted_field_indices\n                )\n            table = table.select(*columns_to_extract)\n            data_format = api.DataFormat(\n                format_type=\"jsonlines\" if format == \"json\" else \"dsv\",\n                key_field_names=[],\n                value_fields=_format_output_value_fields(table),\n                delimiter=delimiter,\n                schema_registry_settings=maybe_schema_registry_settings(","sourceCodeStart":527,"sourceCodeEnd":563,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/_utils.py#L527-L563","documentation":"Raised by MessageQueueOutputFormat.build when writing with format='json' and the table contains columns named 'time' or 'diff'. The json output format additionally writes Pathway's update metadata as top-level JSON fields 'time' and 'diff', so identically named table columns would produce duplicate keys in the emitted JSON object; the writer refuses rather than emit ambiguous payloads.","triggerScenarios":"pw.io.kafka.write(t, ..., format='json') where t already has columns named time or diff (e.g. an events table with a 'time' column, or a table that retains pathway's time/diff columns from a reducer or output-mode table).","commonSituations":"Domain models that legitimately use 'time' as a timestamp column name; tables derived from persistence/output formats that add time/diff columns.","solutions":["Rename the conflicting columns before writing: t = t.rename(event_time=pw.this.time).","Or drop them if not needed: t = t.without(pw.this.time, pw.this.diff).","Or choose format='dsv'/'raw' if the metadata fields are not required and renaming is unacceptable."],"exampleFix":"# before\npw.io.kafka.write(t, ..., format='json')  # t has column 'time'\n\n# after\nt = t.rename(event_time=pw.this.time)\npw.io.kafka.write(t, ..., format='json')","handlingStrategy":"validation","validationCode":"conflicting = {'time', 'diff'} & set(table.column_names())\nassert not conflicting, f\"columns {sorted(conflicting)} clash with reserved json fields; rename or drop them\"","typeGuard":"def json_write_safe(table) -> bool:\n    return not ({'time', 'diff'} & set(table.column_names()))","tryCatchPattern":"try:\n    pw.io.kafka.write(t, ..., format='json')\nexcept ValueError as e:\n    if 'reserved JSON fields' in str(e):\n        renames = {c: c + '_' for c in ('time', 'diff') if c in t.column_names()}\n        pw.io.kafka.write(t.rename(**renames), ..., format='json')\n    else:\n        raise","preventionTips":["Avoid 'time' and 'diff' as column names in tables destined for json-format sinks.","Rename defensively before the write: t.rename(event_time=pw.this.time)."],"tags":["pathway","kafka","json","reserved-names","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}