{"record":{"id":"6bcd73eaed513fd4","repo":"pathwaycom/pathway","slug":"defining-a-primary-key-in-the-schema-is-not-suppor-6bcd73","errorCode":null,"errorMessage":"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.","messagePattern":"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\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/mongodb/__init__.py","lineNumber":278,"sourceCode":"    ``name`` to ``pw.io.mongodb.read()`` so the engine can find the saved offset:\n\n    >>> table = pw.io.mongodb.read(\n    ...     \"mongodb://127.0.0.1:27017/?replicaSet=rs0\",\n    ...     database=\"shop\",\n    ...     collection=\"orders\",\n    ...     schema=OrderSchema,\n    ...     name=\"orders_source\",\n    ... )\n    >>> pw.run(persistence_config=persistence_config)  # doctest: +SKIP\n\n    If the program is restarted, it will resume from the saved oplog position and\n    emit only the changes that arrived after the previous run terminated, without\n    replaying the initial snapshot.\n    \"\"\"\n    _check_entitlements(\"mongodb-oplog-reader\")\n\n    if schema.primary_key_columns():\n        raise ValueError(\n            \"Defining a primary key in the schema is not supported for pw.io.mongodb.read. \"\n            \"The connector maintains a snapshot of the MongoDB collection keyed by the \"\n            \"document's _id field. Using a different primary key could cause mismatches \"\n            \"between Pathway's internal state and the actual collection contents. \"\n            \"If you need to reindex the resulting table by a different column, use \"\n            \"pw.Table.with_id_from() after reading.\"\n        )\n\n    data_storage = api.DataStorage(\n        storage_type=\"mongodb\",\n        connection_string=connection_string,\n        database=database,\n        table_name=collection,\n        mode=internal_connector_mode(mode),\n    )\n\n    schema, api_schema = read_schema(schema)\n    data_format = api.DataFormat(","sourceCodeStart":260,"sourceCodeEnd":296,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/mongodb/__init__.py#L260-L296","documentation":"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.","triggerScenarios":"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))).","commonSituations":"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.","solutions":["Remove primary_key=True from the schema columns passed to pw.io.mongodb.read — the _id becomes the table id automatically","If you need a different key downstream, re-index after reading: table.with_id_from(table.some_column)","Keep two schema classes if the same schema is reused by connectors that do support primary keys"],"exampleFix":"# before\nclass OrderSchema(pw.Schema):\n    order_id: str = pw.column_definition(primary_key=True)\n    total: float\n\nt = pw.io.mongodb.read(uri, db, coll, schema=OrderSchema)\n\n# after\nclass OrderSchema(pw.Schema):\n    order_id: str\n    total: float\n\nt = pw.io.mongodb.read(uri, db, coll, schema=OrderSchema)\nt = t.with_id_from(t.order_id)","handlingStrategy":"validation","validationCode":"if schema.primary_key_columns():\n    raise ValueError(\n        \"MongoDB reader keys by _id; declare the schema without primary keys \"\n        \"and use table.with_id_from() afterwards\"\n    )","typeGuard":"def schema_has_primary_key(schema: type[pw.Schema]) -> bool:\n    return bool(schema.primary_key_columns())","tryCatchPattern":"try:\n    pw.io.mongodb.read(uri, db, coll, schema=MySchema)\nexcept ValueError as e:\n    if \"primary key in the schema is not supported\" in str(e):\n        MySchema = type(\"PlainSchema\", (pw.Schema,), {k: pw.column_definition(dtype=v.type) for k, v in MySchema.typehints().items()})\n    else:\n        raise","preventionTips":["Keep a separate, primary-key-free schema class for the MongoDB reader","Re-index after reading with with_id_from() instead of declaring keys","Add a unit test asserting the passed schema has no primary keys"],"tags":["mongodb","pathway","schema","primary-key","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}