From de38bfd4e956bc4d92b7d9cf58f9fb5f07dbb4f3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 19:16:54 +0000 Subject: [PATCH 1/2] D-STO-5: history folds over resident log records; Lance read only on replay One write costs about a million folds (lance-graph 2026-09-24), and a Lance scan is recovery cost (~320 ms over 1,000 fragments). So the log keeps every record it commits or replays resident, and history runs its Quack program over memory instead of scanning Lance. - Log owns the resident records; commit appends to them after the Lance append and refuses when they are not at the table's version. - replay is the one Lance read; on any error nothing becomes resident. - history is synchronous. A corrupt record cannot become resident (replay refuses it), so the per-match required-lane check is structural now. - Tests: a second log replayed from disk gives identical histories; a log without resident records refuses to commit. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- crates/hubspo-store/examples/probe_history.rs | 60 ++-- crates/hubspo-store/src/log.rs | 306 +++++++++++------- crates/hubspo-store/tests/log.rs | 83 +++-- 3 files changed, 265 insertions(+), 184 deletions(-) diff --git a/crates/hubspo-store/examples/probe_history.rs b/crates/hubspo-store/examples/probe_history.rs index 68c7893..cd6125a 100644 --- a/crates/hubspo-store/examples/probe_history.rs +++ b/crates/hubspo-store/examples/probe_history.rs @@ -1,6 +1,7 @@ //! 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`. +//! batch also writing the probed property. History folds over the resident +//! records; replay is the one Lance read, timed separately. //! //! `cargo run --release -p hubspo-store --features lance --example probe_history` @@ -24,7 +25,7 @@ const SPECS: [LaneSpec; 1] = [LaneSpec { 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 log = Log::new(uri.to_str().expect("utf8")); let mut t = Table::new(OWNER, &SPECS).expect("table"); let started = Instant::now(); let seed = Batch { @@ -52,50 +53,41 @@ async fn main() { started.elapsed().as_secs_f64() ); // One fragment per version: the log is never compacted. - measure(&log, uri.to_str().expect("utf8"), "fragmented").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]) -} + let uri = uri.to_str().expect("utf8"); -/// 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; + // Live clock: history folds over the resident records. let mut hist = Vec::new(); - let mut scan = Vec::new(); let mut n = 0; - let mut records = 0; - for _ in 0..7 { + for _ in 0..101 { let s = Instant::now(); n = log .history(&SPECS, PROBE_ROW, Col(0)) - .await .expect("history") .len(); 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(); - 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}: {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" + "history (resident, {} records): {n} entries, median {h:.3} ms ({hmin:.3}..{hmax:.3}), 101 reads", + log.len() ); + + // Recovery clock: replay reads the whole log once from Lance. + let mut rep = Vec::new(); + for _ in 0..5 { + let mut cold = Log::new(uri); + let s = Instant::now(); + let back = cold.replay(OWNER, &SPECS).await.expect("replay"); + rep.push(s.elapsed().as_secs_f64() * 1e3); + assert_eq!(back.version(), t.version()); + assert_eq!(cold.len(), log.len()); + } + let (r, rmin, rmax) = median(rep); + println!("replay (Lance, cold): median {r:.1} ms ({rmin:.1}..{rmax:.1}), 5 runs"); +} + +fn median(mut ms: Vec) -> (f64, f64, f64) { + ms.sort_by(f64::total_cmp); + (ms[ms.len() / 2], ms[0], ms[ms.len() - 1]) } diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs index b310fed..d928d22 100644 --- a/crates/hubspo-store/src/log.rs +++ b/crates/hubspo-store/src/log.rs @@ -27,8 +27,13 @@ //! then applies it to the table. If the append fails the table is //! unchanged. //! -//! **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. +//! **Resident records.** Lance is the persistence clock: [`Log::replay`] +//! reads it once, [`Log::commit`] writes one version per batch. The records +//! either touches stay resident, so reads fold over memory. +//! +//! **History (D-STO-5)** is a fold over those records: [`Log::history`] +//! reads one property of one record back as the list of writes that touched +//! it, without opening the dataset. //! //! Not yet: an index on `row`, and object-store URIs beyond what `lance` //! resolves by itself. @@ -54,9 +59,6 @@ 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)] pub enum LogError { @@ -106,7 +108,7 @@ fn schema() -> Arc { ])) } -#[derive(Default)] +#[derive(Default, Clone)] struct Columns { version: Vec, owner: Vec, @@ -132,17 +134,33 @@ impl Columns { }); } - fn batch(self) -> Result { + fn len(&self) -> usize { + self.version.len() + } + + fn extend(&mut self, other: Columns) { + self.version.extend(other.version); + self.owner.extend(other.owner); + self.op.extend(other.op); + self.row.extend(other.row); + self.lane.extend(other.lane); + self.present.extend(other.present); + self.value.extend(other.value); + } + + /// The records as one Arrow batch. Copies them: the copy goes to Lance, + /// the original stays resident. + fn batch(&self) -> Result { RecordBatch::try_new( schema(), vec![ - Arc::new(UInt64Array::from(self.version)), - Arc::new(UInt32Array::from(self.owner)), - Arc::new(UInt32Array::from(self.op)), - Arc::new(UInt32Array::from(self.row)), - Arc::new(UInt32Array::from(self.lane)), - Arc::new(BooleanArray::from(self.present)), - Arc::new(UInt32Array::from(self.value)), + Arc::new(UInt64Array::from(self.version.clone())), + Arc::new(UInt32Array::from(self.owner.clone())), + Arc::new(UInt32Array::from(self.op.clone())), + Arc::new(UInt32Array::from(self.row.clone())), + Arc::new(UInt32Array::from(self.lane.clone())), + Arc::new(BooleanArray::from(self.present.clone())), + Arc::new(UInt32Array::from(self.value.clone())), ], ) .map_err(lance_err) @@ -190,16 +208,53 @@ pub struct Entry { pub change: Change, } -/// An append-only Lance log at `uri`. -#[derive(Debug, Clone)] +/// An append-only Lance log at `uri`, with its records resident. +/// +/// Lance is the persistence clock: it is read once, by [`Log::replay`], and +/// written once per batch, by [`Log::commit`]. Every record either one +/// touches stays resident, so reads ([`Log::history`]) fold over memory and +/// never open the dataset. A Lance scan is what recovery pays (P-HIST-1: +/// ~320 ms over 1,000 fragments); a fold over the same records costs +/// nanoseconds each (lance-graph, 2026-09-24: one write buys about a +/// million folds). +#[derive(Default)] pub struct Log { uri: String, + /// Every record of versions `1..=version`, in append order. + records: Columns, + /// The last version the resident records hold. + version: u64, +} + +impl std::fmt::Debug for Log { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Log") + .field("uri", &self.uri) + .field("version", &self.version) + .field("records", &self.records.len()) + .finish() + } } impl Log { - /// The log at `uri` (created by the first commit). + /// The log at `uri` (created by the first commit). Nothing is resident + /// until [`Log::replay`]; a log that already holds batches must be + /// replayed before it can be committed to. pub fn new(uri: impl Into) -> Self { - Self { uri: uri.into() } + Self { + uri: uri.into(), + ..Self::default() + } + } + + /// How many records are resident. + pub fn len(&self) -> usize { + self.records.len() + } + + /// Whether no record is resident. + pub fn is_empty(&self) -> bool { + self.records.len() == 0 } async fn latest(&self) -> Result { @@ -211,17 +266,25 @@ impl Log { } } - /// Validate `batch`, append it as one Lance version, then apply it. + /// Validate `batch`, append it as one Lance version, then apply it to + /// the table and to the resident records. /// /// # 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::Diverged`]: the resident records, or the log on disk, + /// were not at the table's version before the append, or the log was + /// 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 { + pub async fn commit(&mut self, table: &mut Table, batch: &Batch) -> Result { table.check(batch)?; let before = table.version(); + if self.version != before { + return Err(LogError::Diverged { + expected: before, + found: self.version, + }); + } let found = self.latest().await?; if found != before { return Err(LogError::Diverged { @@ -230,7 +293,7 @@ impl Log { }); } let first = u32::try_from(table.n_rows()).map_err(|_| StoreError::Full)?; - let records = encode(batch, before + 1, first).batch()?; + let records = encode(batch, before + 1, first); let params = WriteParams { mode: if before == 0 { WriteMode::Create @@ -240,7 +303,7 @@ impl Log { ..WriteParams::default() }; let ds = Dataset::write( - RecordBatchIterator::new(vec![Ok(records)], schema()), + RecordBatchIterator::new(vec![Ok(records.batch()?)], schema()), &self.uri, Some(params), ) @@ -257,7 +320,10 @@ impl Log { found: after, }); } - Ok(table.apply(batch)?) + let applied = table.apply(batch)?; + self.records.extend(records); + self.version = applied.version; + Ok(applied) } /// Every record of the log, in storage order. @@ -277,31 +343,25 @@ impl Log { /// 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. + /// batch wrote on behalf of. Nothing is stored for history and nothing is + /// read from Lance: one Quack program, + /// `row = r AND ((lane = c AND op = SET) OR op = DELETE)`, runs on + /// mask-risc over the resident records, and only matches 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`. + /// Only records that passed [`Table::check`] (through `commit`) or were + /// applied by a successful [`Log::replay`] are resident, so every match + /// decodes against `specs` the table was built with. /// /// # Errors /// - /// [`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> { + /// [`LogError::Corrupt`] if `col` is not a lane of `specs`. + pub fn history(&self, specs: &[LaneSpec], row: u32, col: Col) -> Result, LogError> { let spec = *specs .get(usize::from(col.0)) .ok_or(LogError::Corrupt("lane"))?; - if self.latest().await? == 0 { + let r = &self.records; + let n = r.len(); + if n == 0 { return Ok(Vec::new()); } // Lanes of the program: 0 = row, 1 = lane, 2 = op. @@ -317,67 +377,52 @@ impl Log { 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 { + let lanes = [ + LaneRef::U32(&r.row), + LaneRef::U32(&r.lane), + LaneRef::U32(&r.op), + ]; + 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"))?; + // Append order is version order (the log is never compacted), so the + // matches are already oldest first. + Ok(materialize_rows(&hits, n) + .into_iter() + .map(|i| Entry { + version: r.version[i], + owner: r.owner[i], + change: if r.op[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 spec.kind { - LaneKind::U32 => Value::U32(value.value(i)), - LaneKind::I32 => Value::I32(value.value(i).cast_signed()), + Change::Set(r.present[i].then_some(match spec.kind { + LaneKind::U32 => Value::U32(r.value[i]), + LaneKind::I32 => Value::I32(r.value[i].cast_signed()), })) - }; - entries.push(Entry { - version: version.value(i), - owner: owner.value(i), - change, - }); - } - } - // No sort: the scan yields records in append order, which is version - // order and, within a version, written order. - Ok(entries) + }, + }) + .collect()) } /// Rebuild the table of `owner` with `specs` from the log, batch by - /// batch, through the same [`Table::apply`] a live writer uses. + /// batch, through the same [`Table::apply`] a live writer uses, and make + /// the log's records resident. + /// + /// This is the one place Lance is read. On any error nothing becomes + /// resident: the log keeps what it held before. /// /// # Errors /// @@ -386,15 +431,20 @@ impl Log { /// [`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 { + pub async fn replay( + &mut self, + owner: MailboxId, + specs: &[LaneSpec], + ) -> Result { let mut table = Table::new(owner, specs)?; let latest = self.latest().await?; if latest == 0 { + self.records = Columns::default(); + self.version = 0; return Ok(table); } - let batches = self.scan().await?; - let mut facts: Vec = Vec::new(); - for b in &batches { + let mut records = Columns::default(); + for b in &self.scan().await? { let version = column::(b, 0, "version")?; let own = column::(b, 1, "owner")?; let op = column::(b, 2, "op")?; @@ -402,29 +452,26 @@ impl Log { 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), - )); - } + records.version.extend_from_slice(version.values()); + records.owner.extend_from_slice(own.values()); + records.op.extend_from_slice(op.values()); + records.row.extend_from_slice(row.values()); + records.lane.extend_from_slice(lane.values()); + records.present.extend(present.values().iter()); + records.value.extend_from_slice(value.values()); } - // No sort: records come back in append order, so each batch's facts + // No sort: records come back in append order, so each batch's records // 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; - let end = facts[i..] + let n = records.len(); + while i < n { + let version = records.version[i]; + let end = records.version[i..] .iter() - .position(|f| f.0 != version) - .map_or(facts.len(), |p| i + p); - let batch = decode(&facts[i..end], owner, &table)?; + .position(|&v| v != version) + .map_or(n, |p| i + p); + let batch = decode(&records, i..end, owner, &table)?; let applied = table.apply(&batch)?; if applied.version != version { return Err(LogError::Diverged { @@ -440,6 +487,8 @@ impl Log { found: table.version(), }); } + self.records = records; + self.version = table.version(); Ok(table) } } @@ -459,14 +508,27 @@ fn column<'a, T: Array + 'static>( .ok_or(LogError::Corrupt(name)) } -fn decode(facts: &[Fact], owner: MailboxId, table: &Table) -> Result { +fn decode( + records: &Columns, + range: std::ops::Range, + 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 { + for k in range { + let (own, op, row, lane, present, value) = ( + records.owner[k], + records.op[k], + records.row[k], + records.lane[k], + records.present[k], + records.value[k], + ); let lane = u16::try_from(lane).map_err(|_| LogError::Corrupt("lane"))?; if own != owner { return Err(LogError::Owner { diff --git a/crates/hubspo-store/tests/log.rs b/crates/hubspo-store/tests/log.rs index 73a0d61..49b6026 100644 --- a/crates/hubspo-store/tests/log.rs +++ b/crates/hubspo-store/tests/log.rs @@ -93,7 +93,7 @@ fn random_batch(r: &mut Rng, t: &Table) -> Batch { #[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 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 { @@ -118,7 +118,7 @@ async fn replay_rebuilds_the_committed_table_exactly() { #[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 mut 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)); } @@ -127,7 +127,7 @@ async fn an_empty_log_replays_to_an_empty_table() { 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 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); @@ -155,7 +155,7 @@ async fn a_second_writer_is_caught_before_it_applies() { 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 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); @@ -193,7 +193,7 @@ async fn a_refused_batch_writes_nothing() { #[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 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); @@ -211,7 +211,7 @@ async fn replay_refuses_a_foreign_owner() { #[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 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. @@ -247,15 +247,39 @@ async fn history_is_every_write_of_one_property_in_order() { } 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 got = log.history(&SPECS, row, Col(lane)).expect("history"); let exp = want.get(&(row, lane)).cloned().unwrap_or_default(); assert_eq!(got, exp, "row {row} lane {lane}"); } } + // Replayable: a second log replayed from disk holds the same records, + // so every history read from it is identical. + let mut again = Log::new(dir.path().join("deals.lance").to_str().expect("utf8")); + let back = again.replay(OWNER, &SPECS).await.expect("replay"); + assert_eq!(again.len(), log.len()); + for row in 0..t.n_rows() as u32 { + for lane in 0..2u16 { + assert_eq!( + again.history(&SPECS, row, Col(lane)).expect("again"), + log.history(&SPECS, row, Col(lane)).expect("log"), + ); + } + } + // A log whose records are not resident at the table's version refuses + // to commit: its history would miss the batches before it. + let mut cold = Log::new(dir.path().join("deals.lance").to_str().expect("utf8")); + let mut t2 = back; + let extra = random_batch(&mut r, &t2); + match cold.commit(&mut t2, &extra).await { + Err(LogError::Diverged { + expected: 30, + found: 0, + }) => {} + other => panic!("expected Diverged {{30, 0}}, got {other:?}"), + } // 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 @@ -273,18 +297,13 @@ async fn history_is_every_write_of_one_property_in_order() { #[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 log = Log::new(dir.path().join("deals.lance").to_str().expect("utf8")); + assert!(log.history(&SPECS, 0, Col(0)).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 { + match log.history(&SPECS, 0, Col(2)) { Err(LogError::Corrupt("lane")) => {} other => panic!("expected Corrupt(lane), got {other:?}"), } @@ -330,14 +349,15 @@ async fn append_raw(uri: &str, version: u64, op: u32, row: u32, lane: u32, prese .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)`. +/// An absent value in a required lane is a state the table can never hold. +/// Replay refuses such a record, and a refused replay leaves the resident +/// records as they were, so history can never report it. #[tokio::test(flavor = "multi_thread")] -async fn history_refuses_an_absent_value_in_a_required_lane() { +async fn a_corrupt_record_never_becomes_resident() { 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 log = Log::new(uri); let mut t = Table::new(OWNER, &SPECS).expect("table"); let b = Batch { owner: OWNER, @@ -345,23 +365,30 @@ async fn history_refuses_an_absent_value_in_a_required_lane() { ..Batch::default() }; log.commit(&mut t, &b).await.expect("commit"); - // Silence half: the nullable lane's absent value is a valid history. + let before = log.history(&SPECS, 0, Col(0)).expect("history"); + assert_eq!( + before, + vec![Entry { + version: 1, + owner: OWNER, + change: Change::Set(Some(Value::U32(1))), + }] + ); + // The nullable lane's absent value is a valid history. assert_eq!( - log.history(&SPECS, 0, Col(1)).await.expect("history"), + log.history(&SPECS, 0, Col(1)).expect("history"), vec![Entry { version: 1, owner: OWNER, change: Change::Set(None), }] ); - // SET row 0, lane 0 (required), absent. + // SET row 0, lane 0 (required), absent — written around `commit`. 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" + "replay refuses it" ); + assert_eq!(log.history(&SPECS, 0, Col(0)).expect("history"), before); + assert_eq!(log.len(), 3, "the resident records are unchanged"); } From ffc56bf92b136a8a879c18048c658565d961e778 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 19:22:03 +0000 Subject: [PATCH 2/2] Board: D-STO-5 passes P-HIST-1 as a resident fold (0.72 ms) ISS-HIST-SCAN-COST resolved without an index or projection. Plan 5.2 records the decision: history folds over resident records; Lance is read once, by replay; only learning adds persistent state. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PPcakjk7ioQsnasmNDtc5z --- .claude/board/ISSUES.md | 2 +- .claude/board/LATEST_STATE.md | 17 +++++++++++++---- .claude/board/STATUS_BOARD.md | 2 +- .claude/plans/hubspo-integration-v1.md | 13 ++++++++++--- 4 files changed, 25 insertions(+), 9 deletions(-) diff --git a/.claude/board/ISSUES.md b/.claude/board/ISSUES.md index 5f191db..f5e81c2 100644 --- a/.claude/board/ISSUES.md +++ b/.claude/board/ISSUES.md @@ -5,7 +5,7 @@ ## ISS-HIST-SCAN-COST — property history scans the whole log -- **Status:** OPEN (D-STO-5 budget; correctness shipped) +- **Status:** RESOLVED 2026-10-10 without either lever: history folds over the resident log records instead of scanning Lance (0.72 ms vs 336 ms); Lance is read once, by replay. No index, no projection, no Lance filter expression. - **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, 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.) diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index a4dfef4..c79ad99 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -145,9 +145,13 @@ - `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, + No history store and no Lance read: one Quack program, `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` + log's resident records; only matches are decoded. `Log` keeps every record + it commits or replays resident (29 B each). Lance is the persistence + clock: `replay` reads it once, `commit` writes one version per batch. + Rule applied: one write costs about a million folds (lance-graph + 2026-09-24); a Lance scan is recovery cost. 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, @@ -156,8 +160,13 @@ 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. + as a batch whose version is not the next one. Resident: a second log + replayed from disk gives identical histories; a log whose records are not + at the table's version refuses to commit; a refused replay leaves the + resident records unchanged, so a corrupt record can never reach history. + Disable runs 2026-10-10, all red: resident-version check off, records + installed before replay validates, commit not appending to them. +- Timing (release, `examples/probe_history.rs`, 100k records, 1,000 versions × 1,000 updates, 1.2M log records, 2026-10-10): history **0.72 ms** median over 101 reads (0.64–1.99), inside P-HIST-1's < 20 ms; replay from Lance 532 ms, once at startup. Superseded: reading history from Lance took 336 ms (#24), 320 ms of it the scan. ## Board view (D-QRY-3, board half) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 50eaeda..19daddb 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. 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-5 | Property history via version log + P-HIST-1 probe | M | In PR | `Log::history` folds one Quack program over the resident log records; Lance is read once, by replay. **0.72 ms** for 1,001 entries over 1.2M records (release, `examples/probe_history.rs`, 101 reads, 2026-10-10) — passes P-HIST-1's < 20 ms; replay 532 ms once at startup. #24 shipped the read (Lance scan, 336 ms); this PR makes it resident | | 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 1598a85..e1fffad 100644 --- a/.claude/plans/hubspo-integration-v1.md +++ b/.claude/plans/hubspo-integration-v1.md @@ -861,8 +861,15 @@ 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. +sorting. + +*Decision 2026-10-10:* history is a fold over the log records held +resident, not a read of Lance. One write costs about a million folds +(lance-graph 2026-09-24), and a Lance scan is recovery cost; so `replay` +reads Lance once and every read after it folds over memory. Measured: +0.72 ms for 1,001 entries over 1.2M records, against 336 ms scanning Lance. +New persistent state comes only from learning, recorded after as one +merged, auditable outcome; derived reads like history write nothing. ### 5.3 Timeline @@ -1108,7 +1115,7 @@ with identical responses modulo ids/timestamps. |---|---|---| | P-CUSTOM-1 | custom property layout | fold over 500 custom properties × 1M rows within 2× of a standard property fold — **lanes pass (1.61× at 500), packed fails (18×)**, 2026-10-10 | | P-TEN-1 | dataset per tenant×type overhead | open + first read at 10k tenants < 50 ms p99 — **passes: 7.8 ms fresh / 5.0 ms shared session, local disk**, 2026-10-10 | -| P-HIST-1 | property history from version log | one property, one record, 1,000 versions < 20 ms — **fails, 2026-10-10: best 46 ms (address index, 101 changes, flat in N); version-log reads 3–147 s**. **Passes with the split-tunnel sparse delta (P-SPARSE-1): 16.9 ms indexed, 3.3–5.9 ms compacted** | +| P-HIST-1 | property history from version log | one property, one record, 1,000 versions < 20 ms — **fails, 2026-10-10: best 46 ms (address index, 101 changes, flat in N); version-log reads 3–147 s**. **Passes with the split-tunnel sparse delta (P-SPARSE-1): 16.9 ms indexed, 3.3–5.9 ms compacted**. **Passes as shipped (D-STO-5): 0.72 ms as a fold over resident log records** | | P-LIST-1 | active list incremental re-eval | delta of 1k changed rows on 10M population < 10 ms; allocation independent of population — **time passes (2.6 ms at 10M); allocation grows with fragments touched (~200 KiB each), not rows**, 2026-10-10. **Both pass with the sparse delta (P-SPARSE-1): 1.93 ms, 257 KiB flat at 10M** | | P-QDB-1 | Quack as the database for tickets / mail / delivery / documents | every read through a bound Quack program over Lance-resident columns, answers equal a plain-loop oracle, < 50 ms at 3M tickets — **passes with fk-ordered storage (all ≤ 25 ms at 3M except Q9 58.6 ms, whose fk is random); random fk layouts fail (253 ms / 1.9 s)**, 2026-10-10 | | P-DEDUP-1 | blocking recall | ≥ 99% of labelled duplicates share a block |