When to Use
- Sales revenue reporting: A scheduled weekly workflow runs an aggregation pipeline over the
orders collection — $match on date range and status, $group by region and product category, $project to shape the output — then feeds the result into a report generation and email delivery workflow. No intermediate data export required.
- Customer order history with product details: Use
$lookup to join the orders collection with the products collection on sku, then $unwind the line items and $project a flat enriched view. This replaces multiple sequential find queries with a single efficient pipeline execution.
- Atlas Full-Text Search with scoring: Combine a
$search stage (available on Atlas M10+) with $match, $project, and $limit to perform intelligent full-text search across product descriptions or support ticket content, ranked by relevance. Feed results directly into an AI-assisted response workflow.
- Real-time inventory shortage detection: A nightly aggregation groups inventory records by
warehouseId, sums stockQty per sku, then filters groups where total stock falls below reorderThreshold. The output list feeds a procurement notification workflow — all without a separate BI system.
- Cohort analysis for SaaS metrics: Aggregate subscription records to compute monthly active users, churn rate by plan tier, and average revenue per account. Group by
planTier and signupMonth, compute $avg and $sum on revenue fields, and project the KPIs needed for the executive dashboard update workflow.
Configuration
Connection
| Field | Required | Description |
|---|
ConnectionUri | Required | MongoDB connection string. Reference via {{ $credentials.mongodbAtlas }}. |
Database | Required | Target database name. |
Collection | Required | The source collection for the first pipeline stage. |
Operation
| Field | Required | Description |
|---|
Pipeline | Required | Array of pipeline stage objects. Each element is a single-key object where the key is the stage operator (e.g. $match, $group). Minimum one stage required. BizFirst expressions are evaluated within stage values before execution. Example: [{"$match": {"tenantId": "{{ $json.tenantId }}"}}, {"$group": {"_id": "$status", "count": {"$sum": 1}}}]. |
AllowDiskUse | Optional | Default false. When true, MongoDB is permitted to spill intermediate results to disk if the in-memory sort or group buffer exceeds 100 MB. Enable for large aggregations involving many documents or complex multi-stage pipelines. Has a performance cost — enable only when needed. |
Pipeline stage reference. All standard MongoDB aggregation stages are supported: $addFields, $bucket, $count, $facet, $geoNear, $graphLookup, $group, $limit, $lookup, $match, $merge, $out, $project, $redact, $replaceRoot, $sample, $search (Atlas), $set, $skip, $sort, $sortByCount, $unset, $unwind.
Sample Configuration
Weekly revenue summary by product category for a specific tenant and date range:
{
"resource": "document",
"operation": "aggregate",
"ConnectionUri": "{{ $credentials.mongodbAtlas }}",
"Database": "ecommerce",
"Collection": "orders",
"Pipeline": [
{
"$match": {
"tenantId": "{{ $json.tenantId }}",
"status": "completed",
"completedAt": {
"$gte": "{{ $json.weekStart }}",
"$lt": "{{ $json.weekEnd }}"
}
}
},
{ "$unwind": "$lineItems" },
{
"$group": {
"_id": "$lineItems.category",
"totalRevenue": { "$sum": { "$multiply": ["$lineItems.qty", "$lineItems.unitPrice"] } },
"orderCount": { "$sum": 1 },
"avgOrderValue":{ "$avg": "$total" }
}
},
{ "$sort": { "totalRevenue": -1 } },
{
"$project": {
"_id": 0,
"category": "$_id",
"totalRevenue": { "$round": ["$totalRevenue", 2] },
"orderCount": 1,
"avgOrderValue": { "$round": ["$avgOrderValue", 2] }
}
}
],
"AllowDiskUse": false
}
Validation Errors
| Error | Cause |
|---|
connectionUri is required | ConnectionUri is empty or not provided. |
database is required | Database field is empty. |
collection is required | Collection field is empty. |
pipeline array cannot be empty | Pipeline is null, an empty array [], or not an array. At least one stage is required. |
Output — Success Port
| Field | Type | Description |
|---|
count | integer | Number of documents in the pipeline result set. For $group operations this is the number of groups; for $match + $project it is the number of matching documents. |
documents | array | The full result set as an array of objects. Each element represents one output document from the final pipeline stage. Empty array [] when the pipeline produces no results. |
status | string | Always "success". |
resource | string | Always "document". |
operation | string | Always "aggregate". |
Sample Output
{
"count": 3,
"documents": [
{
"category": "Footwear",
"totalRevenue": 42180.50,
"orderCount": 287,
"avgOrderValue": 94.22
},
{
"category": "Accessories",
"totalRevenue": 18740.00,
"orderCount": 412,
"avgOrderValue": 45.48
},
{
"category": "Apparel",
"totalRevenue": 11230.75,
"orderCount": 189,
"avgOrderValue": 59.42
}
],
"status": "success",
"resource": "document",
"operation": "aggregate"
}
Expression Reference
| Expression | Returns | Description |
|---|
{{ $output.WeeklySummary.count }} | integer | Number of result documents. Use in IfCondition to branch when the pipeline returns no data. |
{{ $output.WeeklySummary.documents }} | array | Full results array. Pass to a Loop node to process each category row individually (e.g. send per-category alerts). |
{{ $output.WeeklySummary.documents[0] }} | object | First result document — e.g. the top revenue category after sorting. |
{{ $output.WeeklySummary.documents[0].category }} | string | The category field from the top result. |
{{ $output.WeeklySummary.documents[0].totalRevenue }} | number | Numeric field from the aggregation output — use directly in calculations or report templates. |
{{ $output.WeeklySummary.documents | length }} | integer | Equivalent to count — inline length computation. |
Node Policies & GuardRails
| Policy | Recommendation |
|---|
| Credential storage | Always reference ConnectionUri from BizFirst Credentials Manager. Never embed raw credentials in pipeline expressions. |
| Always $match first | Place a $match stage as the first pipeline stage to leverage indexes and reduce the document set before expensive $group, $lookup, or $unwind stages. A pipeline that starts with $group on a large collection performs a full collection scan. |
| AllowDiskUse for large datasets | Enable AllowDiskUse: true for pipelines that sort or group more than 100 MB of intermediate data. Without it, MongoDB aborts the operation with an ExceededMemoryLimit error. Disk spillage is slower — also consider adding a $limit stage to reduce result size. |
| Index the $match fields | All fields used in the initial $match stage should have supporting indexes. Check index usage with explain() in MongoDB Compass or Atlas before deploying aggregations on production collections. |
| Multi-tenant $match | Always include tenantId as the first condition in the $match stage to scope aggregations to the correct tenant's data in shared collections. |
| Limit result set | Add a $limit stage at the end of exploration pipelines. Aggregations that return thousands of documents consume significant memory in the workflow execution context. Project only the fields needed for downstream steps. |
| $lookup performance | Ensure the localField and foreignField in $lookup stages are indexed on both collections. Unindexed lookups perform nested loop scans and become exponentially slow as collection size grows. |
| Atlas $search stage | The $search stage is available only on MongoDB Atlas M10+ clusters. It requires an Atlas Search index on the collection. Verify cluster tier and index existence with searchindex/list before using this stage in production. |
Workflow Examples
Example 1 — Order-Product Join with $lookup
{
"resource": "document",
"operation": "aggregate",
"ConnectionUri": "{{ $credentials.mongodbAtlas }}",
"Database": "ecommerce",
"Collection": "orders",
"Pipeline": [
{ "$match": { "customerId": "{{ $json.customerId }}", "tenantId": "{{ $json.tenantId }}" } },
{ "$unwind": "$lineItems" },
{
"$lookup": {
"from": "products",
"localField": "lineItems.sku",
"foreignField": "sku",
"as": "productDetail"
}
},
{ "$unwind": "$productDetail" },
{
"$project": {
"_id": 0,
"orderId": 1,
"sku": "$lineItems.sku",
"productName": "$productDetail.name",
"qty": "$lineItems.qty",
"unitPrice": "$lineItems.unitPrice",
"lineTotal": { "$multiply": ["$lineItems.qty", "$lineItems.unitPrice"] }
}
},
{ "$sort": { "orderId": -1 } },
{ "$limit": 100 }
],
"AllowDiskUse": false
}
Example 2 — Atlas Full-Text Search on Support Tickets
{
"resource": "document",
"operation": "aggregate",
"ConnectionUri": "{{ $credentials.mongodbAtlas }}",
"Database": "support",
"Collection": "tickets",
"Pipeline": [
{
"$search": {
"index": "ticket_search_index",
"text": {
"query": "{{ $json.searchQuery }}",
"path": ["title", "description", "resolution"]
}
}
},
{ "$match": { "tenantId": "{{ $json.tenantId }}", "status": { "$ne": "deleted" } } },
{ "$limit": 20 },
{
"$project": {
"_id": 0,
"ticketId": 1,
"title": 1,
"status": 1,
"priority": 1,
"score": { "$meta": "searchScore" }
}
}
],
"AllowDiskUse": false
}
Example 3 — Monthly Revenue by Tenant KPI Report
{
"resource": "document",
"operation": "aggregate",
"ConnectionUri": "{{ $credentials.mongodbAtlas }}",
"Database": "billing",
"Collection": "invoices",
"Pipeline": [
{
"$match": {
"status": "paid",
"paidAt": {
"$gte": "{{ $json.monthStart }}",
"$lt": "{{ $json.monthEnd }}"
}
}
},
{
"$group": {
"_id": "$tenantId",
"totalRevenue": { "$sum": "$amount" },
"invoiceCount": { "$sum": 1 },
"avgInvoiceValue": { "$avg": "$amount" }
}
},
{ "$sort": { "totalRevenue": -1 } },
{
"$project": {
"_id": 0,
"tenantId": "$_id",
"totalRevenue": { "$round": ["$totalRevenue", 2] },
"invoiceCount": 1,
"avgInvoiceValue": { "$round": ["$avgInvoiceValue", 2] }
}
}
],
"AllowDiskUse": true
}
