pathwaycom/pathway · error · ValueError

Defining a primary key in the schema is not supported for pw

Error message

Defining a primary key in the schema is not supported for pw.io.mongodb.read. The connector maintains a snapshot of the MongoDB collection keyed by the document's _id field. Using a different primary key could cause mismatches between Pathway's internal state and the actual collection contents. If you need to reindex the resulting table by a different column, use pw.Table.with_id_from() after reading.

What it means

The MongoDB change-stream reader keys its output by MongoDB's own _id field, so a user-declared primary key in the Pathway schema would desynchronize Pathway's snapshot state from the actual collection. pw.io.mongodb.read therefore rejects any schema with primary_key_columns() and suggests pw.Table.with_id_from() for re-indexing after the read.

Source

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

    ``name`` to ``pw.io.mongodb.read()`` so the engine can find the saved offset:

    >>> table = pw.io.mongodb.read(
    ...     "mongodb://127.0.0.1:27017/?replicaSet=rs0",
    ...     database="shop",
    ...     collection="orders",
    ...     schema=OrderSchema,
    ...     name="orders_source",
    ... )
    >>> pw.run(persistence_config=persistence_config)  # doctest: +SKIP

    If the program is restarted, it will resume from the saved oplog position and
    emit only the changes that arrived after the previous run terminated, without
    replaying the initial snapshot.
    """
    _check_entitlements("mongodb-oplog-reader")

    if schema.primary_key_columns():
        raise ValueError(
            "Defining a primary key in the schema is not supported for pw.io.mongodb.read. "
            "The connector maintains a snapshot of the MongoDB collection keyed by the "
            "document's _id field. Using a different primary key could cause mismatches "
            "between Pathway's internal state and the actual collection contents. "
            "If you need to reindex the resulting table by a different column, use "
            "pw.Table.with_id_from() after reading."
        )

    data_storage = api.DataStorage(
        storage_type="mongodb",
        connection_string=connection_string,
        database=database,
        table_name=collection,
        mode=internal_connector_mode(mode),
    )

    schema, api_schema = read_schema(schema)
    data_format = api.DataFormat(

View on GitHub (pinned to fa2f74a464)

Solutions

1. Remove primary_key=True from the schema columns passed to pw.io.mongodb.read — the _id becomes the table id automatically
2. If you need a different key downstream, re-index after reading: table.with_id_from(table.some_column)
3. Keep two schema classes if the same schema is reused by connectors that do support primary keys

Example fix

# before
class OrderSchema(pw.Schema):
    order_id: str = pw.column_definition(primary_key=True)
    total: float

t = pw.io.mongodb.read(uri, db, coll, schema=OrderSchema)

# after
class OrderSchema(pw.Schema):
    order_id: str
    total: float

t = pw.io.mongodb.read(uri, db, coll, schema=OrderSchema)
t = t.with_id_from(t.order_id)

if schema.primary_key_columns():
    raise ValueError(
        "MongoDB reader keys by _id; declare the schema without primary keys "
        "and use table.with_id_from() afterwards"
    )

def schema_has_primary_key(schema: type[pw.Schema]) -> bool:
    return bool(schema.primary_key_columns())

try:
    pw.io.mongodb.read(uri, db, coll, schema=MySchema)
except ValueError as e:
    if "primary key in the schema is not supported" in str(e):
        MySchema = type("PlainSchema", (pw.Schema,), {k: pw.column_definition(dtype=v.type) for k, v in MySchema.typehints().items()})
    else:
        raise

When it happens

Trigger: pw.io.mongodb.read(uri, db, coll, schema=MySchema) where MySchema defines primary_key_in_column() or marks a column as primary (e.g. class MySchema(pw.Schema, id: int = pw.column_definition(primary_key=True))).

Common situations: Porting a schema from another connector (e.g. csv/kafka) that requires primary keys; assuming Pathway needs the primary key declared for deduplication the way other connectors do.

Related errors

• Column name '_id' is reserved: MongoDB uses '_id' as the pri
• Failed to detect the region of S3 bucket {bucket!r} (HTTP st
• Incorrect S3 path: {s3_path}
• SchemaRegistryHeader.key must be a str, got {type(self.key).
• Schema.with_types() argument name has to be an existing colu

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