Aggregation Pipelines

PocketDB supports a full aggregation pipeline for data analysis and transformation. Pipelines process documents through a sequence of stages, where each stage's output becomes the next stage's input.

Installation

Add PocketDB to your Cargo.toml:

[dependencies]
pocketdb = "0.1"

Or via the command line:

cargo add pocketdb

Pipeline Stages

$match — Filter Documents

Filters documents using a Filter expression:

use pocketdb::query::AggregationPipeline;
use pocketdb::{filter, PocketValue};

let results = collection.aggregate()
    .match_(filter! { "age" => >= 18, "active" => == true })
    .execute(&collection.all_documents());

$sort — Order Documents

Sort by one or more fields:

use pocketdb::query::{AggregationPipeline, SortField};

let results = collection.aggregate()
    .sort(vec![
        SortField::asc("name"),
        SortField::desc("age"),
    ])
    .execute(&collection.all_documents());

$limit — Limit Results

let results = collection.aggregate()
    .limit(5)
    .execute(&collection.all_documents());

$skip — Skip Documents

let results = collection.aggregate()
    .skip(10)
    .limit(10) // Pagination: page 2 with 10 items per page
    .execute(&collection.all_documents());

$project — Select Fields

Control which fields appear in results:

use pocketdb::query::Projection;

// Include only specific fields
let results = collection.aggregate()
    .project(Projection::include(&["name", "email"]))
    .execute(&collection.all_documents());

// Exclude specific fields
let results = collection.aggregate()
    .project(Projection::exclude(&["password", "internal_id"]))
    .execute(&collection.all_documents());

$unwind — Deconstruct Arrays

Creates one document per array element:

let results = collection.aggregate()
    .unwind("tags")
    .execute(&collection.all_documents());

// Input:  { "name": "Alice", "tags": ["rust", "db"] }
// Output: { "name": "Alice", "tags": "rust", "tags_index": 0 }
//         { "name": "Alice", "tags": "db", "tags_index": 1 }

$count — Count Documents

let results = collection.aggregate()
    .count("total")
    .execute(&collection.all_documents());
// Returns: [{ "total": 42 }]

$sample — Random Selection

Randomly select N documents from the input:

let results = collection.aggregate()
    .sample(5)
    .execute(&collection.all_documents());
// Returns 5 random documents

$set — Add/Modify Fields

Add new fields or overwrite existing ones:

use pocketdb::PocketValue;
use std::collections::HashMap;

let mut fields = HashMap::new();
fields.insert("full_name".to_string(), PocketValue::String("Alice Smith".into()));

let results = collection.aggregate()
    .set(fields)
    .execute(&collection.all_documents());

$unset — Remove Fields

Remove specified fields from documents:

let results = collection.aggregate()
    .unset(vec!["password".to_string(), "internal_id".to_string()])
    .execute(&collection.all_documents());

$replaceRoot — Replace Document Root

Replace the entire document with a field value:

let results = collection.aggregate()
    .replace_root("address")
    .execute(&collection.all_documents());
// Input:  { "_id": "...", "name": "Alice", "address": { "city": "NYC" } }
// Output: { "city": "NYC" }

$sortByCount — Sort by Count

Sort documents by count in descending order (useful after $group):

let results = collection.aggregate()
    .unwind("tags")
    .add_stage(PipelineStage::SortByCount)
    .execute(&collection.all_documents());

Grouping

$group — Group by Field

Group documents by a field and apply accumulators:

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

let results = collection.aggregate()
    .add_stage(PipelineStage::Group {
        id: Some("department".to_string()),
        accumulators: vec![
            Accumulator {
                field: "avg_age".to_string(),
                op: AccumulatorOp::Avg,
                input_field: "age".to_string(),
            },
            Accumulator {
                field: "count".to_string(),
                op: AccumulatorOp::Count,
                input_field: "age".to_string(),
            },
        ],
    })
    .execute(&collection.all_documents());

Accumulator Operations

