From 64b72393e0f643f36989ab08e24eb4dfe4b164cc Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:37:12 +0000 Subject: [PATCH 1/4] D-QRY-3 (board): count and amount per pipeline stage as two grouped programs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Board::count_query / amount_query lower to GroupReduce over the composite address pipeline × stages + stage; columns() normalises both sinks so a stage whose records all lack an amount reads None, not 0. weighted() is exact in minor units × basis points. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-query/src/board.rs | 143 ++++++++++++ crates/hubspo-query/src/lib.rs | 3 + crates/hubspo-query/tests/board_oracle.rs | 256 ++++++++++++++++++++++ 3 files changed, 402 insertions(+) create mode 100644 crates/hubspo-query/src/board.rs create mode 100644 crates/hubspo-query/tests/board_oracle.rs diff --git a/crates/hubspo-query/src/board.rs b/crates/hubspo-query/src/board.rs new file mode 100644 index 0000000..7db2f92 --- /dev/null +++ b/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)>, +} + +/// 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, +} + +impl Board { + /// The group universe `pipelines × stages`. + /// + /// # Errors + /// + /// [`BoardError::GroupUniverse`] if it is zero or overflows `u32`. + pub fn groups(&self) -> Result { + 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 { + 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 { + 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) +} diff --git a/crates/hubspo-query/src/lib.rs b/crates/hubspo-query/src/lib.rs index c89599a..45c1a71 100644 --- a/crates/hubspo-query/src/lib.rs +++ b/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; diff --git a/crates/hubspo-query/tests/board_oracle.rs b/crates/hubspo-query/tests/board_oracle.rs new file mode 100644 index 0000000..bcce0b1 --- /dev/null +++ b/crates/hubspo-query/tests/board_oracle.rs @@ -0,0 +1,256 @@ +//! D-QRY-3 board half against a plain loop: per (pipeline, stage), the +//! grouped programs' counts and amount sums equal a row-by-row tally over +//! the same filtered records, with "no amount" kept apart from "zero". + +use std::collections::BTreeMap; + +use hubspo_query::board::{Board, BoardError, Column, weighted}; +use hubspo_query::filter::{FilterGroup, Operator, PropertyFilter, compile}; +use hubspo_store::catalog::{Catalog, Codebook, FieldSpec}; +use lance_graph_mask_risc::{Foreign, LaneRef, Out, Planes, Scratch, execute_into, words_for}; +use lance_graph_quack::bind::{FieldKind, create_table}; +use lance_graph_quack::{Col, Filter, Mask, Query, lower}; + +const N: usize = 6_000; +const PIPELINES: u32 = 3; +const STAGES: u32 = 7; +const OWNERS: &[&str] = &["ana", "ben", "cy", "dee"]; + +struct Rng(u64); +impl Rng { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + fn below(&mut self, n: u64) -> u64 { + self.next() % n + } +} + +struct Deals { + live: Vec, + pipeline: Vec, + stage: Vec, + owner: Vec, + amount: Vec>, + amount_lane: Vec, +} + +fn deals() -> Deals { + let mut r = Rng(0x0B0A_4D00); + let mut d = Deals { + live: Vec::new(), + pipeline: Vec::new(), + stage: Vec::new(), + owner: Vec::new(), + amount: Vec::new(), + amount_lane: Vec::new(), + }; + for _ in 0..N { + let pipeline = r.below(u64::from(PIPELINES)) as u32; + let stage = r.below(u64::from(STAGES)) as u32; + d.live.push(r.below(100) >= 4); + d.pipeline.push(pipeline); + d.stage.push(stage); + d.owner.push(r.below(OWNERS.len() as u64) as u32); + // Pipeline 2, stage 6 never carries an amount: a stage with records + // and no amount must read None, not 0. Elsewhere 25 % lack one. + let has = !(pipeline == 2 && stage == 6) && r.below(100) >= 25; + let raw = r.below(200_000) as i32 - 20_000; + d.amount.push(has.then_some(raw)); + d.amount_lane.push(raw); + } + d +} + +fn bits(f: impl Fn(usize) -> bool) -> Vec { + let mut w = vec![0u64; words_for(N)]; + for i in 0..N { + if f(i) { + w[i / 64] |= 1 << (i % 64); + } + } + w +} + +fn catalog() -> Catalog { + let mut c = Catalog::new(); + let t = create_table("deals") + .width(16) + .register(&mut c) + .expect("register") + .id; + c.declare( + t, + FieldSpec { + name: "hubspot_owner_id".into(), + col: Col(2), + kind: FieldKind::Code, + validity: None, + codebook: Some(Codebook::new(OWNERS.iter().copied()).expect("codebook")), + }, + ) + .expect("owner"); + c.declare( + t, + FieldSpec { + name: "amount".into(), + col: Col(3), + kind: FieldKind::I32, + validity: Some(Mask(1)), + codebook: None, + }, + ) + .expect("amount"); + c +} + +const BOARD: Board = Board { + pipeline: Col(0), + stage: Col(1), + pipelines: PIPELINES, + stages: STAGES, + amount: Some((Col(3), Some(Mask(1)))), +}; + +fn exec(d: &Deals, q: &Query, sink: &mut [i64]) { + let live = bits(|i| d.live[i]); + let valid = bits(|i| d.amount[i].is_some()); + let masks: [&[u64]; 2] = [&live, &valid]; + let lanes = [ + LaneRef::U32(&d.pipeline), + LaneRef::U32(&d.stage), + LaneRef::U32(&d.owner), + LaneRef::I32(&d.amount_lane), + ]; + let planes = Planes { + n_rows: N, + masks: &masks, + lanes: &lanes, + }; + let program = lower(q).expect("lower"); + let mut scratch = Scratch::for_program(&program, N).expect("scratch"); + execute_into( + &program, + &planes, + &Foreign::NONE, + &mut scratch, + Out::I64(sink), + ) + .expect("execute"); +} + +fn board(d: &Deals, filter: &Filter) -> Vec { + let groups = BOARD.groups().expect("groups") as usize; + let mut counts = vec![0i64; groups]; + let mut amounts = vec![0i64; groups]; + exec(d, &BOARD.count_query(filter), &mut counts); + exec( + d, + &BOARD.amount_query(filter).expect("amount"), + &mut amounts, + ); + BOARD.columns(&mut counts, Some(&mut amounts)) +} + +fn tally(d: &Deals, keep: impl Fn(usize) -> bool) -> Vec { + let mut cells: BTreeMap<(u32, u32), (u64, Option)> = BTreeMap::new(); + for i in (0..N).filter(|&i| d.live[i] && keep(i)) { + let cell = cells + .entry((d.pipeline[i], d.stage[i])) + .or_insert((0, None)); + cell.0 += 1; + if let Some(a) = d.amount[i] { + cell.1 = Some(cell.1.unwrap_or(0) + i64::from(a)); + } + } + cells + .into_iter() + .map(|((pipeline, stage), (count, amount))| Column { + pipeline, + stage, + count, + amount, + }) + .collect() +} + +#[test] +fn every_stage_matches_a_plain_tally() { + let d = deals(); + let cat = catalog(); + let all = compile("deals", &[], &cat).expect("all"); + let got = board(&d, &all); + let want = tally(&d, |_| true); + assert_eq!(got, want); + assert_eq!( + got.len(), + (PIPELINES * STAGES) as usize, + "every stage is occupied" + ); + let empty_sum = got + .iter() + .find(|c| c.pipeline == 2 && c.stage == 6) + .expect("stage 2/6"); + assert!( + empty_sum.count > 0 && empty_sum.amount.is_none(), + "{empty_sum:?}" + ); +} + +#[test] +fn a_filtered_board_matches_and_leaves_empty_stages_out() { + let d = deals(); + let cat = catalog(); + let groups = [ + FilterGroup { + filters: vec![ + PropertyFilter::new("hubspot_owner_id", Operator::Eq, "ben"), + PropertyFilter::new("amount", Operator::Gte, "150000"), + ], + }, + FilterGroup { + filters: vec![PropertyFilter::new("hubspot_owner_id", Operator::Eq, "dee")], + }, + ]; + let filter = compile("deals", &groups, &cat).expect("compile"); + let got = board(&d, &filter); + let want = tally(&d, |i| { + (d.owner[i] == 1 && d.amount[i].is_some_and(|a| a >= 150_000)) || d.owner[i] == 3 + }); + assert_eq!(got, want); + // Anti-vacuity: the filter must change both counts and sums. + let all = tally(&d, |_| true); + assert!( + got.iter().map(|c| c.count).sum::() < all.iter().map(|c| c.count).sum::() / 2 + ); +} + +#[test] +fn a_board_needs_a_group_universe() { + let mut b = BOARD; + b.stages = 0; + assert_eq!(b.groups(), Err(BoardError::GroupUniverse)); + b.stages = u32::MAX; + assert_eq!(b.groups(), Err(BoardError::GroupUniverse)); +} + +#[test] +fn weighting_the_stage_total_equals_summing_weighted_records() { + let d = deals(); + let cat = catalog(); + let all = compile("deals", &[], &cat).expect("all"); + let probability_bp = |stage: u32| -> u16 { (stage as u16 * 1_437) % 10_001 }; + for c in board(&d, &all) { + let Some(total) = c.amount else { continue }; + let per_record: i128 = (0..N) + .filter(|&i| d.live[i] && d.pipeline[i] == c.pipeline && d.stage[i] == c.stage) + .filter_map(|i| d.amount[i]) + .map(|a| weighted(i64::from(a), probability_bp(c.stage))) + .sum(); + assert_eq!(weighted(total, probability_bp(c.stage)), per_record); + } +} From 065397cff82ebefcef3d56d43a0a051f3ad131f3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:37:29 +0000 Subject: [PATCH 2/4] D-QRY-3 board: test that empty stages are left out Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-query/tests/board_oracle.rs | 28 +++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/crates/hubspo-query/tests/board_oracle.rs b/crates/hubspo-query/tests/board_oracle.rs index bcce0b1..bd70167 100644 --- a/crates/hubspo-query/tests/board_oracle.rs +++ b/crates/hubspo-query/tests/board_oracle.rs @@ -254,3 +254,31 @@ fn weighting_the_stage_total_equals_summing_weighted_records() { assert_eq!(weighted(total, probability_bp(c.stage)), per_record); } } + +#[test] +fn stages_no_kept_record_reaches_are_left_out() { + let d = deals(); + let cat = catalog(); + let filter = compile( + "deals", + &[FilterGroup { + filters: vec![ + PropertyFilter::new("hubspot_owner_id", Operator::Eq, "ben"), + PropertyFilter::new("amount", Operator::Gte, "178000"), + ], + }], + &cat, + ) + .expect("compile"); + let got = board(&d, &filter); + let want = tally(&d, |i| { + d.owner[i] == 1 && d.amount[i].is_some_and(|a| a >= 178_000) + }); + assert_eq!(got, want); + // Anti-vacuity: the filter must leave stages empty, and keep some. + assert!( + !got.is_empty() && got.len() < (PIPELINES * STAGES) as usize / 2, + "{} stages", + got.len() + ); +} From 1af01a2c5b601ebd54fc77b989fde22591788311 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:37:54 +0000 Subject: [PATCH 3/4] Board: D-QRY-3 board half Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- .claude/board/LATEST_STATE.md | 20 ++++++++++++++++++++ .claude/board/STATUS_BOARD.md | 2 +- 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index c82a8e1..af7cb78 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.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 diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 27f9fa9..4bf163a 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -48,7 +48,7 @@ |---|---|---|---|---| | 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-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 | | From bf2e447f601d4aa8984cbb07922be00a7bee07a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:38:14 +0000 Subject: [PATCH 4/4] Board: D-QRY-2 shipped (#20) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- .claude/board/STATUS_BOARD.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 4bf163a..e5125c8 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -47,7 +47,7 @@ | 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-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 | |