{"record":{"id":"a03b0375848a48f5","repo":"pathwaycom/pathway","slug":"column-name-s-sorted-reserved-r-collide-with-t-a03b03","errorCode":null,"errorMessage":"Column name(s) {sorted(reserved)!r} collide with the reserved fields written by pw.io.mongodb.write in 'stream_of_changes' mode. Rename the column(s) or use output_table_type='snapshot'.","messagePattern":"Column name\\(s\\) (.+?) collide with the reserved fields written by pw\\.io\\.mongodb\\.write in 'stream_of_changes' mode\\. Rename the column\\(s\\) or use output_table_type='snapshot'\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/mongodb/__init__.py","lineNumber":657,"sourceCode":"    (``pathway spawn -n N``), the write is distributed across them, and write\n    throughput grows with the worker count up to the capacity of the target\n    MongoDB/Atlas deployment. Each document is written by a single worker, so the\n    result is the same as with one worker. The exception is ``sort_by``: requesting\n    a global order within a minibatch makes the connector write from a single\n    worker, so a sorted output does not benefit from additional workers.\n    \"\"\"\n    is_snapshot_mode = output_table_type == SNAPSHOT_OUTPUT_TABLE_TYPE\n    column_names = set(table.schema.column_names())\n    if \"_id\" in column_names:\n        raise ValueError(\n            \"Column name '_id' is reserved: MongoDB uses '_id' as the primary key \"\n            \"for every document, so pw.io.mongodb.write cannot accept a column with \"\n            \"this name. Rename the column before writing.\"\n        )\n    if not is_snapshot_mode:\n        reserved = {\"diff\", \"time\"} & column_names\n        if reserved:\n            raise ValueError(\n                f\"Column name(s) {sorted(reserved)!r} collide with the reserved \"\n                f\"fields written by pw.io.mongodb.write in 'stream_of_changes' mode. \"\n                f\"Rename the column(s) or use output_table_type='snapshot'.\"\n            )\n    data_storage = api.DataStorage(\n        storage_type=\"mongodb\",\n        connection_string=connection_string,\n        database=database,\n        table_name=collection,\n        max_batch_size=max_batch_size,\n        snapshot_maintenance_on_output=is_snapshot_mode,\n    )\n    data_format = api.DataFormat(\n        format_type=\"bson\",\n        key_field_names=[],\n        value_fields=_format_output_value_fields(table),\n        with_special_fields=not is_snapshot_mode,\n    )","sourceCodeStart":639,"sourceCodeEnd":675,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/mongodb/__init__.py#L639-L675","documentation":"In output_table_type='stream_of_changes' mode, pw.io.mongodb.write adds its own 'diff' and 'time' fields to each document to express row changes. Input columns named 'diff' or 'time' would be overwritten (or shadow those fields), so the connector rejects such schemas unless you switch to snapshot mode, which writes plain rows without change metadata.","triggerScenarios":"pw.io.mongodb.write(table, uri, db, coll, output_table_type='stream_of_changes') (the default for change output) with table columns named 'diff' or 'time'.","commonSituations":"CDC-style pipelines where the source naturally has a 'time' timestamp column or a 'diff' delta column; tables produced from audit logs with both names.","solutions":["Rename the conflicting columns: table.rename(time='event_time', diff='row_diff')","Switch to output_table_type='snapshot' if you want whole-collection snapshots without diff/time metadata","Drop the columns if they are redundant with the connector's own change fields"],"exampleFix":"# before\npw.io.mongodb.write(t, uri, db, coll)  # t has 'time' column, stream mode\n\n# after\nt = t.rename(time='event_time')\npw.io.mongodb.write(t, uri, db, coll)\n# or: pw.io.mongodb.write(t, uri, db, coll, output_table_type='snapshot')","handlingStrategy":"validation","validationCode":"names = set(table.schema.column_names())\nif output_table_type != \"snapshot\":\n    clash = {\"diff\", \"time\"} & names\n    assert not clash, f\"Rename {sorted(clash)} or use output_table_type='snapshot'\"","typeGuard":"def stream_safe_columns(table: pw.Table) -> bool:\n    return not ({\"diff\", \"time\"} & set(table.schema.column_names()))","tryCatchPattern":"try:\n    pw.io.mongodb.write(table, uri, db, coll)\nexcept ValueError as e:\n    if \"collide with the reserved fields\" in str(e):\n        table = table.rename(time=\"event_time\", diff=\"row_diff\")\n        pw.io.mongodb.write(table, uri, db, coll)\n    else:\n        raise","preventionTips":["Avoid naming columns 'diff' or 'time' in tables destined for stream_of_changes output","Prefer explicit timestamps names like event_time/updated_at","Choose snapshot mode when you don't need change metadata in MongoDB"],"tags":["mongodb","pathway","reserved-name","change-data-capture","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}