D-QRY-3 (board half): count and amount per pipeline stage as two grouped programs

.claude/board/LATEST_STATE.md

@@ -85,6 +85,26 @@
 - Quack became consumable by git with lance-graph #1457
   (`ISS-LG-MASKRISC-PATH-NDARRAY`, resolved).
 
+## Board view (D-QRY-3, board half)
+
+- `hubspo-query::board::Board`: per (pipeline, stage), record count and
+  amount sum as two `GroupReduce` programs over the composite address
+  `pipeline × stages + stage` (no row ids, no composite key lane).
+  Records without an amount are counted, not summed; a stage whose records
+  all lack one reads `amount: None`, never `0`. `weighted()` is exact in
+  minor units × basis points (stage total × p = Σ record × p, tested).
+- `tests/board_oracle.rs`: unfiltered, filtered and narrow boards equal a
+  plain per-cell tally over 6,000 deals (3 pipelines × 7 stages, 4 % deleted,
+  25 % without amount, one stage with no amounts at all).
+  Disable runs 2026-10-10: sum without the amount validity (3 red), empty
+  sum read as 0 (2 red), pipeline/stage swapped (3 red), zero group
+  universe accepted (1 red), empty stages reported (1 red — the narrow-board
+  test, added after this disable first stayed green: every earlier fixture
+  occupied all 21 stages).
+- Open (the saved-view half): view definitions as data, and sorting.
+  Quack has no `ORDER BY` (order is storage order), so a table view sorted
+  by a property needs its own decision.
+
 ## Lists (D-QRY-2)
 
 - `hubspo-query::list`: every list operation is a Quack filter over planes

.claude/board/STATUS_BOARD.md

@@ -47,8 +47,8 @@
 | D-id | Deliverable | Size | Status | Evidence |
 |---|---|---|---|---|
 | D-QRY-1 | Filter compiler: HubSpot filter groups (AND within group, OR across groups; operators EQ/NEQ/LT/…/HAS_PROPERTY/CONTAINS_TOKEN/IN/BETWEEN) → Quack program | L | Shipped | `hubspo-store::catalog` + `hubspo-query::filter`; 600 random filter sets equal a plain-loop oracle on mask-risc; #19 |
-| D-QRY-2 | Lists: active (incremental on version delta) + static (stored mask); membership history | L | In PR | `hubspo-query::list` (refresh, static add/remove/members, added/removed); tests vs full recomputation. P-LIST-1 2.6 ms at 10M; P-SPARSE-1 1.93 ms, 257 KiB flat at 10M. Storage of list masks lands with D-STO-1 |
-| D-QRY-3 | Saved views + board view (report pivot over pipeline × stage) | M | Queued |  |
+| D-QRY-2 | Lists: active (incremental on version delta) + static (stored mask); membership history | L | Shipped | `hubspo-query::list` (refresh, static add/remove/members, added/removed); tests vs full recomputation. P-LIST-1 2.6 ms at 10M; P-SPARSE-1 1.93 ms, 257 KiB flat at 10M. Storage of list masks lands with D-STO-1 |
+| D-QRY-3 | Saved views + board view (report pivot over pipeline × stage) | M | In progress | board half in PR: `hubspo-query::board` (count + amount per stage as two `GroupReduce` programs); saved views + sorting open |
 | D-QRY-4 | Quack lowering of callcenter's `ParsedQuery` (the upstream DataFusion `parsed_query_to_plan` is an unimplemented stub) | M | Queued |  |
 | D-QRY-5 | Association-traversing filters ("contacts whose company has an open deal"): Quack `Semijoin` over edge lanes for one hop; multi-hop needs cypher-quack shapes that do not exist yet → upstream request first | L | Queued |  |
 | D-QRY-6 | Reports: single-object, cross-object, funnel, time-series, attribution skeleton on `lance-graph-report` | XL | Queued |  |

crates/hubspo-query/src/board.rs

