Document Transformation Pipelines for Schema Migration

Within the broader Zero-Downtime Schema Migration Automation in MongoDB framework, a document transformation pipeline is the machinery that converts a collection full of legacy-shaped documents into ones that satisfy a new $jsonSchema contract — without pausing writes, saturating the oplog, or ever loading the collection into memory. It is the workhorse that runs in the gap between “the new schema is defined” and “the validator is enforcing in strict + error”: a resumable, idempotent job that reads the exact inventory of non-compliant documents, rewrites each one, confirms the result against the target schema before it lands, and diverts anything it cannot fix to a dead-letter sink instead of dropping it. This guide is the complete implementation reference for platform engineers building that job in pymongo. It covers where the pipeline sits relative to the write path, the prerequisites that keep a live replica set healthy under a bulk rewrite, an idempotent workflow, a full production module built around bulk_write(ordered=False) and a resumable cursor with oplog throttling, the diagnostic fingerprints a stalled or drifting run produces, and the verification and rollback commands that make the whole operation reversible. Two transformations recur often enough to warrant their own playbooks: backfilling new required fields in batches when the schema adds a mandatory field the historical documents lack, and coercing bsonType mismatches in place when fields were stored under the wrong storage type. This page is the framework both of those specialise.

Architectural Context & Enforcement Boundaries

A transformation pipeline is not part of the database’s write-path validator; it is a client-side batch process that runs before enforcement tightens, whose job is to make the eventual collMod to strict a no-op in terms of rejections. MongoDB evaluates a collection validator synchronously on insert, update, replace, and findAndModify only, and never re-scans data when a validator changes — so the population of documents that would fail a newly-tightened schema stays invisible to the engine until an application write touches each one. The pipeline’s entire reason to exist is to touch every one of those documents first, on your schedule rather than a user’s, and normalise it while the collection is still in a permissive posture.

The safe posture to run under is validationLevel: "moderate" with validationAction: "warn", because a partial $set against a document that is currently non-compliant is grandfathered through — the exact mechanics are developed in strict vs moderate validation levels. The pipeline itself has four stages, and each stage exists to keep bad data off the hot path: it reads a bounded batch driven by a $nor + $jsonSchema inventory filter (so only non-compliant documents are ever fetched), transforms each document in Python, re-checks the transformed image against the target schema client-side with Draft4Validator, and then either stages a compliant rewrite into an unordered bulk_write or routes an un-fixable document to a dead-letter collection for human review.

The client-side re-check is what separates a transformation pipeline from a naive updateMany: because it applies the same JSON Schema the server will enforce, a document that passes the pipeline’s own gate is guaranteed to survive the eventual strict promotion, and a document that fails it is caught here — cheaply, in the pipeline — rather than surfacing as a code 121 on a user’s write months later. Documents that reach the dead-letter branch join the disposition machinery described in fallback routing for invalid documents, so tightening a schema never means silent data loss.

Prerequisites & Operational Requirements

Confirm every item below before pointing a transformation job at a production collection.

• MongoDB version: 5.0 or later. The structured errInfo.details.schemaRulesNotSatisfied object you will read to classify residual failures, and the $convert error-handling used by the coercion child page, both require 5.0+. Aggregation-pipeline updates (updateMany with a pipeline stage array) require 4.2+, which is comfortably below the floor.
• Topology: a replica set. The throttling in this pipeline keys off replication lag; on a standalone there is no oplog window to protect and no secondary to fall behind, but you also lose the failover safety the rollback depends on. Never run a bulk rewrite against a standalone you cannot afford to lose.
• Driver: PyMongo 4.x, pinned (pip install "pymongo>=4.6,<5"). Pin jsonschema too (pip install "jsonschema>=4.0") so the Draft4Validator semantics used in the client-side gate stay stable across builds.
• Permissions: the automation principal needs the find and update actions on the target collection and insert on the dead-letter collection. It does not need collMod — the pipeline never changes the validator; the level and action transitions are a separate, more-privileged step. Keep the transformation principal least-privilege.
• Inventory baseline: compute the exact non-compliant count before you start, so you have a burn-down denominator. Because $jsonSchema is a valid query operator, db.coll.countDocuments({ $nor: [{ $jsonSchema: <target> }] }) returns the backlog with no validator active and no data mutation.
• Idempotency key: every transformation must be safe to apply twice. The $nor filter naturally provides this — a document that has already been fixed no longer matches the inventory and is never fetched again — but any computed default must be deterministic so a retried batch produces an identical result.

Idempotent Implementation Workflow

The sequence below is deterministic and safe to re-run at any point; an interruption only ever costs the in-flight batch.

