diff --git a/.claude/board/ISSUES.md b/.claude/board/ISSUES.md index 793846b..5f191db 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, 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 - **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..a4dfef4 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -137,7 +137,27 @@ 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`, 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`) + +- `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. 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 79c8575..50eaeda 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. 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/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..68c7893 --- /dev/null +++ b/crates/hubspo-store/examples/probe_history.rs @@ -0,0 +1,101 @@ +//! 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 in {:.1} s", + 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; +} + +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; + let mut records = 0; + for _ in 0..7 { + 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" + ); +} diff --git a/crates/hubspo-store/src/log.rs b/crates/hubspo-store/src/log.rs index 631b85d..b310fed 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` | @@ -26,27 +27,35 @@ //! 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`, 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 +98,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 +110,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 +138,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 +169,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 +260,122 @@ 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 (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 spec = *specs + .get(usize::from(col.0)) + .ok_or(LogError::Corrupt("lane"))?; + 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 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()), + })) + }; + 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) + } + /// Rebuild the table of `owner` with `specs` from the log, batch by /// batch, through the same [`Table::apply`] a live writer uses. /// @@ -246,22 +392,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() { @@ -276,8 +414,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; @@ -320,11 +459,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 +467,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..73a0d61 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,163 @@ 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:?}"), + } +} + +/// 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" + ); +}