From 3ca92a2943f9c5bb4d6b20843fef5f7c2c9b9e9f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 17:15:58 +0000 Subject: [PATCH 1/6] D-STO-5: property history read from the Lance log Log::history(specs, row, col) returns every write of one property on one record, oldest first, with version and owner: the inserted value, each update, and the record's deletion. It runs one Quack program per log batch over the log's own Arrow buffers and decodes only the matches. The log's op and lane columns widen to u32 so the whole predicate is one mask-risc program over three U32 lanes. The log format shipped in #23 with no data written yet, so no migration. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-store/src/log.rs | 198 +++++++++++++++++++++++++------ crates/hubspo-store/tests/log.rs | 86 +++++++++++++- 2 files changed, 250 insertions(+), 34 deletions(-) diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs index 631b85d..b35d2b2 100644 --- a/crates/hubspo-store/src/log.rs +++ b/crates/hubspo-store/src/log.rs @@ -26,27 +26,37 @@ //! 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. +//! **History (D-STO-5)** is the log itself: [`Log::history`] reads one +//! property of one record back as the list of writes that touched it. +//! +//! Not yet: an index on `row`, compaction (which would add Lance versions +//! and so needs the table version taken from the `version` column rather +//! than from the Lance version), 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, + Array, BooleanArray, RecordBatch, RecordBatchIterator, 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 lance_graph_mask_risc::{ + Foreign, LaneRef, Out, Planes, Scratch, execute_into, materialize_rows, words_for, +}; +use lance_graph_quack::{Agg, Cmp, Col, Filter, Query, lower}; use crate::table::{Applied, Batch, LaneKind, LaneSpec, StoreError, Table, Value}; -const BORN: u8 = 0; -const SET: u8 = 1; -const DELETE: u8 = 2; +const BORN: u32 = 0; +const SET: u32 = 1; +const DELETE: u32 = 2; + +/// One decoded log record: `(version, owner, op, row, lane, present, value)`. +type Fact = (u64, u32, u32, u32, u32, bool, u32); /// Why a log operation failed. #[derive(Debug)] @@ -89,9 +99,9 @@ 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("op", DataType::UInt32, false), Field::new("row", DataType::UInt32, false), - Field::new("lane", DataType::UInt16, false), + Field::new("lane", DataType::UInt32, false), Field::new("present", DataType::Boolean, false), Field::new("value", DataType::UInt32, false), ])) @@ -101,20 +111,20 @@ fn schema() -> Arc { struct Columns { version: Vec, owner: Vec, - op: Vec, + op: Vec, row: Vec, - lane: 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) { + fn push(&mut self, version: u64, owner: u32, op: u32, 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.lane.push(u32::from(lane)); self.present.push(v.is_some()); self.value.push(match v { Some(Value::U32(x)) => x, @@ -129,9 +139,9 @@ impl Columns { vec![ Arc::new(UInt64Array::from(self.version)), Arc::new(UInt32Array::from(self.owner)), - Arc::new(UInt8Array::from(self.op)), + Arc::new(UInt32Array::from(self.op)), Arc::new(UInt32Array::from(self.row)), - Arc::new(UInt16Array::from(self.lane)), + Arc::new(UInt32Array::from(self.lane)), Arc::new(BooleanArray::from(self.present)), Arc::new(UInt32Array::from(self.value)), ], @@ -160,6 +170,27 @@ fn encode(batch: &Batch, version: u64, first: u32) -> Columns { c } +/// What one write did to a property (D-STO-5). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Change { + /// The property was set; `None` is absent (inserted without a value, or + /// cleared). + Set(Option), + /// The record was deleted. Ordinals are never reused, so nothing follows. + Deleted, +} + +/// One entry of a property's history. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Entry { + /// The table version (= Lance version) the write landed in. + pub version: u64, + /// The mailbox the batch wrote on behalf of. + pub owner: MailboxId, + /// What the write did. + pub change: Change, +} + /// An append-only Lance log at `uri`. #[derive(Debug, Clone)] pub struct Log { @@ -230,6 +261,118 @@ impl Log { Ok(table.apply(batch)?) } + /// Every record of the log, in storage order. + async fn scan(&self) -> Result, LogError> { + let ds = Dataset::open(&self.uri).await.map_err(lance_err)?; + ds.scan() + .try_into_stream() + .await + .map_err(lance_err)? + .try_collect() + .await + .map_err(lance_err) + } + + /// D-STO-5: the history of property `col` on record `row`, oldest first. + /// + /// Every write of the property is one [`Entry`]: the value given when the + /// record was inserted (absent included), every later update, and the + /// record's deletion. Each entry carries the version and the owner the + /// batch wrote on behalf of. Nothing is stored for history: it is the + /// log, read through one Quack program per Lance fragment batch, + /// `row = r AND ((lane = c AND op = SET) OR op = DELETE)`, run on + /// mask-risc over the log's own Arrow buffers (borrowed, not copied). + /// Only the matching records are decoded. + /// + /// The read scans the whole log. P-SPARSE-1 measured 16.9 ms for 1,000 + /// versions with a BTree index on `row`; this version has no index, and + /// its cost is measured by `examples/probe_history.rs`. + /// + /// # Errors + /// + /// [`LogError::Corrupt`] if `col` is not a lane of `specs` or a matching + /// record does not decode, and [`LogError::Lance`] if the scan fails. + pub async fn history( + &self, + specs: &[LaneSpec], + row: u32, + col: Col, + ) -> Result, LogError> { + let kind = specs + .get(usize::from(col.0)) + .ok_or(LogError::Corrupt("lane"))? + .kind; + if self.latest().await? == 0 { + return Ok(Vec::new()); + } + // Lanes of the program: 0 = row, 1 = lane, 2 = op. + let eq = |c: u16, v: u32| Filter::cmp(Col(c), Cmp::EqU32(v)); + let program = lower(&Query { + filter: Filter::and([ + eq(0, row), + Filter::or([ + Filter::and([eq(1, u32::from(col.0)), eq(2, SET)]), + eq(2, DELETE), + ]), + ]), + agg: Agg::Rows, + }) + .map_err(|_| LogError::Corrupt("history program"))?; + let mut entries = Vec::new(); + for b in &self.scan().await? { + let n = b.num_rows(); + if n == 0 { + continue; + } + let version = column::(b, 0, "version")?; + let owner = column::(b, 1, "owner")?; + let op = column::(b, 2, "op")?; + let rows = column::(b, 3, "row")?; + let lane = column::(b, 4, "lane")?; + let present = column::(b, 5, "present")?; + let value = column::(b, 6, "value")?; + let lanes = [ + LaneRef::U32(rows.values()), + LaneRef::U32(lane.values()), + LaneRef::U32(op.values()), + ]; + let planes = Planes { + n_rows: n, + masks: &[], + lanes: &lanes, + }; + let mut hits = vec![0u64; words_for(n)]; + let mut scratch = Scratch::for_program(&program, n) + .map_err(|_| LogError::Corrupt("history scratch"))?; + execute_into( + &program, + &planes, + &Foreign::NONE, + &mut scratch, + Out::Mask(&mut hits), + ) + .map_err(|_| LogError::Corrupt("history program"))?; + for i in materialize_rows(&hits, n) { + let change = if op.value(i) == DELETE { + Change::Deleted + } else { + Change::Set(present.value(i).then_some(match kind { + LaneKind::U32 => Value::U32(value.value(i)), + LaneKind::I32 => Value::I32(value.value(i).cast_signed()), + })) + }; + entries.push(Entry { + version: version.value(i), + owner: owner.value(i), + change, + }); + } + } + // Stable: within one version the written order is kept. + entries.sort_by_key(|e| e.version); + Ok(entries) + } + /// Rebuild the table of `owner` with `specs` from the log, batch by /// batch, through the same [`Table::apply`] a live writer uses. /// @@ -246,22 +389,14 @@ impl Log { 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(); + let batches = self.scan().await?; + let mut facts: Vec = 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 op = column::(b, 2, "op")?; let row = column::(b, 3, "row")?; - let lane = column::(b, 4, "lane")?; + 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() { @@ -320,11 +455,7 @@ fn column<'a, T: Array + 'static>( .ok_or(LogError::Corrupt(name)) } -fn decode( - facts: &[(u64, u32, u8, u32, u16, bool, u32)], - owner: MailboxId, - table: &Table, -) -> Result { +fn decode(facts: &[Fact], 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 { @@ -332,6 +463,7 @@ fn decode( ..Batch::default() }; for &(_, own, op, row, lane, present, value) in facts { + let lane = u16::try_from(lane).map_err(|_| LogError::Corrupt("lane"))?; if own != owner { return Err(LogError::Owner { expected: owner, diff --git a/crates/hubspo-store/tests/log.rs b/crates/hubspo-store/tests/log.rs index 413b73f..2c74f3d 100644 --- a/crates/hubspo-store/tests/log.rs +++ b/crates/hubspo-store/tests/log.rs @@ -3,7 +3,7 @@ //! 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::log::{Change, Entry, Log, LogError}; use hubspo_store::table::{Batch, LaneKind, LaneSpec, StoreError, Table, Value}; use lance_graph_mask_risc::LaneRef; use lance_graph_quack::Col; @@ -205,3 +205,87 @@ async fn replay_refuses_a_foreign_owner() { other => panic!("expected Owner, got {:?}", other.map(|t| t.version())), } } + +/// D-STO-5: the history read back from the log equals the history written +/// down independently from the batches, for every record and both lanes. +#[tokio::test(flavor = "multi_thread")] +async fn history_is_every_write_of_one_property_in_order() { + 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_5157); + // Oracle: per (row, lane), the writes in order. + let mut want: std::collections::BTreeMap<(u32, u16), Vec> = Default::default(); + for _ in 0..30 { + let b = random_batch(&mut r, &t); + let first = t.n_rows() as u32; + let applied = log.commit(&mut t, &b).await.expect("commit"); + let e = |change| Entry { + version: applied.version, + owner: OWNER, + change, + }; + for (row, record) in (first..).zip(&b.inserts) { + for (lane, v) in (0u16..).zip(record) { + want.entry((row, lane)) + .or_default() + .push(e(Change::Set(*v))); + } + } + for &(row, col, v) in &b.updates { + want.entry((row, col.0)) + .or_default() + .push(e(Change::Set(v))); + } + for &row in &b.deletes { + for lane in 0..2 { + want.entry((row, lane)) + .or_default() + .push(e(Change::Deleted)); + } + } + } + for row in 0..t.n_rows() as u32 { + for lane in 0..2u16 { + let got = log.history(&SPECS, row, Col(lane)).await.expect("history"); + let exp = want.get(&(row, lane)).cloned().unwrap_or_default(); + assert_eq!(got, exp, "row {row} lane {lane}"); + } + } + // A row that does not exist has no history. + let none = log + .history(&SPECS, t.n_rows() as u32 + 5, Col(0)) + .await + .expect("history"); + assert!(none.is_empty()); + // Anti-vacuity: histories with updates, with a deletion, and with a + // negative value were compared. + let all: Vec<&Entry> = want.values().flatten().collect(); + assert!(want.values().any(|h| h.len() >= 3)); + assert!(all.iter().any(|e| e.change == Change::Deleted)); + assert!( + all.iter() + .any(|e| matches!(e.change, Change::Set(Some(Value::I32(x))) if x < 0)) + ); + assert!(all.iter().any(|e| e.change == Change::Set(None))); +} + +#[tokio::test(flavor = "multi_thread")] +async fn history_refuses_an_unknown_lane_and_reads_an_empty_log() { + let dir = tempfile::tempdir().expect("tempdir"); + let log = Log::new(dir.path().join("deals.lance").to_str().expect("utf8")); + assert!( + log.history(&SPECS, 0, Col(0)) + .await + .expect("empty") + .is_empty() + ); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let mut r = Rng(0x0000_6A11); + let b = random_batch(&mut r, &t); + log.commit(&mut t, &b).await.expect("commit"); + match log.history(&SPECS, 0, Col(2)).await { + Err(LogError::Corrupt("lane")) => {} + other => panic!("expected Corrupt(lane), got {other:?}"), + } +} From acc8e548832b85aa9625a8c7855f3a4ceab05b6c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 17:19:51 +0000 Subject: [PATCH 2/6] D-STO-5: history timing probe; note why the sort is kept examples/probe_history.rs times one property's history over 1,000 versions of 1,000 updates each on 100k records, with no index on row. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-store/Cargo.toml | 4 ++ crates/hubspo-store/examples/probe_history.rs | 72 +++++++++++++++++++ crates/hubspo-store/src/log.rs | 3 + 3 files changed, 79 insertions(+) create mode 100644 crates/hubspo-store/examples/probe_history.rs diff --git a/crates/hubspo-store/Cargo.toml b/crates/hubspo-store/Cargo.toml index 53598b6..14951de 100644 --- a/crates/hubspo-store/Cargo.toml +++ b/crates/hubspo-store/Cargo.toml @@ -29,3 +29,7 @@ lance = ["dep:lance", "dep:arrow", "dep:futures"] [dev-dependencies] tokio = { version = "1", features = ["rt-multi-thread", "macros"] } tempfile = "3" + +[[example]] +name = "probe_history" +required-features = ["lance"] diff --git a/crates/hubspo-store/examples/probe_history.rs b/crates/hubspo-store/examples/probe_history.rs new file mode 100644 index 0000000..c6ff7e7 --- /dev/null +++ b/crates/hubspo-store/examples/probe_history.rs @@ -0,0 +1,72 @@ +//! D-STO-5 timing: history of one property of one record, read from a log of +//! `VERSIONS` batches of `UPDATES` updates each over `ROWS` records, every +//! batch also writing the probed property. No index on `row`. +//! +//! `cargo run --release -p hubspo-store --features lance --example probe_history` + +use std::time::Instant; + +use hubspo_store::log::Log; +use hubspo_store::table::{Batch, LaneKind, LaneSpec, Table, Value}; +use lance_graph_quack::Col; + +const OWNER: u32 = 7; +const ROWS: u32 = 100_000; +const VERSIONS: u32 = 1_000; +const UPDATES: u32 = 1_000; +const PROBE_ROW: u32 = 4_242; +const SPECS: [LaneSpec; 1] = [LaneSpec { + kind: LaneKind::U32, + nullable: false, +}]; + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + let dir = tempfile::tempdir().expect("tempdir"); + let uri = dir.path().join("probe.lance"); + let log = Log::new(uri.to_str().expect("utf8")); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let started = Instant::now(); + let seed = Batch { + owner: OWNER, + inserts: (0..ROWS).map(|i| vec![Some(Value::U32(i))]).collect(), + ..Batch::default() + }; + log.commit(&mut t, &seed).await.expect("seed"); + for v in 0..VERSIONS { + let mut updates: Vec<_> = (0..UPDATES - 1) + .map(|k| ((v * 7_919 + k * 97) % ROWS, Col(0), Some(Value::U32(v)))) + .filter(|(r, _, _)| *r != PROBE_ROW) + .collect(); + updates.push((PROBE_ROW, Col(0), Some(Value::U32(v)))); + let b = Batch { + owner: OWNER, + updates, + ..Batch::default() + }; + log.commit(&mut t, &b).await.expect("commit"); + } + println!( + "wrote {} versions ({} records) in {:.1} s", + VERSIONS + 1, + ROWS + VERSIONS * UPDATES, + started.elapsed().as_secs_f64() + ); + let mut times = Vec::new(); + let mut n = 0; + for _ in 0..7 { + let s = Instant::now(); + n = log + .history(&SPECS, PROBE_ROW, Col(0)) + .await + .expect("history") + .len(); + times.push(s.elapsed().as_secs_f64() * 1e3); + } + times.sort_by(f64::total_cmp); + assert_eq!(n, VERSIONS as usize + 1, "one entry per version"); + println!( + "history: {n} entries, median {:.1} ms (min {:.1}, max {:.1}) over 7 reads", + times[3], times[0], times[6] + ); +} diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs index b35d2b2..31594a2 100644 --- a/crates/hubspo-store/src/log.rs +++ b/crates/hubspo-store/src/log.rs @@ -368,6 +368,9 @@ impl Log { }); } } + // Lance's default scan already yields fragments in append order, so a + // disable run of this sort stays green; it is kept so the result + // stays ordered if the scan is ever made unordered or parallel. // Stable: within one version the written order is kept. entries.sort_by_key(|e| e.version); Ok(entries) From e8e0c3af1a0f2a75916289c040f7f53984ab4d4c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 17:49:08 +0000 Subject: [PATCH 3/6] D-STO-5 probe: split raw scan from history; measure after compaction Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-store/examples/probe_history.rs | 53 +++++++++++++++++-- 1 file changed, 48 insertions(+), 5 deletions(-) diff --git a/crates/hubspo-store/examples/probe_history.rs b/crates/hubspo-store/examples/probe_history.rs index c6ff7e7..87ef7ea 100644 --- a/crates/hubspo-store/examples/probe_history.rs +++ b/crates/hubspo-store/examples/probe_history.rs @@ -52,7 +52,34 @@ async fn main() { ROWS + VERSIONS * UPDATES, started.elapsed().as_secs_f64() ); - let mut times = Vec::new(); + measure(&log, uri.to_str().expect("utf8"), "fragmented").await; + // The same log after compacting its fragments. `history` reads the + // `version` column, not the Lance version, so it stays correct; `commit` + // would refuse the compacted log (its Lance version moved). + let mut ds = lance::Dataset::open(uri.to_str().expect("utf8")) + .await + .expect("open"); + let before = ds.get_fragments().len(); + lance::dataset::optimize::compact_files(&mut ds, Default::default(), None) + .await + .expect("compact"); + println!( + "compacted {before} fragments into {}", + ds.get_fragments().len() + ); + measure(&log, uri.to_str().expect("utf8"), "compacted").await; +} + +fn median(mut ms: Vec) -> (f64, f64, f64) { + ms.sort_by(f64::total_cmp); + (ms[ms.len() / 2], ms[0], ms[ms.len() - 1]) +} + +/// Time `history`, and a raw scan of every column with no program, 7 times. +async fn measure(log: &Log, uri: &str, label: &str) { + use futures::TryStreamExt; + let mut hist = Vec::new(); + let mut scan = Vec::new(); let mut n = 0; for _ in 0..7 { let s = Instant::now(); @@ -61,12 +88,28 @@ async fn main() { .await .expect("history") .len(); - times.push(s.elapsed().as_secs_f64() * 1e3); + hist.push(s.elapsed().as_secs_f64() * 1e3); + let s = Instant::now(); + let ds = lance::Dataset::open(uri).await.expect("open"); + let rows: usize = ds + .scan() + .try_into_stream() + .await + .expect("stream") + .try_collect::>() + .await + .expect("collect") + .iter() + .map(|b| b.num_rows()) + .sum(); + assert_eq!(rows, (ROWS + VERSIONS * UPDATES) as usize); + scan.push(s.elapsed().as_secs_f64() * 1e3); } - times.sort_by(f64::total_cmp); assert_eq!(n, VERSIONS as usize + 1, "one entry per version"); + let (h, hmin, hmax) = median(hist); + let (r, rmin, rmax) = median(scan); println!( - "history: {n} entries, median {:.1} ms (min {:.1}, max {:.1}) over 7 reads", - times[3], times[0], times[6] + "{label}: history {n} entries, median {h:.1} ms ({hmin:.1}..{hmax:.1}); \ + raw scan alone median {r:.1} ms ({rmin:.1}..{rmax:.1}); 7 reads each" ); } From de52dd31dbb73e67e733a292f8e275adf87e3559 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 17:52:03 +0000 Subject: [PATCH 4/6] Board: D-STO-1 shipped; D-STO-5 history read with its measured cost History is correct but fails P-HIST-1's 20 ms budget: 336 ms fragmented, 57 ms compacted, almost all of it the raw scan. ISS-HIST-SCAN-COST names the levers (compaction after decoupling the version check, column projection, an index on row). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- .claude/board/ISSUES.md | 10 +++++++++ .claude/board/LATEST_STATE.md | 21 ++++++++++++++++++- .claude/board/STATUS_BOARD.md | 4 ++-- crates/hubspo-store/examples/probe_history.rs | 8 +++---- 4 files changed, 36 insertions(+), 7 deletions(-) diff --git a/.claude/board/ISSUES.md b/.claude/board/ISSUES.md index 793846b..4017266 100644 --- a/.claude/board/ISSUES.md +++ b/.claude/board/ISSUES.md @@ -3,6 +3,16 @@ > Open blockers and cross-repo follow-ups. Append new entries at the top; > change only the `Status` line of an existing entry. +## ISS-HIST-SCAN-COST — property history scans the whole log + +- **Status:** OPEN (D-STO-5 budget; correctness shipped) +- **Where:** `hubspo-store::log::Log::history`. +- **What:** history reads every log record and runs the Quack program over them. Measured 2026-10-10 (`examples/probe_history.rs`, release, 100k records, 1,000 versions × 1,000 updates, 1.2M log records): 336 ms with one fragment per version, 320 ms of it the raw scan; 57 ms after compacting to one fragment, 54 ms of it the scan. P-HIST-1's budget is < 20 ms. The program is not the cost. +- **Levers, in order:** + 1. Compaction (6× here). It adds Lance versions, so `commit`'s check "Lance version = table version" must change first: take the table version from the log's `version` column (or record the Lance version a table version maps to), and refuse a second writer by that instead. + 2. Read only `row`, `lane`, `op` for the program and fetch the other columns for the matches by position. + 3. An index on `row`. P-SPARSE-1 measured 3.3–5.9 ms compacted with Lance's BTree. Using it takes a Lance filter expression, which runs Lance's internal DataFusion; whether that sits within iron rule 5 (no `datafusion*` dependency, queries are Quack) is an operator decision. The plan's alternative (§5.2) is our own change index — positions per record, an index, not a second truth. + ## ISS-LG-MASKRISC-PATH-NDARRAY — Quack could not be consumed by git - **Status:** RESOLVED 2026-10-10 by lance-graph #1457 (`e724d83`): mask-risc names ndarray by git, and lance-graph's workspaces patch it back to the sibling checkout. diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index 42aed4c..919673b 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -137,7 +137,26 @@ 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. +- Not yet: an index on `row`, compaction, object stores. Compaction adds + Lance versions, so it needs the table version read from the `version` + column instead of the Lance version. + +## Property history (D-STO-5, feature `lance`) + +- `Log::history(specs, row, col) -> Vec`: every write of one + property on one record, oldest first — the inserted value (absent + included), each update, the deletion — each with its version and owner. + No history store: one Quack program per log batch, + `row = r ∧ ((lane = c ∧ op = SET) ∨ op = DELETE)`, on mask-risc over the + log's Arrow buffers (borrowed); only matches are decoded. The log's `op` + and `lane` columns widened to `u32` for this (no data existed yet). +- Tests: for every record and both lanes over 30 random batches, the + history equals one written down from the batches (anti-vacuity: updates, + deletions, negative and absent values); unknown lane refused; empty log + reads empty. Disable runs 2026-10-10, red: drop the row term, drop the + DELETE arm, drop `op = SET`, lose the I32 sign. Green, kept: the sort by + version (Lance's default scan is already in append order). +- Timing: fragmented log (1,001 fragments, 1.2M records): **336 ms**, of which the raw scan is 320 ms; compacted to one fragment: **57 ms** (scan 54 ms). Fails P-HIST-1's < 20 ms either way; the Quack program is 3–15 ms of it (release, `examples/probe_history.rs`, 7 reads, 2026-10-10). The scan, not the program, is the cost. Open: ISS-HIST-SCAN-COST. ## Board view (D-QRY-3, board half) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 79c8575..44ae9cb 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -35,11 +35,11 @@ | 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 PR | resident half shipped (#22); persistence half in PR: `hubspo-store::log` (one Lance version per batch, replay, feature `lance`) | +| D-STO-1 | `hubspo-store` write lane: batch image → one Lance version per publish; owner-stamped (`mailbox_owner`) | L | Shipped | resident half #22; persistence half #23: `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) | -| D-STO-5 | Property history via version log + P-HIST-1 probe | M | Queued | P-HIST-1 failed on in-place updates (46 ms); P-SPARSE-1 passes with the split-tunnel sparse delta: 16.9 ms indexed, 3.3–5.9 ms compacted | +| D-STO-5 | Property history via version log + P-HIST-1 probe | M | In progress | read shipped in PR: `Log::history` (one Quack program per log batch, no index); fragmented log (1,001 fragments, 1.2M records): **336 ms**, of which the raw scan is 320 ms; compacted to one fragment: **57 ms** (scan 54 ms). Fails P-HIST-1's < 20 ms either way; the Quack program is 3–15 ms of it (release, `examples/probe_history.rs`, 7 reads, 2026-10-10). Budget open: ISS-HIST-SCAN-COST. P-HIST-1 failed on in-place updates (46 ms); P-SPARSE-1 passes with the split-tunnel sparse delta: 16.9 ms indexed, 3.3–5.9 ms compacted | | D-STO-6 | Importer (value oracle, part 1): sandbox portal → crm-sim plan → commit; counts compared per object type | L | Queued | | ## Wave 3 — Query, lists, views, reports diff --git a/crates/hubspo-store/examples/probe_history.rs b/crates/hubspo-store/examples/probe_history.rs index 87ef7ea..42a3b0a 100644 --- a/crates/hubspo-store/examples/probe_history.rs +++ b/crates/hubspo-store/examples/probe_history.rs @@ -47,9 +47,8 @@ async fn main() { log.commit(&mut t, &b).await.expect("commit"); } println!( - "wrote {} versions ({} records) in {:.1} s", + "wrote {} versions in {:.1} s", VERSIONS + 1, - ROWS + VERSIONS * UPDATES, started.elapsed().as_secs_f64() ); measure(&log, uri.to_str().expect("utf8"), "fragmented").await; @@ -81,6 +80,7 @@ async fn measure(log: &Log, uri: &str, label: &str) { let mut hist = Vec::new(); let mut scan = Vec::new(); let mut n = 0; + let mut records = 0; for _ in 0..7 { let s = Instant::now(); n = log @@ -102,14 +102,14 @@ async fn measure(log: &Log, uri: &str, label: &str) { .iter() .map(|b| b.num_rows()) .sum(); - assert_eq!(rows, (ROWS + VERSIONS * UPDATES) as usize); + records = rows; scan.push(s.elapsed().as_secs_f64() * 1e3); } assert_eq!(n, VERSIONS as usize + 1, "one entry per version"); let (h, hmin, hmax) = median(hist); let (r, rmin, rmax) = median(scan); println!( - "{label}: history {n} entries, median {h:.1} ms ({hmin:.1}..{hmax:.1}); \ + "{label}: {records} log records; history {n} entries, median {h:.1} ms ({hmin:.1}..{hmax:.1}); \ raw scan alone median {r:.1} ms ({rmin:.1}..{rmax:.1}); 7 reads each" ); } From a1a3963c3b0cf222b78fee39bd7cc227637ff957 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 18:10:49 +0000 Subject: [PATCH 5/6] Never compact the log: readers rely on append order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The append order is the version order, so history and replay drop their sorts (history's was inert under a disable run). Replay still refuses an out-of-order log as a batch whose version is not the next one. The probe drops its compaction arm; ISS-HIST-SCAN-COST and plan §5.2 record the decision, leaving projection and an index on row as the levers. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- .claude/board/ISSUES.md | 10 +++++----- .claude/board/LATEST_STATE.md | 13 +++++++------ .claude/board/STATUS_BOARD.md | 2 +- .claude/plans/hubspo-integration-v1.md | 5 +++++ crates/hubspo-store/examples/probe_history.rs | 16 +--------------- crates/hubspo-store/src/log.rs | 19 ++++++++----------- 6 files changed, 27 insertions(+), 38 deletions(-) diff --git a/.claude/board/ISSUES.md b/.claude/board/ISSUES.md index 4017266..5f191db 100644 --- a/.claude/board/ISSUES.md +++ b/.claude/board/ISSUES.md @@ -7,11 +7,11 @@ - **Status:** OPEN (D-STO-5 budget; correctness shipped) - **Where:** `hubspo-store::log::Log::history`. -- **What:** history reads every log record and runs the Quack program over them. Measured 2026-10-10 (`examples/probe_history.rs`, release, 100k records, 1,000 versions × 1,000 updates, 1.2M log records): 336 ms with one fragment per version, 320 ms of it the raw scan; 57 ms after compacting to one fragment, 54 ms of it the scan. P-HIST-1's budget is < 20 ms. The program is not the cost. -- **Levers, in order:** - 1. Compaction (6× here). It adds Lance versions, so `commit`'s check "Lance version = table version" must change first: take the table version from the log's `version` column (or record the Lance version a table version maps to), and refuse a second writer by that instead. - 2. Read only `row`, `lane`, `op` for the program and fetch the other columns for the matches by position. - 3. An index on `row`. P-SPARSE-1 measured 3.3–5.9 ms compacted with Lance's BTree. Using it takes a Lance filter expression, which runs Lance's internal DataFusion; whether that sits within iron rule 5 (no `datafusion*` dependency, queries are Quack) is an operator decision. The plan's alternative (§5.2) is our own change index — positions per record, an index, not a second truth. +- **What:** history reads every log record and runs the Quack program over them. Measured 2026-10-10 (`examples/probe_history.rs`, release, 100k records, 1,000 versions × 1,000 updates, 1.2M log records, one fragment per version): 336 ms, 320 ms of it the raw scan. P-HIST-1's budget is < 20 ms. The program is not the cost. +- **Not a lever:** compaction. DECISION 2026-10-10 (operator): the log is never compacted, because the append order gives version order for free. (A one-off measurement compacted to one fragment read in 57 ms, still over budget.) +- **Levers:** + 1. Read only `row`, `lane`, `op` for the program and fetch the other columns for the matches by position. + 2. An index on `row`. P-SPARSE-1 measured 16.9 ms with Lance's BTree on an uncompacted log of 1,000 fragments, inside the budget. Using it takes a Lance filter expression, which runs Lance's internal DataFusion; whether that sits within iron rule 5 (no `datafusion*` dependency, queries are Quack) is an operator decision. The plan's alternative (§5.2) is our own change index — positions per record, an index, not a second truth. ## ISS-LG-MASKRISC-PATH-NDARRAY — Quack could not be consumed by git diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index 919673b..a4dfef4 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -137,9 +137,8 @@ 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: an index on `row`, compaction, object stores. Compaction adds - Lance versions, so it needs the table version read from the `version` - column instead of the Lance version. +- Not yet: an index on `row`, object stores. The log is never compacted + (decision 2026-10-10): the append order is the version order. ## Property history (D-STO-5, feature `lance`) @@ -154,9 +153,11 @@ history equals one written down from the batches (anti-vacuity: updates, deletions, negative and absent values); unknown lane refused; empty log reads empty. Disable runs 2026-10-10, red: drop the row term, drop the - DELETE arm, drop `op = SET`, lose the I32 sign. Green, kept: the sort by - version (Lance's default scan is already in append order). -- Timing: fragmented log (1,001 fragments, 1.2M records): **336 ms**, of which the raw scan is 320 ms; compacted to one fragment: **57 ms** (scan 54 ms). Fails P-HIST-1's < 20 ms either way; the Quack program is 3–15 ms of it (release, `examples/probe_history.rs`, 7 reads, 2026-10-10). The scan, not the program, is the cost. Open: ISS-HIST-SCAN-COST. + DELETE arm, drop `op = SET`, lose the I32 sign. A sort by version was + inert under a disable run and is removed: append order is version order. + Replay dropped its sort too; a log out of order is still refused there, + as a batch whose version is not the next one. +- Timing: fragmented log (1,001 fragments, 1.2M records): **336 ms**, of which the raw scan is 320 ms. Fails P-HIST-1's < 20 ms; the Quack program is 3–15 ms of it (release, `examples/probe_history.rs`, 7 reads, 2026-10-10). The scan, not the program, is the cost. Open: ISS-HIST-SCAN-COST. ## Board view (D-QRY-3, board half) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 44ae9cb..50eaeda 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -39,7 +39,7 @@ | 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) | -| D-STO-5 | Property history via version log + P-HIST-1 probe | M | In progress | read shipped in PR: `Log::history` (one Quack program per log batch, no index); fragmented log (1,001 fragments, 1.2M records): **336 ms**, of which the raw scan is 320 ms; compacted to one fragment: **57 ms** (scan 54 ms). Fails P-HIST-1's < 20 ms either way; the Quack program is 3–15 ms of it (release, `examples/probe_history.rs`, 7 reads, 2026-10-10). Budget open: ISS-HIST-SCAN-COST. P-HIST-1 failed on in-place updates (46 ms); P-SPARSE-1 passes with the split-tunnel sparse delta: 16.9 ms indexed, 3.3–5.9 ms compacted | +| D-STO-5 | Property history via version log + P-HIST-1 probe | M | In progress | read shipped in PR: `Log::history` (one Quack program per log batch, no index); fragmented log (1,001 fragments, 1.2M records): **336 ms**, of which the raw scan is 320 ms. Fails P-HIST-1's < 20 ms; the Quack program is 3–15 ms of it (release, `examples/probe_history.rs`, 7 reads, 2026-10-10). Budget open: ISS-HIST-SCAN-COST. P-HIST-1 failed on in-place updates (46 ms); P-SPARSE-1 passes with the split-tunnel sparse delta: 16.9 ms indexed, 3.3–5.9 ms compacted | | D-STO-6 | Importer (value oracle, part 1): sandbox portal → crm-sim plan → commit; counts compared per object type | L | Queued | | ## Wave 3 — Query, lists, views, reports diff --git a/.claude/plans/hubspo-integration-v1.md b/.claude/plans/hubspo-integration-v1.md index c81c369..1598a85 100644 --- a/.claude/plans/hubspo-integration-v1.md +++ b/.claude/plans/hubspo-integration-v1.md @@ -859,6 +859,11 @@ Writing the 1,000 versions takes 6.6–12.4 s instead of ~300 s. The delta is append-only, so compacting it loses no history; the base is never rewritten, so nothing depends on keeping old Lance versions. +*Decision 2026-10-10:* the log is never compacted. The append order gives +version order for free, and readers (history, replay) rely on it instead of +sorting. History speed comes from reading fewer columns or an index on +`row` (ISS-HIST-SCAN-COST), not from compaction. + ### 5.3 Timeline The record timeline is spear's `activity` table (D-HUB-4) joined on diff --git a/crates/hubspo-store/examples/probe_history.rs b/crates/hubspo-store/examples/probe_history.rs index 42a3b0a..68c7893 100644 --- a/crates/hubspo-store/examples/probe_history.rs +++ b/crates/hubspo-store/examples/probe_history.rs @@ -51,22 +51,8 @@ async fn main() { VERSIONS + 1, started.elapsed().as_secs_f64() ); + // One fragment per version: the log is never compacted. measure(&log, uri.to_str().expect("utf8"), "fragmented").await; - // The same log after compacting its fragments. `history` reads the - // `version` column, not the Lance version, so it stays correct; `commit` - // would refuse the compacted log (its Lance version moved). - let mut ds = lance::Dataset::open(uri.to_str().expect("utf8")) - .await - .expect("open"); - let before = ds.get_fragments().len(); - lance::dataset::optimize::compact_files(&mut ds, Default::default(), None) - .await - .expect("compact"); - println!( - "compacted {before} fragments into {}", - ds.get_fragments().len() - ); - measure(&log, uri.to_str().expect("utf8"), "compacted").await; } fn median(mut ms: Vec) -> (f64, f64, f64) { diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs index 31594a2..dd090e0 100644 --- a/crates/hubspo-store/src/log.rs +++ b/crates/hubspo-store/src/log.rs @@ -2,7 +2,8 @@ //! 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. +//! there are no deletion vectors to read. It is never compacted: the +//! append order is the version order, and readers rely on it. //! Each record is one fact of one batch: //! //! | `op` | meaning | `row` | `lane` | `present` / `value` | @@ -29,9 +30,7 @@ //! **History (D-STO-5)** is the log itself: [`Log::history`] reads one //! property of one record back as the list of writes that touched it. //! -//! Not yet: an index on `row`, compaction (which would add Lance versions -//! and so needs the table version taken from the `version` column rather -//! than from the Lance version), and object-store URIs beyond what `lance` +//! Not yet: an index on `row`, and object-store URIs beyond what `lance` //! resolves by itself. use std::sync::Arc; @@ -368,11 +367,8 @@ impl Log { }); } } - // Lance's default scan already yields fragments in append order, so a - // disable run of this sort stays green; it is kept so the result - // stays ordered if the scan is ever made unordered or parallel. - // Stable: within one version the written order is kept. - entries.sort_by_key(|e| e.version); + // No sort: the scan yields records in append order, which is version + // order and, within a version, written order. Ok(entries) } @@ -414,8 +410,9 @@ impl Log { )); } } - // Stable sort by version keeps each batch's facts in written order. - facts.sort_by_key(|f| f.0); + // No sort: records come back in append order, so each batch's facts + // are contiguous and in written order. A log out of order is caught + // below, as a batch whose version is not the next one. let mut i = 0; while i < facts.len() { let version = facts[i].0; From 9871f64ae38aa368661a4baf5dced4cdeed9297d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 18:12:29 +0000 Subject: [PATCH 6/6] History refuses an absent value in a required lane Replay refuses that record through Table::apply, so history must not report it as Set(None). Found by review on #24. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-store/src/log.rs | 16 ++++--- crates/hubspo-store/tests/log.rs | 76 ++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 6 deletions(-) diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs index dd090e0..b310fed 100644 --- a/crates/hubspo-store/src/log.rs +++ b/crates/hubspo-store/src/log.rs @@ -289,18 +289,18 @@ impl Log { /// /// # Errors /// - /// [`LogError::Corrupt`] if `col` is not a lane of `specs` or a matching - /// record does not decode, and [`LogError::Lance`] if the scan fails. + /// [`LogError::Corrupt`] if `col` is not a lane of `specs`, or a matching + /// record does not decode (an absent value in a required lane included), + /// and [`LogError::Lance`] if the scan fails. pub async fn history( &self, specs: &[LaneSpec], row: u32, col: Col, ) -> Result, LogError> { - let kind = specs + let spec = *specs .get(usize::from(col.0)) - .ok_or(LogError::Corrupt("lane"))? - .kind; + .ok_or(LogError::Corrupt("lane"))?; if self.latest().await? == 0 { return Ok(Vec::new()); } @@ -354,8 +354,12 @@ impl Log { for i in materialize_rows(&hits, n) { let change = if op.value(i) == DELETE { Change::Deleted + } else if !present.value(i) && !spec.nullable { + // Replay refuses this record (`Table::apply`); history + // must not report a state the table can never hold. + return Err(LogError::Corrupt("absent value in a required lane")); } else { - Change::Set(present.value(i).then_some(match kind { + Change::Set(present.value(i).then_some(match spec.kind { LaneKind::U32 => Value::U32(value.value(i)), LaneKind::I32 => Value::I32(value.value(i).cast_signed()), })) diff --git a/crates/hubspo-store/tests/log.rs b/crates/hubspo-store/tests/log.rs index 2c74f3d..73a0d61 100644 --- a/crates/hubspo-store/tests/log.rs +++ b/crates/hubspo-store/tests/log.rs @@ -289,3 +289,79 @@ async fn history_refuses_an_unknown_lane_and_reads_an_empty_log() { other => panic!("expected Corrupt(lane), got {other:?}"), } } + +/// Append one raw record to the log, bypassing `commit` (a corrupt writer). +async fn append_raw(uri: &str, version: u64, op: u32, row: u32, lane: u32, present: bool) { + use arrow::array::{BooleanArray, RecordBatch, RecordBatchIterator, UInt32Array, UInt64Array}; + use arrow::datatypes::{DataType, Field, Schema}; + use std::sync::Arc; + let schema = Arc::new(Schema::new(vec![ + Field::new("version", DataType::UInt64, false), + Field::new("owner", DataType::UInt32, false), + Field::new("op", DataType::UInt32, false), + Field::new("row", DataType::UInt32, false), + Field::new("lane", DataType::UInt32, false), + Field::new("present", DataType::Boolean, false), + Field::new("value", DataType::UInt32, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(UInt64Array::from(vec![version])), + Arc::new(UInt32Array::from(vec![OWNER])), + Arc::new(UInt32Array::from(vec![op])), + Arc::new(UInt32Array::from(vec![row])), + Arc::new(UInt32Array::from(vec![lane])), + Arc::new(BooleanArray::from(vec![present])), + Arc::new(UInt32Array::from(vec![0u32])), + ], + ) + .expect("batch"); + let params = lance::dataset::WriteParams { + mode: lance::dataset::WriteMode::Append, + ..Default::default() + }; + lance::Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema), + uri, + Some(params), + ) + .await + .expect("append"); +} + +/// An absent value in a required lane is a state the table can never hold: +/// replay refuses it, so history must too, rather than report `Set(None)`. +#[tokio::test(flavor = "multi_thread")] +async fn history_refuses_an_absent_value_in_a_required_lane() { + let dir = tempfile::tempdir().expect("tempdir"); + let uri = dir.path().join("deals.lance"); + let uri = uri.to_str().expect("utf8"); + let log = Log::new(uri); + let mut t = Table::new(OWNER, &SPECS).expect("table"); + let b = Batch { + owner: OWNER, + inserts: vec![vec![Some(Value::U32(1)), None]], + ..Batch::default() + }; + log.commit(&mut t, &b).await.expect("commit"); + // Silence half: the nullable lane's absent value is a valid history. + assert_eq!( + log.history(&SPECS, 0, Col(1)).await.expect("history"), + vec![Entry { + version: 1, + owner: OWNER, + change: Change::Set(None), + }] + ); + // SET row 0, lane 0 (required), absent. + append_raw(uri, 2, 1, 0, 0, false).await; + match log.history(&SPECS, 0, Col(0)).await { + Err(LogError::Corrupt("absent value in a required lane")) => {} + other => panic!("expected Corrupt, got {other:?}"), + } + assert!( + log.replay(OWNER, &SPECS).await.is_err(), + "replay refuses it too" + ); +}