From 4caa1caac184e38686e7828defdb8abc7c7cbfc6 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Tue, 13 Oct 2026 20:26:19 +0200 Subject: [PATCH] feat(scoring): expose group_by on tri quad and penta joins Add the missing fluent grouping methods to cross-source and same-source arity streams. Adapt borrowed tuples to the existing GroupNode and retain authored filters with their semantic source indexes; no new grouping engine or copied collector payloads are introduced. Cover grouping, source mutations, empty groups, reset and move-only collected values with seven executed regressions and a macro compile fixture. Update the scoring and facade wireframes and README. Workspace tests, workspace all-target clippy, formatting and independent review pass. Fixes #73 --- README.md | 11 +- .../solverforge_constraints_tri_grouped.rs | 28 ++ crates/solverforge-scoring/WIREFRAME.md | 20 ++ .../arity_stream_macros/nary_stream/shared.rs | 38 +++ .../src/stream/cross_tri_stream.rs | 2 +- .../src/stream/cross_tri_stream/base.rs | 4 +- .../src/stream/group_by_penta.rs | 98 ++++++ .../src/stream/group_by_quad.rs | 91 +++++ .../src/stream/group_by_tri.rs | 80 +++++ .../src/stream/joined_grouped_stream.rs | 120 +++++++ crates/solverforge-scoring/src/stream/mod.rs | 5 + .../src/stream/relational/mod.rs | 3 + .../src/stream/relational/tuple_collector.rs | 85 +++++ .../tests/group_by_joined.rs | 319 ++++++++++++++++++ crates/solverforge/WIREFRAME.md | 6 + 15 files changed, 903 insertions(+), 7 deletions(-) create mode 100644 crates/solverforge-macros/tests/ui/pass/solverforge_constraints_tri_grouped.rs create mode 100644 crates/solverforge-scoring/src/stream/group_by_penta.rs create mode 100644 crates/solverforge-scoring/src/stream/group_by_quad.rs create mode 100644 crates/solverforge-scoring/src/stream/group_by_tri.rs create mode 100644 crates/solverforge-scoring/src/stream/joined_grouped_stream.rs create mode 100644 crates/solverforge-scoring/src/stream/relational/tuple_collector.rs create mode 100644 crates/solverforge-scoring/tests/group_by_joined.rs diff --git a/README.md b/README.md index 4a4892c6..1262d854 100644 --- a/README.md +++ b/README.md @@ -170,10 +170,13 @@ ConstraintFactory::::new() ``` Collectors retain mapped values by ownership. The same collector protocol covers -unary rows and joined pairs: direct cross-join grouping passes `(&A, &B)` to the -collector mapper, while both direct cross-join grouped streams and projected -grouped streams can continue into `complement(...)` for supply/demand-style -scoring. `collect_vec` exposes a +unary rows and joined tuples. `Tri`, `Quad`, and `Penta` cross streams and the +same-source arity streams expose `.group_by(key, collector)`: the key receives +three, four, or five entity arguments, and the collector mapper receives the +corresponding borrowed tuple. Grouped scoring weights receive `(&key, &result)`. +Upstream filters remain binding during full and incremental scoring. Direct +cross-join grouped pairs and projected grouped streams can continue into +`complement(...)` for supply/demand-style scoring. `collect_vec` exposes a `CollectedVec` result view, so grouped payloads such as `String` labels do not need `Copy`, `Clone`, or `PartialEq` just to be collected. diff --git a/crates/solverforge-macros/tests/ui/pass/solverforge_constraints_tri_grouped.rs b/crates/solverforge-macros/tests/ui/pass/solverforge_constraints_tri_grouped.rs new file mode 100644 index 00000000..503a9e15 --- /dev/null +++ b/crates/solverforge-macros/tests/ui/pass/solverforge_constraints_tri_grouped.rs @@ -0,0 +1,28 @@ +use solverforge::prelude::*; +use solverforge::stream::relational::{operator::Pair, Leaf}; + +struct Model { a: Vec, b: Vec, c: Vec } +fn a(m: &Model) -> &[u32] { &m.a } +fn b(m: &Model) -> &[u32] { &m.b } +fn c(m: &Model) -> &[u32] { &m.c } +fn pair_key(row: &Pair, Leaf<'_, u32>>) -> u32 { *row.left.entity } +fn leaf_key(row: &Leaf<'_, u32>) -> u32 { *row.entity } + +#[solverforge_constraints] +fn constraints() -> impl ConstraintSet { + let factory = ConstraintFactory::::new(); + ( + factory.for_each(a as fn(&Model) -> &[u32]) + .join((b as fn(&Model) -> &[u32], joiner::equal_bi(|a: &u32| *a, |b: &u32| *b))) + .join((c as fn(&Model) -> &[u32], joiner::equal_on(pair_key, leaf_key))) + .group_by(|a: &u32, _: &u32, _: &u32| *a, count()) + .penalize(|_: &u32, n: &usize| SoftScore::of(*n as i64)) + .named("tri grouped count"), + ) +} +fn main() { + let mut constraints = constraints(); + let model = Model { a: vec![1, 1], b: vec![1], c: vec![1] }; + assert_eq!(constraints.evaluate_all(&model), SoftScore::of(-2)); + assert_eq!(constraints.initialize_all(&model), SoftScore::of(-2)); +} diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index 0205a49b..e3cbcef1 100644 --- a/crates/solverforge-scoring/WIREFRAME.md +++ b/crates/solverforge-scoring/WIREFRAME.md @@ -5,6 +5,26 @@ Zero-erasure incremental constraint scoring infrastructure for SolverForge. **Location:** `crates/solverforge-scoring/` **Workspace Release:** `0.20.4` +## Joined grouping fluent surface + +- `stream::cross::{Tri, Quad, Penta}` and + `stream::{TriConstraintStream, QuadConstraintStream, PentaConstraintStream}` + expose `group_by(key, collector)`. Key closures take three/four/five entity + references; collector mappers receive the corresponding borrowed tuple. +- `stream::JoinedGroupedStream` exposes + `penalize(weight)` and `reward(weight)` over `(&K, &R)`. + `stream::JoinedGroupedBuilder<..., W, Sc>::named(name)` produces an + incremental `OperatorTerminal`. Collector results are borrowed, not cloned; + `collect_vec` retains move-only payloads and exposes `CollectedVec`. +- All these methods preserve upstream filtering and semantic source indexes. + Joined tuple grouping uses the existing `GroupNode` and `FilterNode`, not + arity-specific retention engines. +- File map: `stream/group_by_{tri,quad,penta}.rs`, + `stream/joined_grouped_stream.rs`, and + `stream/relational/tuple_collector.rs`; same-source methods are generated in + `stream/arity_stream_macros/nary_stream/shared.rs`. The doc-hidden + `stream::relational::TupleCollector` is the borrowed-row adapter. + ## Typed relational operator surface These are module-level low-level exports, not facade-root exports or a completed diff --git a/crates/solverforge-scoring/src/stream/arity_stream_macros/nary_stream/shared.rs b/crates/solverforge-scoring/src/stream/arity_stream_macros/nary_stream/shared.rs index 586bd3dc..d9c84b43 100644 --- a/crates/solverforge-scoring/src/stream/arity_stream_macros/nary_stream/shared.rs +++ b/crates/solverforge-scoring/src/stream/arity_stream_macros/nary_stream/shared.rs @@ -167,6 +167,44 @@ macro_rules! impl_nary_arity_stream_common { } } + /// Groups each matching tuple, preserving upstream filters. + pub fn group_by(self, key: GF, collector: Co) + -> super::JoinedGroupedStream, K, KE, $arity>, + impl Fn(&S, &[crate::stream::relational::Leaf<'_, A>; $arity]) -> bool + Send + Sync>, + impl Fn(&[crate::stream::relational::Leaf<'_, A>; $arity]) -> GK + Send + Sync, + crate::stream::relational::TupleCollector, GK, V, R, Acc, Sc> + where + E: 'static, + KE: 'static, + F: 'static, + K: 'static, + GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static, + GF: Fn($(repeat_tokens!($entity => &A)),+) -> GK + Send + Sync + 'static, + Co: for<'i> super::collector::Collector< + ($(repeat_tokens!($entity => &'i A)),+), Value = V, Result = R, Accumulator = Acc> + 'static, + V: Send + Sync + 'static, + R: Send + Sync + 'static, + Acc: super::collector::Accumulator + 'static, + for<'i> crate::stream::relational::TupleCollector: + super::collector::Collector<[crate::stream::relational::Leaf<'i, A>; $arity], Value = V, Result = R, Accumulator = Acc>, + { + let filter = self.filter; + let input = crate::stream::relational::operator::SelfJoinNode::new( + crate::stream::relational::operator::CollectionNode::new(self.extractor, 0), self.key_extractor); + let filtered = crate::stream::relational::operator::FilterNode::new(input, + move |s: &S, row: &[crate::stream::relational::Leaf<'_, A>; $arity]| { + filter.test(s, $(row[repeat_nary_arity_position!($entity)].entity,)+ + $(row[repeat_nary_arity_position!($filter_idx)].index),+) + }); + super::JoinedGroupedStream::new(filtered, + move |row: &[crate::stream::relational::Leaf<'_, A>; $arity]| { + key($(row[repeat_nary_arity_position!($entity)].entity),+) + }, crate::stream::relational::TupleCollector::new(collector)) + } + pub fn penalize( self, weight: W, diff --git a/crates/solverforge-scoring/src/stream/cross_tri_stream.rs b/crates/solverforge-scoring/src/stream/cross_tri_stream.rs index 5d6f1411..2c1d63c2 100644 --- a/crates/solverforge-scoring/src/stream/cross_tri_stream.rs +++ b/crates/solverforge-scoring/src/stream/cross_tri_stream.rs @@ -7,7 +7,7 @@ domain stays independent. One uniform `.join()` on the Bi stream builds it — no per-arity engine, no shared-key retention at any depth. */ -mod base; +pub(crate) mod base; pub(crate) use base::assemble_tri; pub use base::Tri; diff --git a/crates/solverforge-scoring/src/stream/cross_tri_stream/base.rs b/crates/solverforge-scoring/src/stream/cross_tri_stream/base.rs index 35abc8a3..038f0d01 100644 --- a/crates/solverforge-scoring/src/stream/cross_tri_stream/base.rs +++ b/crates/solverforge-scoring/src/stream/cross_tri_stream/base.rs @@ -129,8 +129,8 @@ where P1: CompileCondition, P2: CompileCondition, { - pub(super) tree: BuiltTri, - pub(super) filter: F, + pub(crate) tree: BuiltTri, + pub(crate) filter: F, pub(super) _phantom: PhantomData<(fn() -> S, fn() -> A, fn() -> B, fn() -> C, fn() -> Sc)>, } diff --git a/crates/solverforge-scoring/src/stream/group_by_penta.rs b/crates/solverforge-scoring/src/stream/group_by_penta.rs new file mode 100644 index 00000000..78b5c7a3 --- /dev/null +++ b/crates/solverforge-scoring/src/stream/group_by_penta.rs @@ -0,0 +1,98 @@ +use super::collector::{Accumulator, Collector}; +use super::filter::PentaFilter; +use super::joiner::plan::{CompileCondition, IndexedPlan}; +use super::relational::operator::{FilterNode, Operator, Pair}; +use super::relational::{Leaf, TupleCollector}; +use super::JoinedGroupedStream; +use solverforge_core::score::Score; + +type Row<'a, A, B, C, D, E> = + Pair, Leaf<'a, B>>, Leaf<'a, C>>, Leaf<'a, D>>, Leaf<'a, E>>; + +impl + super::cross::Penta +where + S: Send + Sync + 'static, + A: Clone + Send + Sync + 'static, + B: Clone + Send + Sync + 'static, + C: Clone + Send + Sync + 'static, + D: Clone + Send + Sync + 'static, + E: Clone + Send + Sync + 'static, + EA: 'static, + EB: 'static, + EC: 'static, + ED: 'static, + EE: 'static, + P1: CompileCondition + 'static, + P2: CompileCondition + 'static, + P3: CompileCondition + 'static, + P4: CompileCondition + 'static, + for<'a> super::cross_penta_stream::BuiltPenta: + Operator = Row<'a, A, B, C, D, E>> + Send + Sync, + F: PentaFilter + 'static, + Sc: Score + 'static, +{ + /// Groups joined tuples by a key and an owned-value collector. + /// Upstream filters remain binding on full and incremental evaluation. + pub fn group_by( + self, + key: GF, + collector: Co, + ) -> JoinedGroupedStream< + S, + FilterNode< + super::cross_penta_stream::BuiltPenta, + impl Fn(&S, &Row<'_, A, B, C, D, E>) -> bool + Send + Sync, + >, + impl Fn(&Row<'_, A, B, C, D, E>) -> GK + Send + Sync, + TupleCollector, + GK, + V, + R, + Acc, + Sc, + > + where + GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static, + GF: Fn(&A, &B, &C, &D, &E) -> GK + Send + Sync + 'static, + Co: for<'i> Collector< + (&'i A, &'i B, &'i C, &'i D, &'i E), + Value = V, + Result = R, + Accumulator = Acc, + > + 'static, + V: Send + Sync + 'static, + R: Send + Sync + 'static, + Acc: Accumulator + 'static, + { + let filter = self.filter; + let input = FilterNode::new(self.tree, move |s: &S, row: &Row<'_, A, B, C, D, E>| { + filter.test( + s, + row.left.left.left.left.entity, + row.left.left.left.right.entity, + row.left.left.right.entity, + row.left.right.entity, + row.right.entity, + row.left.left.left.left.index, + row.left.left.left.right.index, + row.left.left.right.index, + row.left.right.index, + row.right.index, + ) + }); + JoinedGroupedStream::new( + input, + move |row: &Row<'_, A, B, C, D, E>| { + key( + row.left.left.left.left.entity, + row.left.left.left.right.entity, + row.left.left.right.entity, + row.left.right.entity, + row.right.entity, + ) + }, + TupleCollector::new(collector), + ) + } +} diff --git a/crates/solverforge-scoring/src/stream/group_by_quad.rs b/crates/solverforge-scoring/src/stream/group_by_quad.rs new file mode 100644 index 00000000..9c935e51 --- /dev/null +++ b/crates/solverforge-scoring/src/stream/group_by_quad.rs @@ -0,0 +1,91 @@ +use super::collector::{Accumulator, Collector}; +use super::filter::QuadFilter; +use super::joiner::plan::{CompileCondition, IndexedPlan}; +use super::relational::operator::{FilterNode, Operator, Pair}; +use super::relational::{Leaf, TupleCollector}; +use super::JoinedGroupedStream; +use solverforge_core::score::Score; + +type Row<'a, A, B, C, D> = Pair, Leaf<'a, B>>, Leaf<'a, C>>, Leaf<'a, D>>; + +impl + super::cross::Quad +where + S: Send + Sync + 'static, + A: Clone + Send + Sync + 'static, + B: Clone + Send + Sync + 'static, + C: Clone + Send + Sync + 'static, + D: Clone + Send + Sync + 'static, + EA: 'static, + EB: 'static, + EC: 'static, + ED: 'static, + P1: CompileCondition + 'static, + P2: CompileCondition + 'static, + P3: CompileCondition + 'static, + for<'a> super::cross_quad_stream::BuiltQuad: + Operator = Row<'a, A, B, C, D>> + Send + Sync, + F: QuadFilter + 'static, + Sc: Score + 'static, +{ + /// Groups joined tuples by a key and an owned-value collector. + /// Upstream filters remain binding on full and incremental evaluation. + pub fn group_by( + self, + key: GF, + collector: Co, + ) -> JoinedGroupedStream< + S, + FilterNode< + super::cross_quad_stream::BuiltQuad, + impl Fn(&S, &Row<'_, A, B, C, D>) -> bool + Send + Sync, + >, + impl Fn(&Row<'_, A, B, C, D>) -> GK + Send + Sync, + TupleCollector, + GK, + V, + R, + Acc, + Sc, + > + where + GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static, + GF: Fn(&A, &B, &C, &D) -> GK + Send + Sync + 'static, + Co: for<'i> Collector< + (&'i A, &'i B, &'i C, &'i D), + Value = V, + Result = R, + Accumulator = Acc, + > + 'static, + V: Send + Sync + 'static, + R: Send + Sync + 'static, + Acc: Accumulator + 'static, + { + let filter = self.filter; + let input = FilterNode::new(self.tree, move |s: &S, row: &Row<'_, A, B, C, D>| { + filter.test( + s, + row.left.left.left.entity, + row.left.left.right.entity, + row.left.right.entity, + row.right.entity, + row.left.left.left.index, + row.left.left.right.index, + row.left.right.index, + row.right.index, + ) + }); + JoinedGroupedStream::new( + input, + move |row: &Row<'_, A, B, C, D>| { + key( + row.left.left.left.entity, + row.left.left.right.entity, + row.left.right.entity, + row.right.entity, + ) + }, + TupleCollector::new(collector), + ) + } +} diff --git a/crates/solverforge-scoring/src/stream/group_by_tri.rs b/crates/solverforge-scoring/src/stream/group_by_tri.rs new file mode 100644 index 00000000..f5fc1d5f --- /dev/null +++ b/crates/solverforge-scoring/src/stream/group_by_tri.rs @@ -0,0 +1,80 @@ +use super::collector::{Accumulator, Collector}; +use super::filter::TriFilter; +use super::joiner::plan::{CompileCondition, IndexedPlan}; +use super::relational::operator::{FilterNode, Operator, Pair}; +use super::relational::{Leaf, TupleCollector}; +use super::JoinedGroupedStream; +use solverforge_core::score::Score; + +type Row<'a, A, B, C> = Pair, Leaf<'a, B>>, Leaf<'a, C>>; + +impl super::cross::Tri +where + S: Send + Sync + 'static, + A: Clone + Send + Sync + 'static, + B: Clone + Send + Sync + 'static, + C: Clone + Send + Sync + 'static, + EA: 'static, + EB: 'static, + EC: 'static, + P1: CompileCondition + 'static, + P2: CompileCondition + 'static, + for<'a> super::cross_tri_stream::base::BuiltTri: + Operator = Row<'a, A, B, C>> + Send + Sync, + F: TriFilter + 'static, + Sc: Score + 'static, +{ + /// Groups joined tuples by a key and an owned-value collector. + /// Upstream filters remain binding on full and incremental evaluation. + pub fn group_by( + self, + key: GF, + collector: Co, + ) -> JoinedGroupedStream< + S, + FilterNode< + super::cross_tri_stream::base::BuiltTri, + impl Fn(&S, &Row<'_, A, B, C>) -> bool + Send + Sync, + >, + impl Fn(&Row<'_, A, B, C>) -> GK + Send + Sync, + TupleCollector, + GK, + V, + R, + Acc, + Sc, + > + where + GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static, + GF: Fn(&A, &B, &C) -> GK + Send + Sync + 'static, + Co: for<'i> Collector<(&'i A, &'i B, &'i C), Value = V, Result = R, Accumulator = Acc> + + 'static, + V: Send + Sync + 'static, + R: Send + Sync + 'static, + Acc: Accumulator + 'static, + { + let filter = self.filter; + let input = FilterNode::new(self.tree, move |s: &S, row: &Row<'_, A, B, C>| { + filter.test( + s, + row.left.left.entity, + row.left.right.entity, + row.right.entity, + row.left.left.index, + row.left.right.index, + row.right.index, + ) + }); + JoinedGroupedStream::new( + input, + move |row: &Row<'_, A, B, C>| { + key( + row.left.left.entity, + row.left.right.entity, + row.right.entity, + ) + }, + TupleCollector::new(collector), + ) + } +} diff --git a/crates/solverforge-scoring/src/stream/joined_grouped_stream.rs b/crates/solverforge-scoring/src/stream/joined_grouped_stream.rs new file mode 100644 index 00000000..298e989f --- /dev/null +++ b/crates/solverforge-scoring/src/stream/joined_grouped_stream.rs @@ -0,0 +1,120 @@ +use std::hash::Hash; +use std::marker::PhantomData; + +use super::collector::{Accumulator, Collector}; +use super::relational::operator::{ExplainRow, GroupNode, GroupView, Operator}; +use super::weighting_support::ConstraintWeight; +use crate::constraint::relational::OperatorTerminal; +use solverforge_core::{score::Score, ConstraintRef, ImpactType}; + +/// Grouped joined rows, backed by the common grouping operator. +pub struct JoinedGroupedStream, Sc> { + node: GroupNode, + marker: PhantomData (S, Sc)>, +} + +impl, Sc> + JoinedGroupedStream +{ + pub(crate) fn new(input: O, key: F, collector: C) -> Self { + Self { + node: GroupNode::new(input, key, collector), + marker: PhantomData, + } + } +} + +impl JoinedGroupedStream +where + S: Send + Sync + 'static, + O: Operator + Send + Sync, + F: for<'a> Fn(&O::View<'a>) -> K + Send + Sync + 'static, + C: for<'a> Collector, Value = V, Result = R, Accumulator = Acc> + 'static, + K: Eq + Hash + Clone + Send + Sync + 'static, + V: Send + Sync + 'static, + R: Send + Sync + 'static, + Acc: Accumulator + 'static, + Sc: Score + 'static, +{ + /// Penalizes each group using its key and borrowed collector result. + pub fn penalize( + self, + weight: W, + ) -> JoinedGroupedBuilder Sc + Send + Sync, Sc> + where + W: for<'a> ConstraintWeight<(&'a K, &'a R), Sc> + Send + Sync, + { + let hard = weight.is_hard(); + JoinedGroupedBuilder { + node: self.node, + weight: move |k: &K, r: &R| weight.score((k, r)), + impact: ImpactType::Penalty, + hard, + marker: PhantomData, + } + } + + /// Rewards each group using its key and borrowed collector result. + pub fn reward( + self, + weight: W, + ) -> JoinedGroupedBuilder Sc + Send + Sync, Sc> + where + W: for<'a> ConstraintWeight<(&'a K, &'a R), Sc> + Send + Sync, + { + let hard = weight.is_hard(); + JoinedGroupedBuilder { + node: self.node, + weight: move |k: &K, r: &R| weight.score((k, r)), + impact: ImpactType::Reward, + hard, + marker: PhantomData, + } + } +} + +/// Finalizer for grouped joined rows. +pub struct JoinedGroupedBuilder, W, Sc> { + node: GroupNode, + weight: W, + impact: ImpactType, + hard: bool, + marker: PhantomData (S, Sc)>, +} + +impl JoinedGroupedBuilder +where + S: Send + Sync + 'static, + O: Operator + Send + Sync, + for<'a> O::View<'a>: ExplainRow, + F: for<'a> Fn(&O::View<'a>) -> K + Send + Sync + 'static, + C: for<'a> Collector, Value = V, Result = R, Accumulator = Acc> + 'static, + K: Eq + Hash + Clone + Send + Sync + 'static, + V: Send + Sync + 'static, + R: Send + Sync + 'static, + Acc: Accumulator + 'static, + W: Fn(&K, &R) -> Sc + Send + Sync, + Sc: Score + 'static, +{ + /// Finalizes the grouped constraint without copying collector results. + pub fn named( + self, + name: &str, + ) -> OperatorTerminal< + S, + GroupNode, + impl Fn(&S, &GroupView<'_, S, O, K, Acc, V, R>) -> Sc + Send + Sync, + Sc, + > { + let weight = self.weight; + OperatorTerminal::new( + ConstraintRef::new("", name), + self.impact, + self.node, + move |_: &S, row: &GroupView<'_, S, O, K, Acc, V, R>| { + row.with_result(|r| weight(row.key, r)) + }, + self.hard, + ) + } +} diff --git a/crates/solverforge-scoring/src/stream/mod.rs b/crates/solverforge-scoring/src/stream/mod.rs index 97c7cf88..465e4a60 100644 --- a/crates/solverforge-scoring/src/stream/mod.rs +++ b/crates/solverforge-scoring/src/stream/mod.rs @@ -78,7 +78,12 @@ mod existence_target; mod factory; pub mod filter; mod flattened_bi_stream; +mod group_by_penta; +mod group_by_quad; +mod group_by_tri; mod grouped_stream; +mod joined_grouped_stream; +pub use joined_grouped_stream::{JoinedGroupedBuilder, JoinedGroupedStream}; pub mod join_target; pub mod joiner; pub mod key_extract; diff --git a/crates/solverforge-scoring/src/stream/relational/mod.rs b/crates/solverforge-scoring/src/stream/relational/mod.rs index 76165f7f..d1f1650b 100644 --- a/crates/solverforge-scoring/src/stream/relational/mod.rs +++ b/crates/solverforge-scoring/src/stream/relational/mod.rs @@ -16,6 +16,9 @@ pub(crate) mod leaf_collector; pub mod operator; mod row; mod storage; +mod tuple_collector; +#[doc(hidden)] +pub use tuple_collector::TupleCollector; pub mod view_plan; pub(crate) use handle_map::HandleMap; diff --git a/crates/solverforge-scoring/src/stream/relational/tuple_collector.rs b/crates/solverforge-scoring/src/stream/relational/tuple_collector.rs new file mode 100644 index 00000000..18e68c27 --- /dev/null +++ b/crates/solverforge-scoring/src/stream/relational/tuple_collector.rs @@ -0,0 +1,85 @@ +use super::{operator::Pair, Leaf}; +use crate::stream::collector::{Accumulator, Collector}; +use std::marker::PhantomData; + +/// Adapts a tuple collector to borrowed operator rows without copying entities. +pub struct TupleCollector { + inner: C, + marker: PhantomData (V, R, Acc)>, +} +impl TupleCollector { + pub fn new(inner: C) -> Self { + Self { + inner, + marker: PhantomData, + } + } +} + +macro_rules! impl_tuple_collector { + ([$($ty:ident),+], $row:ty, |$input:ident| $tuple:expr) => { + impl<'a, $($ty: 'a),+, C, V, R, Acc> Collector<$row> for TupleCollector + where + C: Collector<($(&'a $ty),+), Value = V, Result = R, Accumulator = Acc>, + Acc: Accumulator, + R: Send + Sync, + { + type Value = V; + type Result = R; + type Accumulator = Acc; + fn extract(&self, $input: $row) -> V { self.inner.extract($tuple) } + fn create_accumulator(&self) -> Acc { self.inner.create_accumulator() } + } + }; +} +impl_tuple_collector!( + [A, B, D], + Pair, Leaf<'a, B>>, Leaf<'a, D>>, + |row| ( + row.left.left.entity, + row.left.right.entity, + row.right.entity + ) +); +impl_tuple_collector!( + [A, B, D, E], + Pair, Leaf<'a, B>>, Leaf<'a, D>>, Leaf<'a, E>>, + |row| ( + row.left.left.left.entity, + row.left.left.right.entity, + row.left.right.entity, + row.right.entity + ) +); +impl_tuple_collector!( + [A, B, D, E, F], + Pair, Leaf<'a, B>>, Leaf<'a, D>>, Leaf<'a, E>>, Leaf<'a, F>>, + |row| ( + row.left.left.left.left.entity, + row.left.left.left.right.entity, + row.left.left.right.entity, + row.left.right.entity, + row.right.entity + ) +); + +macro_rules! impl_array_collector { + ($n:literal, [$($index:literal),+]) => { + impl<'a, A: 'a, C, V, R, Acc> Collector<[Leaf<'a, A>; $n]> for TupleCollector + where + C: Collector<($(impl_array_collector!(@ty $index, &'a A)),+), Value = V, Result = R, Accumulator = Acc>, + Acc: Accumulator, + R: Send + Sync, + { + type Value = V; + type Result = R; + type Accumulator = Acc; + fn extract(&self, row: [Leaf<'a, A>; $n]) -> V { self.inner.extract(($(row[$index].entity),+)) } + fn create_accumulator(&self) -> Acc { self.inner.create_accumulator() } + } + }; + (@ty $index:literal, $ty:ty) => { $ty }; +} +impl_array_collector!(3, [0, 1, 2]); +impl_array_collector!(4, [0, 1, 2, 3]); +impl_array_collector!(5, [0, 1, 2, 3, 4]); diff --git a/crates/solverforge-scoring/tests/group_by_joined.rs b/crates/solverforge-scoring/tests/group_by_joined.rs new file mode 100644 index 00000000..fa34a28f --- /dev/null +++ b/crates/solverforge-scoring/tests/group_by_joined.rs @@ -0,0 +1,319 @@ +use solverforge_core::score::SoftScore; +use solverforge_scoring::api::constraint_set::IncrementalConstraint; +use solverforge_scoring::stream::collector::count; +use solverforge_scoring::stream::joiner::{equal_bi, equal_on}; +use solverforge_scoring::stream::relational::{operator::Pair, Leaf}; +use solverforge_scoring::stream::{source, ChangeSource, ConstraintFactory}; + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct Item { + key: u32, + value: i64, +} +struct Model { + a: Vec, + b: Vec, + c: Vec, + d: Vec, + e: Vec, +} +fn a(m: &Model) -> &[Item] { + &m.a +} +fn b(m: &Model) -> &[Item] { + &m.b +} +fn c(m: &Model) -> &[Item] { + &m.c +} +fn pair_key(row: &Pair, Leaf<'_, Item>>) -> u32 { + row.left.entity.key +} +fn leaf_key(row: &Leaf<'_, Item>) -> u32 { + row.entity.key +} + +#[test] +fn self_join_tri_group_by_counts_combinations() { + let constraint = solverforge_scoring::stream::TriConstraintStream::< + Model, + Item, + u32, + _, + _, + _, + SoftScore, + >::new_self_join( + a as fn(&Model) -> &[Item], + |_: &Model, i: &Item, _: usize| i.key, + ) + .group_by(|i: &Item, _: &Item, _: &Item| i.key, count()) + .reward(|_: &u32, n: &usize| SoftScore::of(*n as i64)) + .named("self tri groups"); + let model = Model { + a: vec![Item { key: 1, value: 1 }; 5], + b: vec![], + c: vec![], + d: vec![], + e: vec![], + }; + assert_eq!(constraint.evaluate(&model), SoftScore::of(10)); +} + +#[test] +fn self_join_quad_group_by_counts_combinations() { + let mut constraint = solverforge_scoring::stream::QuadConstraintStream::< + Model, + Item, + u32, + _, + _, + _, + SoftScore, + >::new_self_join( + source(a as fn(&Model) -> &[Item], ChangeSource::Descriptor(0)), + |_: &Model, i: &Item, _: usize| i.key, + ) + .group_by(|i: &Item, _: &Item, _: &Item, _: &Item| i.key, count()) + .reward(|_: &u32, n: &usize| SoftScore::of(*n as i64)) + .named("self quad groups"); + let mut m = model(); + m.a = vec![Item { key: 1, value: 1 }; 5]; + let mut score = constraint.initialize(&m); + assert_eq!(score, SoftScore::of(5)); + score = score + constraint.on_retract(&m, 0, 0); + m.a[0].key = 2; + score = score + constraint.on_insert(&m, 0, 0); + assert_eq!(score, SoftScore::of(1)); + assert_eq!(score, constraint.evaluate(&m)); +} + +#[test] +fn self_join_penta_group_by_counts_combinations() { + let mut constraint = solverforge_scoring::stream::PentaConstraintStream::< + Model, + Item, + u32, + _, + _, + _, + SoftScore, + >::new_self_join( + source(a as fn(&Model) -> &[Item], ChangeSource::Descriptor(0)), + |_: &Model, i: &Item, _: usize| i.key, + ) + .group_by( + |i: &Item, _: &Item, _: &Item, _: &Item, _: &Item| i.key, + count(), + ) + .reward(|_: &u32, n: &usize| SoftScore::of(*n as i64)) + .named("self penta groups"); + let mut m = model(); + m.a = vec![Item { key: 1, value: 1 }; 5]; + let mut score = constraint.initialize(&m); + assert_eq!(score, SoftScore::of(1)); + score = score + constraint.on_retract(&m, 0, 0); + m.a[0].key = 2; + score = score + constraint.on_insert(&m, 0, 0); + assert_eq!(score, SoftScore::of(0)); + assert_eq!(score, constraint.evaluate(&m)); +} + +#[test] +fn tri_group_by_preserves_filter_and_incremental_counts() { + let mut constraint = ConstraintFactory::::new() + .for_each(source( + a as fn(&Model) -> &[Item], + ChangeSource::Descriptor(0), + )) + .join(( + source(b as fn(&Model) -> &[Item], ChangeSource::Descriptor(1)), + equal_bi(|a: &Item| a.key, |b: &Item| b.key), + )) + .join(( + source(c as fn(&Model) -> &[Item], ChangeSource::Descriptor(2)), + equal_on(pair_key, leaf_key), + )) + .filter(|a: &Item, _: &Item, _: &Item| a.value > 0) + .group_by(|a: &Item, _: &Item, _: &Item| a.key, count()) + .penalize(|_: &u32, n: &usize| SoftScore::of(*n as i64)) + .named("tri groups"); + let mut model = Model { + a: vec![Item { key: 1, value: 1 }, Item { key: 1, value: -1 }], + b: vec![Item { key: 1, value: 2 }, Item { key: 1, value: 2 }], + c: vec![Item { key: 1, value: 3 }], + d: vec![], + e: vec![], + }; + assert_eq!(constraint.evaluate(&model), SoftScore::of(-2)); + assert_eq!(constraint.match_count(&model), 1); + let mut score = constraint.initialize(&model); + score = score + constraint.on_retract(&model, 0, 0); + model.a[0].key = 2; + score = score + constraint.on_insert(&model, 0, 0); + assert_eq!(score, SoftScore::of(0)); + assert_eq!(score, constraint.evaluate(&model)); + score = score + constraint.on_retract(&model, 0, 0); + model.a[0].key = 1; + score = score + constraint.on_insert(&model, 0, 0); + assert_eq!(score, SoftScore::of(-2)); + constraint.reset(); + assert_eq!(constraint.initialize(&model), score); +} + +fn d(m: &Model) -> &[Item] { + &m.d +} +fn e(m: &Model) -> &[Item] { + &m.e +} +type TriRow<'a> = Pair, Leaf<'a, Item>>, Leaf<'a, Item>>; +type QuadRow<'a> = Pair, Leaf<'a, Item>>; +fn tri_key(row: &TriRow<'_>) -> u32 { + row.right.entity.key +} +fn quad_key(row: &QuadRow<'_>) -> u32 { + row.right.entity.key +} + +macro_rules! tri { + () => { + ConstraintFactory::::new() + .for_each(source( + a as fn(&Model) -> &[Item], + ChangeSource::Descriptor(0), + )) + .join(( + source(b as fn(&Model) -> &[Item], ChangeSource::Descriptor(1)), + equal_bi(|a: &Item| a.key, |b: &Item| b.key), + )) + .join(( + source(c as fn(&Model) -> &[Item], ChangeSource::Descriptor(2)), + equal_on(pair_key, leaf_key), + )) + }; +} +macro_rules! quad { + () => { + tri!().join(( + source(d as fn(&Model) -> &[Item], ChangeSource::Descriptor(3)), + equal_on(tri_key, leaf_key), + )) + }; +} +macro_rules! penta { + () => { + quad!().join(( + source(e as fn(&Model) -> &[Item], ChangeSource::Descriptor(4)), + equal_on(quad_key, leaf_key), + )) + }; +} + +fn model() -> Model { + Model { + a: vec![Item { key: 1, value: 1 }], + b: vec![Item { key: 1, value: 2 }; 2], + c: vec![Item { key: 1, value: 3 }], + d: vec![Item { key: 1, value: 4 }], + e: vec![Item { key: 1, value: 5 }], + } +} +fn verify_sources( + mut constraint: impl IncrementalConstraint, + sources: usize, + expected: SoftScore, +) { + let mut m = model(); + assert_eq!(constraint.evaluate(&m), expected); + assert_eq!(constraint.match_count(&m), 1); + let mut score = constraint.initialize(&m); + assert_eq!(score, expected); + for descriptor in 0..sources { + score = score + constraint.on_retract(&m, 0, descriptor); + assert_eq!(constraint.on_retract(&m, 0, descriptor), SoftScore::of(0)); + let rows = match descriptor { + 0 => &mut m.a, + 1 => &mut m.b, + 2 => &mut m.c, + 3 => &mut m.d, + _ => &mut m.e, + }; + rows[0].key = 2; + score = score + constraint.on_insert(&m, 0, descriptor); + assert_eq!(score, constraint.evaluate(&m)); + score = score + constraint.on_retract(&m, 0, descriptor); + let rows = match descriptor { + 0 => &mut m.a, + 1 => &mut m.b, + 2 => &mut m.c, + 3 => &mut m.d, + _ => &mut m.e, + }; + rows[0].key = 1; + score = score + constraint.on_insert(&m, 0, descriptor); + assert_eq!(score, expected); + assert_eq!(score, constraint.evaluate(&m)); + } + constraint.reset(); + assert_eq!(constraint.initialize(&m), expected); + assert_eq!(constraint.get_matches(&m).len(), 1); +} + +#[test] +fn quad_group_by_retracts_each_source() { + let constraint = quad!() + .group_by(|_: &Item, _: &Item, _: &Item, d: &Item| d.key, count()) + .reward(|_: &u32, n: &usize| SoftScore::of(*n as i64)) + .named("quad groups"); + verify_sources(constraint, 4, SoftScore::of(2)); +} + +struct Payload(i64); +#[test] +fn penta_group_by_collects_move_only_payloads_and_retracts_each_source() { + use solverforge_scoring::stream::collector::{collect_vec, CollectedVec}; + let constraint = penta!() + .group_by( + |_: &Item, _: &Item, _: &Item, _: &Item, e: &Item| e.key, + collect_vec(|(a, b, c, d, e): (&Item, &Item, &Item, &Item, &Item)| { + Payload(a.value + b.value + c.value + d.value + e.value) + }), + ) + .penalize(|_: &u32, result: &CollectedVec| { + SoftScore::of(result.iter().map(|p| p.0).sum()) + }) + .named("penta payload groups"); + verify_sources(constraint, 5, SoftScore::of(-30)); +} + +#[test] +fn joined_group_by_preserves_reject_all_filters_at_every_arity() { + let tri = tri!() + .filter(|_: &Item, _: &Item, _: &Item| false) + .group_by(|a: &Item, _: &Item, _: &Item| a.key, count()) + .penalize(SoftScore::of(1)) + .named("empty tri"); + let quad = quad!() + .filter(|_: &Item, _: &Item, _: &Item, _: &Item| false) + .group_by(|a: &Item, _: &Item, _: &Item, _: &Item| a.key, count()) + .penalize(SoftScore::of(1)) + .named("empty quad"); + let penta = penta!() + .filter(|_: &Item, _: &Item, _: &Item, _: &Item, _: &Item| false) + .group_by( + |a: &Item, _: &Item, _: &Item, _: &Item, _: &Item| a.key, + count(), + ) + .penalize(SoftScore::of(1)) + .named("empty penta"); + fn assert_empty(mut constraint: impl IncrementalConstraint) { + let m = model(); + assert_eq!(constraint.evaluate(&m), SoftScore::of(0)); + assert_eq!(constraint.match_count(&m), 0); + assert_eq!(constraint.initialize(&m), SoftScore::of(0)); + } + assert_empty(tri); + assert_empty(quad); + assert_empty(penta); +} diff --git a/crates/solverforge/WIREFRAME.md b/crates/solverforge/WIREFRAME.md index 713fe54d..70ead1c9 100644 --- a/crates/solverforge/WIREFRAME.md +++ b/crates/solverforge/WIREFRAME.md @@ -308,6 +308,12 @@ grouping and passes joined pairs as `(&A, &B)`. - `examples/mixed-job-shop` — mixed scalar/list planning model - `examples/nqueens` — scalar assignment model +Extractor ergonomics: Tri/Quad/Penta cross streams and same-source arity streams +also support `.group_by(key, collector)`. Key closures receive three/four/five +entity references; collector mappers receive borrowed tuples, and scoring +weights receive `(&group_key, &collector_result)`. These methods retain upstream +filters and incremental retraction semantics through the common grouping operator. + Extractor ergonomics: `for_each` accepts `CollectionExtract`. Raw keyed join targets accept any `CollectionExtract` target, including slice closures, `vec(|s| &s.field)`, custom extractor structs, and macro-generated solution source methods. Passing a right-hand `UniConstraintStream` remains the preferred form when the source needs its own `.filter(...)`. Model-owned keyed joins use solution source methods generated by `#[planning_solution]`, preserving hidden descriptor/static metadata: