pathwaycom/pathway · error · ValueError

Column name(s) {sorted(reserved)!r} collide with the reserve

Error message

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'.

What it means

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.

Source

Thrown at python/pathway/io/mongodb/__init__.py:657

    (``pathway spawn -n N``), the write is distributed across them, and write
    throughput grows with the worker count up to the capacity of the target
    MongoDB/Atlas deployment. Each document is written by a single worker, so the
    result is the same as with one worker. The exception is ``sort_by``: requesting
    a global order within a minibatch makes the connector write from a single
    worker, so a sorted output does not benefit from additional workers.
    """
    is_snapshot_mode = output_table_type == SNAPSHOT_OUTPUT_TABLE_TYPE
    column_names = set(table.schema.column_names())
    if "_id" in column_names:
        raise ValueError(
            "Column name '_id' is reserved: MongoDB uses '_id' as the primary key "
            "for every document, so pw.io.mongodb.write cannot accept a column with "
            "this name. Rename the column before writing."
        )
    if not is_snapshot_mode:
        reserved = {"diff", "time"} & column_names
        if reserved:
            raise ValueError(
                f"Column name(s) {sorted(reserved)!r} collide with the reserved "
                f"fields written by pw.io.mongodb.write in 'stream_of_changes' mode. "
                f"Rename the column(s) or use output_table_type='snapshot'."
            )
    data_storage = api.DataStorage(
        storage_type="mongodb",
        connection_string=connection_string,
        database=database,
        table_name=collection,
        max_batch_size=max_batch_size,
        snapshot_maintenance_on_output=is_snapshot_mode,
    )
    data_format = api.DataFormat(
        format_type="bson",
        key_field_names=[],
        value_fields=_format_output_value_fields(table),
        with_special_fields=not is_snapshot_mode,
    )

View on GitHub (pinned to fa2f74a464)

Solutions

1. Rename the conflicting columns: table.rename(time='event_time', diff='row_diff')
2. Switch to output_table_type='snapshot' if you want whole-collection snapshots without diff/time metadata
3. Drop the columns if they are redundant with the connector's own change fields

Example fix

# before
pw.io.mongodb.write(t, uri, db, coll)  # t has 'time' column, stream mode

# after
t = t.rename(time='event_time')
pw.io.mongodb.write(t, uri, db, coll)
# or: pw.io.mongodb.write(t, uri, db, coll, output_table_type='snapshot')

names = set(table.schema.column_names())
if output_table_type != "snapshot":
    clash = {"diff", "time"} & names
    assert not clash, f"Rename {sorted(clash)} or use output_table_type='snapshot'"

def stream_safe_columns(table: pw.Table) -> bool:
    return not ({"diff", "time"} & set(table.schema.column_names()))

try:
    pw.io.mongodb.write(table, uri, db, coll)
except ValueError as e:
    if "collide with the reserved fields" in str(e):
        table = table.rename(time="event_time", diff="row_diff")
        pw.io.mongodb.write(table, uri, db, coll)
    else:
        raise

When it happens

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

Common situations: 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.

Related errors

• Column name '_id' is reserved: MongoDB uses '_id' as the pri
• Defining a primary key in the schema is not supported for pw
• Failed to install dependencies
• Column {pseudocolumn} has to contain integers only.
• Column {api.TIME_PSEUDOCOLUMN} cannot contain negative times

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