SolverForge
Search docs Ctrl K

Documentation

Collectors

Aggregation functions for group_by operations in constraint streams.

Collectors aggregate values within groups created by group_by(...). They transform a stream of individual matches into grouped summaries. For simple fairness rules, prefer balance(...); use collectors when you need explicit counts, totals, or custom imbalance data.

In the current release, the collector trait is Collector<Input>. Input is the borrowed stream match shape: unary and projected grouping pass &A or &Out, while direct cross-join grouping passes (&A, &B). Stock collectors such as count, sum, collect_vec, consecutive_runs, and indexed_presence use the same protocol across those shapes.

Using Collectors

Pass a collector as the second argument to group_by:

factory.for_each(Schedule::shifts())
    .group_by(
        |shift: &Shift| shift.employee_idx,   // grouping key
        count(),                              // collector
    )
    // Result: grouped stream of (key, usize)

Available Collectors

count()

Counts the number of matches in each group. Returns usize.

.group_by(|s: &Shift| s.employee_idx, count())
// → (key, usize)

sum(mapper)

Sums numeric values in each group. The mapper extracts the value to sum.

.group_by(|s: &Shift| s.employee_idx, sum(|s: &Shift| s.hours))
// → (key, i64)

For direct cross-join grouping, the mapper receives the joined pair as a tuple:

type Streams = ConstraintFactory<Plan, HardSoftScore>;

Streams::new()
    .for_each(Plan::assignments())
    .join((
        Streams::new().for_each(Plan::capacities()),
        equal_bi(
            |assignment: &Assignment| assignment.capacity_id,
            |capacity: &Capacity| Some(capacity.id),
        ),
    ))
    .group_by(
        |assignment: &Assignment, _capacity: &Capacity| assignment.bucket,
        sum(|(assignment, capacity): (&Assignment, &Capacity)| {
            capacity.amount - assignment.demand
        }),
    )

load_balance(key_fn, metric_fn)

Measures load imbalance across a grouping key. Returns a LoadBalance<K> with unfairness metric.

.group_by(
    |s: &Shift| s.department_idx,
    load_balance(|s: &Shift| s.employee_idx, |s: &Shift| 1i64),
)

consecutive_runs(index_fn)

Groups integer points into consecutive runs. Duplicate points increase item_count() for the run but still count as one unique point for point_count().

factory.for_each(Schedule::shifts())
    .filter(|shift: &Shift| shift.employee_idx.is_some())
    .group_by(
        |shift: &Shift| shift.employee_idx.unwrap_or(usize::MAX),
        consecutive_runs(|shift: &Shift| shift.date as i64),
    )
    .penalize(|_employee_idx: &usize, runs: &Runs| {
        let excess_days = runs
            .runs()
            .iter()
            .map(|run| run.point_count().saturating_sub(5) as i64)
            .sum();
        HardSoftScore::of_soft(excess_days)
    })
    .named("Long work streaks")

Run exposes start(), end(), point_count(), and item_count(). Runs exposes runs(), point_count(), item_count(), len(), and is_empty().

collect_vec(mapper)

Collects mapped values for each group and returns a CollectedVec<T>. The collector owns values once and can retract them exactly during incremental updates, so T does not need Copy, Clone, or PartialEq just to participate in grouped scoring.

.group_by(
    |shift: &Shift| shift.employee_idx.unwrap_or(usize::MAX),
    collect_vec(|shift: &Shift| shift.required_skill.clone()),
)
.penalize(|_employee_idx: &usize, labels: &CollectedVec<String>| {
    HardSoftScore::of_soft(labels.len().saturating_sub(5) as i64)
})

indexed_presence(index_fn)

Tracks active ordinal positions and exposes both presence and complement runs. Use it when a rule needs to know which slots are covered and which slots are missing.

.group_by(
    |shift: &Shift| shift.employee_idx.unwrap_or(usize::MAX),
    indexed_presence(|shift: &Shift| shift.date as i64),
)
.penalize(|_employee_idx: &usize, presence: &IndexedPresence| {
    HardSoftScore::of_soft(presence.complement_runs(0..7).len() as i64)
})

Complemented Groups

Use complement(...) after group_by(...) when a grouped rule needs rows for keys that have no source matches. For example, workload fairness often needs a zero-count row for employees with no assigned shifts:

factory.for_each(Schedule::shifts())
    .filter(|shift: &Shift| shift.employee_idx.is_some())
    .group_by(
        |shift: &Shift| shift.employee_idx.unwrap_or(usize::MAX),
        count(),
    )
    .complement(Schedule::employees(), |employee: &Employee| employee.index, |_employee| 0usize)
    .penalize(|_employee_idx: &usize, count: &usize| {
        HardSoftScore::of_soft((*count as i64 - 4).abs())
    })
    .named("Balanced workload")

In v0.19.9 unary complemented groups use a relational ComplementNode and the shared OperatorTerminal, rather than a separate complemented scoring engine. The node relates the target domain to grouped results and owns default values for targets with no matching result. The public complement(source, key, default) authoring shape is unchanged. Direct cross-join and projected complemented groups still use their respective grouped terminal families.

Projected grouped streams can also use complement(...) after grouping when the source rows were created by project(...).

Direct cross-join grouped streams can also use complement(...) after grouping. This is useful when the group key comes from a joined target and the rule needs default rows for targets that have no matching left-side rows:

type Streams = ConstraintFactory<Plan, HardSoftScore>;

Streams::new()
    .for_each(Plan::assignments())
    .join((
        Streams::new().for_each(Plan::capacities()),
        equal_bi(
            |assignment: &Assignment| assignment.capacity_id,
            |capacity: &Capacity| Some(capacity.id),
        ),
    ))
    .group_by(
        |_assignment: &Assignment, capacity: &Capacity| capacity.id,
        sum(|(assignment, _capacity): (&Assignment, &Capacity)| assignment.demand),
    )
    .complement(
        Plan::capacities(),
        |capacity: &Capacity| capacity.id,
        |_capacity: &Capacity| 0i64,
    )

Balance Stream Operation

For simple load balancing without group_by, use the balance stream operation directly:

factory.for_each(Schedule::shifts())
    .balance(|shift: &Shift| shift.employee_idx)
    .penalize(HardSoftScore::ONE_SOFT)
    .named("Fair distribution")

The key function returns Option<K> — None values (unassigned entities) are skipped.

See Also