@@ -0,0 +1,143 @@
+//! D-QRY-3 (board half): the pipeline board as two grouped Quack programs.
+//!
+//! HubSpot's board view shows, per pipeline stage, how many records sit in
+//! it and the total (and probability-weighted) amount. That is a `GROUP BY
+//! (pipeline, stage)` over the records a view's filter keeps, so it is two
+//! [`Agg::GroupReduce`] programs over the composite address
+//! `pipeline × stages + stage` ([`GroupAddr::Pair`]): one counts, one sums
+//! the amount. Neither materialises a row or a composite key lane.
+//!
+//! Records without an amount are counted but not summed. A stage whose
+//! records all lack an amount reports `amount: None`, never `0`: an empty sum
+//! is not a zero sum (`normalize_group_sink`).
+//!
+//! Weighted amounts are exact: `amount × probability` in minor units ×
+//! basis points (`i128`), so summing per record and multiplying the stage
+//! total agree to the last unit. Rounding is the display's business.
+//!
+//! Not here yet (D-QRY-3's saved-view half): view definitions as data and
+//! sorting. Quack has no `ORDER BY` (order is storage order), so a sorted
+//! table view needs its own decision.
+
+use lance_graph_quack::{Agg, Col, Filter, GroupAddr, GroupAgg, Mask, Query, normalize_group_sink};
+
+/// Which lanes a board reads.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Board {
+    /// The pipeline code lane (`u32`, `0..pipelines`).
+    pub pipeline: Col,
+    /// The stage code lane (`u32`, `0..stages`, numbered within a pipeline).
+    pub stage: Col,
+    /// Number of pipelines.
+    pub pipelines: u32,
+    /// Stage slots per pipeline (the largest pipeline's stage count).
+    pub stages: u32,
+    /// The amount lane (`i32`, minor units) and its validity plane when
+    /// nullable.
+    pub amount: Option<(Col, Option<Mask>)>,
+}
+
+/// Why a board cannot be built.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BoardError {
+    /// `pipelines × stages` is zero or does not fit a group universe.
+    GroupUniverse,
+}
+
+/// One board column: a stage of a pipeline.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Column {
+    /// Pipeline code.
+    pub pipeline: u32,
+    /// Stage code within the pipeline.
+    pub stage: u32,
+    /// Records in the stage.
+    pub count: u64,
+    /// Sum of their amounts; `None` when none of them has an amount.
+    pub amount: Option<i64>,
+}
+
+impl Board {
+    /// The group universe `pipelines × stages`.
+    ///
+    /// # Errors
+    ///
+    /// [`BoardError::GroupUniverse`] if it is zero or overflows `u32`.
+    pub fn groups(&self) -> Result<u32, BoardError> {
+        match self.pipelines.checked_mul(self.stages) {
+            Some(0) | None => Err(BoardError::GroupUniverse),
+            Some(g) => Ok(g),
+        }
+    }
+
+    fn addr(&self) -> GroupAddr {
+        GroupAddr::Pair {
+            hi: self.pipeline,
+            lo: self.stage,
+            stride: self.stages,
+        }
+    }
+
+    /// Records per stage, over the records `filter` keeps. The caller runs it
+    /// with an `Out::I64` sink of [`Board::groups`] slots.
+    pub fn count_query(&self, filter: &Filter) -> Query {
+        Query {
+            filter: filter.clone(),
+            agg: Agg::GroupReduce {
+                key: self.addr(),
+                agg: GroupAgg::Count,
+            },
+        }
+    }
+
+    /// Amount per stage over the kept records that have an amount; `None`
+    /// when the board has no amount lane.
+    pub fn amount_query(&self, filter: &Filter) -> Option<Query> {
+        let (col, validity) = self.amount?;
+        let filter = match validity {
+            Some(valid) => Filter::and([filter.clone(), Filter::plane(valid)]),
+            None => filter.clone(),
+        };
+        Some(Query {
+            filter,
+            agg: Agg::GroupReduce {
+                key: self.addr(),
+                agg: GroupAgg::SumI32(col),
+            },
+        })
+    }
+
+    /// The board's columns from the two executed sinks, in group order;
+    /// stages no kept record reached are left out.
+    ///
+    /// Both sinks are normalised in place (`normalize_group_sink`).
+    pub fn columns(&self, counts: &mut [i64], amounts: Option<&mut [i64]>) -> Vec<Column> {
+        let present = normalize_group_sink(GroupAgg::Count.fold(), counts);
+        let amount_present = match (self.amount, amounts) {
+            (Some((col, _)), Some(sink)) => Some((
+                normalize_group_sink(GroupAgg::SumI32(col).fold(), sink),
+                sink,
+            )),
+            _ => None,
+        };
+        let bit = |mask: &[u64], g: usize| mask[g / 64] >> (g % 64) & 1 == 1;
+        let stride = self.stages as usize;
+        (0..counts.len())
+            .filter(|&g| bit(&present, g))
+            .map(|g| Column {
+                pipeline: (g / stride) as u32,
+                stage: (g % stride) as u32,
+                count: counts[g].unsigned_abs(),
+                amount: amount_present
+                    .as_ref()
+                    .and_then(|(mask, sink)| bit(mask, g).then(|| sink[g])),
+            })
+            .collect()
+    }
+}
+
+/// `amount × probability` in minor units × basis points (exact). Divide by
+/// 10,000 for minor units.
+pub fn weighted(amount_minor: i64, probability_bp: u16) -> i128 {
+    i128::from(amount_minor) * i128::from(probability_bp)
+}

crates/hubspo-query/src/lib.rs

@@ -7,8 +7,11 @@
 //! - [`filter`] (D-QRY-1): HubSpot filter groups → one Quack [`Filter`].
 //! - [`list`] (D-QRY-2): active / static lists and membership history as
 //!   mask programs.
+//! - [`board`] (D-QRY-3, board half): count and amount per pipeline stage
+//!   as two grouped programs.
 //!
 //! [`Filter`]: lance_graph_quack::Filter
 
+pub mod board;
 pub mod filter;
 pub mod list;