diff --git a/examples/sync-demo/src/colnDocType.ts b/examples/sync-demo/src/colnDocType.ts index a27f520a..2c352cfb 100644 --- a/examples/sync-demo/src/colnDocType.ts +++ b/examples/sync-demo/src/colnDocType.ts @@ -93,7 +93,7 @@ export function colnDocType< state: ColnState, body: (tx: ColnFfiTransaction) => void ): ColnState => { - const tx = state.store.beginTransaction() + const tx = state.store.transaction() try { const typedTx = new ffi.Transaction(state.store, tx) body(typedTx) diff --git a/packages/coln-js-runtime/package.json b/packages/coln-js-runtime/package.json index 3efbd11f..f6aebe19 100644 --- a/packages/coln-js-runtime/package.json +++ b/packages/coln-js-runtime/package.json @@ -16,7 +16,7 @@ "copy-license": "cp -r ../../LICENSES LICENSES", "build": "wasm-bodge build", "test": "npm run test:unit && npm run test:basic-ir", - "test:unit": "tsx --test tests/id-resolution.test.ts", + "test:unit": "tsx --test tests/id-resolution.test.ts tests/store-access.test.ts", "test:basic-ir": "npm run typecheck:basic-ir && npm run test:basic-ir:runtime", "typecheck:basic-ir": "tsx --tsconfig tests/basic-ir/tsconfig.json --test tests/basic-ir/typecheck.ts", "test:basic-ir:runtime": "tsx --tsconfig tests/basic-ir/tsconfig.json --test \"tests/basic-ir/*.test.ts\"" diff --git a/packages/coln-js-runtime/src/rust/handles.rs b/packages/coln-js-runtime/src/rust/handles.rs index 6446a8f6..ebf4297d 100644 --- a/packages/coln-js-runtime/src/rust/handles.rs +++ b/packages/coln-js-runtime/src/rust/handles.rs @@ -5,7 +5,7 @@ use coln_flir_rs::ir; use coln_store::{ commit::{chunk::Chunk, hash::CommitHash as StoreCommitHash}, - store::Store, + store::{Store, read::StoreRead}, table::RowId as StoreRowId, txn::{OwnedTransaction, RowHandle as StoreRowHandle}, }; @@ -43,6 +43,20 @@ pub struct TransactionHandle { pending_handles: Vec<(StoreRowHandle, JsValue)>, } +impl TransactionHandle { + fn read_tx(&self) -> Result<&OwnedTransaction, JsValue> { + self.tx + .as_ref() + .ok_or_else(|| js_error("txn has been committed")) + } + + fn write_tx(&mut self) -> Result<&mut OwnedTransaction, JsValue> { + self.tx + .as_mut() + .ok_or_else(|| js_error("txn has been committed")) + } +} + /* This function turns something like { @@ -76,12 +90,37 @@ fn resolve_value_id(js_value: &JsValue, row_id: RowId) -> Result<(), JsValue> { Ok(()) } +// TODO we might want to distinguish between read and write txns +// This is tricky because read transaction also needs an OwnedTransaction, which +// does not make much sense. #[wasm_bindgen] impl TransactionHandle { + // read methods + #[wasm_bindgen(js_name = scanTable)] + pub fn scan_table(&self, path: String) -> Result, JsValue> { + let path = ir::Path::from(path); + let rows = self + .read_tx()? + .scan_table(&path) + .map(|rows| rows.map(RowView::from).collect::>()) + .unwrap_or_default(); + + Ok(rows) + } + + #[wasm_bindgen(js_name = rowById)] + pub fn row_by_id(&self, path: String, row_id: RowRef) -> Result, JsValue> { + let path = ir::Path::from(path); + let row_id = StoreRowId::try_from(row_id).map_err(js_error)?; + + Ok(self.read_tx()?.row_by_id(&path, row_id).map(RowView::from)) + } + + // Write methods pub fn add(&mut self, path: String, values: Vec) -> Result { let path = ir::Path::from(path); let values = values.into_iter().map(|v| v.into()).collect::>(); - let handle = self.tx()?.add(&path, values).map_err(js_error)?; + let handle = self.write_tx()?.add(&path, values).map_err(js_error)?; let (tx_id, counter) = handle.pending_ids().map_err(js_error)?; let temp_id = Value::temp_id(tx_id, counter); @@ -116,14 +155,12 @@ impl TransactionHandle { Err((err, store)) => { self.recovered_store = Some(store); Err(js_error(format!( - "{err}; recover the store with TransactionHandle.takeStore()" + "{err}; recover the store with the read/write handle.take_store()" ))) } } } - // TODO adjust this API to not use take_store to recover but return the store - // after committing #[wasm_bindgen(js_name = takeStore)] pub fn take_store(&mut self) -> Result { if let Some(tx) = self.tx.take() { @@ -175,6 +212,18 @@ impl StoreHandle { self.store()?.json_ir().map_err(js_error) } + pub fn transaction(&mut self) -> Result { + let (store, pending_chunks) = self.owned_store()?; + Ok(TransactionHandle { + tx: Some(store.into_transaction()), + recovered_store: None, + pending_chunks, + + pending_handles: Vec::new(), + }) + } + + // TODO DUPLICATE CODE! Remove after redesigning the RW interface #[wasm_bindgen(js_name = scanTable)] pub fn scan_table(&self, path: String) -> Result, JsValue> { let path = ir::Path::from(path); @@ -187,6 +236,7 @@ impl StoreHandle { Ok(rows) } + // TODO DUPLICATE CODE! Remove after redesigning the RW interface #[wasm_bindgen(js_name = rowById)] pub fn row_by_id(&self, path: String, row_id: RowRef) -> Result, JsValue> { let path = ir::Path::from(path); @@ -195,8 +245,19 @@ impl StoreHandle { Ok(self.store()?.row_by_id(&path, row_id).map(RowView::from)) } - #[wasm_bindgen(js_name = beginTransaction)] - pub fn begin_transaction(&mut self) -> Result { + fn store(&self) -> Result<&Store, JsValue> { + match &self.state { + StoreHandleState::Uninitialized { .. } => { + Err(js_error("store handle has not been initialized")) + } + StoreHandleState::Ready { store, .. } => Ok(store), + StoreHandleState::Moved => Err(js_error( + "store handle has already been moved into a transaction", + )), + } + } + + fn owned_store(&mut self) -> Result<(Store, Vec>), JsValue> { let state = std::mem::replace(&mut self.state, StoreHandleState::Moved); let (store, pending_chunks) = match state { StoreHandleState::Ready { @@ -214,13 +275,69 @@ impl StoreHandle { } }; - Ok(TransactionHandle { - tx: Some(store.into_transaction()), - recovered_store: None, - pending_chunks, + Ok((store, pending_chunks)) + } +} - pending_handles: Vec::new(), - }) +impl StoreHandle { + fn ready(store: Store) -> Self { + Self::ready_with_pending(store, Vec::new()) + } + + fn ready_with_pending(store: Store, pending_chunks: Vec>) -> Self { + Self { + state: StoreHandleState::Ready { + store: Box::new(store), + pending_chunks, + }, + } + } + + // TODO this function is doing causal order delivery. This logic should NOT + // be here, and should be moved to somewhere else in the future. + fn apply_chunks(&mut self, chunk_bytes: Vec>) -> Result<(), String> { + match &mut self.state { + StoreHandleState::Uninitialized { chunks, has_root } => { + let decoded = chunk_bytes + .iter() + .map(|bytes| Chunk::decode(bytes)) + .collect::, _>>() + .map_err(|error| error.to_string())?; + let previous_len = chunks.len(); + let previously_had_root = *has_root; + *has_root |= decoded.iter().any(Chunk::is_root); + chunks.extend(chunk_bytes); + if *has_root { + match Store::try_from_commit_bytes(chunks.iter()) { + Ok((store, pending)) => { + self.state = StoreHandle::ready_with_pending(store, pending).state + } + Err(error) => { + chunks.truncate(previous_len); + *has_root = previously_had_root; + return Err(error.to_string()); + } + } + } + Ok(()) + } + StoreHandleState::Ready { + store, + pending_chunks, + } => { + pending_chunks.extend(chunk_bytes); + match store.apply_chunk_bytes(pending_chunks.iter().cloned()) { + Ok(pending) => { + *pending_chunks = pending; + Ok(()) + } + Err(error) => Err(error.to_string()), + } + } + StoreHandleState::Moved => { + Err("store handle has already been moved into a transaction".into()) + } + } } } @@ -292,89 +409,6 @@ impl CommitResult { .ok_or_else(|| js_error("commit result store has already been taken")) } } - -impl StoreHandle { - fn ready(store: Store) -> Self { - Self::ready_with_pending(store, Vec::new()) - } - - fn ready_with_pending(store: Store, pending_chunks: Vec>) -> Self { - Self { - state: StoreHandleState::Ready { - store: Box::new(store), - pending_chunks, - }, - } - } - - // TODO this function is doing causal order delivery. This logic should NOT - // be here, and should be moved to somewhere else in the future. - fn apply_chunks(&mut self, chunk_bytes: Vec>) -> Result<(), String> { - match &mut self.state { - StoreHandleState::Uninitialized { chunks, has_root } => { - let decoded = chunk_bytes - .iter() - .map(|bytes| Chunk::decode(bytes)) - .collect::, _>>() - .map_err(|error| error.to_string())?; - let previous_len = chunks.len(); - let previously_had_root = *has_root; - *has_root |= decoded.iter().any(Chunk::is_root); - chunks.extend(chunk_bytes); - if *has_root { - match Store::try_from_commit_bytes(chunks.iter()) { - Ok((store, pending)) => { - self.state = StoreHandle::ready_with_pending(store, pending).state - } - Err(error) => { - chunks.truncate(previous_len); - *has_root = previously_had_root; - return Err(error.to_string()); - } - } - } - Ok(()) - } - StoreHandleState::Ready { - store, - pending_chunks, - } => { - pending_chunks.extend(chunk_bytes); - match store.apply_chunk_bytes(pending_chunks.iter().cloned()) { - Ok(pending) => { - *pending_chunks = pending; - Ok(()) - } - Err(error) => Err(error.to_string()), - } - } - StoreHandleState::Moved => { - Err("store handle has already been moved into a transaction".into()) - } - } - } - - fn store(&self) -> Result<&Store, JsValue> { - match &self.state { - StoreHandleState::Uninitialized { .. } => { - Err(js_error("store handle has not been initialized")) - } - StoreHandleState::Ready { store, .. } => Ok(store), - StoreHandleState::Moved => Err(js_error( - "store handle has already been moved into a transaction", - )), - } - } -} - -impl TransactionHandle { - fn tx(&mut self) -> Result<&mut OwnedTransaction, JsValue> { - self.tx - .as_mut() - .ok_or_else(|| js_error("transaction has already been committed")) - } -} - #[cfg(test)] mod tests { use coln_flir_rs::ir::{ @@ -495,8 +529,12 @@ mod tests { let child = data.pop().expect("child commit"); handle.apply_chunks(vec![child]).expect("buffer child"); - let mut transaction = handle.begin_transaction().expect("begin transaction"); - handle = transaction.take_store().expect("abort transaction"); + let mut transaction = handle.transaction().expect("transaction"); + handle = transaction + .commit() + .expect("empty commit") + .take_store() + .expect("store"); handle.apply_chunks(data).expect("retry with parent"); let table = handle @@ -510,9 +548,9 @@ mod tests { #[test] fn active_transaction_can_return_its_store_without_committing() { let mut handle = StoreHandle::ready(source_store()); - let mut transaction = handle.begin_transaction().expect("transaction"); + let mut transaction = handle.transaction().expect("transaction"); transaction - .tx() + .write_tx() .expect("owned transaction") .add(&Path::from("T"), vec![84_i64.into()]) .expect("stage row"); diff --git a/packages/coln-js-runtime/src/ts/RowIdSet.ts b/packages/coln-js-runtime/src/ts/RowIdSet.ts index a89b3c11..759bfcb3 100644 --- a/packages/coln-js-runtime/src/ts/RowIdSet.ts +++ b/packages/coln-js-runtime/src/ts/RowIdSet.ts @@ -2,9 +2,15 @@ // // SPDX-License-Identifier: Apache-2.0 OR MIT -import * as ColnSet from "./ColnSet" -import { Value, StoreHandle, RowView, TransactionHandle, getRowRef } from "#wasm-bodge/bindings" -import { Tuple, tupleEqual } from "./tuple" +import * as ColnSet from "./ColnSet"; +import { + Value, + StoreHandle, + RowView, + TransactionHandle, + getRowRef, +} from "#wasm-bodge/bindings"; +import { Tuple, tupleEqual } from "./tuple"; export class View implements ColnSet.View { store: StoreHandle; @@ -18,7 +24,7 @@ export class View implements ColnSet.View { } has(x: Value): boolean { - const rowRef = getRowRef(x) + const rowRef = getRowRef(x); if (rowRef == undefined) return false; const row = this.store.rowById(this.path, rowRef); @@ -36,12 +42,36 @@ export class View implements ColnSet.View { export class Transaction extends View implements ColnSet.Transaction { transaction: TransactionHandle; - constructor(store_handle: StoreHandle, path: string, params: Tuple, transaction: TransactionHandle) { + constructor( + store_handle: StoreHandle, + path: string, + params: Tuple, + transaction: TransactionHandle, + ) { super(store_handle, path, params); this.transaction = transaction; } - + add(): Value { return this.transaction.add(this.path, this.params); } + + // TODO DUP CODE, but a transaction needs to have its own `rowById`, rather than + // calling on the store + // TODO will redesign the RW interface + has(x: Value): boolean { + const rowRef = getRowRef(x); + if (rowRef == undefined) return false; + + const row = this.transaction.rowById(this.path, rowRef); + + return row !== undefined && tupleEqual(row.values, this.params); + } + + // TODO DUP CODE + values(): Iterator { + const rows = this.transaction.scanTable(this.path); + + return rows.filter((row) => tupleEqual(row.values, this.params)).values(); + } } diff --git a/packages/coln-js-runtime/tests/basic-ir/helpers.ts b/packages/coln-js-runtime/tests/basic-ir/helpers.ts index be9e2327..d29fdc03 100644 --- a/packages/coln-js-runtime/tests/basic-ir/helpers.ts +++ b/packages/coln-js-runtime/tests/basic-ir/helpers.ts @@ -8,7 +8,7 @@ export function beginRealm( realm: RealmBindings, ) { let store = StoreHandle.fromTheory(JSON.stringify(realm.schema)); - const transaction = store.beginTransaction(); + const transaction = store.transaction(); const root = new realm.Transaction(store, transaction).root; return { diff --git a/packages/coln-js-runtime/tests/id-resolution.test.ts b/packages/coln-js-runtime/tests/id-resolution.test.ts index 20c376f2..83a03b9d 100644 --- a/packages/coln-js-runtime/tests/id-resolution.test.ts +++ b/packages/coln-js-runtime/tests/id-resolution.test.ts @@ -7,13 +7,11 @@ import test from "node:test"; import { StoreHandle } from "#wasm-bodge/bindings"; -import theory from "../../coln-compiler/test/golden/basic-ir/set.ts.output/TRealm.json" with { - type: "json", -}; +import theory from "../../coln-compiler/test/golden/basic-ir/set.ts.output/TRealm.json" with { type: "json" }; test("resolve pending row id to existing on commit", () => { const store = StoreHandle.fromTheory(JSON.stringify(theory)); - let txn = store.beginTransaction(); + let txn = store.transaction(); let vertex = txn.add("TRealm.V", []); assert.ok("pending" in vertex.value, "is pending"); diff --git a/packages/coln-js-runtime/tests/store-access.test.ts b/packages/coln-js-runtime/tests/store-access.test.ts index bba92f1a..aa43d83b 100644 --- a/packages/coln-js-runtime/tests/store-access.test.ts +++ b/packages/coln-js-runtime/tests/store-access.test.ts @@ -11,7 +11,7 @@ import theory from "../../coln-compiler/test/golden/graph.ts.output/GraphRealm.j test("Add vertices and edges to a store", () => { let store = StoreHandle.fromTheory(JSON.stringify(theory)); - let txn = store.beginTransaction(); + let txn = store.transaction(); // adding two vertices let v1 = txn.add("GraphRealm.V", []); @@ -26,13 +26,14 @@ test("Add vertices and edges to a store", () => { store = txn.takeStore(); throw e; } - let vs = store.scanTable("GraphRealm.V"); - let es = store.scanTable("GraphRealm.E"); - // We have two vertices and one edge + + txn = store.transaction(); + let vs = txn.scanTable("GraphRealm.V"); + let es = txn.scanTable("GraphRealm.E"); + // Committed rows are visible on a later transaction assert.equal(vs.length, 2); assert.equal(es.length, 1); - txn = store.beginTransaction(); let v3 = txn.add("GraphRealm.V", []); let v4 = txn.add("GraphRealm.V", []); @@ -72,5 +73,5 @@ test("Add vertices and edges to a store", () => { } const expected_edges = [e1, e3]; - assert.deepStrictEqual([...v1v2_edges].sort(), [...expected_edges].sort()) + assert.deepStrictEqual([...v1v2_edges].sort(), [...expected_edges].sort()); }); diff --git a/packages/coln-store/src/commit/hash.rs b/packages/coln-store/src/commit/hash.rs index 26532c29..e84e18cf 100644 --- a/packages/coln-store/src/commit/hash.rs +++ b/packages/coln-store/src/commit/hash.rs @@ -6,6 +6,7 @@ use std::fmt; /// The number of bytes in a commit hash. pub(crate) const HASH_SIZE: usize = 32; +pub(crate) static ALL_ZERO_HASH: CommitHash = CommitHash([0; HASH_SIZE]); #[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug)] pub struct CommitHash(pub [u8; HASH_SIZE]); diff --git a/packages/coln-store/src/store/mod.rs b/packages/coln-store/src/store/mod.rs index dd1729dd..e0b73b04 100644 --- a/packages/coln-store/src/store/mod.rs +++ b/packages/coln-store/src/store/mod.rs @@ -3,6 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 OR MIT pub mod error; +pub mod read; use std::collections::{BTreeSet, HashMap, HashSet}; @@ -21,6 +22,7 @@ use crate::solver::compile::{CompRule, CompileError}; use crate::solver::validate::RuleViolation; use crate::solver::{self}; use crate::store::error::{CommitApplyError, StoreError}; +use crate::store::read::StoreRead; use crate::table::{ CellValue, RowId, RowView, Table, TableMeta, TableOid, TableRef, TableSnapshot, ValidationError, }; @@ -159,10 +161,6 @@ impl Store { &self.rule_entries } - pub fn scan_table(&self, table_path: &ir::Path) -> Option + '_> { - self.table_at(table_path).map(|table| table.table_scan()) - } - pub fn json_ir(&self) -> Result { let realm = self.commits.root_commit()?.root_payload()?; Ok(serde_json::to_string(&realm).map_err(CodecError::from)?) @@ -173,8 +171,14 @@ impl Store { let canonical = self.rowing.canonical_id(&packed, &self.id_packer); Some(self.id_packer.unpack_row_id(canonical)) } +} + +impl StoreRead for Store { + fn scan_table(&self, table_path: &ir::Path) -> Option + '_> { + self.table_at(table_path).map(|table| table.table_scan()) + } - pub fn row_by_handle(&self, table: &ir::Path, row_handle: RowHandle) -> Option { + fn row_by_handle(&self, table: &ir::Path, row_handle: &RowHandle) -> Option { let row_id = row_handle.row_id().ok()?; let con_rowid = self.canonical_row_id(row_id)?; // replace the rowid in the row_handle so it stays canonical @@ -187,7 +191,7 @@ impl Store { // This function will canonicalise the row_id on read, but will not change it // See `row_by_handle` which will actually canonicalise the handle. // We need both because the TS FFI does not deal with handles. - pub fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { + fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { let row_id = self.canonical_row_id(row_id)?; self.table_at(table) .and_then(|table| table.row_at(table.row_position(row_id)?)) diff --git a/packages/coln-store/src/store/read.rs b/packages/coln-store/src/store/read.rs new file mode 100644 index 00000000..7a366979 --- /dev/null +++ b/packages/coln-store/src/store/read.rs @@ -0,0 +1,20 @@ +// SPDX-FileCopyrightText: 2026 Coln contributors +// +// SPDX-License-Identifier: Apache-2.0 OR MIT + +use coln_flir_rs::ir; + +use crate::{ + table::{RowId, RowView}, + txn::RowHandle, +}; + +pub trait StoreRead { + fn scan_table(&self, table: &ir::Path) -> Option + '_>; + + fn row_by_handle(&self, table: &ir::Path, handle: &RowHandle) -> Option; + + fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { + self.row_by_handle(table, &RowHandle::from_existing(row_id)) + } +} diff --git a/packages/coln-store/src/txn/inner.rs b/packages/coln-store/src/txn/inner.rs index f75b6920..8491dc15 100644 --- a/packages/coln-store/src/txn/inner.rs +++ b/packages/coln-store/src/txn/inner.rs @@ -8,7 +8,12 @@ use coln_flir_rs::ir; use tracing::info; use crate::{ - commit::{Commit, author::Author, hash::CommitHash, wire::CommitData}, + commit::{ + Commit, + author::Author, + hash::{self, CommitHash}, + wire::CommitData, + }, store::{Store, error::StoreError}, table::ValidationError, txn::{PendingOp, RowHandle, TempRowId, TxnCellValue, TxnId, TxnValue, timestamp::Timestamp}, @@ -119,6 +124,13 @@ impl TxnInner { pending_handles, .. } = self; + + // If we received an empty commit, then do nothing, return a all-zero hash + // TODO we could add an option to allow empty commit + if pending.is_empty() { + return Ok(hash::ALL_ZERO_HASH); + } + let cmt = Commit::from_commit_data( CommitData::new(deps, author, *timestamp.as_ref(), message, pending), |oid| store.table_meta(oid), diff --git a/packages/coln-store/src/txn/mod.rs b/packages/coln-store/src/txn/mod.rs index f28b8222..b54aba0a 100644 --- a/packages/coln-store/src/txn/mod.rs +++ b/packages/coln-store/src/txn/mod.rs @@ -3,6 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 OR MIT mod inner; +mod owned; mod row_handle; mod timestamp; @@ -10,10 +11,12 @@ use coln_flir_rs::ir; use crate::{ commit::hash::CommitHash, - store::{Store, error::StoreError}, + store::{Store, error::StoreError, read::StoreRead}, + table::{RowId, RowView}, }; use inner::TxnInner; +pub use owned::OwnedTransaction; pub(crate) use row_handle::{PendingOp, RowRef, TempRowId, TxnCellValue}; pub use row_handle::{RowHandle, TxnId, TxnValue}; @@ -61,40 +64,17 @@ impl<'a> Transaction<'a> { } } -pub struct OwnedTransaction { - inner: TxnInner, - store: Store, -} - -impl OwnedTransaction { - pub fn new(store: Store) -> Self { - let deps = store.commits().heads().copied().collect(); - Self { - inner: TxnInner::new(deps), - store, - } - } - - pub fn add( - &mut self, - table: &ir::Path, - values: Vec, - ) -> Result { - self.inner.add(&self.store, table, values) +impl StoreRead for Transaction<'_> { + fn scan_table(&self, table: &ir::Path) -> Option + '_> { + self.store.scan_table(table) } - pub fn abort(self) -> Store { - self.inner.abort(); - self.store + fn row_by_handle(&self, table: &ir::Path, handle: &RowHandle) -> Option { + self.store.row_by_handle(table, handle) } - // We need to return Store to the user for roll back purposes, so the Err variant must be large - #[allow(clippy::result_large_err)] - pub fn commit(mut self) -> Result<(CommitHash, Store), (StoreError, Store)> { - match self.inner.commit(&mut self.store) { - Ok(hash) => Ok((hash, self.store)), - Err(err) => Err((err, self.store)), - } + fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { + self.store.row_by_id(table, row_id) } } @@ -128,47 +108,6 @@ mod tests { } } - #[test] - fn owned_transaction_commits_and_returns_updated_store() { - let path = Path::from("T"); - let schema = table_schema(vec![int_col("c0")], None); - let mut store = Store::new(); - store - .create_table(path.clone(), schema) - .expect("create table"); - - let mut tx = OwnedTransaction::new(store); - tx.add(&path, vec![42_i64.into()]).expect("add"); - - let (_hash, committed) = tx.commit().expect("commit"); - assert_eq!(committed.table_at(&path).expect("T").row_count(), 1); - } - - #[test] - fn owned_transaction_add_validates_table_and_column_count() { - let path = Path::from("T"); - let schema = table_schema(vec![int_col("c0")], None); - let mut store = Store::new(); - store - .create_table(path.clone(), schema) - .expect("create table"); - - let mut tx = OwnedTransaction::new(store); - let err = tx - .add(&Path::from("missing"), vec![1_i64.into()]) - .unwrap_err(); - assert!(matches!( - err, - StoreError::Validation(ValidationError::UnknownTable { .. }) - )); - - let err = tx.add(&path, vec![1_i64.into(), 2_i64.into()]).unwrap_err(); - assert!(matches!( - err, - StoreError::Validation(ValidationError::ColumnCount { .. }) - )); - } - #[test] fn transaction_resolves_pending_row_references_with_commit_hash() { let nodes = Path::from("Nodes"); @@ -347,7 +286,7 @@ mod tests { // The first handle resolves through the store even if its id went // stale, and the read writes the canonical id back into the handle. let view = store - .row_by_handle(&term, first.clone()) + .row_by_handle(&term, &first) .expect("class row is stored"); assert_eq!(view.row_id, stored); assert_eq!(first.row_id().expect("finalized"), stored); @@ -390,4 +329,63 @@ mod tests { vec![second] ); } + + /// We provide read-committed isolation guarantee. So uncommitted data will + /// not be seen. + #[test] + fn transaction_store_read_sees_committed_rows_not_pending() { + let path = Path::from("T"); + let schema = table_schema(vec![int_col("c0")], None); + let mut store = Store::new(); + store + .create_table(path.clone(), schema) + .expect("create table"); + + let mut tx = store.transaction(); + tx.add(&path, vec![1_i64.into()]).expect("add"); + tx.commit().expect("commit"); + + let mut tx = store.transaction(); + let rows: Vec<_> = tx.scan_table(&path).expect("T").collect(); + assert_eq!(rows.len(), 1); + assert!(tx.row_by_id(&path, rows[0].row_id).is_some()); + assert!( + tx.row_by_handle(&path, &RowHandle::from_existing(rows[0].row_id)) + .is_some() + ); + + tx.add(&path, vec![2_i64.into()]).expect("add pending"); + assert_eq!(tx.scan_table(&path).expect("T").count(), 1); + tx.abort(); + } + + /// Committing an empty transaction does not modify the commit graph + #[test] + fn txn_empty_commit_not_added() { + let mut store = Store::new(); + let root = store.commits().root_commit().expect("root commit").hash(); + let heads: Vec<_> = store.commits().heads().copied().collect(); + let commits: Vec<_> = store + .commits() + .iter_topological() + .map(|c| c.hash()) + .collect(); + + let empty = store.transaction().commit().expect("empty commit"); + + assert!(!store.commits().contains(&empty)); + assert_eq!( + store.commits().root_commit().expect("root commit").hash(), + root + ); + assert_eq!(store.commits().heads().copied().collect::>(), heads); + assert_eq!( + store + .commits() + .iter_topological() + .map(|c| c.hash()) + .collect::>(), + commits + ); + } } diff --git a/packages/coln-store/src/txn/owned.rs b/packages/coln-store/src/txn/owned.rs new file mode 100644 index 00000000..355dc1b3 --- /dev/null +++ b/packages/coln-store/src/txn/owned.rs @@ -0,0 +1,152 @@ +// SPDX-FileCopyrightText: 2026 Coln contributors +// +// SPDX-License-Identifier: Apache-2.0 OR MIT + +use coln_flir_rs::ir; + +use crate::{ + commit::hash::CommitHash, + store::{Store, error::StoreError, read::StoreRead}, + table::{RowId, RowView}, +}; + +use super::{RowHandle, TxnInner, TxnValue}; + +pub struct OwnedTransaction { + inner: TxnInner, + store: Store, +} + +impl OwnedTransaction { + pub fn new(store: Store) -> Self { + let deps = store.commits().heads().copied().collect(); + Self { + inner: TxnInner::new(deps), + store, + } + } + + pub fn add( + &mut self, + table: &ir::Path, + values: Vec, + ) -> Result { + self.inner.add(&self.store, table, values) + } + + pub fn abort(self) -> Store { + self.inner.abort(); + self.store + } + + // We need to return Store to the user for roll back purposes, so the Err variant must be large + #[allow(clippy::result_large_err)] + pub fn commit(mut self) -> Result<(CommitHash, Store), (StoreError, Store)> { + match self.inner.commit(&mut self.store) { + Ok(hash) => Ok((hash, self.store)), + Err(err) => Err((err, self.store)), + } + } +} + +impl StoreRead for OwnedTransaction { + fn scan_table(&self, table: &ir::Path) -> Option + '_> { + self.store.scan_table(table) + } + + fn row_by_handle(&self, table: &ir::Path, handle: &RowHandle) -> Option { + self.store.row_by_handle(table, handle) + } + + fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { + self.store.row_by_id(table, row_id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ir::{BuiltinTy, ColType, ColumnEntry, EntityVariant, Path, Schema}; + use crate::table::ValidationError; + + fn table_schema(columns: Vec, primary_key: Option>) -> Schema { + Schema { + entity_variant: EntityVariant::Table, + columns, + primary_key, + } + } + + fn int_col(name: &str) -> ColumnEntry { + ColumnEntry { + path: Path::from(name), + col_type: ColType::BuiltinTy { + builtin_ty: BuiltinTy::BuiltinInt, + }, + } + } + + #[test] + fn owned_transaction_commits_and_returns_updated_store() { + let path = Path::from("T"); + let schema = table_schema(vec![int_col("c0")], None); + let mut store = Store::new(); + store + .create_table(path.clone(), schema) + .expect("create table"); + + let mut tx = OwnedTransaction::new(store); + tx.add(&path, vec![42_i64.into()]).expect("add"); + + let (_hash, committed) = tx.commit().expect("commit"); + assert_eq!(committed.table_at(&path).expect("T").row_count(), 1); + } + + #[test] + fn owned_transaction_add_validates_table_and_column_count() { + let path = Path::from("T"); + let schema = table_schema(vec![int_col("c0")], None); + let mut store = Store::new(); + store + .create_table(path.clone(), schema) + .expect("create table"); + + let mut tx = OwnedTransaction::new(store); + let err = tx + .add(&Path::from("missing"), vec![1_i64.into()]) + .unwrap_err(); + assert!(matches!( + err, + StoreError::Validation(ValidationError::UnknownTable { .. }) + )); + + let err = tx.add(&path, vec![1_i64.into(), 2_i64.into()]).unwrap_err(); + assert!(matches!( + err, + StoreError::Validation(ValidationError::ColumnCount { .. }) + )); + } + + #[test] + fn owned_transaction_store_read_sees_committed_rows_not_pending() { + let path = Path::from("T"); + let schema = table_schema(vec![int_col("c0")], None); + let mut store = Store::new(); + store + .create_table(path.clone(), schema) + .expect("create table"); + + let mut tx = OwnedTransaction::new(store); + tx.add(&path, vec![1_i64.into()]).expect("add"); + let (_hash, store) = tx.commit().expect("commit"); + + let mut tx = OwnedTransaction::new(store); + let rows: Vec<_> = tx.scan_table(&path).expect("T").collect(); + assert_eq!(rows.len(), 1); + assert!(tx.row_by_id(&path, rows[0].row_id).is_some()); + + tx.add(&path, vec![2_i64.into()]).expect("add pending"); + assert_eq!(tx.scan_table(&path).expect("T").count(), 1); + let _store = tx.abort(); + } +}