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..f049456 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) | @@ -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 | | 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() + ); +}