pathwaycom/pathway · error · ValueError

Pathway schema column(s) {offending} collide with the 'time'

Error message

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.

What it means

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.

Source

Thrown at python/pathway/io/postgres/__init__.py:856

    is_snapshot_mode = output_table_type == SNAPSHOT_OUTPUT_TABLE_TYPE

    # Stream-of-changes mode appends `time BIGINT NOT NULL, diff SMALLINT
    # NOT NULL` metadata columns to the generated CREATE TABLE. A user
    # column with one of those names (case-insensitive — PostgreSQL
    # folds unquoted identifiers to lowercase) would otherwise land
    # twice in the DDL and the engine worker would panic at pipeline
    # start with an opaque `db error`. Snapshot mode does not append
    # these columns, so the check only applies in stream mode.
    if not is_snapshot_mode:
        offending = sorted(
            {
                name
                for name in table.schema.column_names()
                if name.lower() in ("time", "diff")
            }
        )
        if offending:
            raise ValueError(
                f"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."
            )

    data_storage = api.DataStorage(
        storage_type="postgres",
        connection_string=_connection_string_from_settings(postgres_settings),
        max_batch_size=max_batch_size,
        table_name=table_name,
        schema_name=schema_name,
        table_writer_init_mode=init_mode_from_str(init_mode),
        snapshot_maintenance_on_output=is_snapshot_mode,
        tls_settings=tls.settings,
    )

View on GitHub (pinned to fa2f74a464)

Solutions

1. 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()}).
2. Or switch to output_table_type="snapshot" with a primary_key, which appends no time/diff columns (but changes semantics to current-state upserts).

Example fix

# before
class Events(pw.Schema):
    time: int
    value: str
pw.io.postgres.write(t, parts, "events")
# after
class Events(pw.Schema):
    event_time: int
    value: str
pw.io.postgres.write(t, parts, "events")

reserved = {"time", "diff"}
clash = [c for c in table.schema.column_names() if c.lower() in reserved]
assert not clash, f"rename {clash} before stream_of_changes write"

When it happens

Trigger: 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.

Common situations: Writing domain tables that naturally have a time or diff column (timestamps, change/diff flags) — e.g. audit logs, event tables, diff-analysis results.

Related errors

• Failed to detect the region of S3 bucket {bucket!r} (HTTP st
• SchemaRegistryHeader.value must be a str, got {type(self.val
• argument {name} has incorrect schema
• DateTimeNaive cannot contain timezone information. Use pw.Da
• DateTimeUtc must contain timezone information. Use pw.DateTi

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/066cf04dc3e3c44d. Report an issue: GitHub.