diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index 4798dc7..42aed4c 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -113,6 +113,32 @@ - Not yet: persistence (one Lance version per published batch, P-SPARSE-1 base + delta) — the next half of D-STO-1. +## Store, persistence half (D-STO-1, feature `lance`) + +- `hubspo-store::log::Log`: every batch is one Lance version of an + append-only log (P-SPARSE-1's delta shape; nothing updated in place). + Records: `BORN` (row), `SET` (row, lane, present, value as the lane's 32 + bits), `DELETE` (row), each stamped with version and owner. +- `commit(table, batch)`: check → append one Lance version → apply. Lance + version = table version; the log's version is checked before the append + (and after it). A refused batch writes nothing; an empty batch is + refused (`StoreError::EmptyBatch`). +- `replay(owner, specs)`: rebuilds the table batch by batch through the + same `Table::apply`; refuses a foreign owner, an undecodable record, and + a log whose Lance version is not its batch count. +- Tests (`cargo test -p hubspo-store --features lance`, CI job + `store-lance`, which installs `protoc`): 25 committed batches replay to + an identical image (deletes and negative values included); one Lance + version per batch; a second, stale writer is refused before it applies + anything; refused and empty batches leave the log at its version; + replay under another owner is refused. + Disable runs 2026-10-10, red: no pre-append version check, owner ignored + on replay, the I32 sign bit lost, deletes not logged, apply before the + append. Green, and kept on purpose: the post-append version check — it + catches only a writer that appends between the check and the write, a + race no sequential test can schedule. +- Not yet: folding the log into a base dataset (D-STO-5), object stores. + ## 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 f049456..79c8575 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 | In progress | resident half in PR: `hubspo-store::table` (owner-stamped atomic batches, one version each, `changed` plane); Lance persistence next | +| D-STO-1 | `hubspo-store` write lane: batch image → one Lance version per publish; owner-stamped (`mailbox_owner`) | L | In PR | resident half shipped (#22); persistence half in PR: `hubspo-store::log` (one Lance version per batch, replay, feature `lance`) | | 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) | diff --git a/.claude/plans/hubspo-integration-v1.md b/.claude/plans/hubspo-integration-v1.md index 9bf4f91..c81c369 100644 --- a/.claude/plans/hubspo-integration-v1.md +++ b/.claude/plans/hubspo-integration-v1.md @@ -1167,7 +1167,7 @@ below. Cross-repo follow-ups are tracked in `.claude/board/ISSUES.md`. | OD-9 | Edition 2024 for HubSPO-rs | yes | | OD-10 | Which HubSpot sandbox portal is the value oracle, and who holds its token | operator | | OD-11 | lance-graph `ogar-consumer-preflight.md` still whitelists `lance =9.0.0 / lancedb =0.33.0`; lance-graph `CLAUDE.md` floats `12.*` / `0.39.*` | follow `CLAUDE.md`; fix the preflight upstream | -| OD-12 | How stalwart gets users and teams (`ISS-STALWART-DIRSIM-WIRING`): (a) build the dir-sim backend outside stalwart's config model, (b) dir-sim persistence + stalwart's schema generator first, (c) stalwart keeps its own directory and the CRM join key comes from spear's placement | (c) — approved 2026-10-10 The fork's `crates/directory/src/backend/dirsim/` exists behind feature `dir-sim` but nothing constructs it and the registry `Directory` enum has no variant for it (re-read 2026-10-10, `.claude/knowledge/comms-api-surface.md` §3). | +| OD-12 | How stalwart gets users and teams (`ISS-STALWART-DIRSIM-WIRING`): (a) build the dir-sim backend outside stalwart's config model, (b) dir-sim persistence + stalwart's schema generator first, (c) stalwart keeps its own directory and the CRM join key comes from spear's placement | (c) — approved 2026-10-10 The fork's `crates/directory/src/backend/dirsim/` exists behind feature `dir-sim` but nothing constructs it and the registry `Directory` enum has no variant for it (re-read 2026-10-10, `.claude/knowledge/comms-api-surface.md` §3). *Option to evaluate, added 2026-10-10:* (d) stalwart's own LDAP directory backend reading dir-sim through lance-graph #1453's read-only LDAP v3 handler (`lance-graph-dir-sim::ldap::Server`; the host owns the socket, so `hubspo-server` would host it). No stalwart code change. Open before it can replace (c): password checks go through the host's `Authority`, not dir-sim, and the served version is in memory until dir-sim persistence lands. | --- diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2c4254d..b5a8cdc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,3 +39,23 @@ jobs: # identity shapes (Gate G0). - name: Test run: cargo test --workspace + + # hubspo-store's Lance log (feature `lance`) needs `protoc` to build + # lance-encoding, so it runs in its own job and the default job stays fast. + store-lance: + name: hubspo-store (lance) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Install pinned toolchain + run: rustup show active-toolchain || rustup toolchain install + + - name: Install protoc + run: sudo apt-get update && sudo apt-get install -y protobuf-compiler + + - name: Clippy + run: cargo clippy -p hubspo-store --features lance --all-targets --no-deps -- -D warnings + + - name: Test + run: cargo test -p hubspo-store --features lance diff --git a/crates/hubspo-store/Cargo.toml b/crates/hubspo-store/Cargo.toml index d929fcd..53598b6 100644 --- a/crates/hubspo-store/Cargo.toml +++ b/crates/hubspo-store/Cargo.toml @@ -15,3 +15,17 @@ workspace = true lance-graph-quack.workspace = true lance-graph-mask-risc.workspace = true lance-graph-contract.workspace = true + +# One Lance version per published batch (D-STO-1, persistence half). Off by +# default: lance needs `protoc` to build, and only the `store-lance` CI job +# installs it. +lance = { workspace = true, optional = true } +arrow = { workspace = true, optional = true } +futures = { version = "0.3", optional = true } + +[features] +lance = ["dep:lance", "dep:arrow", "dep:futures"] + +[dev-dependencies] +tokio = { version = "1", features = ["rt-multi-thread", "macros"] } +tempfile = "3" diff --git a/crates/hubspo-store/src/lib.rs b/crates/hubspo-store/src/lib.rs index f95cd21..acb62de 100644 --- a/crates/hubspo-store/src/lib.rs +++ b/crates/hubspo-store/src/lib.rs @@ -8,6 +8,10 @@ //! 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. +//! - [`log`] (D-STO-1, persistence half, feature `lance`): every batch is one +//! Lance version of an append-only log; a table is rebuilt by replay. pub mod catalog; +#[cfg(feature = "lance")] +pub mod log; pub mod table; diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs new file mode 100644 index 0000000..631b85d --- /dev/null +++ b/crates/hubspo-store/src/log.rs @@ -0,0 +1,377 @@ +//! D-STO-1 (persistence half): every batch is one Lance version of an +//! append-only log, and a table is rebuilt by replaying it. +//! +//! The shape is P-SPARSE-1's delta: nothing is ever updated in place, so +//! there are no deletion vectors to read and the log is safe to compact. +//! Each record is one fact of one batch: +//! +//! | `op` | meaning | `row` | `lane` | `present` / `value` | +//! |---|---|---|---|---| +//! | 0 `BORN` | a record is inserted | its ordinal | — | — | +//! | 1 `SET` | a property is written | the record | the lane | the value, or absent | +//! | 2 `DELETE` | a record is deleted | the record | — | — | +//! +//! Every record carries the batch's `version` and `owner`. Values are the +//! lane's 32 bits (`I32` stored as its bit pattern). +//! +//! **One Lance version per batch, and equal numbers.** The first batch +//! creates the dataset (Lance version 1, table version 1) and every later +//! one appends exactly one Lance version. [`Log::commit`] checks the log's +//! latest version before and after the append; anything else means a second +//! writer, which breaks one-writer-per-mailbox and is reported as +//! [`LogError::Diverged`], never accepted. Lance appends do not conflict +//! with each other, so this check is what enforces the rule. +//! +//! **Durable first.** `commit` validates the batch, appends it, and only +//! then applies it to the table. If the append fails the table is +//! unchanged. +//! +//! Not yet: folding the log into a base dataset (D-STO-5, history), and +//! object-store URIs beyond what `lance` resolves by itself. + +use std::sync::Arc; + +use arrow::array::{ + Array, BooleanArray, RecordBatch, RecordBatchIterator, UInt8Array, UInt16Array, UInt32Array, + UInt64Array, +}; +use arrow::datatypes::{DataType, Field, Schema}; +use futures::TryStreamExt; +use lance::Dataset; +use lance::dataset::{WriteMode, WriteParams}; +use lance_graph_contract::collapse_gate::MailboxId; +use lance_graph_quack::Col; + +use crate::table::{Applied, Batch, LaneKind, LaneSpec, StoreError, Table, Value}; + +const BORN: u8 = 0; +const SET: u8 = 1; +const DELETE: u8 = 2; + +/// Why a log operation failed. +#[derive(Debug)] +pub enum LogError { + /// The batch was refused by the table; nothing was written. + Store(StoreError), + /// Lance or Arrow failed. + Lance(String), + /// The log's version does not match the table's: another writer + /// appended, or the log is not this table's. The table must be + /// replayed. + Diverged { + /// Version the log should be at. + expected: u64, + /// Version it is at. + found: u64, + }, + /// A record names another owner than the one replaying. + Owner { + /// Owner asked for. + expected: MailboxId, + /// Owner found. + found: MailboxId, + }, + /// A record does not decode against the lane declarations. + Corrupt(&'static str), +} + +impl From for LogError { + fn from(e: StoreError) -> Self { + LogError::Store(e) + } +} + +fn lance_err(e: impl std::fmt::Display) -> LogError { + LogError::Lance(e.to_string()) +} + +fn schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("version", DataType::UInt64, false), + Field::new("owner", DataType::UInt32, false), + Field::new("op", DataType::UInt8, false), + Field::new("row", DataType::UInt32, false), + Field::new("lane", DataType::UInt16, false), + Field::new("present", DataType::Boolean, false), + Field::new("value", DataType::UInt32, false), + ])) +} + +#[derive(Default)] +struct Columns { + version: Vec, + owner: Vec, + op: Vec, + row: Vec, + lane: Vec, + present: Vec, + value: Vec, +} + +impl Columns { + fn push(&mut self, version: u64, owner: u32, op: u8, row: u32, lane: u16, v: Option) { + self.version.push(version); + self.owner.push(owner); + self.op.push(op); + self.row.push(row); + self.lane.push(lane); + self.present.push(v.is_some()); + self.value.push(match v { + Some(Value::U32(x)) => x, + Some(Value::I32(x)) => x.cast_unsigned(), + None => 0, + }); + } + + fn batch(self) -> Result { + RecordBatch::try_new( + schema(), + vec![ + Arc::new(UInt64Array::from(self.version)), + Arc::new(UInt32Array::from(self.owner)), + Arc::new(UInt8Array::from(self.op)), + Arc::new(UInt32Array::from(self.row)), + Arc::new(UInt16Array::from(self.lane)), + Arc::new(BooleanArray::from(self.present)), + Arc::new(UInt32Array::from(self.value)), + ], + ) + .map_err(lance_err) + } +} + +/// The records of `batch` as version `version`, inserting from ordinal +/// `first`. +fn encode(batch: &Batch, version: u64, first: u32) -> Columns { + let mut c = Columns::default(); + let o = batch.owner; + for (i, record) in (first..).zip(&batch.inserts) { + c.push(version, o, BORN, i, 0, None); + for (lane, v) in (0..=u16::MAX).zip(record) { + c.push(version, o, SET, i, lane, *v); + } + } + for &(row, lane, v) in &batch.updates { + c.push(version, o, SET, row, lane.0, v); + } + for &row in &batch.deletes { + c.push(version, o, DELETE, row, 0, None); + } + c +} + +/// An append-only Lance log at `uri`. +#[derive(Debug, Clone)] +pub struct Log { + uri: String, +} + +impl Log { + /// The log at `uri` (created by the first commit). + pub fn new(uri: impl Into) -> Self { + Self { uri: uri.into() } + } + + async fn latest(&self) -> Result { + match Dataset::open(&self.uri).await { + Ok(ds) => Ok(ds.version().version), + // No dataset yet is version 0, like an empty table. + Err(lance::Error::DatasetNotFound { .. }) => Ok(0), + Err(e) => Err(lance_err(e)), + } + } + + /// Validate `batch`, append it as one Lance version, then apply it. + /// + /// # Errors + /// + /// - [`LogError::Store`]: the batch was refused; nothing was written. + /// - [`LogError::Diverged`]: the log was not at the table's version + /// before the append, or not one past it after; nothing was applied. + /// - [`LogError::Lance`]: the append failed; nothing was applied. + pub async fn commit(&self, table: &mut Table, batch: &Batch) -> Result { + table.check(batch)?; + let before = table.version(); + let found = self.latest().await?; + if found != before { + return Err(LogError::Diverged { + expected: before, + found, + }); + } + let first = u32::try_from(table.n_rows()).map_err(|_| StoreError::Full)?; + let records = encode(batch, before + 1, first).batch()?; + let params = WriteParams { + mode: if before == 0 { + WriteMode::Create + } else { + WriteMode::Append + }, + ..WriteParams::default() + }; + let ds = Dataset::write( + RecordBatchIterator::new(vec![Ok(records)], schema()), + &self.uri, + Some(params), + ) + .await + .map_err(lance_err)?; + let after = ds.version().version; + // Reached only when another writer appends between the check above + // and this write. No test can schedule that race; a disable run + // stays green (the sequential case is caught above). Kept because + // Lance appends do not conflict, so nothing else would notice. + if after != before + 1 { + return Err(LogError::Diverged { + expected: before + 1, + found: after, + }); + } + Ok(table.apply(batch)?) + } + + /// Rebuild the table of `owner` with `specs` from the log, batch by + /// batch, through the same [`Table::apply`] a live writer uses. + /// + /// # Errors + /// + /// [`LogError::Owner`] for a record of another owner, + /// [`LogError::Corrupt`] for a record that does not decode, + /// [`LogError::Store`] if a replayed batch is refused, and + /// [`LogError::Diverged`] if the log's Lance version is not the number + /// of batches in it. + pub async fn replay(&self, owner: MailboxId, specs: &[LaneSpec]) -> Result { + let mut table = Table::new(owner, specs)?; + let latest = self.latest().await?; + if latest == 0 { + return Ok(table); + } + let ds = Dataset::open(&self.uri).await.map_err(lance_err)?; + let batches: Vec = ds + .scan() + .try_into_stream() + .await + .map_err(lance_err)? + .try_collect() + .await + .map_err(lance_err)?; + let mut facts: Vec<(u64, u32, u8, u32, u16, bool, u32)> = Vec::new(); + for b in &batches { + let version = column::(b, 0, "version")?; + let own = column::(b, 1, "owner")?; + let op = column::(b, 2, "op")?; + let row = column::(b, 3, "row")?; + let lane = column::(b, 4, "lane")?; + let present = column::(b, 5, "present")?; + let value = column::(b, 6, "value")?; + for i in 0..b.num_rows() { + facts.push(( + version.value(i), + own.value(i), + op.value(i), + row.value(i), + lane.value(i), + present.value(i), + value.value(i), + )); + } + } + // Stable sort by version keeps each batch's facts in written order. + facts.sort_by_key(|f| f.0); + let mut i = 0; + while i < facts.len() { + let version = facts[i].0; + let end = facts[i..] + .iter() + .position(|f| f.0 != version) + .map_or(facts.len(), |p| i + p); + let batch = decode(&facts[i..end], owner, &table)?; + let applied = table.apply(&batch)?; + if applied.version != version { + return Err(LogError::Diverged { + expected: applied.version, + found: version, + }); + } + i = end; + } + if table.version() != latest { + return Err(LogError::Diverged { + expected: latest, + found: table.version(), + }); + } + Ok(table) + } +} + +/// Column `i` of a log batch as `T`, with no nulls. +fn column<'a, T: Array + 'static>( + b: &'a RecordBatch, + i: usize, + name: &'static str, +) -> Result<&'a T, LogError> { + let a = b.column(i); + if a.null_count() > 0 { + return Err(LogError::Corrupt(name)); + } + a.as_any() + .downcast_ref::() + .ok_or(LogError::Corrupt(name)) +} + +fn decode( + facts: &[(u64, u32, u8, u32, u16, bool, u32)], + owner: MailboxId, + table: &Table, +) -> Result { + let specs = table.specs(); + let first = u32::try_from(table.n_rows()).map_err(|_| StoreError::Full)?; + let mut batch = Batch { + owner, + ..Batch::default() + }; + for &(_, own, op, row, lane, present, value) in facts { + if own != owner { + return Err(LogError::Owner { + expected: owner, + found: own, + }); + } + let v = |lane: u16| -> Result, LogError> { + let spec = specs + .get(usize::from(lane)) + .ok_or(LogError::Corrupt("lane"))?; + Ok(present.then_some(match spec.kind { + LaneKind::U32 => Value::U32(value), + LaneKind::I32 => Value::I32(value.cast_signed()), + })) + }; + match op { + BORN => { + // Ordinals are born in order: the next one, every time. + let next = + first + u32::try_from(batch.inserts.len()).map_err(|_| StoreError::Full)?; + if row != next { + return Err(LogError::Corrupt("born out of order")); + } + batch.inserts.push(vec![None; specs.len()]); + } + SET if row >= first => { + let k = usize::try_from(row - first).map_err(|_| LogError::Corrupt("row"))?; + let record = batch + .inserts + .get_mut(k) + .ok_or(LogError::Corrupt("set before born"))?; + let slot = record + .get_mut(usize::from(lane)) + .ok_or(LogError::Corrupt("lane"))?; + *slot = v(lane)?; + } + SET => batch.updates.push((row, Col(lane), v(lane)?)), + DELETE => batch.deletes.push(row), + _ => return Err(LogError::Corrupt("op")), + } + } + Ok(batch) +} diff --git a/crates/hubspo-store/src/table.rs b/crates/hubspo-store/src/table.rs index de0ebd2..1fa9046 100644 --- a/crates/hubspo-store/src/table.rs +++ b/crates/hubspo-store/src/table.rs @@ -97,6 +97,9 @@ pub enum StoreError { Full, /// More lanes than a `Col` can address. TooManyLanes, + /// A batch with no inserts, updates or deletes: it would be a version + /// that changes nothing. + EmptyBatch, } /// What one applied batch did. @@ -205,6 +208,11 @@ impl Table { u16::try_from(plane).ok().map(Mask) } + /// The lane declarations, in lane order. + pub fn specs(&self) -> &[LaneSpec] { + &self.specs + } + /// 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) @@ -237,13 +245,23 @@ impl Table { } } - fn check(&self, batch: &Batch) -> Result<(), StoreError> { + /// Whether [`Table::apply`] would accept `batch`, without changing + /// anything. A writer that persists first checks, then persists, then + /// applies (`log::Log::commit`). + /// + /// # Errors + /// + /// The [`StoreError`] `apply` would return. + pub fn check(&self, batch: &Batch) -> Result<(), StoreError> { if batch.owner != self.owner { return Err(StoreError::OwnerMismatch { table: self.owner, batch: batch.owner, }); } + if batch.inserts.is_empty() && batch.updates.is_empty() && batch.deletes.is_empty() { + return Err(StoreError::EmptyBatch); + } if self .n_rows .checked_add(batch.inserts.len()) diff --git a/crates/hubspo-store/tests/log.rs b/crates/hubspo-store/tests/log.rs new file mode 100644 index 0000000..413b73f --- /dev/null +++ b/crates/hubspo-store/tests/log.rs @@ -0,0 +1,207 @@ +//! D-STO-1 persistence half: a table committed batch by batch to the Lance +//! log is rebuilt exactly by replay; the log is one Lance version per batch; +//! a second writer, a foreign owner and a refused batch are all caught. +#![cfg(feature = "lance")] + +use hubspo_store::log::{Log, LogError}; +use hubspo_store::table::{Batch, LaneKind, LaneSpec, StoreError, Table, Value}; +use lance_graph_mask_risc::LaneRef; +use lance_graph_quack::Col; + +const OWNER: u32 = 11; +const SPECS: [LaneSpec; 2] = [ + LaneSpec { + kind: LaneKind::U32, + nullable: false, + }, + LaneSpec { + kind: LaneKind::I32, + 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 + } +} + +/// Everything a reader can see of a table, for equality. +fn image(t: &Table) -> (u64, usize, Vec>, Vec, Vec) { + let (masks, lanes) = t.planes(); + let u = match lanes[0] { + LaneRef::U32(v) => v.to_vec(), + _ => unreachable!(), + }; + let i = match lanes[1] { + LaneRef::I32(v) => v.to_vec(), + _ => unreachable!(), + }; + ( + t.version(), + t.n_rows(), + masks.iter().map(|m| m.to_vec()).collect(), + u, + i, + ) +} + +fn random_batch(r: &mut Rng, t: &Table) -> Batch { + let mut b = Batch { + owner: OWNER, + ..Batch::default() + }; + for _ in 0..1 + r.below(10) { + b.inserts.push(vec![ + Some(Value::U32(r.below(7) as u32)), + (r.below(3) > 0).then(|| Value::I32(r.below(2_000) as i32 - 1_000)), + ]); + } + let mut touched = std::collections::BTreeSet::new(); + for _ in 0..r.below(8) { + 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(3) { + 0 => b.deletes.push(row), + 1 => b + .updates + .push((row, Col(0), Some(Value::U32(r.below(7) as u32)))), + // Negative values exercise the I32 bit pattern in the log. + _ => b.updates.push(( + row, + Col(1), + (r.below(3) > 0).then(|| Value::I32(r.below(2_000) as i32 - 1_000)), + )), + } + } + b +} + +#[tokio::test(flavor = "multi_thread")] +async fn replay_rebuilds_the_committed_table_exactly() { + let dir = tempfile::tempdir().expect("tempdir"); + let log = Log::new(dir.path().join("deals.lance").to_str().expect("utf8")); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let mut r = Rng(0x0000_10C1); + for round in 1..=25u64 { + let b = random_batch(&mut r, &t); + let applied = log.commit(&mut t, &b).await.expect("commit"); + assert_eq!(applied.version, round); + } + let lance_version = lance::Dataset::open(dir.path().join("deals.lance").to_str().unwrap()) + .await + .expect("open") + .version() + .version; + assert_eq!(lance_version, 25, "one Lance version per batch"); + let back = log.replay(OWNER, &SPECS).await.expect("replay"); + assert_eq!(image(&back), image(&t)); + // Anti-vacuity: the run deleted rows and stored negative values. + let (_, _, masks, _, amounts) = image(&t); + assert!(masks[0].iter().map(|w| w.count_zeros()).sum::() > 0); + assert!(amounts.iter().any(|a| *a < 0)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_empty_log_replays_to_an_empty_table() { + let dir = tempfile::tempdir().expect("tempdir"); + let log = Log::new(dir.path().join("none.lance").to_str().expect("utf8")); + let t = log.replay(OWNER, &SPECS).await.expect("replay"); + assert_eq!((t.version(), t.n_rows()), (0, 0)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_second_writer_is_caught_before_it_applies() { + let dir = tempfile::tempdir().expect("tempdir"); + let uri = dir.path().join("deals.lance"); + let log = Log::new(uri.to_str().expect("utf8")); + let mut r = Rng(0x0000_2A17); + let mut a = Table::new(OWNER, &SPECS).expect("a"); + let first = random_batch(&mut r, &a); + log.commit(&mut a, &first).await.expect("a commits"); + // `b` is a second in-memory table for the same log: it replays, then a + // commits again, so b's view is one version behind. + let mut b = log.replay(OWNER, &SPECS).await.expect("b replays"); + let second = random_batch(&mut r, &a); + log.commit(&mut a, &second).await.expect("a commits again"); + let before = image(&b); + let third = random_batch(&mut r, &b); + match log.commit(&mut b, &third).await { + Err(LogError::Diverged { + expected: 1, + found: 2, + }) => {} + other => panic!("expected Diverged {{1, 2}}, got {other:?}"), + } + assert_eq!(image(&b), before, "a diverged commit must not apply"); + let back = log.replay(OWNER, &SPECS).await.expect("replay"); + assert_eq!(image(&back), image(&a), "the log holds a's batches only"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_refused_batch_writes_nothing() { + let dir = tempfile::tempdir().expect("tempdir"); + let uri = dir.path().join("deals.lance"); + let log = Log::new(uri.to_str().expect("utf8")); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let mut r = Rng(0x0000_3E55); + let ok = random_batch(&mut r, &t); + log.commit(&mut t, &ok).await.expect("commit"); + let bad = Batch { + owner: OWNER, + inserts: vec![vec![None, None]], + ..Batch::default() + }; + match log.commit(&mut t, &bad).await { + Err(LogError::Store(StoreError::Required(Col(0)))) => {} + other => panic!("expected Required(0), got {other:?}"), + } + match log + .commit( + &mut t, + &Batch { + owner: OWNER, + ..Batch::default() + }, + ) + .await + { + Err(LogError::Store(StoreError::EmptyBatch)) => {} + other => panic!("expected EmptyBatch, got {other:?}"), + } + let v = lance::Dataset::open(uri.to_str().unwrap()) + .await + .expect("open") + .version() + .version; + assert_eq!(v, 1, "refused batches must not reach the log"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn replay_refuses_a_foreign_owner() { + let dir = tempfile::tempdir().expect("tempdir"); + let log = Log::new(dir.path().join("deals.lance").to_str().expect("utf8")); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let mut r = Rng(0x0000_4F0E); + let b = random_batch(&mut r, &t); + log.commit(&mut t, &b).await.expect("commit"); + match log.replay(OWNER + 1, &SPECS).await { + Err(LogError::Owner { expected, found }) => { + assert_eq!((expected, found), (OWNER + 1, OWNER)); + } + other => panic!("expected Owner, got {:?}", other.map(|t| t.version())), + } +} diff --git a/crates/hubspo-store/tests/table.rs b/crates/hubspo-store/tests/table.rs index df90f8f..153a61c 100644 --- a/crates/hubspo-store/tests/table.rs +++ b/crates/hubspo-store/tests/table.rs @@ -291,6 +291,12 @@ fn refused_batches_change_nothing() { assert_eq!(t.version(), 2, "{err:?} moved the version"); assert_matches(&t, &model, &last); } + // A batch that changes nothing is not a version. + assert_eq!( + t.apply(&base(Batch::default())), + Err(StoreError::EmptyBatch) + ); + assert_eq!(t.version(), 2); // Two updates of different lanes of one row are fine. assert!( t.apply(&base(Batch {