{"record":{"id":"066cf04dc3e3c44d","repo":"pathwaycom/pathway","slug":"pathway-schema-column-s-offending-collide-with-066cf0","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":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/postgres/__init__.py","lineNumber":856,"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. A user\n    # column with one of those names (case-insensitive — PostgreSQL\n    # folds unquoted identifiers to lowercase) would otherwise land\n    # twice in the DDL and the engine worker would panic at pipeline\n    # start with an opaque `db error`. Snapshot mode does not append\n    # these columns, so the check only applies in stream mode.\n    if not is_snapshot_mode:\n        offending = sorted(\n            {\n                name\n                for name in table.schema.column_names()\n                if 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=\"postgres\",\n        connection_string=_connection_string_from_settings(postgres_settings),\n        max_batch_size=max_batch_size,\n        table_name=table_name,\n        schema_name=schema_name,\n        table_writer_init_mode=init_mode_from_str(init_mode),\n        snapshot_maintenance_on_output=is_snapshot_mode,\n        tls_settings=tls.settings,\n    )\n","sourceCodeStart":838,"sourceCodeEnd":874,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/postgres/__init__.py#L838-L874","documentation":"Raised by pw.io.postgres.write() in the default stream_of_changes output mode when the Pathway schema already contains a column named time or diff (case-insensitive). The connector appends its own time and diff bookkeeping columns in that mode, so a same-named user column would appear twice in the generated DDL and crash the engine with an opaque 'db error' — hence the up-front rejection.","triggerScenarios":"pw.io.postgres.write(table_with_time_or_diff_column, parts, \"tbl\") with the default output_table_type (stream_of_changes); schema fields named time, Time, diff, DIFF, etc.","commonSituations":"Writing domain tables that naturally have a time or diff column (timestamps, change/diff flags) — e.g. audit logs, event tables, diff-analysis results.","solutions":["Rename the schema column (e.g. time -> event_time) before writing: table.select(**{('event_time' if c=='time' else c): table[c] for c in table.column_names()}).","Or switch to output_table_type=\"snapshot\" with a primary_key, which appends no time/diff columns (but changes semantics to current-state upserts)."],"exampleFix":"# before\nclass Events(pw.Schema):\n    time: int\n    value: str\npw.io.postgres.write(t, parts, \"events\")\n# after\nclass Events(pw.Schema):\n    event_time: int\n    value: str\npw.io.postgres.write(t, parts, \"events\")","handlingStrategy":"validation","validationCode":"reserved = {\"time\", \"diff\"}\nclash = [c for c in table.schema.column_names() if c.lower() in reserved]\nassert not clash, f\"rename {clash} before stream_of_changes write\"","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Avoid time/diff as schema field names in tables destined for a postgres stream write; prefer event_time/changed.","Add a lint step in your schema definitions for reserved names.","If the names are fixed by upstream contracts, use output_table_type=\"snapshot\"."],"tags":["postgres","pathway","schema","naming-collision","ddl"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}