Keyboard shortcuts

Press or to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

mongo_aggregate

Runs an aggregation pipeline, the surface find() cannot reach: $group, $unwind, $lookup, $facet, and the rest. Stages are extended JSON, and stage names are allowlisted. The read-only stages are always available.

Synopsis

{
  "name": "mongo_aggregate",
  "input": {
    "connector": "mongo",
    "database": "shop",
    "collection": "recordings",
    "pipeline": [],
    "allow_disk_use": false,
    "output": "temp_data.by_quality"
  }
}

Description

mongo_aggregate is a connector function. It names a connector for its credentials and endpoint. Orion validates its input when the workflow is saved, and the call runs through the connector’s circuit breaker.

Retry safety: read. See Retry safety for what the answer costs.

The write stages $out/$merge run only on a connector that sets aggregate_write_stages: true; the default is false, because an aggregation must not silently write. An unknown stage is refused by name, at authoring time for a literal pipeline and again at runtime after {"var": ..} substitution. Message data therefore cannot smuggle a stage in.

Fields

FieldTypeRequiredDefaultDescription
connectorstringyesName of the MongoDB connector
databasestringyesDatabase name
collectionstringyesCollection name
pipelinearrayyesAggregation stages, each {"$stage": …} (extended JSON)
allow_disk_useboolnofalseLet the server spill large stages to disk
outputstring | JSONLogicno"data"Dotted path where result documents are written

Results are bounded by query.max_limit like mongo_read; a $out/$merge pipeline returns an empty array (Mongo’s own contract for those stages).

Examples

{
  "name": "mongo_aggregate",
  "input": {
    "connector": "mongo",
    "database": "shop",
    "collection": "recordings",
    "pipeline": [
      { "$match": { "meetingId": { "var": "data.meeting_id" } } },
      { "$unwind": "$videos" },
      { "$group": { "_id": "$videos.quality", "count": { "$sum": 1 } } }
    ],
    "output": "temp_data.by_quality"
  }
}

Last verified 14 September 2026