fix(scoring)!: count changed group keys in changed_key_count

crates/solverforge-scoring/WIREFRAME.md

@@ -57,6 +57,10 @@ migration of the fluent stream families:
   accumulator-owned result, including non-Clone `CollectedVec<T>` payloads.
   Either join input can be a grouped producer. Contributor changes replace
   aggregate rows on both callbacks and coalesce within one root notification.
+- `GroupedOperator<S>: Operator<S>` 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<S, L, R, K, LK, RK>::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
 

crates/solverforge-scoring/public-api.txt

@@ -577,6 +577,8 @@ trait stream::relational::operator::FilterPredicate: pub trait FilterPredicate<S
 trait stream::relational::operator::FlattenSource: pub trait FlattenSource<S: 'static, O: Operator<S>> { 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<S: 'static>: Operator<S> { // 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<S: 'static>: '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<RowHandle>; fn resolve<'a>( &'a self, solution: &'a S, handle: RowHandle, ) -> Option<Self::View<'a>>; 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<RowHandle>, inserted: &mut Vec<RowHandle>, ); fn insert( &mut self, solution: &S, descriptor: usize, index: usize, removed: &mut Vec<RowHandle>, inserted: &mut Vec<RowHandle>, ); // 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

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,
 }
 

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<S, O, A, W, Sc> SharedOperatorBuilder<S, O, A, (), W, Sc> {
 impl<S, O, A, W, Sc> SharedOperatorBuilder<S, O, A, (), W, Sc>
 where
     S: Send + Sync + 'static,
-    O: Operator<S> + Send + Sync,
+    O: GroupedOperator<S> + Send + Sync,
     A: AdaptWeight<S, O, W, Sc>,
     Sc: Score,
 {
@@ -130,7 +130,7 @@ where
 impl<S, O, A, Scorers, Sc> SharedOperatorTerminal<S, O, A, Scorers, Sc>
 where
     S: Send + Sync + 'static,
-    O: Operator<S> + Send + Sync,
+    O: GroupedOperator<S> + 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<S, O, A, Scorers, W, Sc> SharedOperatorBuilder<S, O, A, Scorers, W, Sc>
 where
     S: Send + Sync + 'static,
-    O: Operator<S> + Send + Sync,
+    O: GroupedOperator<S> + Send + Sync,
     A: AdaptWeight<S, O, W, Sc>,
     Scorers: OperatorScorerSet<S, O, Sc>,
     Sc: Score,
@@ -301,7 +301,7 @@ impl<S, O, A, Left, Right, Sc> ConstraintSet<S, Sc>
     for SharedOperatorTerminal<S, O, A, ScorerChain<(Left, Right)>, Sc>
 where
     S: Send + Sync + 'static,
-    O: Operator<S> + Send + Sync,
+    O: GroupedOperator<S> + Send + Sync,
     A: Send + Sync,
     Left: OperatorScorerSet<S, O, Sc>,
     Right: OperatorScorerSet<S, O, Sc>,
@@ -433,7 +433,7 @@ impl<S, O, A, W, Sc> IncrementalConstraint<S, Sc>
     for SharedOperatorTerminal<S, O, A, OperatorScorer<W, Sc>, Sc>
 where
     S: Send + Sync + 'static,
-    O: Operator<S> + Send + Sync,
+    O: GroupedOperator<S> + Send + Sync,
     A: Send + Sync,
     W: RowWeight<S, O, Sc>,
     Sc: Score,

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::<Schedule, SoftScore>::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);
+}

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::<Plan, SoftScore>::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);
+}

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::<Schedule, SoftScore>::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);
+}

crates/solverforge-scoring/src/stream/relational/operator.rs

@@ -117,3 +117,12 @@ pub trait Operator<S: 'static>: 'static {
         inserted: &mut Vec<RowHandle>,
     );
 }
+
+/// 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<S: 'static>: Operator<S> {
+    /// 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;
+}

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<S>,
+        R: GroupedOperator<S>,
+        P: IndexedPlan + KeyOf + RightKeyOf<Key = <P as KeyOf>::Key> + 'static,
+        F: 'static,
+        T: 'static,
+    > GroupedOperator<S> for ComplementNode<S, L, R, P, F, T>
+where
+    for<'a> P: ExecutablePlan<L::View<'a>, R::View<'a>>,
+    F: for<'a> Fn(&S, &L::View<'a>) -> T,
+{
+    fn changed_group_count(&self) -> usize {
+        self.results.changed_group_count()
+    }
+}

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<O, F, C, K, A: Accumulator<V, R>, V, R> {
     rows: DenseRowStore<RowHandle>,
     output_of: HandleMap<RowHandle>,
     changed: Vec<RowHandle>,
+    /// Groups the last `insert` or `retract` changed.
+    changed_group_count: usize,
     /// Child notification scratch, reused across events.
     child_removed: Vec<RowHandle>,
     child_inserted: Vec<RowHandle>,
@@ -59,6 +61,7 @@ impl<O, F, C, K, A: Accumulator<V, R>, V, R> GroupNode<O, F, C, K, A, V, R> {
             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<O, F, C, K, A: Accumulator<V, R>, V, R> GroupNode<O, F, C, K, A, V, R> {
                 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<S: 'static, O, F, C, K, A, V: 'static, R: 'static> GroupedOperator<S>
+    for GroupNode<O, F, C, K, A, V, R>
+where
+    O: Operator<S>,
+    F: for<'a> Fn(&O::View<'a>) -> K + 'static,
+    for<'a> C: Collector<O::View<'a>, Value = V, Result = R, Accumulator = A> + 'static,
+    K: Eq + Hash + 'static,
+    A: Accumulator<V, R> + 'static,
+{
+    fn changed_group_count(&self) -> usize {
+        self.changed_group_count
+    }
+}

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<S: 'static, L: GroupedOperator<S>, R: Operator<S>, P: IndexedPlan + 'static> GroupedOperator<S>
+    for JoinNode<S, L, R, P>
+where
+    for<'a> P: ExecutablePlan<L::View<'a>, R::View<'a>>,
+{
+    fn changed_group_count(&self) -> usize {
+        self.left.changed_group_count()
+    }
+}