MongoDB

Document operations against a MongoDB connection. Reach for it when the shape of what you are storing varies — event records, payloads from a third party, anything you would otherwise model as a table full of nullable columns.

Type: mongodb · Category: Datasource · Ports: one input · one output

When to use it

• Storing records whose fields differ from one to the next
• Keeping a raw copy of an inbound integration payload
• Reading recent events back for a screen or a report

Settings

Setting	Key	Default	Purpose
Connection	connectionId	—	Which MongoDB connection.
Database	database	—	Target database.
Collection	collection	—	Target collection.
Operation	operation	find	find, insert, update, delete.
Filter	filter	—	The query document, templated from the message.
Update	update	—	The update document, for update.
Limit	limit	0 (no limit)	Cap on documents returned.
Timeout	timeout	0 (connection default)	Per-operation timeout.

Example A — reading

Find the last ten downtime events for a line.

Operation:  find
Database:   plant
Collection: downtime
Limit:      10

{
  "line": "{{payload.line}}",
  "startedAt": { "$gte": "{{payload.since}}" }
}

In

{ "payload": { "line": "LINE1", "since": "2026-08-12T06:00:00Z" } }

Out

{
  "payload": [
    {
      "_id": "66b9…",
      "line": "LINE1",
      "reason": "Infeed jam",
      "startedAt": "2026-08-12T06:41:12Z",
      "durationSeconds": 214
    }
  ]
}

As with SQL, find returns an array and no match returns [].

Example B — writing

Operation:  insert
Database:   plant
Collection: downtime

{
  "line": "{{payload.line}}",
  "reason": "{{payload.reason}}",
  "startedAt": "{{payload.startedAt}}",
  "durationSeconds": {{payload.durationSeconds}},
  "recordedBy": "pipeline"
}

Out

{ "payload": { "insertedIds": ["66ba…"], "insertedCount": 1 } }

Example C — update, and why it should be an upsert

Filter:

{ "line": "{{payload.line}}", "shift": "{{payload.shift}}", "date": "{{payload.date}}" }

Update:

{ "$set": { "units": {{payload.units}}, "oee": {{payload.oee}} },
  "$setOnInsert": { "createdBy": "pipeline" } }

Enable upsert on the operation. Store-and-forward may replay a write after a timeout, and an upsert keyed on the natural identity of the row is what makes that replay harmless.

Gotchas

• Mind the quoting. "{{payload.line}}" is a string; {{payload.durationSeconds}} is a number. Leaving the quotes on a number stores it as text, and it will not compare or sum later.
• A path that does not exist is left as literal text, not blanked — "{{payload.line}}" on a message with no line field stays the eight characters {{payload.line}} and the filter matches nothing. That is the safe failure for a filter (a null would match every document with no line, and on a delete that is the whole collection), but it is silent: a find returns zero documents and looks like an empty result rather than a typo. Unquoted, the same miss leaves {{payload.units}} where a number belongs and the operation fails to parse instead.
• A collection with no index is a full scan on every read. Index whatever your filter uses.

See also

• MongoDB connections