{"record":{"id":"4feca197d9e4bf45","repo":"pathwaycom/pathway","slug":"pathway-schema-column-s-offending-collide-with","errorCode":null,"errorMessage":"Pathway schema column(s) {offending} collide with the 'time' and 'diff' metadata columns appended in stream_of_changes mode. Rename the column(s) in your schema or switch to output_table_type='snapshot' which does not append these metadata columns.","messagePattern":"Pathway schema column\\(s\\) (.+?) collide with the 'time' and 'diff' metadata columns appended in stream_of_changes mode\\. Rename the column\\(s\\) in your schema or switch to output_table_type='snapshot' which does not append these metadata columns\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/mysql/__init__.py","lineNumber":402,"sourceCode":"    is_snapshot_mode = output_table_type == SNAPSHOT_OUTPUT_TABLE_TYPE\n\n    # Stream-of-changes mode appends ``time BIGINT NOT NULL, diff SMALLINT\n    # NOT NULL`` metadata columns to the generated CREATE TABLE / INSERT. A\n    # user column with one of those names (case-insensitive -- MySQL column\n    # names are not case-sensitive) would otherwise be emitted twice and\n    # the engine worker would fail mid-run with an opaque \"Duplicate column\n    # name\" error. Snapshot mode does not append these columns, so the\n    # check only applies in stream mode.\n    if not is_snapshot_mode:\n        offending = sorted(\n            {\n                column_name\n                for column_name in table.schema.column_names()\n                if column_name.lower() in (\"time\", \"diff\")\n            }\n        )\n        if offending:\n            raise ValueError(\n                f\"Pathway schema column(s) {offending} collide with \"\n                \"the 'time' and 'diff' metadata columns appended in \"\n                \"stream_of_changes mode. Rename the column(s) in your \"\n                \"schema or switch to output_table_type='snapshot' \"\n                \"which does not append these metadata columns.\"\n            )\n\n    data_storage = api.DataStorage(\n        storage_type=\"mysql\",\n        connection_string=connection_string,\n        max_batch_size=max_batch_size,\n        table_name=table_name,\n        table_writer_init_mode=init_mode_from_str(init_mode),\n        snapshot_maintenance_on_output=is_snapshot_mode,\n    )\n\n    key_field_names = None\n    if primary_key is not None:","sourceCodeStart":384,"sourceCodeEnd":420,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/mysql/__init__.py#L384-L420","documentation":"In stream-of-changes mode, the MySQL sink appends 'time' and 'diff' metadata columns to the destination table; a schema column with the same name (case-insensitively) would make the engine fail mid-run with an opaque 'Duplicate column name' error. pw.io.mysql.write raises this ValueError at call time naming the colliding columns. Snapshot mode does not append these columns, so the check only applies in stream mode.","triggerScenarios":"Calling pw.io.mysql.write(table, table_name) with the default output_table_type while the table schema contains a column named time, Time, diff, or DIFF.","commonSituations":"Streaming event tables with a 'time' timestamp column; audit/change tables with a 'diff' column; porting pipelines from other sinks where these names are allowed.","solutions":["Rename the colliding column(s) before the write, e.g. time -> event_time via table.select() or table.rename().","Or pass output_table_type=\"snapshot\" to skip appending metadata columns.","Adopt a naming convention that avoids bare 'time'/'diff' in sink-bound tables."],"exampleFix":"# before\ntable = table.select(table.id, table.time, table.amount)\npw.io.mysql.write(table, \"payments\")\n\n# after\ntable = table.select(table.id, event_time=table.time, table.amount)\npw.io.mysql.write(table, \"payments\")","handlingStrategy":"validation","validationCode":"RESERVED = {\"time\", \"diff\"}\n\nbad = sorted(n for n in table.schema.column_names() if n.lower() in RESERVED)\nif bad and output_table_type != \"snapshot\":\n    table = table.select(**{\n        **{c: pw.this[c] for c in table.schema.column_names()},\n        **{c: pw.this[c] for c in bad},\n    }).rename(**{c: f\"src_{c}\" for c in bad})","typeGuard":"def is_safe_for_stream_sink(column_names: list[str]) -> bool:\n    return all(n.lower() not in (\"time\", \"diff\") for n in column_names)","tryCatchPattern":"try:\n    pw.io.mysql.write(table, \"events\")\nexcept ValueError as e:\n    if \"metadata columns\" in str(e):\n        pw.io.mysql.write(table, \"events\", output_table_type=\"snapshot\")\n    else:\n        raise","preventionTips":["Rename 'time'/'diff' columns (event_time, change_diff) before MySQL stream sinks.","Centralize final-projection selects before sinks so renaming is one obvious place.","Snapshot mode avoids the reserved columns entirely if renaming is not an option."],"tags":["mysql","reserved-columns","stream-mode","sink","pathway"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}