Operation	Description	Input Type
Sum	Sum of numeric values	Int, Float
Avg	Average of numeric values	Int, Float
Min	Minimum value	Any ordered type
Max	Maximum value	Any ordered type
Push	Collect values into an array	Any
First	First value in the group	Any
Last	Last value in the group	Any
Count	Count of documents	Any (uses _id)

Global Group (No ID)

Group all documents into a single result:

let results = collection.aggregate()
    .add_stage(PipelineStage::Group {
        id: None, // No grouping field = global group
        accumulators: vec![
            Accumulator {
                field: "total".to_string(),
                op: AccumulatorOp::Count,
                input_field: "_id".to_string(),
            },
        ],
    })
    .execute(&collection.all_documents());

Group with Multiple Accumulators

let results = collection.aggregate()
    .add_stage(PipelineStage::Group {
        id: Some("category".to_string()),
        accumulators: vec![
            Accumulator { field: "total_sales".to_string(), op: AccumulatorOp::Sum, input_field: "amount".to_string() },
            Accumulator { field: "avg_sale".to_string(), op: AccumulatorOp::Avg, input_field: "amount".to_string() },
            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() },
            Accumulator { field: "deal_count".to_string(), op: AccumulatorOp::Count, input_field: "amount".to_string() },
        ],
    })
    .sort(vec![SortField::desc("total_sales")])
    .execute(&collection.all_documents());

Lookup (Joins)

$lookup — Join with Another Collection

use pocketdb::query::relationship_lookup::{LookupStage, JoinType};

// Left outer join (default)
let results = collection.aggregate()
    .add_stage(PipelineStage::Lookup(
        LookupStage::new("orders", "user_id", "customer_id", "purchases")
            .with_join_type(JoinType::LeftOuter)
    ))
    .execute(&collection.all_documents());

See Joins Guide for all 4 join types.

Chaining Stages

Pipeline stages are executed in order — each stage's output becomes the next stage's input:

let results = collection.aggregate()
    .match_(filter! { "status" => == "active" })  // 1. Filter
    .unwind("tags")                                // 2. Unwind arrays
    .add_stage(PipelineStage::Group {              // 3. Group
        id: Some("tags".to_string()),
        accumulators: vec![
            Accumulator {
                field: "count".to_string(),
                op: AccumulatorOp::Count,
                input_field: "tags".to_string(),
            },
        ],
    })
    .sort(vec![SortField::desc("count")])          // 4. Sort
    .limit(10)                                      // 5. Limit
    .execute(&collection.all_documents());

Complete Example

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

fn analyze_sales(db: &Database) -> pocketdb::Result<()> {
    let sales = db.collection("sales")?;

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

    // Group by region, sum amounts, sort by total descending
    let results = sales.aggregate()
        .add_stage(PipelineStage::Group {
            id: Some("region".to_string()),
            accumulators: vec![
                Accumulator {
                    field: "total_sales".to_string(),
                    op: AccumulatorOp::Sum,
                    input_field: "amount".to_string(),
                },
                Accumulator {
                    field: "deal_count".to_string(),
                    op: AccumulatorOp::Count,
                    input_field: "amount".to_string(),
                },
            ],
        })
        .sort(vec![SortField::desc("total_sales")])
        .execute(&sales.all_documents());

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

    Ok(())
}

Pipeline Stage Reference

Stage	Method	Description
$match	.match_(filter)	Filter documents
$sort	.sort(fields)	Order by fields
$limit	.limit(n)	Take first N
$skip	.skip(n)	Skip first N
$project	.project(proj)	Select/exclude fields
$unwind	.unwind(field)	Deconstruct arrays
$count	.count(name)	Count documents
$sample	.sample(n)	Random N documents
$set	.set(fields)	Add/modify fields
$unset	.unset(fields)	Remove fields
$replaceRoot	.replace_root(field)	Replace doc root
$sortByCount	.add_stage(SortByCount)	Sort by count desc
$group	.add_stage(Group{...})	Group by field
$lookup	.add_stage(Lookup(...))	Join collections