1. Freeze the target schema. Pin the exact $jsonSchema you are converging on as a version-controlled artifact, and version it alongside your schema versioning strategies for NoSQL record so the contract is auditable. The pipeline uses this same object twice — negated as the inventory filter, and directly as the client-side gate.

2. Attach the validator in observe mode. Apply the schema with validationLevel: "moderate" and validationAction: "warn" so writes are never blocked while you remediate, and non-compliant writes are logged for telemetry:

   db.runCommand({
     collMod: "events",
     validator: { $jsonSchema: {
       bsonType: "object",
       required: ["tenant_id", "event_type", "payload_version"],
       properties: {
         tenant_id: { bsonType: "string" },
         event_type: { bsonType: "string" },
         payload_version: { bsonType: "int", minimum: 2 }
       }
     }},
     validationLevel: "moderate",
     validationAction: "warn"
   })
   // Expected output: { ok: 1 }

3. Measure the backlog. Count the inventory so you can track burn-down:

   db.events.countDocuments({ $nor: [{ $jsonSchema: /* target schema */ {} }] })
   // Expected output: an integer, e.g. 48211

4. Run the transformation pipeline. Execute the Python module below. It reads only non-compliant documents, transforms and re-checks each, applies compliant rewrites in unordered batches, and dead-letters the rest. It throttles between batches to keep secondaries within the oplog window.

5. Re-measure and verify. Re-run the step 3 count until it returns 0 (ignoring documents intentionally parked in the dead-letter collection). Only a zero backlog authorises promoting validationAction to "error" — the promotion mechanics themselves live under strict vs moderate validation levels.

Production-Ready Automation Implementation

The module below is the reference transformation runner. It selects work through the $nor + $jsonSchema inventory filter, so it is inherently resumable — a fixed document never re-enters the stream. It batches corrected documents into bulk_write(ordered=False) so a single un-fixable document never aborts a batch, gates every rewrite through a Draft4Validator mirror of the server schema, routes rejects to a dead-letter collection, and throttles between batches by polling replication lag so a bulk rewrite cannot push a secondary off the oplog window. The transform callable is the one part you customise per migration; the two child pages supply concrete versions of it.

import logging
import time
from typing import Any, Callable, Dict, List, Optional

from bson import ObjectId
from jsonschema import Draft4Validator
from pymongo import UpdateOne
from pymongo.collection import Collection
from pymongo.errors import BulkWriteError, PyMongoError

logger = logging.getLogger("transform_pipeline")

# Sentinel returned by a transform to mean "cannot be fixed — dead-letter this doc".
UNFIXABLE = object()


