diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index 51aae76d..bc38665d 100644 --- a/crates/solverforge-scoring/WIREFRAME.md +++ b/crates/solverforge-scoring/WIREFRAME.md @@ -57,6 +57,10 @@ migration of the fluent stream families: accumulator-owned result, including non-Clone `CollectedVec` payloads. Either join input can be a grouped producer. Contributor changes replace aggregate rows on both callbacks and coalesce within one root notification. +- `GroupedOperator: Operator` reports `changed_group_count()`, the number + of group keys the last `insert` or `retract` changed. `GroupNode` reports its + own groups; `ComplementNode` and `JoinNode` with a grouped results or left + input report that input's count. Shared terminals require it. - `ExistenceNode::new(left, right, left_key, right_keys, exists)` answers a pure-key semi/anti join from per-key occurrence counts (dense arrays for `usize` planning ids via `KeyState` in @@ -712,6 +716,8 @@ compiler node. **`SharedNodeDiagnostics`** — Public fields `{ id, fingerprint, operation, terminal_consumers, update_count, changed_key_count }`; construct with `new(...)`. +`update_count` counts notifications that changed at least one row; +`changed_key_count` sums the group keys each notification changed. ### Analysis Types diff --git a/crates/solverforge-scoring/public-api.txt b/crates/solverforge-scoring/public-api.txt index b6b10542..bad2e947 100644 --- a/crates/solverforge-scoring/public-api.txt +++ b/crates/solverforge-scoring/public-api.txt @@ -577,6 +577,8 @@ trait stream::relational::operator::FilterPredicate: pub trait FilterPredicate> { type Item: 'static; // Required method fn flatten<'a>( &self, solution: &'a S, view: O::View<'a>, ) -> &'a [Self::Item]; } stream::relational::operator::FlattenSource: type Item: 'static stream::relational::operator::FlattenSource: fn flatten<'a>(&self, solution: &'a S, view: O::View<'a>) -> &'a [Self::Item] +trait stream::relational::operator::GroupedOperator: pub trait GroupedOperator: Operator { // Required method fn changed_group_count(&self) -> usize; } + stream::relational::operator::GroupedOperator: fn changed_group_count(&self) -> usize trait stream::relational::operator::Operator: pub trait Operator: 'static { type View<'a>: Copy; type Evaluation; // Required methods fn prepare_evaluation(&self, solution: &S) -> Self::Evaluation; fn visit_evaluation<'a>( &'a self, solution: &'a S, evaluation: &'a Self::Evaluation, visitor: &mut impl FnMut(Self::View<'a>), ); fn clear(&mut self); fn initialize(&mut self, solution: &S); fn handles(&self) -> Vec; fn resolve<'a>( &'a self, solution: &'a S, handle: RowHandle, ) -> Option>; fn visit_provenance( &self, handle: RowHandle, visitor: &mut impl FnMut(u32, usize, usize), ); fn retract( &mut self, solution: &S, descriptor: usize, index: usize, removed: &mut Vec, inserted: &mut Vec, ); fn insert( &mut self, solution: &S, descriptor: usize, index: usize, removed: &mut Vec, inserted: &mut Vec, ); // Provided method fn visit_all( &self, solution: &S, visitor: &mut impl for<'a> FnMut(Self::View<'a>), ) { ... } } stream::relational::operator::Operator: type View<'a>: Copy stream::relational::operator::Operator: type Evaluation diff --git a/crates/solverforge-scoring/src/api/node_sharing.rs b/crates/solverforge-scoring/src/api/node_sharing.rs index 80fc0ede..f0ceb1af 100644 --- a/crates/solverforge-scoring/src/api/node_sharing.rs +++ b/crates/solverforge-scoring/src/api/node_sharing.rs @@ -40,8 +40,10 @@ pub struct SharedNodeDiagnostics { /// Number of entity notifications that changed at least one of the node's /// rows. pub update_count: usize, - /// Count of row-changing notifications. `SharedOperatorTerminal` increments - /// it together with `update_count`, so the two are currently equal. + /// Number of group keys whose collector result a notification changed, + /// summed over notifications. One notification can change several keys, + /// or none while still changing rows (a complemented or joined node's + /// non-grouped side). pub changed_key_count: usize, } diff --git a/crates/solverforge-scoring/src/constraint/relational/shared_terminal.rs b/crates/solverforge-scoring/src/constraint/relational/shared_terminal.rs index 60b54afd..012835e7 100644 --- a/crates/solverforge-scoring/src/constraint/relational/shared_terminal.rs +++ b/crates/solverforge-scoring/src/constraint/relational/shared_terminal.rs @@ -14,7 +14,7 @@ use crate::api::{ SingleConstraint, }, }; -use crate::stream::relational::operator::Operator; +use crate::stream::relational::operator::GroupedOperator; use crate::stream::relational::RowHandle; /// Scoring terminal that owns one operator tree and scores it for several named @@ -95,7 +95,7 @@ impl SharedOperatorBuilder { impl SharedOperatorBuilder where S: Send + Sync + 'static, - O: Operator + Send + Sync, + O: GroupedOperator + Send + Sync, A: AdaptWeight, Sc: Score, { @@ -130,7 +130,7 @@ where impl SharedOperatorTerminal where S: Send + Sync + 'static, - O: Operator + Send + Sync, + O: GroupedOperator + Send + Sync, Sc: Score, { /// Starts another penalizing terminal over the same operator. @@ -194,8 +194,8 @@ where pub fn update_count(&self) -> usize { self.update_count } - /// Count of row-changing notifications; currently incremented together with - /// [`update_count`](Self::update_count). + /// Number of group keys whose collector result a notification changed, + /// summed over notifications. pub fn changed_key_count(&self) -> usize { self.changed_key_count } @@ -255,15 +255,15 @@ where } if !self.removed.is_empty() || !self.inserted.is_empty() { self.update_count += 1; - self.changed_key_count += 1; } + self.changed_key_count += self.operator.changed_group_count(); } } impl SharedOperatorBuilder where S: Send + Sync + 'static, - O: Operator + Send + Sync, + O: GroupedOperator + Send + Sync, A: AdaptWeight, Scorers: OperatorScorerSet, Sc: Score, @@ -301,7 +301,7 @@ impl ConstraintSet for SharedOperatorTerminal, Sc> where S: Send + Sync + 'static, - O: Operator + Send + Sync, + O: GroupedOperator + Send + Sync, A: Send + Sync, Left: OperatorScorerSet, Right: OperatorScorerSet, @@ -433,7 +433,7 @@ impl IncrementalConstraint for SharedOperatorTerminal, Sc> where S: Send + Sync + 'static, - O: Operator + Send + Sync, + O: GroupedOperator + Send + Sync, A: Send + Sync, W: RowWeight, Sc: Score, diff --git a/crates/solverforge-scoring/src/constraint/tests/complemented.rs b/crates/solverforge-scoring/src/constraint/tests/complemented.rs index b4a83c6b..34641b25 100644 --- a/crates/solverforge-scoring/src/constraint/tests/complemented.rs +++ b/crates/solverforge-scoring/src/constraint/tests/complemented.rs @@ -186,3 +186,38 @@ fn duplicate_complement_keys_emit_one_row_per_target() { assert_eq!(running, c.evaluate(&s)); assert_eq!(c.match_count(&s), 3); } + +#[test] +fn only_grouped_side_notifications_change_group_keys() { + let mut constraint = ConstraintFactory::::new() + .for_each(source( + shifts as fn(&Schedule) -> &[Shift], + ChangeSource::Descriptor(0), + )) + .group_by(|s: &Shift| s.employee_id.unwrap_or(NO_KEY), count()) + .complement( + source( + employees as fn(&Schedule) -> &[Employee], + ChangeSource::Descriptor(1), + ), + |e: &Employee| e.id, + |_e| 0usize, + ) + .penalize(|_id: &usize, count: &usize| SoftScore::of(*count as i64)) + .named("Shift count"); + let mut s = schedule(); + constraint.initialize(&s); + + // A new employee adds a default row but changes no group. + s.employees.push(Employee { id: 2 }); + constraint.on_insert(&s, 2, 1); + assert_eq!(constraint.update_count(), 1); + assert_eq!(constraint.changed_key_count(), 0); + + // Moving a shift changes employee 0's group, then creates employee 1's. + constraint.on_retract(&s, 0, 0); + s.shifts[0].employee_id = Some(1); + constraint.on_insert(&s, 0, 0); + assert_eq!(constraint.update_count(), 3); + assert_eq!(constraint.changed_key_count(), 2); +} diff --git a/crates/solverforge-scoring/src/constraint/tests/projected/updates.rs b/crates/solverforge-scoring/src/constraint/tests/projected/updates.rs index ed95ec88..6342795b 100644 --- a/crates/solverforge-scoring/src/constraint/tests/projected/updates.rs +++ b/crates/solverforge-scoring/src/constraint/tests/projected/updates.rs @@ -388,3 +388,33 @@ fn projected_merged_descriptor_sources_keep_same_entity_index_slots_distinct() { assert_eq!(total, SoftScore::of(-5)); assert_eq!(total, constraint.evaluate(&plan)); } + +#[test] +fn one_notification_can_change_several_group_keys() { + // Each work item contributes to its bucket and the next one. + let mut constraint = ConstraintFactory::::new() + .for_each(source( + work as fn(&Plan) -> &[Work], + ChangeSource::Descriptor(0), + )) + .project(WorkTwoEntries) + .group_by( + |entry: &Entry| entry.bucket, + sum(|entry: &Entry| entry.delta), + ) + .penalize(|_bucket: &usize, demand: &i64| SoftScore::of(*demand)) + .named("demand"); + let plan = Plan { + work: vec![Work { + bucket: 0, + demand: 3, + enabled: true, + }], + capacity: Vec::new(), + }; + constraint.initialize(&plan); + + constraint.on_retract(&plan, 0, 0); + assert_eq!(constraint.update_count(), 1); + assert_eq!(constraint.changed_key_count(), 2); +} diff --git a/crates/solverforge-scoring/src/constraint/tests/relational/grouped_join_fluent.rs b/crates/solverforge-scoring/src/constraint/tests/relational/grouped_join_fluent.rs index eec1e702..f5aa2f0f 100644 --- a/crates/solverforge-scoring/src/constraint/tests/relational/grouped_join_fluent.rs +++ b/crates/solverforge-scoring/src/constraint/tests/relational/grouped_join_fluent.rs @@ -129,3 +129,38 @@ fn source_filter_stays_binding_after_grouping() { assert_eq!(c.evaluate(&s), SoftScore::of(6)); assert_eq!(c.match_count(&s), 1); } + +#[test] +fn only_grouped_side_notifications_change_group_keys() { + let mut constraint = ConstraintFactory::::new() + .for_each(source( + shifts as fn(&Schedule) -> &[Shift], + ChangeSource::Descriptor(0), + )) + .group_by(|s: &Shift| s.employee_id, count()) + .join(( + source( + employees as fn(&Schedule) -> &[Employee], + ChangeSource::Descriptor(1), + ), + |e: &Employee| e.id, + )) + .reward(|_emp_id: &u32, count: &usize, employee: &Employee| { + SoftScore::of(*count as i64 * employee.budget) + }) + .named("budget weighted coverage"); + let mut s = schedule(); + constraint.initialize(&s); + + // A budget change rescores employee 1's joined row; no group changes. + constraint.on_retract(&s, 1, 1); + s.employees[1].budget = 7; + constraint.on_insert(&s, 1, 1); + assert_eq!(constraint.update_count(), 2); + assert_eq!(constraint.changed_key_count(), 0); + + // Removing a shift changes employee 0's group. + constraint.on_retract(&s, 0, 0); + assert_eq!(constraint.update_count(), 3); + assert_eq!(constraint.changed_key_count(), 1); +} diff --git a/crates/solverforge-scoring/src/stream/relational/operator.rs b/crates/solverforge-scoring/src/stream/relational/operator.rs index 52e2e45f..cf59e954 100644 --- a/crates/solverforge-scoring/src/stream/relational/operator.rs +++ b/crates/solverforge-scoring/src/stream/relational/operator.rs @@ -117,3 +117,12 @@ pub trait Operator: 'static { inserted: &mut Vec, ); } + +/// An operator whose rows come from a group-by, so a notification changes a +/// countable set of group keys. Grouped nodes report their own groups; +/// complemented and joined grouped nodes report their group input's. +pub trait GroupedOperator: Operator { + /// Number of group keys whose collector result the last `insert` or + /// `retract` changed, including groups it created or emptied. + fn changed_group_count(&self) -> usize; +} diff --git a/crates/solverforge-scoring/src/stream/relational/operator/complement.rs b/crates/solverforge-scoring/src/stream/relational/operator/complement.rs index 9dae9fdd..107c85eb 100644 --- a/crates/solverforge-scoring/src/stream/relational/operator/complement.rs +++ b/crates/solverforge-scoring/src/stream/relational/operator/complement.rs @@ -1,4 +1,4 @@ -use super::{ComplementView, Operator, Pair, PairDelta, ProjectView}; +use super::{ComplementView, GroupedOperator, Operator, Pair, PairDelta, ProjectView}; use crate::stream::joiner::plan::{CompileCondition, ExecutablePlan, IndexedPlan}; use crate::stream::relational::hash::FastMap; use crate::stream::relational::keyed_plan::{KeyOf, RightKeyOf}; @@ -437,3 +437,20 @@ where self.update(solution, descriptor, index, false, removed, inserted) } } + +impl< + S: 'static, + L: Operator, + R: GroupedOperator, + P: IndexedPlan + KeyOf + RightKeyOf::Key> + 'static, + F: 'static, + T: 'static, + > GroupedOperator for ComplementNode +where + for<'a> P: ExecutablePlan, R::View<'a>>, + F: for<'a> Fn(&S, &L::View<'a>) -> T, +{ + fn changed_group_count(&self) -> usize { + self.results.changed_group_count() + } +} diff --git a/crates/solverforge-scoring/src/stream/relational/operator/group.rs b/crates/solverforge-scoring/src/stream/relational/operator/group.rs index f5cf6e53..0c150103 100644 --- a/crates/solverforge-scoring/src/stream/relational/operator/group.rs +++ b/crates/solverforge-scoring/src/stream/relational/operator/group.rs @@ -1,4 +1,4 @@ -use super::{group_view::Contributors, GroupView, Operator}; +use super::{group_view::Contributors, GroupView, GroupedOperator, Operator}; use crate::stream::collector::{Accumulator, Collector}; use crate::stream::relational::{DenseRowStore, FastMap, HandleMap, RowHandle}; use solverforge_core::fast_hash::FastHasher; @@ -39,6 +39,8 @@ pub struct GroupNode, V, R> { rows: DenseRowStore, output_of: HandleMap, changed: Vec, + /// Groups the last `insert` or `retract` changed. + changed_group_count: usize, /// Child notification scratch, reused across events. child_removed: Vec, child_inserted: Vec, @@ -59,6 +61,7 @@ impl, V, R> GroupNode { rows: DenseRowStore::new(), output_of: HandleMap::new(), changed: Vec::new(), + changed_group_count: 0, child_removed: Vec::new(), child_inserted: Vec::new(), marker: PhantomData, @@ -163,6 +166,7 @@ impl, V, R> GroupNode { inserted.push(output); } } + self.changed_group_count = changed.len(); changed.clear(); self.changed = changed; } @@ -234,6 +238,7 @@ where self.rows.clear(); self.output_of.clear(); self.changed.clear(); + self.changed_group_count = 0; self.child_removed.clear(); self.child_inserted.clear(); } @@ -326,3 +331,17 @@ where self.child_inserted = in_inserted; } } + +impl GroupedOperator + for GroupNode +where + O: Operator, + F: for<'a> Fn(&O::View<'a>) -> K + 'static, + for<'a> C: Collector, Value = V, Result = R, Accumulator = A> + 'static, + K: Eq + Hash + 'static, + A: Accumulator + 'static, +{ + fn changed_group_count(&self) -> usize { + self.changed_group_count + } +} diff --git a/crates/solverforge-scoring/src/stream/relational/operator/join.rs b/crates/solverforge-scoring/src/stream/relational/operator/join.rs index 000431ba..7242a044 100644 --- a/crates/solverforge-scoring/src/stream/relational/operator/join.rs +++ b/crates/solverforge-scoring/src/stream/relational/operator/join.rs @@ -1,5 +1,5 @@ use super::super::{DenseRowStore, HandleMap, JoinedIdentity, RowHandle}; -use super::{links::remove_link, Operator, Pair, PairDelta}; +use super::{links::remove_link, GroupedOperator, Operator, Pair, PairDelta}; use crate::stream::joiner::plan::{CompileCondition, ExecutablePlan, IndexedPlan}; use crate::stream::relational::hash::FastMap; use std::marker::PhantomData; @@ -397,3 +397,13 @@ where self.child_r_inserted = ri; } } + +impl, R: Operator, P: IndexedPlan + 'static> GroupedOperator + for JoinNode +where + for<'a> P: ExecutablePlan, R::View<'a>>, +{ + fn changed_group_count(&self) -> usize { + self.left.changed_group_count() + } +}