Aggregation Examples

This guide demonstrates aggregation pipeline patterns for data analysis.

Setup

use pocketdb::{Database, doc, filter, PocketValue};
use pocketdb::query::{
    AggregationPipeline, PipelineStage, SortField, SortDirection,
    Accumulator, AccumulatorOp, Projection,
};

fn main() -> pocketdb::Result<()> {
    let db = Database::open("aggregation_example.pdb")?;
    let sales = db.collection("sales")?;

    // Insert sample sales data
    sales.insert(doc! { "region" => "US", "product" => "A", "amount" => 100, "qty" => 10 })?;
    sales.insert(doc! { "region" => "US", "product" => "B", "amount" => 200, "qty" => 5 })?;
    sales.insert(doc! { "region" => "EU", "product" => "A", "amount" => 150, "qty" => 8 })?;
    sales.insert(doc! { "region" => "EU", "product" => "B", "amount" => 250, "qty" => 12 })?;
    sales.insert(doc! { "region" => "US", "product" => "A", "amount" => 120, "qty" => 6 })?;

    // ... examples ...

    Ok(())
}

Filtering with $match

let results = sales.aggregate()
    .match_(filter! { "region" => == "US" })
    .execute(&sales.all_documents());

println!("US sales: {}", results.len());

Sorting with $sort

use pocketdb::query::SortField;

// Sort by amount descending
let results = sales.aggregate()
    .sort(vec![SortField::desc("amount")])
    .execute(&sales.all_documents());

for doc in &results {
    let product = doc.get("product").unwrap();
    let amount = doc.get("amount").unwrap();
    println!("{}: ${}", product, amount);
}

Limit and Skip

// Top 3 sales
let results = sales.aggregate()
    .sort(vec![SortField::desc("amount")])
    .limit(3)
    .execute(&sales.all_documents());

// Skip the first 2, return the rest
let results = sales.aggregate()
    .sort(vec![SortField::desc("amount")])
    .skip(2)
    .execute(&sales.all_documents());

Projection

// Only show product and amount
let results = sales.aggregate()
    .project(Projection::include(&["product", "amount"]))
    .execute(&sales.all_documents());

// Hide the region field
let results = sales.aggregate()
    .project(Projection::exclude(&["region"]))
    .execute(&sales.all_documents());

Counting

let results = sales.aggregate()
    .count("total_sales")
    .execute(&sales.all_documents());

// Returns: [{ "total_sales": 5 }]

Unwind Arrays

let orders = db.collection("orders")?;
orders.insert(doc! { "customer" => "Alice", "items" => ["Laptop", "Mouse"] })?;
orders.insert(doc! { "customer" => "Bob", "items" => ["Keyboard"] })?;

let results = orders.aggregate()
    .unwind("items")
    .execute(&orders.all_documents());

// Output:
// { "customer": "Alice", "items": "Laptop", "items_index": 0 }
// { "customer": "Alice", "items": "Mouse", "items_index": 1 }
// { "customer": "Bob", "items": "Keyboard", "items_index": 0 }

Grouping

Group by Field

use pocketdb::query::{Accumulator, AccumulatorOp, PipelineStage};

let results = sales.aggregate()
    .add_stage(PipelineStage::Group {
        id: Some("region".to_string()),
        accumulators: vec![
            Accumulator {
                field: "total_amount".to_string(),
                op: AccumulatorOp::Sum,
                input_field: "amount".to_string(),
            },
            Accumulator {
                field: "avg_amount".to_string(),
                op: AccumulatorOp::Avg,
                input_field: "amount".to_string(),
            },
        ],
    })
    .execute(&sales.all_documents());

for doc in &results {
    let region = doc.get("_id").unwrap();
    let total = doc.get("total_amount").unwrap();
    let avg = doc.get("avg_amount").unwrap();
    println!("{}: total={}, avg={}", region, total, avg);
}

Global Group (All Documents)

let results = sales.aggregate()
    .add_stage(PipelineStage::Group {
        id: None,  // No grouping = single group for all documents
        accumulators: vec![
            Accumulator {
                field: "grand_total".to_string(),
                op: AccumulatorOp::Sum,
                input_field: "amount".to_string(),
            },
            Accumulator {
                field: "count".to_string(),
                op: AccumulatorOp::Count,
                input_field: "amount".to_string(),
            },
        ],
    })
    .execute(&sales.all_documents());

// Returns: [{ "_id": null, "grand_total": 820, "count": 5 }]

Min and Max

let results = sales.aggregate()
    .add_stage(PipelineStage::Group {
        id: Some("product".to_string()),
        accumulators: vec![
            Accumulator {
                field: "min_sale".to_string(),
                op: AccumulatorOp::Min,
                input_field: "amount".to_string(),
            },
            Accumulator {
                field: "max_sale".to_string(),
                op: AccumulatorOp::Max,
                input_field: "amount".to_string(),
            },
        ],
    })
    .execute(&sales.all_documents());

Collect into Array

let results = sales.aggregate()
    .add_stage(PipelineStage::Group {
        id: Some("region".to_string()),
        accumulators: vec![
            Accumulator {
                field: "amounts".to_string(),
                op: AccumulatorOp::Push,
                input_field: "amount".to_string(),
            },
        ],
    })
    .execute(&sales.all_documents());
// { "_id": "US", "amounts": [100, 200, 120] }

Complete Pipeline

let results = sales.aggregate()
    // Filter to US region
    .match_(filter! { "region" => == "US" })
    // Group by product, calculate totals
    .add_stage(PipelineStage::Group {
        id: Some("product".to_string()),
        accumulators: vec![
            Accumulator {
                field: "total".to_string(),
                op: AccumulatorOp::Sum,
                input_field: "amount".to_string(),
            },
            Accumulator {
                field: "units".to_string(),
                op: AccumulatorOp::Sum,
                input_field: "qty".to_string(),
            },
            Accumulator {
                field: "deal_count".to_string(),
                op: AccumulatorOp::Count,
                input_field: "amount".to_string(),
            },
        ],
    })
    // Sort by total descending
    .sort(vec![SortField::desc("total")])
    // Return top 5
    .limit(5)
    // Execute
    .execute(&sales.all_documents());

for doc in &results {
    println!("{}", doc);
}