class TransformationPipeline:
    """Resumable, idempotent, oplog-throttled document rewriter for schema migration."""

    def __init__(
        self,
        collection: Collection,
        target_schema: Dict[str, Any],
        transform: Callable[[Dict[str, Any]], Any],
        dead_letter: Collection,
        batch_size: int = 1000,
        max_lag_seconds: float = 10.0,
        pause_seconds: float = 0.5,
    ):
        self.collection = collection
        self.target_schema = target_schema
        self.transform = transform
        self.dead_letter = dead_letter
        self.batch_size = batch_size
        self.max_lag_seconds = max_lag_seconds
        self.pause_seconds = pause_seconds
        self.validator = Draft4Validator(target_schema)
        # Only ever fetch documents that still violate the target schema.
        self.inventory_filter = {"$nor": [{"$jsonSchema": target_schema}]}

    def replication_lag_seconds(self) -> float:
        """Largest secondary-to-primary lag in seconds, from replSetGetStatus."""
        status = self.collection.database.client.admin.command("replSetGetStatus")
        members = status["members"]
        primary = next(m for m in members if m["stateStr"] == "PRIMARY")
        secondaries = [m for m in members if m["stateStr"] == "SECONDARY"]
        if not secondaries:
            return 0.0
        p_ts = primary["optimeDate"]
        return max((p_ts - s["optimeDate"]).total_seconds() for s in secondaries)

    def _throttle(self) -> None:
        """Block until replication has caught up, protecting the oplog window."""
        while self.replication_lag_seconds() > self.max_lag_seconds:
            logger.warning("Replication lag above %.1fs; pausing batch dispatch.",
                           self.max_lag_seconds)
            time.sleep(self.pause_seconds * 4)
        time.sleep(self.pause_seconds)

    def _flush(self, ops: List[UpdateOne]) -> int:
        """Apply a batch of corrected documents; ordered=False isolates failures."""
        if not ops:
            return 0
        try:
            return self.collection.bulk_write(ops, ordered=False).modified_count
        except BulkWriteError as bwe:
            # A residual code-121 here means the client-side gate and the server
            # validator disagree — log the offending docs, keep the batch alive.
            for we in bwe.details.get("writeErrors", []):
                logger.error("bulk_write rejected _id=%s code=%s: %s",
                             we.get("op", {}).get("q"), we.get("code"), we.get("errmsg"))
            return bwe.details.get("nModified", 0)

    def _dead_letter(self, doc: Dict[str, Any], reason: str) -> None:
        self.dead_letter.insert_one({
            "source_id": doc["_id"],
            "document": doc,
            "reason": reason,
            "quarantined_at": time.time(),
        })

    def run(self, projection: Optional[Dict[str, int]] = None) -> Dict[str, int]:
        """Drive the pipeline to completion. Returns a run summary."""
        fixed = dead_lettered = scanned = 0
        ops: List[UpdateOne] = []
        cursor = self.collection.find(
            self.inventory_filter, projection, no_cursor_timeout=False
        ).batch_size(self.batch_size)

        try:
            for doc in cursor:
                scanned += 1
                try:
                    corrected = self.transform(dict(doc))
                except Exception as exc:  # a transform bug must not lose the doc
                    self._dead_letter(doc, f"transform raised: {exc!r}")
                    dead_lettered += 1
                    continue

                if corrected is UNFIXABLE:
                    self._dead_letter(doc, "transform returned UNFIXABLE")
                    dead_lettered += 1
                    continue

                # Client-side gate: mirror the server validator exactly.
                errors = sorted(self.validator.iter_errors(corrected),
                                key=lambda e: list(e.path))
                if errors:
                    self._dead_letter(doc, f"still invalid: {errors[0].message}")
                    dead_lettered += 1
                    continue

                corrected.pop("_id", None)  # never $set an immutable _id
                ops.append(UpdateOne({"_id": doc["_id"]}, {"$set": corrected}))
                if len(ops) >= self.batch_size:
                    fixed += self._flush(ops)
                    ops = []
                    self._throttle()

            fixed += self._flush(ops)
        except PyMongoError as exc:
            logger.error("Pipeline aborted (resumable on restart): %s", exc)
            raise
        finally:
            cursor.close()

        summary = {"scanned": scanned, "fixed": fixed, "dead_lettered": dead_lettered}
        logger.info("Transformation run complete: %s", summary)
        return summary


if __name__ == "__main__":
    from pymongo import MongoClient

    logging.basicConfig(level=logging.INFO)
    client = MongoClient("mongodb://primary:27017/?replicaSet=rs0")
    db = client.production_db

    TARGET = {
        "bsonType": "object",
        "required": ["tenant_id", "event_type", "payload_version"],
        "properties": {
            "tenant_id": {"bsonType": "string"},
            "event_type": {"bsonType": "string"},
            "payload_version": {"bsonType": "int", "minimum": 2},
        },
    }

    def upgrade_payload(doc: Dict[str, Any]) -> Any:
        # Example: stamp the new required payload_version onto legacy events.
        if doc.get("payload_version") is None:
            doc["payload_version"] = 2
        if not isinstance(doc.get("tenant_id"), str):
            return UNFIXABLE  # cannot invent a tenant; quarantine for review
        return doc

    pipeline = TransformationPipeline(
        collection=db.events,
        target_schema=TARGET,
        transform=upgrade_payload,
        dead_letter=db.events_deadletter,
        batch_size=1000,
        max_lag_seconds=10.0,
    )
    pipeline.run(projection=None)

Note the two invariants that make this safe to run against live traffic: the client-side Draft4Validator gate guarantees every staged rewrite will pass the server validator, and the $nor inventory filter guarantees that re-running the job resumes exactly where an interruption left off. The Draft4Validator here is the same tool documented in the python-jsonschema reference; it is deliberately configured with the schema’s Draft 4 semantics to match MongoDB’s engine rather than a newer draft.

Diagnostic Fingerprints & Fast Resolution

A transformation run misbehaves in a small number of recognisable ways. Match the signature and act.

