Pipeline API Reference

The AggregationPipeline struct provides a fluent builder for aggregation pipelines.

Installation

Add PocketDB to your Cargo.toml:

[dependencies]
pocketdb = "0.1"

Or via the command line:

cargo add pocketdb

Import

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

Creating a Pipeline

let pipeline = AggregationPipeline::new();

Pipeline Methods

match_

pub fn match_(self, filter: Filter) -> Self

Adds a $match stage to filter documents.

use pocketdb::filter;

let pipeline = pipeline.match_(filter! { "age" => >= 18 });

---

sort

pub fn sort(self, fields: Vec<SortField>) -> Self

Adds a $sort stage to order documents.

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

let pipeline = pipeline.sort(vec![
    SortField::asc("name"),
    SortField::desc("age"),
]);

---

limit

pub fn limit(self, n: usize) -> Self

Adds a $limit stage to limit the number of documents.

---

skip

pub fn skip(self, n: usize) -> Self

Adds a $skip stage to skip the first n documents.

---

project

pub fn project(self, projection: Projection) -> Self

Adds a $project stage to select fields.

use pocketdb::query::Projection;

let pipeline = pipeline.project(Projection::include(&["name", "email"]));

---

unwind

pub fn unwind(self, field: &str) -> Self

Adds an $unwind stage to deconstruct an array field, creating one document per element.

---

count

pub fn count(self, field: &str) -> Self

Adds a $count stage to count documents.

let pipeline = pipeline.count("total");
// Returns: [{ "total": 42 }]

---

lookup

pub fn lookup(self, stage: LookupStage) -> Self

Adds a $lookup stage to join with another collection.

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

let pipeline = pipeline.lookup(
    LookupStage::new("orders", "user_id", "customer_id", "purchases")
        .with_join_type(JoinType::LeftOuter)
);

---

add_stage

pub fn add_stage(mut self, stage: PipelineStage) -> Self

Adds an arbitrary pipeline stage. Used for $group and other advanced stages.

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

let pipeline = pipeline.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(),
        },
    ],
});

---

stages

pub fn stages(&self) -> &[PipelineStage]

Returns the list of pipeline stages.

---

execute

pub fn execute(&self, docs: &[Document]) -> Vec<Document>

Executes the aggregation pipeline against the given documents and returns the results.

Parameters:

• docs — The input documents

Returns: Vec<Document> — The aggregated results

Example:

let results = pipeline.execute(&collection.all_documents());

PipelineStage Variants

pub enum PipelineStage {
    Match(Filter),
    Group {
        id: Option<String>,
        accumulators: Vec<Accumulator>,
    },
    Sort(Vec<SortField>),
    Limit(usize),
    Skip(usize),
    Project(Projection),
    Unwind(String),
    Count(String),
    Lookup(LookupStage),
    AddFields(Vec<(String, PocketValue)>),
}

Accumulator Operations

pub enum AccumulatorOp {
    Sum,     // Sum of numeric values
    Avg,     // Average of numeric values
    Min,     // Minimum value
    Max,     // Maximum value
    Push,    // Collect values into an array
    First,   // First value in the group
    Last,    // Last value in the group
    Count,   // Count of documents
}

SortField

pub struct SortField {
    pub field: String,
    pub direction: SortDirection,
}

impl SortField {
    pub fn new(field: &str, direction: SortDirection) -> Self
    pub fn asc(field: &str) -> Self
    pub fn desc(field: &str) -> Self
}

Projection

pub struct Projection { ... }

impl Projection {
    pub fn include(fields: &[&str]) -> Self
    pub fn exclude(fields: &[&str]) -> Self
    pub fn none() -> Self
    pub fn apply(&self, doc: &Document) -> Document
}

Complete Pipeline Example

use pocketdb::{filter, PocketValue};
use pocketdb::query::{
    AggregationPipeline, PipelineStage, SortField,
    Accumulator, AccumulatorOp, Projection,
};
use pocketdb::query::relationship_lookup::{LookupStage, JoinType};

let results = collection.aggregate()
    // Filter active users
    .match_(filter! { "active" => == true })
    // Join with orders
    .lookup(
        LookupStage::new("orders", "user_id", "customer_id", "purchases")
            .with_join_type(JoinType::LeftOuter)
    )
    // Unwind orders
    .unwind("purchases")
    // Group by user, sum order amounts
    .add_stage(PipelineStage::Group {
        id: Some("name".to_string()),
        accumulators: vec![
            Accumulator {
                field: "total_spent".to_string(),
                op: AccumulatorOp::Sum,
                input_field: "purchases.amount".to_string(),
            },
            Accumulator {
                field: "order_count".to_string(),
                op: AccumulatorOp::Count,
                input_field: "purchases.amount".to_string(),
            },
        ],
    })
    // Sort by total spent descending
    .sort(vec![SortField::desc("total_spent")])
    // Limit to top 10
    .limit(10)
    // Execute against the collection
    .execute(&collection.all_documents());