From 6e4a3b15b37a35a398eda83adbd29e0318d04171 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:41:28 +0000 Subject: [PATCH 1/3] D-STO-1 (resident half): owner-stamped batches over resident lanes hubspo-store::table::Table holds one dataset's records as resident lanes with a live plane and one validity plane per nullable lane; reads borrow them. A Batch (inserts, updates, deletes) is validated whole and lands as exactly one version or not at all, only for the table's owner (MailboxId, write on behalf). apply() returns the plane of touched rows, which is what list::refresh takes. Ordinals are never reused. Tests: 60 random batches match a plain model, lanes and planes alike; twelve refusal kinds leave the table untouched; a list maintained from the store's changed planes equals a full recomputation over 50 batches. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-query/tests/store_list.rs | 179 ++++++++++++ crates/hubspo-store/Cargo.toml | 2 + crates/hubspo-store/src/lib.rs | 7 +- crates/hubspo-store/src/table.rs | 352 ++++++++++++++++++++++++ crates/hubspo-store/tests/table.rs | 302 ++++++++++++++++++++ 5 files changed, 840 insertions(+), 2 deletions(-) create mode 100644 crates/hubspo-query/tests/store_list.rs create mode 100644 crates/hubspo-store/src/table.rs create mode 100644 crates/hubspo-store/tests/table.rs diff --git a/crates/hubspo-query/tests/store_list.rs b/crates/hubspo-query/tests/store_list.rs new file mode 100644 index 0000000..40e943b --- /dev/null +++ b/crates/hubspo-query/tests/store_list.rs @@ -0,0 +1,179 @@ +//! End to end on a real `Table`: batches applied to the store feed the +//! filter compiler and the active-list refresh, and the refreshed list +//! equals a full recomputation after every batch. + +use hubspo_query::filter::{FilterGroup, Operator, PropertyFilter, compile}; +use hubspo_query::list; +use hubspo_store::catalog::{Catalog, Codebook, FieldSpec}; +use hubspo_store::table::{Batch, LaneKind, LaneSpec, Table, Value}; +use lance_graph_mask_risc::{ + Foreign, 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 OWNER: u32 = 3; +const STAGES: &[&str] = &["lead", "opportunity", "customer"]; + +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 + } +} + +/// Run `filter` over the table's planes plus two extra planes (stored list, +/// changed rows) appended after the table's own. +fn run(t: &Table, filter: &Filter, extra: [&[u64]; 2]) -> Vec { + let (mut masks, lanes) = t.planes(); + masks.extend(extra); + let planes = Planes { + n_rows: t.n_rows(), + masks: &masks, + lanes: &lanes, + }; + let program = lower(&Query { + filter: filter.clone(), + agg: Agg::Rows, + }) + .expect("lower"); + let mut out = vec![0u64; words_for(t.n_rows())]; + let mut scratch = Scratch::for_program(&program, t.n_rows()).expect("scratch"); + execute_into( + &program, + &planes, + &Foreign::NONE, + &mut scratch, + Out::Mask(&mut out), + ) + .expect("execute"); + materialize_rows(&out, t.n_rows()) +} + +fn bits(n: usize, rows: &[usize]) -> Vec { + let mut w = vec![0u64; words_for(n)]; + for &r in rows { + w[r / 64] |= 1 << (r % 64); + } + w +} + +#[test] +fn a_list_on_the_store_follows_its_batches() { + let mut t = Table::new( + OWNER, + &[ + LaneSpec { + kind: LaneKind::U32, + nullable: false, + }, + LaneSpec { + kind: LaneKind::I32, + nullable: true, + }, + ], + ) + .expect("table"); + let mut cat = Catalog::new(); + let id = create_table("deals") + .width(8) + .register(&mut cat) + .expect("register") + .id; + cat.declare( + id, + FieldSpec { + name: "dealstage".into(), + col: Col(0), + kind: FieldKind::Code, + validity: t.validity(Col(0)), + codebook: Some(Codebook::new(STAGES.iter().copied()).expect("codebook")), + }, + ) + .expect("stage"); + cat.declare( + id, + FieldSpec { + name: "amount".into(), + col: Col(1), + kind: FieldKind::I32, + validity: t.validity(Col(1)), + codebook: None, + }, + ) + .expect("amount"); + // The table owns planes 0 (live) and 1 (amount validity); the list and + // the changed rows go after them. + let (prev, changed) = (Mask(2), Mask(3)); + let pred = compile( + "deals", + &[FilterGroup { + filters: vec![ + PropertyFilter::new("dealstage", Operator::Neq, "lead"), + PropertyFilter::new("amount", Operator::Gt, "5000"), + ], + }], + &cat, + ) + .expect("compile"); + + let mut r = Rng(0x0005_7011); + let mut stored: Vec = Vec::new(); + let (mut joined, mut left) = (0usize, 0usize); + for round in 0..50 { + let mut b = Batch { + owner: OWNER, + ..Batch::default() + }; + for _ in 0..r.below(30) { + b.inserts.push(vec![ + Some(Value::U32(r.below(3) as u32)), + (r.below(4) > 0).then(|| Value::I32(r.below(10_000) as i32)), + ]); + } + let mut touched = std::collections::BTreeSet::new(); + for _ in 0..r.below(25) { + if t.n_rows() == 0 { + break; + } + let row = r.below(t.n_rows() as u64) as u32; + if !t.is_live(row) || !touched.insert(row) { + continue; + } + match r.below(4) { + 0 => b.deletes.push(row), + 1 => b + .updates + .push((row, Col(0), Some(Value::U32(r.below(3) as u32)))), + _ => b.updates.push(( + row, + Col(1), + (r.below(4) > 0).then(|| Value::I32(r.below(10_000) as i32)), + )), + } + } + let applied = t.apply(&b).expect("apply"); + let n = t.n_rows(); + let prev_bits = bits(n, &stored); + let next = run( + &t, + &list::refresh(&pred, prev, changed), + [&prev_bits, &applied.changed], + ); + let full = run(&t, &pred, [&prev_bits, &applied.changed]); + assert_eq!(next, full, "round {round}"); + let before: std::collections::BTreeSet<_> = stored.iter().copied().collect(); + let after: std::collections::BTreeSet<_> = next.iter().copied().collect(); + joined += after.difference(&before).count(); + left += before.difference(&after).count(); + stored = next; + } + assert!(joined > 50 && left > 20, "joined {joined}, left {left}"); +} diff --git a/crates/hubspo-store/Cargo.toml b/crates/hubspo-store/Cargo.toml index c08b64f..d929fcd 100644 --- a/crates/hubspo-store/Cargo.toml +++ b/crates/hubspo-store/Cargo.toml @@ -13,3 +13,5 @@ workspace = true [dependencies] lance-graph-quack.workspace = true +lance-graph-mask-risc.workspace = true +lance-graph-contract.workspace = true diff --git a/crates/hubspo-store/src/lib.rs b/crates/hubspo-store/src/lib.rs index 7ce19e2..f95cd21 100644 --- a/crates/hubspo-store/src/lib.rs +++ b/crates/hubspo-store/src/lib.rs @@ -4,7 +4,10 @@ //! //! - Zone: zone 2 (no serde, no wire formats). //! - Deliverables: D-STO-1..6 (`.claude/plans/hubspo-integration-v1.md`). -//! - First module: [`catalog`], the table and field catalog every Quack read -//! binds through (H-ADR-6). Storage lands with D-STO-1. +//! - [`catalog`]: the table and field catalog every Quack read binds +//! through (H-ADR-6). +//! - [`table`] (D-STO-1, resident half): one dataset's resident lanes, changed +//! only by whole owner-stamped batches, one version each. pub mod catalog; +pub mod table; diff --git a/crates/hubspo-store/src/table.rs b/crates/hubspo-store/src/table.rs new file mode 100644 index 0000000..de0ebd2 --- /dev/null +++ b/crates/hubspo-store/src/table.rs @@ -0,0 +1,352 @@ +//! D-STO-1 (resident half): one dataset's records as resident lanes, changed +//! only by whole batches stamped with the dataset's owner. +//! +//! A [`Table`] is the in-memory image of one dataset (tenant × object type, +//! plan §5.5): a live plane, one validity plane per nullable lane, and one +//! value lane per property. Every read borrows it ([`Table::planes`]); no +//! read copies a lane. +//! +//! Writes arrive as a [`Batch`]: inserts, updates and deletes that land +//! together or not at all, and become exactly one new version. The batch +//! names the mailbox it writes for; a table accepts only batches for its own +//! owner (write on behalf, one writer per mailbox — the contract's +//! `SoaEnvelope::mailbox_owner`). Applying a batch returns a plane of the +//! rows it touched, which is what an active list's refresh needs +//! (`hubspo-query::list::refresh`). +//! +//! Ordinals are never reused: inserts append, deletes clear the live bit and +//! leave the row in place. Static lists rely on that. +//! +//! Persistence (one Lance version per published batch, P-SPARSE-1's +//! base + sparse delta) is the other half of D-STO-1. + +use lance_graph_contract::collapse_gate::MailboxId; +use lance_graph_mask_risc::{LaneRef, words_for}; +use lance_graph_quack::{Col, Mask}; + +/// What a lane holds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LaneKind { + /// Codes and foreign keys. + U32, + /// Signed integers (amounts in minor units, days, seconds). + I32, +} + +/// One lane's declaration. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LaneSpec { + /// What the lane holds. + pub kind: LaneKind, + /// Whether a record may have no value. + pub nullable: bool, +} + +/// One property value. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Value { + /// For a `U32` lane. + U32(u32), + /// For an `I32` lane. + I32(i32), +} + +/// A write that lands as one version or not at all. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct Batch { + /// The mailbox this batch writes on behalf of. + pub owner: MailboxId, + /// New records; one value (or none) per lane, in lane order. + pub inserts: Vec>>, + /// `(row, lane, value)`; `None` clears a nullable property. + pub updates: Vec<(u32, Col, Option)>, + /// Rows to delete. + pub deletes: Vec, +} + +/// Why a batch was refused. Nothing is applied when a batch is refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StoreError { + /// The batch writes for another mailbox than the table's owner. + OwnerMismatch { + /// The table's owner. + table: MailboxId, + /// The batch's owner. + batch: MailboxId, + }, + /// An insert does not give one entry per lane. + WrongArity { + /// Lanes the table has. + expected: usize, + /// Entries given. + got: usize, + }, + /// No such lane. + UnknownLane(Col), + /// A value of the wrong kind for its lane. + KindMismatch(Col), + /// No value for a lane that is not nullable. + Required(Col), + /// The row does not exist or is deleted. + RowNotLive(u32), + /// The same row is written twice in one batch (two updates of one lane, + /// or an update and a delete). Order inside a batch carries no meaning, + /// so the batch would be ambiguous. + Conflict(u32), + /// The table would exceed `u32::MAX` rows. + Full, + /// More lanes than a `Col` can address. + TooManyLanes, +} + +/// What one applied batch did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Applied { + /// The version the batch became. + pub version: u64, + /// The first ordinal it inserted (equal to the old row count). + pub first_insert: u32, + /// One bit per row it inserted, updated or deleted, over the new row + /// count. + pub changed: Vec, +} + +enum Lane { + U32(Vec), + I32(Vec), +} + +/// One dataset's records, resident. See the module docs. +pub struct Table { + owner: MailboxId, + version: u64, + n_rows: usize, + specs: Vec, + lanes: Vec, + /// Plane 0 is `live`; plane `1 + k` is the validity of the `k`-th + /// nullable lane. + planes: Vec>, + validity: Vec>, +} + +/// The live plane of every table. +pub const LIVE: Mask = Mask(0); + +fn set(words: &mut [u64], row: usize, on: bool) { + let bit = 1u64 << (row % 64); + if on { + words[row / 64] |= bit; + } else { + words[row / 64] &= !bit; + } +} + +fn get(words: &[u64], row: usize) -> bool { + words[row / 64] >> (row % 64) & 1 == 1 +} + +impl Table { + /// An empty table owned by `owner`, at version 0. + /// + /// # Errors + /// + /// [`StoreError::TooManyLanes`] past `u16::MAX` lanes, or past + /// `u16::MAX` planes. + pub fn new(owner: MailboxId, specs: &[LaneSpec]) -> Result { + if specs.len() > usize::from(u16::MAX) { + return Err(StoreError::TooManyLanes); + } + let mut validity = Vec::with_capacity(specs.len()); + let mut planes = vec![Vec::new()]; + for s in specs { + if s.nullable { + validity.push(Some(planes.len())); + planes.push(Vec::new()); + } else { + validity.push(None); + } + } + Ok(Self { + owner, + version: 0, + n_rows: 0, + specs: specs.to_vec(), + lanes: specs + .iter() + .map(|s| match s.kind { + LaneKind::U32 => Lane::U32(Vec::new()), + LaneKind::I32 => Lane::I32(Vec::new()), + }) + .collect(), + planes, + validity, + }) + } + + /// The mailbox every batch must write for. + pub fn owner(&self) -> MailboxId { + self.owner + } + + /// The last applied version (0 before the first batch). + pub fn version(&self) -> u64 { + self.version + } + + /// Rows ever inserted, deleted ones included. + pub fn n_rows(&self) -> usize { + self.n_rows + } + + /// The validity plane of a nullable lane, for the catalog + /// (`FieldSpec::validity`). + pub fn validity(&self, col: Col) -> Option { + let plane = (*self.validity.get(usize::from(col.0))?)?; + u16::try_from(plane).ok().map(Mask) + } + + /// Whether `row` is live. + pub fn is_live(&self, row: u32) -> bool { + (row as usize) < self.n_rows && get(&self.planes[0], row as usize) + } + + /// The planes and lanes for mask-risc, borrowed. + pub fn planes(&self) -> (Vec<&[u64]>, Vec>) { + let masks = self.planes.iter().map(Vec::as_slice).collect(); + let lanes = self + .lanes + .iter() + .map(|l| match l { + Lane::U32(v) => LaneRef::U32(v), + Lane::I32(v) => LaneRef::I32(v), + }) + .collect(); + (masks, lanes) + } + + fn check_value(&self, col: Col, v: Option) -> Result<(), StoreError> { + let spec = self + .specs + .get(usize::from(col.0)) + .ok_or(StoreError::UnknownLane(col))?; + match (v, spec.kind) { + (None, _) if spec.nullable => Ok(()), + (None, _) => Err(StoreError::Required(col)), + (Some(Value::U32(_)), LaneKind::U32) | (Some(Value::I32(_)), LaneKind::I32) => Ok(()), + _ => Err(StoreError::KindMismatch(col)), + } + } + + fn check(&self, batch: &Batch) -> Result<(), StoreError> { + if batch.owner != self.owner { + return Err(StoreError::OwnerMismatch { + table: self.owner, + batch: batch.owner, + }); + } + if self + .n_rows + .checked_add(batch.inserts.len()) + .is_none_or(|n| u32::try_from(n).is_err()) + { + return Err(StoreError::Full); + } + for record in &batch.inserts { + if record.len() != self.specs.len() { + return Err(StoreError::WrongArity { + expected: self.specs.len(), + got: record.len(), + }); + } + // `new` caps the lane count at `u16::MAX`, so every index fits. + for (k, v) in (0..=u16::MAX).zip(record) { + self.check_value(Col(k), *v)?; + } + } + let mut written: Vec<(u32, u16)> = Vec::with_capacity(batch.updates.len()); + for &(row, col, v) in &batch.updates { + if !self.is_live(row) { + return Err(StoreError::RowNotLive(row)); + } + self.check_value(col, v)?; + written.push((row, col.0)); + } + written.sort_unstable(); + if let Some(w) = written.windows(2).find(|w| w[0] == w[1]) { + return Err(StoreError::Conflict(w[0].0)); + } + let mut deleted = batch.deletes.clone(); + deleted.sort_unstable(); + if let Some(w) = deleted.windows(2).find(|w| w[0] == w[1]) { + return Err(StoreError::Conflict(w[0])); + } + for &row in &deleted { + if !self.is_live(row) { + return Err(StoreError::RowNotLive(row)); + } + if written.binary_search_by(|w| w.0.cmp(&row)).is_ok() { + return Err(StoreError::Conflict(row)); + } + } + Ok(()) + } + + fn write(&mut self, row: usize, k: usize, v: Option) { + if let Some(plane) = self.validity[k] { + set(&mut self.planes[plane], row, v.is_some()); + } + // A cleared value keeps its old payload: the validity plane, not the + // lane, says it is absent. + match (&mut self.lanes[k], v) { + (Lane::U32(lane), Some(Value::U32(x))) => lane[row] = x, + (Lane::I32(lane), Some(Value::I32(x))) => lane[row] = x, + _ => {} + } + } + + /// Apply `batch` as one new version. + /// + /// # Errors + /// + /// The first [`StoreError`]; the table is unchanged when a batch is + /// refused. + pub fn apply(&mut self, batch: &Batch) -> Result { + self.check(batch)?; + let first = self.n_rows; + let n = first + batch.inserts.len(); + let words = words_for(n); + for plane in &mut self.planes { + plane.resize(words, 0); + } + for lane in &mut self.lanes { + match lane { + Lane::U32(v) => v.resize(n, 0), + Lane::I32(v) => v.resize(n, 0), + } + } + self.n_rows = n; + let mut changed = vec![0u64; words]; + for (i, record) in batch.inserts.iter().enumerate() { + let row = first + i; + set(&mut self.planes[0], row, true); + for (k, v) in record.iter().enumerate() { + self.write(row, k, *v); + } + set(&mut changed, row, true); + } + for &(row, col, v) in &batch.updates { + self.write(row as usize, usize::from(col.0), v); + set(&mut changed, row as usize, true); + } + for &row in &batch.deletes { + set(&mut self.planes[0], row as usize, false); + set(&mut changed, row as usize, true); + } + self.version += 1; + Ok(Applied { + version: self.version, + first_insert: u32::try_from(first).unwrap_or(u32::MAX), + changed, + }) + } +} diff --git a/crates/hubspo-store/tests/table.rs b/crates/hubspo-store/tests/table.rs new file mode 100644 index 0000000..df90f8f --- /dev/null +++ b/crates/hubspo-store/tests/table.rs @@ -0,0 +1,302 @@ +//! D-STO-1 resident half: batches land whole or not at all, one version +//! each, for the table's owner only; the lanes and planes always equal a +//! plain model of the records. + +use hubspo_store::table::{Batch, LaneKind, LaneSpec, StoreError, Table, Value}; +use lance_graph_mask_risc::LaneRef; +use lance_graph_quack::Col; + +const OWNER: u32 = 7; +const SPECS: [LaneSpec; 3] = [ + LaneSpec { + kind: LaneKind::U32, + nullable: false, + }, + LaneSpec { + kind: LaneKind::I32, + nullable: true, + }, + LaneSpec { + kind: LaneKind::U32, + nullable: true, + }, +]; + +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 + } +} + +type Record = [Option; 3]; + +fn record(r: &mut Rng) -> Record { + [ + Some(Value::U32(r.below(5) as u32)), + (r.below(4) > 0).then(|| Value::I32(r.below(10_000) as i32 - 100)), + (r.below(3) > 0).then(|| Value::U32(r.below(1_000) as u32)), + ] +} + +/// What the table must show for `model`: per row, live and per-lane value. +fn assert_matches(t: &Table, model: &[Option], last: &[Option]) { + assert_eq!(t.n_rows(), model.len()); + let (masks, lanes) = t.planes(); + let bit = |plane: &[u64], row: usize| plane[row / 64] >> (row % 64) & 1 == 1; + for (row, rec) in model.iter().enumerate() { + assert_eq!(bit(masks[0], row), rec.is_some(), "live {row}"); + // A deleted row keeps its last values in place, inert. + let shown = rec + .as_ref() + .or(last[row].as_ref()) + .expect("row was inserted"); + for (k, want) in shown.iter().enumerate() { + let valid = match t.validity(Col(k as u16)) { + Some(m) => bit(masks[usize::from(m.0)], row), + None => true, + }; + assert_eq!(valid, want.is_some(), "validity {row}/{k}"); + if let Some(v) = want { + let got = match lanes[k] { + LaneRef::U32(l) => Value::U32(l[row]), + LaneRef::I32(l) => Value::I32(l[row]), + _ => unreachable!(), + }; + assert_eq!(got, *v, "value {row}/{k}"); + } + } + } +} + +#[test] +fn random_batches_match_a_plain_model() { + let mut r = Rng(0x0000_57E0); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let mut model: Vec> = Vec::new(); + let mut last: Vec> = Vec::new(); + let (mut updates, mut deletes) = (0, 0); + for round in 1..=60u64 { + let mut b = Batch { + owner: OWNER, + ..Batch::default() + }; + let mut touched = std::collections::BTreeSet::new(); + for _ in 0..r.below(20) { + b.inserts.push(record(&mut r).to_vec()); + } + let live: Vec = (0..model.len()).filter(|&i| model[i].is_some()).collect(); + for _ in 0..r.below(15) { + if live.is_empty() { + break; + } + let row = live[r.below(live.len() as u64) as usize]; + if !touched.insert(row) { + continue; + } + if r.below(5) == 0 { + b.deletes.push(row as u32); + } else { + let k = r.below(3) as usize; + // `record` never leaves lane 0 empty, so only the nullable + // lanes are ever cleared. + let v = record(&mut r)[k]; + b.updates.push((row as u32, Col(k as u16), v)); + } + } + let first = model.len(); + let applied = t.apply(&b).expect("apply"); + assert_eq!(applied.version, round); + assert_eq!(applied.first_insert as usize, first); + for rec in &b.inserts { + let rec: Record = [rec[0], rec[1], rec[2]]; + model.push(Some(rec)); + last.push(Some(rec)); + } + for &(row, col, v) in &b.updates { + let rec = model[row as usize].as_mut().expect("live"); + rec[usize::from(col.0)] = v; + last[row as usize] = Some(*rec); + updates += 1; + } + for &row in &b.deletes { + model[row as usize] = None; + deletes += 1; + } + let want_changed: Vec = (first..model.len()) + .chain(b.updates.iter().map(|u| u.0 as usize)) + .chain(b.deletes.iter().map(|&d| d as usize)) + .collect::>() + .into_iter() + .collect(); + assert_eq!( + lance_graph_mask_risc::materialize_rows(&applied.changed, model.len()), + want_changed, + "changed rows, round {round}" + ); + assert_matches(&t, &model, &last); + } + assert!( + updates > 200 && deletes > 30, + "{updates} updates, {deletes} deletes" + ); +} + +#[test] +fn refused_batches_change_nothing() { + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let mut r = Rng(0x00BA_D0B0); + let seed = Batch { + owner: OWNER, + inserts: (0..10).map(|_| record(&mut r).to_vec()).collect(), + ..Batch::default() + }; + t.apply(&seed).expect("seed"); + t.apply(&Batch { + owner: OWNER, + deletes: vec![9], + ..Batch::default() + }) + .expect("delete 9"); + let model: Vec> = seed + .inserts + .iter() + .enumerate() + .map(|(i, rec)| (i != 9).then(|| [rec[0], rec[1], rec[2]])) + .collect(); + let last: Vec> = seed + .inserts + .iter() + .map(|rec| Some([rec[0], rec[1], rec[2]])) + .collect(); + let good = record(&mut r).to_vec(); + let base = |b: Batch| Batch { owner: OWNER, ..b }; + let cases: Vec<(Batch, StoreError)> = vec![ + ( + Batch { + owner: OWNER + 1, + inserts: vec![good.clone()], + ..Batch::default() + }, + StoreError::OwnerMismatch { + table: OWNER, + batch: OWNER + 1, + }, + ), + ( + base(Batch { + inserts: vec![good.clone(), good[..2].to_vec()], + ..Batch::default() + }), + StoreError::WrongArity { + expected: 3, + got: 2, + }, + ), + ( + base(Batch { + inserts: vec![vec![Some(Value::I32(1)), None, None]], + ..Batch::default() + }), + StoreError::KindMismatch(Col(0)), + ), + ( + base(Batch { + inserts: vec![vec![None, None, None]], + ..Batch::default() + }), + StoreError::Required(Col(0)), + ), + ( + base(Batch { + updates: vec![(0, Col(0), None)], + ..Batch::default() + }), + StoreError::Required(Col(0)), + ), + ( + base(Batch { + updates: vec![(0, Col(3), Some(Value::U32(1)))], + ..Batch::default() + }), + StoreError::UnknownLane(Col(3)), + ), + ( + base(Batch { + updates: vec![(9, Col(1), Some(Value::I32(1)))], + ..Batch::default() + }), + StoreError::RowNotLive(9), + ), + ( + base(Batch { + updates: vec![(40, Col(1), Some(Value::I32(1)))], + ..Batch::default() + }), + StoreError::RowNotLive(40), + ), + ( + base(Batch { + deletes: vec![9], + ..Batch::default() + }), + StoreError::RowNotLive(9), + ), + ( + base(Batch { + updates: vec![ + (2, Col(1), Some(Value::I32(1))), + (2, Col(1), Some(Value::I32(2))), + ], + ..Batch::default() + }), + StoreError::Conflict(2), + ), + ( + base(Batch { + updates: vec![(3, Col(1), Some(Value::I32(1)))], + deletes: vec![3], + ..Batch::default() + }), + StoreError::Conflict(3), + ), + ( + base(Batch { + deletes: vec![4, 4], + ..Batch::default() + }), + StoreError::Conflict(4), + ), + ]; + for (batch, err) in cases { + // Every refused batch also carries a valid insert, so a partial + // apply would show up as a new row. + let mut batch = batch; + if err + != (StoreError::WrongArity { + expected: 3, + got: 2, + }) + { + batch.inserts.push(good.clone()); + } + assert_eq!(t.apply(&batch), Err(err)); + assert_eq!(t.version(), 2, "{err:?} moved the version"); + assert_matches(&t, &model, &last); + } + // Two updates of different lanes of one row are fine. + assert!( + t.apply(&base(Batch { + updates: vec![(2, Col(1), Some(Value::I32(1))), (2, Col(2), None),], + ..Batch::default() + })) + .is_ok() + ); +} From 0eb99fcab8e1bfeacff99ef1c09e6b34fd2ad069 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:41:57 +0000 Subject: [PATCH 2/3] Board: D-STO-1 resident half Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- .claude/board/LATEST_STATE.md | 28 ++++++++++++++++++++++++++++ .claude/board/STATUS_BOARD.md | 2 +- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index af7cb78..4798dc7 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -85,6 +85,34 @@ - Quack became consumable by git with lance-graph #1457 (`ISS-LG-MASKRISC-PATH-NDARRAY`, resolved). +## Store, resident half (D-STO-1) + +- `hubspo-store::table::Table`: one dataset (tenant × object type) as + resident lanes (`U32` / `I32`), a live plane (plane 0) and one validity + plane per nullable lane. `planes()` borrows them for mask-risc; no read + copies a lane. +- Writes are `Batch`es (inserts, updates, deletes), validated whole and + applied as exactly one version or not at all, and only for the table's + owner (`MailboxId`, write on behalf). Refused: wrong owner, wrong arity, + wrong kind, missing required value, unknown lane, a deleted or absent + row, and a row written twice in one batch (update + update of one lane, + update + delete, delete + delete; order inside a batch means nothing). +- `apply()` returns `Applied { version, first_insert, changed }`; `changed` + is the plane `list::refresh` takes. Ordinals are never reused: inserts + append, deletes clear the live bit and leave the row inert. +- Tests: 60 random batches match a plain model (every lane and plane, 200+ + updates, 30+ deletes); 12 refusals leave version, lanes and planes + untouched (each refused batch also carries a valid insert, so a partial + apply would show); `hubspo-query/tests/store_list.rs` runs catalog → + filter → `list::refresh` on the store's own `changed` plane over 50 + batches and equals a full recomputation. + Disable runs 2026-10-10, all red: owner unchecked, deletes missing from + `changed` (red in the store and in the list test), delete keeps `live`, + a cleared value keeps validity, version frozen, conflict check off, + updates to deleted rows allowed, lane kind unchecked. +- Not yet: persistence (one Lance version per published batch, P-SPARSE-1 + base + delta) — the next half of D-STO-1. + ## Board view (D-QRY-3, board half) - `hubspo-query::board::Board`: per (pipeline, stage), record count and diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index e5125c8..857eca5 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -35,7 +35,7 @@ | D-id | Deliverable | Size | Status | Evidence | |---|---|---|---|---| -| D-STO-1 | `hubspo-store` write lane: batch image → one Lance version per publish; owner-stamped (`mailbox_owner`) | L | Queued | | +| D-STO-1 | `hubspo-store` write lane: batch image → one Lance version per publish; owner-stamped (`mailbox_owner`) | L | In progress | resident half in PR: `hubspo-store::table` (owner-stamped atomic batches, one version each, `changed` plane); Lance persistence next | | D-STO-2 | KV value store for variable-length properties; zero-copy lens reads | M | Queued | | | D-STO-3 | Edge lanes + association labels (H-ADR-4); cardinality invariants as Quack `GroupReduce` | L | Queued | | | D-STO-4 | Tenant layout + P-TEN-1 probe | M | Queued | P-TEN-1 passed 2026-10-10 (7.8 ms p99 at 10k tenants, local disk) | From c3d67740a08d0ff2cc7d706f2d0c7f394a39825d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:42:14 +0000 Subject: [PATCH 3/3] Board: D-QRY-3 board half shipped (#21) 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 857eca5..f049456 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 | 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-3 | Saved views + board view (report pivot over pipeline × stage) | M | In progress | board half shipped (#21): `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 | |