Signature	Root cause	Resolution
BulkWriteError with per-op code 121 during _flush	The client-side gate and the server validator disagree — usually a Draft-2019 keyword in the server schema that Draft4Validator silently ignores.	Rewrite the server $jsonSchema in Draft 4 combinators (allOf/anyOf/oneOf/not) so both sides agree; re-run.
Backlog count stops falling but the job keeps scanning	The transform returns a document that still fails the gate, so it is dead-lettered but never fixed — an infinite-safe no-op, not a loop.	Inspect the dead-letter reason; fix the transform logic or accept the quarantine.
Replication lag climbs and the job pauses repeatedly	batch_size is too large for the oplog throughput, or a secondary is I/O-bound.	Lower batch_size, raise pause_seconds, or widen the oplog before resuming.
pymongo.errors.CursorNotFound mid-run	The snapshot cursor timed out on a very long run under memory pressure.	The $nor filter makes the job resumable — simply restart it; fixed documents are already excluded.
OperationFailure code 13 (Unauthorized) on insert to dead-letter	The principal lacks insert on the dead-letter collection.	Grant insert on the quarantine namespace; the transform principal needs no collMod.

To read why a specific residual document is still failing after a run, pull the structured reason straight from the server by attempting the write and catching it:

from pymongo.errors import WriteError

try:
    db.events.update_one({"_id": stuck_id}, {"$set": {"_touch": True}})
    db.events.update_one({"_id": stuck_id}, {"$unset": {"_touch": ""}})
except WriteError as exc:
    rules = exc.details["errInfo"]["details"]["schemaRulesNotSatisfied"]
    print("Still failing keywords:", rules)

Edge Cases, Gotchas & Known Limitations

• $merge/$out bypass the validator entirely. If any part of your migration rewrites documents through an aggregation $merge into the same collection, those writes skip validation at any level — a later strict update will then reject them. Route rewrites through the bulk_write path this module uses, not through $merge.
• A snapshot cursor is not a transaction. The find cursor sees documents as they were when each batch was fetched; a document modified by live traffic mid-run may be re-read on a later pass. Idempotent $set transforms make this harmless, but a transform that appends to an array must guard against double-application.
• Never $set _id. _id is immutable; including it in the update sub-document raises an error. The module pops it before staging the op — preserve that if you adapt the code.
• Client-side and server schemas must be the same draft. Draft4Validator ignores keywords it does not know (if/then, $defs, format). If the server schema uses a keyword MongoDB rejects at all, the mismatch surfaces as a residual 121; keep both to strict Draft 4.
• Throttling reads the whole replica set on every batch. replSetGetStatus is cheap but not free; on very small batches the poll can dominate. Size batch_size so the throttle check amortises to a small fraction of batch time.
• The dead-letter collection needs its own lifecycle. Documents parked there are still non-compliant; they must be replayed or purged before the backlog truly reaches zero. Do not count them as “fixed”.

Verification & Rollback Procedures

The pipeline is verified by the same $nor count that measured the backlog — a return of 0 (excluding the dead-letter population) is the only proof the collection is ready for enforcement. Follow it with a positive test that a genuinely non-compliant write is still catchable under the target schema:

# Confirm the backlog is drained (dead-lettered docs excluded by design).
remaining = db.events.count_documents({"$nor": [{"$jsonSchema": TARGET}]})
assert remaining == 0, f"{remaining} documents still fail the target schema"
print("Backlog drained; collection is ready for strict promotion.")

Because the pipeline only ever issues $set updates and inserts into a separate dead-letter namespace, it makes no schema-metadata changes to roll back — but the observe-mode validator attached in the workflow does. If a run misbehaves or the backlog burns down incorrectly, revert enforcement posture with a single metadata-only collMod; time-to-recover is the replication lag of your slowest secondary, typically seconds:

// Soft rollback — keep the schema attached for telemetry, stop logging pressure off the hot path.
db.runCommand({ collMod: "events", validationLevel: "moderate", validationAction: "warn" })

// Hard rollback — detach the validator entirely for emergency recovery.
db.runCommand({ collMod: "events", validator: {}, validationLevel: "off" })

The data written by the pipeline is not itself “rolled back” — a corrected document is more compliant than the original and safe to keep. If a transform produced a semantically wrong value (not merely a schema-valid one), recover from a point-in-time backup or snapshot rather than trying to reverse the $set, which is why every transform should be dry-run against a copied collection before it touches production.

Frequently Asked Questions

Related

• Zero-Downtime Schema Migration Automation in MongoDB — the parent framework defining how transformation, indexing, dual-writes, and rollback compose into a live migration.
• Backfilling New Required Fields in Batches — the specialisation of this pipeline for adding a mandatory field the historical documents lack.
• Coercing bsonType Mismatches In Place — the specialisation for fixing fields stored under the wrong BSON type with an aggregation-pipeline update.
• Strict vs Moderate Validation Levels — the permissive posture the pipeline runs under and the promotion it clears the way for.
• Fallback Routing for Invalid Documents — where the dead-letter branch sends documents the pipeline cannot fix in place.