diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index 7ae7645..c82a8e1 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -85,6 +85,31 @@ - Quack became consumable by git with lance-graph #1457 (`ISS-LG-MASKRISC-PATH-NDARRAY`, resolved). +## Lists (D-QRY-2) + +- `hubspo-query::list`: every list operation is a Quack filter over planes + the caller names, run by mask-risc (no row ids, no population loop). + - `refresh(predicate, prev, changed)` = `(prev ∧ ¬changed) ∨ + (predicate ∧ changed)`: an active list re-reads the predicate only + where records changed. Deletes need no special case (the compiled + predicate carries `live`). + - static lists: `add`, `remove`, and `members` (`∧ live` on every read); + - history: `added` / `removed` between two stored versions. +- Invariant this rests on: **ordinals are never reused** (records are + appended, the base is never rewritten; P-SPARSE-1). A store that reused + ordinals would have to clear a deleted record's bit in every static list. +- `tests/list_oracle.rs`: 40 rounds of updates / inserts / deletes keep a + refreshed list equal to full recomputation and to a plain loop (added / + removed equal plain set differences, both > 100 over the run); a static + list equals a plain set through adds, removes and deletes; refresh keeps + stored bits outside the changed rows (> 1,000 of them disagree with the + predicate on purpose). + Disable runs 2026-10-10 (`cargo test -p hubspo-query --test list_oracle`): + refresh keeping stale bits (2 red), refresh re-reading every row (1 red), + `members` without `live` (1 red), `added` computing `removed` (1 red), + `remove` keeping rows (1 red). A `live` gate inside `add` stayed green + and was removed: `members` already applies it on every read. + ## Harvest (D-ONT-1, public-catalog half) - `scripts/fetch-hubspot-specs.sh` → `.cache/` (gitignored), then diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 93170ef..27f9fa9 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -46,8 +46,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 | In PR | `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 | Queued | P-LIST-1 2026-10-10: 2.6 ms at 10M; alloc ∝ fragments touched. P-SPARSE-1: 1.93 ms, 257 KiB flat at 10M | +| 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-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 | | diff --git a/crates/hubspo-query/src/lib.rs b/crates/hubspo-query/src/lib.rs index 70a9e39..c89599a 100644 --- a/crates/hubspo-query/src/lib.rs +++ b/crates/hubspo-query/src/lib.rs @@ -5,7 +5,10 @@ //! - Zone: zone 2 (no serde, no wire formats). //! - Deliverables: D-QRY-1..7 (`.claude/plans/hubspo-integration-v1.md`). //! - [`filter`] (D-QRY-1): HubSpot filter groups → one Quack [`Filter`]. +//! - [`list`] (D-QRY-2): active / static lists and membership history as +//! mask programs. //! //! [`Filter`]: lance_graph_quack::Filter pub mod filter; +pub mod list; diff --git a/crates/hubspo-query/src/list.rs b/crates/hubspo-query/src/list.rs new file mode 100644 index 0000000..547c104 --- /dev/null +++ b/crates/hubspo-query/src/list.rs @@ -0,0 +1,70 @@ +//! D-QRY-2: active lists, static lists and membership history as mask +//! programs (H-ADR-5). +//! +//! A list's membership is a resident mask plane over record ordinals; the +//! store owns its words. Everything here builds a Quack [`Filter`] over +//! planes the caller names, so every list operation runs on mask-risc like +//! any other query: no row ids are materialised and no Rust loop walks the +//! population. +//! +//! - **Active list.** Membership is the compiled predicate +//! ([`crate::filter::compile`]). After a write, [`refresh`] re-evaluates the +//! predicate only where records changed and keeps the stored bits +//! elsewhere: `(prev ∧ ¬changed) ∨ (predicate ∧ changed)`. A deleted record +//! is a changed record whose live bit is clear, so it leaves the list with +//! no special case — the compiled predicate already conjoins `live`. +//! P-LIST-1 / P-SPARSE-1 measured the step at 0.85–2.6 ms up to 10M rows. +//! - **Static list.** Membership is whatever was added. [`add`] and +//! [`remove`] take a plane of the records named in the request. A static +//! list's stored bits can outlive a deleted record, so readers go through +//! [`members`], which conjoins `live`. This relies on ordinals never being +//! reused (see [`add`]). +//! - **History.** Between two stored versions of one list, [`added`] and +//! [`removed`] are the two differences. +//! +//! [`Filter`]: lance_graph_quack::Filter + +use lance_graph_quack::{Filter, Mask}; + +/// The active list's membership after a write that touched `changed`. +/// +/// `predicate` is the list's compiled filter (it includes `live`); `prev` is +/// the stored membership; `changed` marks every record inserted, updated or +/// deleted since `prev` was stored. +pub fn refresh(predicate: &Filter, prev: Mask, changed: Mask) -> Filter { + Filter::or([ + Filter::and([Filter::plane(prev), Filter::negate(Filter::plane(changed))]), + Filter::and([predicate.clone(), Filter::plane(changed)]), + ]) +} + +/// A static list after adding the records in `rows`. +/// +/// Not gated on `live`: a bit stored for a deleted record is inert, because +/// every read goes through [`members`]. That holds only while ordinals are +/// never reused, which the store guarantees (records are appended, the base +/// is never rewritten; P-SPARSE-1). A store that reused ordinals would have +/// to clear a deleted record's bit in every static list first. +pub fn add(prev: Mask, rows: Mask) -> Filter { + Filter::or([Filter::plane(prev), Filter::plane(rows)]) +} + +/// A static list after removing the records in `rows`. +pub fn remove(prev: Mask, rows: Mask) -> Filter { + Filter::and([Filter::plane(prev), Filter::negate(Filter::plane(rows))]) +} + +/// A list's current members: its stored bits restricted to live records. +pub fn members(list: Mask, live: Mask) -> Filter { + Filter::and([Filter::plane(list), Filter::plane(live)]) +} + +/// Records in `next` that were not in `prev`. +pub fn added(prev: Mask, next: Mask) -> Filter { + Filter::and([Filter::plane(next), Filter::negate(Filter::plane(prev))]) +} + +/// Records in `prev` that are not in `next`. +pub fn removed(prev: Mask, next: Mask) -> Filter { + Filter::and([Filter::plane(prev), Filter::negate(Filter::plane(next))]) +} diff --git a/crates/hubspo-query/tests/list_oracle.rs b/crates/hubspo-query/tests/list_oracle.rs new file mode 100644 index 0000000..3391b81 --- /dev/null +++ b/crates/hubspo-query/tests/list_oracle.rs @@ -0,0 +1,276 @@ +//! D-QRY-2 against full recomputation: an active list maintained by +//! `list::refresh` over 40 rounds of updates, inserts and deletes must equal +//! the predicate evaluated from scratch after every round, and a static list +//! maintained by `add` / `remove` must equal a plain set. + +use std::collections::BTreeSet; + +use hubspo_query::filter::{FilterGroup, Operator, PropertyFilter, compile}; +use hubspo_query::list; +use hubspo_store::catalog::{Catalog, Codebook, FieldSpec}; +use lance_graph_mask_risc::{ + Foreign, LaneRef, Out, Planes, Scratch, execute_into, materialize_rows, words_for, +}; +use lance_graph_quack::bind::{FieldKind, create_table}; +use lance_graph_quack::{Agg, Col, Filter, Mask, Query, lower}; + +const CAP: usize = 4_000; +const STAGES: &[&str] = &["lead", "opportunity", "customer", "evangelist"]; +// Planes: 0 live, 1 stored membership, 2 changed / named rows, 3 second list. +const LIVE: Mask = Mask(0); +const PREV: Mask = Mask(1); +const CHANGED: Mask = Mask(2); +const NEXT: Mask = Mask(3); + +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: usize) -> usize { + (self.next() % n as u64) as usize + } +} + +struct Table { + live: Vec, + stage: Vec, + amount: Vec, +} + +fn bits(rows: impl IntoIterator) -> Vec { + let mut w = vec![0u64; words_for(CAP)]; + for r in rows { + w[r / 64] |= 1 << (r % 64); + } + w +} + +fn catalog() -> Catalog { + let mut c = Catalog::new(); + let t = create_table("contacts") + .width(8) + .register(&mut c) + .expect("register") + .id; + c.declare( + t, + FieldSpec { + name: "lifecyclestage".into(), + col: Col(0), + kind: FieldKind::Code, + validity: None, + codebook: Some(Codebook::new(STAGES.iter().copied()).expect("codebook")), + }, + ) + .expect("stage"); + c.declare( + t, + FieldSpec { + name: "amount".into(), + col: Col(1), + kind: FieldKind::I32, + validity: None, + codebook: None, + }, + ) + .expect("amount"); + c +} + +/// `stage ∈ {opportunity, customer} ∧ amount ≥ 2000`, or `stage = evangelist`. +fn predicate(cat: &Catalog) -> Filter { + compile( + "contacts", + &[ + FilterGroup { + filters: vec![ + PropertyFilter::set( + "lifecyclestage", + Operator::In, + &["opportunity", "customer"], + ), + PropertyFilter::new("amount", Operator::Gte, "2000"), + ], + }, + FilterGroup { + filters: vec![PropertyFilter::new( + "lifecyclestage", + Operator::Eq, + "evangelist", + )], + }, + ], + cat, + ) + .expect("compile") +} + +fn oracle_member(t: &Table, row: usize) -> bool { + t.live[row] && ((matches!(t.stage[row], 1 | 2) && t.amount[row] >= 2000) || t.stage[row] == 3) +} + +/// Run `filter` over the table with planes `[live, p1, p2, p3]`; returns the +/// kept rows. +fn run(t: &Table, filter: Filter, p1: &[u64], p2: &[u64], p3: &[u64]) -> Vec { + let live = bits((0..CAP).filter(|&r| t.live[r])); + let masks: [&[u64]; 4] = [&live, p1, p2, p3]; + let lanes = [LaneRef::U32(&t.stage), LaneRef::I32(&t.amount)]; + let planes = Planes { + n_rows: CAP, + masks: &masks, + lanes: &lanes, + }; + let program = lower(&Query { + filter, + agg: Agg::Rows, + }) + .expect("lower"); + let mut out = vec![0u64; words_for(CAP)]; + let mut scratch = Scratch::for_program(&program, CAP).expect("scratch"); + execute_into( + &program, + &planes, + &Foreign::NONE, + &mut scratch, + Out::Mask(&mut out), + ) + .expect("execute"); + materialize_rows(&out, CAP) +} + +fn table(r: &mut Rng) -> Table { + Table { + live: (0..CAP).map(|i| i < 3_000).collect(), + stage: (0..CAP).map(|_| r.below(STAGES.len()) as u32).collect(), + amount: (0..CAP).map(|_| r.below(5_000) as i32).collect(), + } +} + +#[test] +fn an_active_list_refreshed_on_changes_equals_full_recomputation() { + let mut r = Rng(0x000A_C71E); + let cat = catalog(); + let pred = predicate(&cat); + let mut t = table(&mut r); + let none = bits([]); + let mut stored = run(&t, pred.clone(), &none, &none, &none); + let (mut joined, mut left) = (0usize, 0usize); + for round in 0..40 { + let mut changed = BTreeSet::new(); + for _ in 0..60 { + let row = r.below(CAP); + match r.below(10) { + // update an existing record + 0..=6 if t.live[row] => { + t.stage[row] = r.below(STAGES.len()) as u32; + t.amount[row] = r.below(5_000) as i32; + } + // delete + 7 if t.live[row] => t.live[row] = false, + // insert into a free slot + 8 | 9 if !t.live[row] => { + t.live[row] = true; + t.stage[row] = r.below(STAGES.len()) as u32; + t.amount[row] = r.below(5_000) as i32; + } + _ => continue, + } + changed.insert(row); + } + let prev = bits(stored.iter().copied()); + let ch = bits(changed.iter().copied()); + let next = run(&t, list::refresh(&pred, PREV, CHANGED), &prev, &ch, &none); + let full = run(&t, pred.clone(), &none, &none, &none); + let want: Vec = (0..CAP).filter(|&row| oracle_member(&t, row)).collect(); + assert_eq!(full, want, "round {round}: full evaluation vs plain loop"); + assert_eq!(next, full, "round {round}: incremental vs full"); + + // History of this step. + let next_bits = bits(next.iter().copied()); + let add = run(&t, list::added(PREV, NEXT), &prev, &none, &next_bits); + let rem = run(&t, list::removed(PREV, NEXT), &prev, &none, &next_bits); + let before: BTreeSet = stored.iter().copied().collect(); + let after: BTreeSet = next.iter().copied().collect(); + assert_eq!(add, after.difference(&before).copied().collect::>()); + assert_eq!(rem, before.difference(&after).copied().collect::>()); + joined += add.len(); + left += rem.len(); + stored = next; + } + // Anti-vacuity: membership really moved in both directions. + assert!(joined > 100 && left > 100, "joined {joined}, left {left}"); +} + +#[test] +fn a_static_list_follows_adds_removes_and_deletes() { + // Ordinals are never reused here, as in the store: a deleted record + // stays dead. + let mut r = Rng(0x0057_A71C); + let mut t = table(&mut r); + let none = bits([]); + let mut want: BTreeSet = BTreeSet::new(); + let mut stored: Vec = Vec::new(); + for round in 0..30 { + let named: BTreeSet = (0..40).map(|_| r.below(CAP)).collect(); + let prev = bits(stored.iter().copied()); + let rows = bits(named.iter().copied()); + stored = if round % 3 == 2 { + want.retain(|row| !named.contains(row)); + run(&t, list::remove(PREV, CHANGED), &prev, &rows, &none) + } else { + want.extend(named.iter().copied()); + run(&t, list::add(PREV, CHANGED), &prev, &rows, &none) + }; + // Some members are deleted between requests. + for _ in 0..5 { + let row = r.below(CAP); + t.live[row] = false; + } + let stored_bits = bits(stored.iter().copied()); + let members = run(&t, list::members(PREV, LIVE), &stored_bits, &none, &none); + let expect: Vec = want.iter().copied().filter(|&row| t.live[row]).collect(); + assert_eq!(members, expect, "round {round}"); + } + assert!(want.len() > 200, "the list must grow to mean something"); + assert!( + want.iter().any(|&row| !t.live[row]), + "some stored members must have been deleted" + ); +} + +#[test] +fn refresh_reads_the_predicate_only_where_records_changed() { + // Stored bits that disagree with the predicate on purpose: every even + // row. Only rows 0..100 are reported changed, so only they may be + // re-evaluated; a refresh that re-reads the whole table would rewrite + // the rest too. + let mut r = Rng(0x001C_4A6E); + let t = table(&mut r); + let pred = predicate(&catalog()); + let none = bits([]); + let stored: Vec = (0..CAP).filter(|row| row % 2 == 0).collect(); + let prev = bits(stored.iter().copied()); + let ch = bits(0..100); + let next = run(&t, list::refresh(&pred, PREV, CHANGED), &prev, &ch, &none); + let want: Vec = (0..CAP) + .filter(|&row| { + if row < 100 { + oracle_member(&t, row) + } else { + row % 2 == 0 + } + }) + .collect(); + assert_eq!(next, want); + // Anti-vacuity: outside the changed rows the stored bits really differ + // from the predicate, or keeping them would prove nothing. + let differ = (100..CAP) + .filter(|&row| (row % 2 == 0) != oracle_member(&t, row)) + .count(); + assert!(differ > 1_000, "only {differ} rows differ"); +}