Upsert generator

Upsert into MongoDB: bulkWrite updateOne { upsert: true }

Name your table, key and columns and get the statement MongoDB actually accepts, with a guard so an older row never overwrites a newer one. Below it: how the mechanic works, what breaks, and how Datrise loads MongoDB incrementally.

Generate the statement

Statement · bulkWrite updateOne { upsert: true }

// staged: the fresh documents for this batch
const ops = staged.map((doc) => ({
  updateOne: {
    filter: { id: doc.id, updated_at: { $lt: doc.updated_at } },
    update: {
      $set: {
        name: doc.name,
        stage: doc.stage,
        amount: doc.amount,
        owner_id: doc.owner_id,
        updated_at: doc.updated_at,
      },
      $setOnInsert: { id: doc.id },
    },
    upsert: true,
  },
}));

await db.collection("deals").bulkWrite(ops, { ordered: false });

How the upsert works in MongoDB

MongoDB upserts with updateOne({ filter }, { $set }, { upsert: true }), and the efficient way to run thousands of them is a single bulkWrite with ordered: false so one failure does not stop the batch. The generated code sets only the listed fields, uses $setOnInsert for the key so new documents carry it, and matches on the business key. Every document keeps its own shape, so a field the source stopped sending stays until you $unset it.

Create a unique index on the key before the first load. Without it, two concurrent upserts for a key that does not exist yet can both insert, and you end up with duplicate documents that no later upsert will reconcile. With it, the loser gets an E11000 duplicate key error you retry. Keep field types consistent across documents; a string amount in one and a number in another breaks aggregation.

Before you run it

Create a unique index on the key first. Two concurrent upserts for a new key can both insert; the unique index turns the loser into an E11000 error you retry, instead of a duplicate document.
$set only touches the listed fields, so fields the source stopped sending stay as they were; use $unset (or replaceOne) if a missing field should disappear.
The watermark filter only matches when the stored document is older. If it does not match and the key exists, MongoDB will try to upsert a second document and hit the unique index, so drop the guard when you cannot guarantee monotonic timestamps.

Questions people ask

updateOne with upsert, or replaceOne?

updateOne with $set when the source sends partial documents; replaceOne when each staged document is complete and stale fields should disappear.

Why did I get duplicate documents?

Concurrent upserts raced on a key with no unique index. Add { unique: true } on the key and retry E11000 errors; the second attempt matches and updates.

How do I apply the watermark guard safely?

Only when updated_at is monotonic per key. If the guard does not match because the stored document is newer, the upsert tries to insert and the unique index rejects it, which is the intended outcome.

The same generator for other destinations

Skip writing the merge at all

Datrise lands CRM and SaaS entities into MongoDB with this exact mechanic, a watermark on updated-at, and typed columns, so the statement above is what runs on your behalf. Join the waitlist to get early access.

Browse the integration catalog