From 773b52a63db2a2749dc7fcf1400ef1d1f1af432a Mon Sep 17 00:00:00 2001 From: Shuntian Liu Date: Tue, 1 Sep 2026 22:06:59 +0100 Subject: [PATCH 1/3] feat(coln-store): empty commit short circuits An empty commit will return early with a all-zero hash and not modify the commit graph. In the future we might want to introduce an option of something like --allow-empty. --- packages/coln-store/src/commit/hash.rs | 1 + packages/coln-store/src/store/mod.rs | 16 ++- packages/coln-store/src/store/read.rs | 20 ++++ packages/coln-store/src/txn/inner.rs | 14 ++- packages/coln-store/src/txn/mod.rs | 144 ++++++++++++----------- packages/coln-store/src/txn/owned.rs | 152 +++++++++++++++++++++++++ 6 files changed, 267 insertions(+), 80 deletions(-) create mode 100644 packages/coln-store/src/store/read.rs create mode 100644 packages/coln-store/src/txn/owned.rs 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(); + } +} From 1a145798e9be8a672b6565f57fa4afff859af908 Mon Sep 17 00:00:00 2001 From: Shuntian Liu Date: Wed, 2 Sep 2026 00:04:21 +0100 Subject: [PATCH 2/3] feat(coln-js-runtime): Add txn read methods Make txn able to read things from table. For now these are mostly dup code from the StoreHandle. I want to redesign the RW interface from coln-store so that might change. For now, we can read stuff inside a transaction. --- examples/sync-demo/src/colnDocType.ts | 2 +- packages/coln-js-runtime/package.json | 2 +- packages/coln-js-runtime/src/rust/handles.rs | 238 ++++++++++-------- packages/coln-js-runtime/src/ts/RowIdSet.ts | 42 +++- .../coln-js-runtime/tests/basic-ir/helpers.ts | 2 +- .../tests/id-resolution.test.ts | 6 +- .../tests/store-access.test.ts | 13 +- 7 files changed, 186 insertions(+), 119 deletions(-) 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()); }); From dcab21d78b0e3b44de0a4ee57f620716176994e4 Mon Sep 17 00:00:00 2001 From: Shuntian Liu Date: Wed, 2 Sep 2026 14:09:24 +0100 Subject: [PATCH 3/3] feat(coln-store): Add RW txn Add read/write transactions to coln-store, with typestate. Add convenience methods to do single-shot read/write on the store, internally they will start/finish a transaction for you. --- packages/coln-js-runtime/src/rust/handles.rs | 146 +++++++++++------- .../tests/store-access.test.ts | 52 +++++++ packages/coln-store/docs/primitives.md | 7 +- packages/coln-store/src/commit/author.rs | 6 + packages/coln-store/src/commit/pst.rs | 1 + packages/coln-store/src/solver/bind.rs | 1 + packages/coln-store/src/solver/validate.rs | 1 + packages/coln-store/src/store/mod.rs | 59 +++++-- packages/coln-store/src/store/read.rs | 20 --- packages/coln-store/src/store/tests.rs | 51 +++--- packages/coln-store/src/txn/inner.rs | 34 ++-- packages/coln-store/src/txn/mod.rs | 104 ++++++++++--- packages/coln-store/src/txn/owned.rs | 31 ++-- packages/coln-store/src/txn/rw.rs | 29 ++++ packages/coln-store/tests/test_path.rs | 1 + packages/coln-store/tests/test_subduction.rs | 2 +- 16 files changed, 370 insertions(+), 175 deletions(-) delete mode 100644 packages/coln-store/src/store/read.rs create mode 100644 packages/coln-store/src/txn/rw.rs diff --git a/packages/coln-js-runtime/src/rust/handles.rs b/packages/coln-js-runtime/src/rust/handles.rs index ebf4297d..e56a4d36 100644 --- a/packages/coln-js-runtime/src/rust/handles.rs +++ b/packages/coln-js-runtime/src/rust/handles.rs @@ -5,9 +5,12 @@ use coln_flir_rs::ir; use coln_store::{ commit::{chunk::Chunk, hash::CommitHash as StoreCommitHash}, - store::{Store, read::StoreRead}, + store::Store, table::RowId as StoreRowId, - txn::{OwnedTransaction, RowHandle as StoreRowHandle}, + txn::{ + OwnedTransaction, RowHandle as StoreRowHandle, + rw::{StoreRead, StoreWrite}, + }, }; use js_sys::Reflect; @@ -102,7 +105,7 @@ impl TransactionHandle { let rows = self .read_tx()? .scan_table(&path) - .map(|rows| rows.map(RowView::from).collect::>()) + .map(|rows| rows.into_iter().map(RowView::from).collect::>()) .unwrap_or_default(); Ok(rows) @@ -223,20 +226,19 @@ impl StoreHandle { }) } - // TODO DUPLICATE CODE! Remove after redesigning the RW interface + /// This is a convenience method that will start a transaction, do a scan and immediately close it #[wasm_bindgen(js_name = scanTable)] pub fn scan_table(&self, path: String) -> Result, JsValue> { let path = ir::Path::from(path); let rows = self .store()? .scan_table(&path) - .map(|rows| rows.map(RowView::from).collect::>()) + .map(|rows| rows.into_iter().map(RowView::from).collect::>()) .unwrap_or_default(); 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); @@ -245,6 +247,18 @@ impl StoreHandle { Ok(self.store()?.row_by_id(&path, row_id).map(RowView::from)) } + 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.store_mut()?.add(&path, values).map_err(js_error)?; + + let rid = handle.row_id().map_err(js_error)?; + let existing_id = Value::existing_id(rid.into()); + let js_value = serde_wasm_bindgen::to_value(&existing_id)?; + + Ok(js_value) + } + fn store(&self) -> Result<&Store, JsValue> { match &self.state { StoreHandleState::Uninitialized { .. } => { @@ -257,6 +271,18 @@ impl StoreHandle { } } + fn store_mut(&mut self) -> Result<&mut Store, JsValue> { + match &mut 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 { @@ -279,6 +305,60 @@ impl StoreHandle { } } +#[wasm_bindgen] +impl StoreHandle { + // For automerge-repo interfacing + + pub fn heads(&self) -> Result, JsValue> { + let heads = match &self.state { + StoreHandleState::Uninitialized { .. } => return Ok(Vec::new()), + StoreHandleState::Ready { store, .. } => store, + StoreHandleState::Moved => { + return Err(js_error( + "store handle has already been moved into a transaction", + )); + } + } + .heads() + .into_iter() + .map(CommitHash::from) + .collect::>(); + + Ok(heads) + } + + #[wasm_bindgen(js_name = commitChunksAfter)] + pub fn commit_chunks_after( + &self, + have_heads: Vec, + ) -> Result, JsValue> { + if matches!(self.state, StoreHandleState::Uninitialized { .. }) { + return Ok(Vec::new()); + } + let have_heads = have_heads + .into_iter() + .map(StoreCommitHash::try_from) + .collect::, _>>() + .map_err(js_error)?; + + let chunks = self + .store()? + .commit_chunks_after(&have_heads) + .into_iter() + .map(CommitChunk::from) + .collect::>(); + + Ok(chunks) + } + + #[wasm_bindgen(js_name = applyChunkBytes)] + pub fn apply_chunk_bytes(&mut self, chunk_bytes: JsValue) -> Result<(), JsValue> { + let chunk_bytes = + serde_wasm_bindgen::from_value::>>(chunk_bytes).map_err(js_error)?; + self.apply_chunks(chunk_bytes).map_err(js_error) + } +} + impl StoreHandle { fn ready(store: Store) -> Self { Self::ready_with_pending(store, Vec::new()) @@ -341,60 +421,6 @@ impl StoreHandle { } } -#[wasm_bindgen] -impl StoreHandle { - // For automerge-repo interfacing - - pub fn heads(&self) -> Result, JsValue> { - let heads = match &self.state { - StoreHandleState::Uninitialized { .. } => return Ok(Vec::new()), - StoreHandleState::Ready { store, .. } => store, - StoreHandleState::Moved => { - return Err(js_error( - "store handle has already been moved into a transaction", - )); - } - } - .heads() - .into_iter() - .map(CommitHash::from) - .collect::>(); - - Ok(heads) - } - - #[wasm_bindgen(js_name = commitChunksAfter)] - pub fn commit_chunks_after( - &self, - have_heads: Vec, - ) -> Result, JsValue> { - if matches!(self.state, StoreHandleState::Uninitialized { .. }) { - return Ok(Vec::new()); - } - let have_heads = have_heads - .into_iter() - .map(StoreCommitHash::try_from) - .collect::, _>>() - .map_err(js_error)?; - - let chunks = self - .store()? - .commit_chunks_after(&have_heads) - .into_iter() - .map(CommitChunk::from) - .collect::>(); - - Ok(chunks) - } - - #[wasm_bindgen(js_name = applyChunkBytes)] - pub fn apply_chunk_bytes(&mut self, chunk_bytes: JsValue) -> Result<(), JsValue> { - let chunk_bytes = - serde_wasm_bindgen::from_value::>>(chunk_bytes).map_err(js_error)?; - self.apply_chunks(chunk_bytes).map_err(js_error) - } -} - #[wasm_bindgen] impl CommitResult { #[wasm_bindgen(getter)] diff --git a/packages/coln-js-runtime/tests/store-access.test.ts b/packages/coln-js-runtime/tests/store-access.test.ts index aa43d83b..58834a0f 100644 --- a/packages/coln-js-runtime/tests/store-access.test.ts +++ b/packages/coln-js-runtime/tests/store-access.test.ts @@ -75,3 +75,55 @@ test("Add vertices and edges to a store", () => { const expected_edges = [e1, e3]; assert.deepStrictEqual([...v1v2_edges].sort(), [...expected_edges].sort()); }); + +// Do everything the same, but directly on the store +test("Change store directly", () => { + let store = StoreHandle.fromTheory(JSON.stringify(theory)); + + // adding two vertices + let v1 = store.add("GraphRealm.V", []); + let v2 = store.add("GraphRealm.V", []); + + // add an edge between them + let e1 = store.add("GraphRealm.E", [v1, v2]); + + let vs = store.scanTable("GraphRealm.V"); + let es = store.scanTable("GraphRealm.E"); + // Committed rows are visible on a later transaction + assert.equal(vs.length, 2); + assert.equal(es.length, 1); + + let v3 = store.add("GraphRealm.V", []); + let v4 = store.add("GraphRealm.V", []); + + let e2 = store.add("GraphRealm.E", [v3, v4]); + // Add a second edge between v1 and v2 + let e3 = store.add("GraphRealm.E", [v1, v2]); + + // Now find out all vertices connected to e2 + const e2_vs = []; + es = store.scanTable("GraphRealm.E"); + for (let e of es) { + if (valueEqual(e.rowId, e2)) { + e2_vs.push(e.values[0]); + e2_vs.push(e.values[1]); + } + } + + const expected = [v3, v4]; + assert.deepStrictEqual([...e2_vs].sort(), [...expected].sort()); + + // Find out all edges between v1 and v2 + const v1v2_edges = []; + for (let e of es) { + if ( + (valueEqual(e.values[0], v1) && valueEqual(e.values[1], v2)) || + (valueEqual(e.values[0], v2) && valueEqual(e.values[1], v1)) + ) { + v1v2_edges.push(e.rowId); + } + } + + const expected_edges = [e1, e3]; + assert.deepStrictEqual([...v1v2_edges].sort(), [...expected_edges].sort()); +}); diff --git a/packages/coln-store/docs/primitives.md b/packages/coln-store/docs/primitives.md index e17f889d..eeab49ad 100644 --- a/packages/coln-store/docs/primitives.md +++ b/packages/coln-store/docs/primitives.md @@ -54,13 +54,12 @@ struct RowView { values: Vec, } -Store::scan_table(table_path) -> Option> +Store::scan_table(table_path) -> Option> Store::row_by_id(table_path, row_id: RowId) -> Option ``` -The first one `scan` gives an iterator to the underlying table rows. A known -empty table returns `Some` with an empty iterator, while an unknown table returns -`None`. The second one is a member query scoped to a table path and indexed by +The first one `scan` returns the underlying table rows. A known empty table +returns `Some` with an empty vector, while an unknown table returns `None`. The second one is a member query scoped to a table path and indexed by the row_id. Note this `row_by_id` is only intended to support the most straightforward lookup right now, i.e. a table storing edb. Although I have not thought about this in detail, it is not intended for derived tables that might diff --git a/packages/coln-store/src/commit/author.rs b/packages/coln-store/src/commit/author.rs index b80d8314..e47b6d87 100644 --- a/packages/coln-store/src/commit/author.rs +++ b/packages/coln-store/src/commit/author.rs @@ -22,6 +22,12 @@ impl Author { } } +impl Default for Author { + fn default() -> Self { + Author::foo() + } +} + impl fmt::Display for Author { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}", self.to_hex_string()) diff --git a/packages/coln-store/src/commit/pst.rs b/packages/coln-store/src/commit/pst.rs index 3c2e923e..a39338f9 100644 --- a/packages/coln-store/src/commit/pst.rs +++ b/packages/coln-store/src/commit/pst.rs @@ -101,6 +101,7 @@ mod tests { use crate::commit::wire::CommitData; use crate::ir::{FlatRealm, Path, Schema, TableEntry}; use crate::table::CellValue; + use crate::txn::rw::StoreWrite; fn int_schema() -> Schema { Schema { diff --git a/packages/coln-store/src/solver/bind.rs b/packages/coln-store/src/solver/bind.rs index 6330cdbf..7b6b948b 100644 --- a/packages/coln-store/src/solver/bind.rs +++ b/packages/coln-store/src/solver/bind.rs @@ -151,6 +151,7 @@ mod tests { }, solver::compile::compile_rule, table::CellValue, + txn::rw::StoreWrite, }; fn int_ty() -> ColType { diff --git a/packages/coln-store/src/solver/validate.rs b/packages/coln-store/src/solver/validate.rs index 5e2518c5..19295e3d 100644 --- a/packages/coln-store/src/solver/validate.rs +++ b/packages/coln-store/src/solver/validate.rs @@ -156,6 +156,7 @@ mod tests { }, solver::compile::compile_rule, table::CellValue, + txn::rw::StoreWrite, }; fn int_ty() -> ColType { diff --git a/packages/coln-store/src/store/mod.rs b/packages/coln-store/src/store/mod.rs index e0b73b04..a48ad93e 100644 --- a/packages/coln-store/src/store/mod.rs +++ b/packages/coln-store/src/store/mod.rs @@ -3,7 +3,6 @@ // SPDX-License-Identifier: Apache-2.0 OR MIT pub mod error; -pub mod read; use std::collections::{BTreeSet, HashMap, HashSet}; @@ -22,11 +21,11 @@ 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, }; -use crate::txn::{OwnedTransaction, Transaction}; +use crate::txn::rw::{StoreRead, StoreWrite}; +use crate::txn::{OwnedTransaction, ReadOnly, ReadWrite, Transaction}; use crate::{op::Op, txn::RowHandle}; #[derive(Debug)] @@ -171,14 +170,19 @@ 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 + '_> { + pub(crate) fn scan_table_iter( + &self, + table_path: &ir::Path, + ) -> Option + '_> { self.table_at(table_path).map(|table| table.table_scan()) } - fn row_by_handle(&self, table: &ir::Path, row_handle: &RowHandle) -> Option { + pub(crate) fn row_by_handle_inner( + &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 @@ -191,13 +195,44 @@ impl StoreRead for 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. - fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { + pub(crate) fn row_by_id_inner(&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)?)) } } +// Autocommit method that opens up a txn, does a single operations +// then immediately closes the txn +impl StoreRead for Store { + fn scan_table(&self, table: &ir::Path) -> Option> { + let txn = self.ro_transaction(); + txn.scan_table(table) + } + + fn row_by_handle(&self, table: &ir::Path, handle: &RowHandle) -> Option { + self.ro_transaction().row_by_handle(table, handle) + } + + fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { + self.ro_transaction().row_by_id(table, row_id) + } +} + +impl StoreWrite for Store { + // Opens a single transaction and hands back the RowHandle, does not return hash + fn add( + &mut self, + table: &ir::Path, + values: Vec, + ) -> Result { + let mut txn = self.transaction(); + let h = txn.add(table, values); + txn.commit()?; + h + } +} + impl Store { // create stores from theory and transactions on stores @@ -245,8 +280,12 @@ impl Store { impl Store { // transactions - pub fn transaction(&mut self) -> Transaction<'_> { - Transaction::new(self) + pub fn ro_transaction(&self) -> Transaction> { + Transaction::>::new(self) + } + + pub fn transaction(&mut self) -> Transaction> { + Transaction::>::new(self) } pub fn into_transaction(self) -> OwnedTransaction { diff --git a/packages/coln-store/src/store/read.rs b/packages/coln-store/src/store/read.rs deleted file mode 100644 index 7a366979..00000000 --- a/packages/coln-store/src/store/read.rs +++ /dev/null @@ -1,20 +0,0 @@ -// 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/store/tests.rs b/packages/coln-store/src/store/tests.rs index 2570dd0a..fdbd212b 100644 --- a/packages/coln-store/src/store/tests.rs +++ b/packages/coln-store/src/store/tests.rs @@ -99,7 +99,10 @@ pub(crate) mod test_support { } use super::*; -use crate::ir::{BuiltinTy, ColType, ColumnEntry, EntityVariant, Path, RuleVariant, Schema}; +use crate::{ + ir::{BuiltinTy, ColType, ColumnEntry, EntityVariant, Path, RuleVariant, Schema}, + txn::rw::StoreWrite, +}; fn single_int_store() -> Store { let path = Path::from("T"); @@ -316,6 +319,16 @@ mod transactions { assert_eq!(t.cell_at(0, 0), Some(CellValue::Int(42))); } + #[test] + fn store_add_inserts_row() { + let path = Path::from("T"); + let mut store = single_int_store(); + + store.add(&path, vec![42_i64.into()]).expect("add row"); + + assert_eq!(store.table_at(&path).expect("T").row_count(), 1); + } + #[test] fn leaves_store_unchanged_when_rules_fail() { let theory = link_foreign_key_theory(); @@ -360,22 +373,13 @@ mod query { let path = Path::from("T"); let mut store = single_int_store(); - assert_eq!( - store - .scan_table(&path) - .expect("known table") - .collect::>(), - vec![] - ); + assert_eq!(store.scan_table(&path).expect("known table"), vec![]); assert!(store.scan_table(&Path::from("missing")).is_none()); let commit = commit_int(&mut store, 42); assert_eq!( - store - .scan_table(&path) - .expect("known table") - .collect::>(), + store.scan_table(&path).expect("known table"), vec![RowView { row_id: RowId { commit, counter: 0 }, values: vec![CellValue::Int(42)], @@ -612,8 +616,8 @@ mod rowing { ]) .expect("duplicates merge rather than failing the commit"); - let terms: Vec = store.scan_table(&Path::from("Term")).unwrap().collect(); - let plus: Vec = store.scan_table(&Path::from("Plus")).unwrap().collect(); + let terms: Vec = store.scan_table(&Path::from("Term")).unwrap(); + let plus: Vec = store.scan_table(&Path::from("Plus")).unwrap(); assert_eq!(terms.len(), 1); assert_eq!(plus.len(), 1); @@ -655,7 +659,7 @@ mod rowing { ]) .expect("duplicates merge rather than failing the commit"); - let terms: Vec = store.scan_table(&Path::from("Term")).unwrap().collect(); + let terms: Vec = store.scan_table(&Path::from("Term")).unwrap(); assert_eq!(terms.len(), 2); assert_eq!( store.row_by_id(&Path::from("Plus"), plus), @@ -694,9 +698,9 @@ mod rowing { .unwrap(); txn2.commit().unwrap(); - let terms: Vec = store.scan_table(&term_path).unwrap().collect(); - let plus: Vec = store.scan_table(&plus_path).unwrap().collect(); - let mult: Vec = store.scan_table(&mult_path).unwrap().collect(); + let terms: Vec = store.scan_table(&term_path).unwrap(); + let plus: Vec = store.scan_table(&plus_path).unwrap(); + let mult: Vec = store.scan_table(&mult_path).unwrap(); // The second commit adds no rows: every row it names is structurally // identical to one the first commit already stored. @@ -746,8 +750,8 @@ mod rowing { .expect("F(Term1, Term2)"); first.commit().expect("x is mapped only once"); - let terms_before = store.scan_table(&term).expect("Term").count(); - let f_before = store.scan_table(&f).expect("F").collect::>(); + let terms_before = store.scan_table(&term).expect("Term").len(); + let f_before = store.scan_table(&f).expect("F"); assert_eq!(terms_before, 3); assert_eq!(f_before.len(), 1); @@ -770,11 +774,8 @@ mod rowing { // The rejected commit rolls back whole, including Term(4), which was // legal on its own. - assert_eq!(store.scan_table(&term).expect("Term").count(), terms_before); - assert_eq!( - store.scan_table(&f).expect("F").collect::>(), - f_before - ); + assert_eq!(store.scan_table(&term).expect("Term").len(), terms_before); + assert_eq!(store.scan_table(&f).expect("F"), f_before); } } diff --git a/packages/coln-store/src/txn/inner.rs b/packages/coln-store/src/txn/inner.rs index 8491dc15..440f374b 100644 --- a/packages/coln-store/src/txn/inner.rs +++ b/packages/coln-store/src/txn/inner.rs @@ -98,33 +98,33 @@ impl TxnInner { self.add_cell_values(store, table, values) } - fn invalidate_handles(pending_handles: Vec, reason: &str) { - pending_handles - .into_iter() + fn invalidate_handles(&mut self, reason: &str) { + self.pending_handles + .iter() .for_each(|h| h.invalidate(reason)); } /// Finalize handles to the id the store actually kept: a row that was /// deduplicated against an existing class finalizes to that class's /// canonical id, not to the never-stored raw id. - fn finalize_handles(pending_handles: Vec, h: CommitHash, store: &Store) { - pending_handles.into_iter().for_each(|handle| { + fn finalize_handles(&mut self, h: CommitHash, store: &Store) { + self.pending_handles.iter().for_each(|handle| { handle.finalize(h, |rid| store.canonical_row_id(rid).unwrap_or(rid)) }); } - pub(super) fn commit(self, store: &mut Store) -> Result { - info!(op_count = self.pending.len(), "commit txn"); + pub(super) fn commit(&mut self, store: &mut Store) -> Result { let TxnInner { deps, author, pending, timestamp, message, - pending_handles, .. } = self; + info!(op_count = pending.len(), "commit txn"); + // 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() { @@ -132,13 +132,19 @@ impl TxnInner { } let cmt = Commit::from_commit_data( - CommitData::new(deps, author, *timestamp.as_ref(), message, pending), + CommitData::new( + std::mem::take(deps), + std::mem::take(author), + *timestamp.as_ref(), + message.take(), + std::mem::take(pending), + ), |oid| store.table_meta(oid), ); let cmt = match cmt { Ok(cmt) => cmt, Err(err) => { - Self::invalidate_handles(pending_handles, "txn commit encoding failed"); + self.invalidate_handles("txn commit encoding failed"); return Err(err.into()); } }; @@ -147,20 +153,20 @@ impl TxnInner { match store.apply_commit(cmt) { Ok(None) => { // Everything applied successfully - Self::finalize_handles(pending_handles, h, store); + self.finalize_handles(h, store); Ok(h) } Ok(Some(_)) => { unreachable!("commit a local transaction should always succeed"); } Err(err) => { - Self::invalidate_handles(pending_handles, "txn commit failed"); + self.invalidate_handles("txn commit failed"); Err(err) } } } - pub(super) fn abort(self) { - Self::invalidate_handles(self.pending_handles, "txn abort"); + pub(super) fn abort(&mut self) { + self.invalidate_handles("txn abort"); } } diff --git a/packages/coln-store/src/txn/mod.rs b/packages/coln-store/src/txn/mod.rs index b54aba0a..53b15c98 100644 --- a/packages/coln-store/src/txn/mod.rs +++ b/packages/coln-store/src/txn/mod.rs @@ -5,14 +5,16 @@ mod inner; mod owned; mod row_handle; +pub mod rw; mod timestamp; use coln_flir_rs::ir; use crate::{ commit::hash::CommitHash, - store::{Store, error::StoreError, read::StoreRead}, + store::{Store, error::StoreError}, table::{RowId, RowView}, + txn::rw::{StoreRead, StoreWrite}, }; use inner::TxnInner; @@ -20,28 +22,57 @@ pub use owned::OwnedTransaction; pub(crate) use row_handle::{PendingOp, RowRef, TempRowId, TxnCellValue}; pub use row_handle::{RowHandle, TxnId, TxnValue}; -pub struct Transaction<'a> { - inner: TxnInner, +pub struct ReadOnly<'a> { + store: &'a Store, +} +pub struct ReadWrite<'a> { store: &'a mut Store, } -impl<'a> Transaction<'a> { - pub fn new(store: &'a mut Store) -> Self { +pub trait Mode { + fn store(&self) -> &Store; +} + +impl<'a> Mode for ReadOnly<'a> { + fn store(&self) -> &Store { + self.store + } +} + +impl<'a> Mode for ReadWrite<'a> { + fn store(&self) -> &Store { + self.store + } +} + +pub struct Transaction { + inner: TxnInner, + mode: M, + // Need to know if txn is still open to implement Drop + // but not checking this in every txn because the type system ensures + // that no method can be called on a closed txn + open: bool, +} + +impl<'a> Transaction> { + pub(crate) fn new(store: &'a Store) -> Self { let deps = store.commits().heads().copied().collect(); Self { inner: TxnInner::new(deps), - store, + mode: ReadOnly { store }, + open: true, } } +} - // TODO this API is a bit awkward to use, clients have to call .into() all - // the time on their values - pub fn add( - &mut self, - table: &ir::Path, - values: Vec, - ) -> Result { - self.inner.add(self.store, table, values) +impl<'a> Transaction> { + pub(crate) fn new(store: &'a mut Store) -> Self { + let deps = store.commits().heads().copied().collect(); + Self { + inner: TxnInner::new(deps), + mode: ReadWrite { store }, + open: true, + } } // Used by the REPL only @@ -51,30 +82,52 @@ impl<'a> Transaction<'a> { table: &ir::Path, values: Vec, ) -> Result { - self.inner.add_internal(self.store, table, values) + self.inner.add_internal(self.mode.store, table, values) } - pub fn commit(self) -> Result { - self.inner.commit(self.store) + pub fn commit(mut self) -> Result { + let h = self.inner.commit(self.mode.store); + self.open = false; + h } + // pub fn commit_with(mut self, opts: CommitOptions) -> Result { ... } - pub fn abort(self) { + pub fn abort(mut self) { + self.open = false; self.inner.abort() } } -impl StoreRead for Transaction<'_> { - fn scan_table(&self, table: &ir::Path) -> Option + '_> { - self.store.scan_table(table) +impl StoreRead for Transaction { + fn scan_table(&self, table: &ir::Path) -> Option> { + self.mode + .store() + .scan_table_iter(table) + .map(|rows| rows.collect()) } fn row_by_handle(&self, table: &ir::Path, handle: &RowHandle) -> Option { - self.store.row_by_handle(table, handle) + self.mode.store().row_by_handle_inner(table, handle) } fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { - self.store.row_by_id(table, row_id) + self.mode.store().row_by_id_inner(table, row_id) + } +} + +impl StoreWrite for Transaction> { + fn add(&mut self, table: &ir::Path, values: Vec) -> Result { + self.inner.add(self.mode.store, table, values) + } +} + +impl Drop for Transaction { + fn drop(&mut self) { + if self.open { + // This is fine for RO txn, because there will be no handles + self.inner.abort(); + } } } @@ -203,6 +256,7 @@ mod tests { err, StoreError::Validation(ValidationError::InvalidRowHandle { .. }) )); + tx.abort(); assert_eq!(store.table_at(&nodes).expect("Nodes").row_count(), 0); } @@ -346,7 +400,7 @@ mod tests { tx.commit().expect("commit"); let mut tx = store.transaction(); - let rows: Vec<_> = tx.scan_table(&path).expect("T").collect(); + let rows = tx.scan_table(&path).expect("T"); assert_eq!(rows.len(), 1); assert!(tx.row_by_id(&path, rows[0].row_id).is_some()); assert!( @@ -355,7 +409,7 @@ mod tests { ); tx.add(&path, vec![2_i64.into()]).expect("add pending"); - assert_eq!(tx.scan_table(&path).expect("T").count(), 1); + assert_eq!(tx.scan_table(&path).expect("T").len(), 1); tx.abort(); } diff --git a/packages/coln-store/src/txn/owned.rs b/packages/coln-store/src/txn/owned.rs index 355dc1b3..cb13ef57 100644 --- a/packages/coln-store/src/txn/owned.rs +++ b/packages/coln-store/src/txn/owned.rs @@ -6,8 +6,9 @@ use coln_flir_rs::ir; use crate::{ commit::hash::CommitHash, - store::{Store, error::StoreError, read::StoreRead}, + store::{Store, error::StoreError}, table::{RowId, RowView}, + txn::rw::{StoreRead, StoreWrite}, }; use super::{RowHandle, TxnInner, TxnValue}; @@ -26,15 +27,7 @@ impl OwnedTransaction { } } - pub fn add( - &mut self, - table: &ir::Path, - values: Vec, - ) -> Result { - self.inner.add(&self.store, table, values) - } - - pub fn abort(self) -> Store { + pub fn abort(mut self) -> Store { self.inner.abort(); self.store } @@ -50,16 +43,22 @@ impl OwnedTransaction { } impl StoreRead for OwnedTransaction { - fn scan_table(&self, table: &ir::Path) -> Option + '_> { - self.store.scan_table(table) + fn scan_table(&self, table: &ir::Path) -> Option> { + self.store.scan_table_iter(table).map(|rows| rows.collect()) } fn row_by_handle(&self, table: &ir::Path, handle: &RowHandle) -> Option { - self.store.row_by_handle(table, handle) + self.store.row_by_handle_inner(table, handle) } fn row_by_id(&self, table: &ir::Path, row_id: RowId) -> Option { - self.store.row_by_id(table, row_id) + self.store.row_by_id_inner(table, row_id) + } +} + +impl StoreWrite for OwnedTransaction { + fn add(&mut self, table: &ir::Path, values: Vec) -> Result { + self.inner.add(&self.store, table, values) } } @@ -141,12 +140,12 @@ mod tests { let (_hash, store) = tx.commit().expect("commit"); let mut tx = OwnedTransaction::new(store); - let rows: Vec<_> = tx.scan_table(&path).expect("T").collect(); + let rows = tx.scan_table(&path).expect("T"); 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); + assert_eq!(tx.scan_table(&path).expect("T").len(), 1); let _store = tx.abort(); } } diff --git a/packages/coln-store/src/txn/rw.rs b/packages/coln-store/src/txn/rw.rs new file mode 100644 index 00000000..9aa2cf88 --- /dev/null +++ b/packages/coln-store/src/txn/rw.rs @@ -0,0 +1,29 @@ +// SPDX-FileCopyrightText: 2026 Coln contributors +// +// SPDX-License-Identifier: Apache-2.0 OR MIT + +use coln_flir_rs::ir; + +use crate::{ + store::error::StoreError, + table::{RowId, RowView}, + txn::{RowHandle, TxnValue}, +}; + +pub trait StoreRead { + // Return a vec for external world + // TODO we might want another version of the API which does vectorised processing model for query processing + 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)) + } +} + +pub trait StoreWrite { + // TODO this API is a bit awkward to use, clients have to call .into() all + // the time on their values + fn add(&mut self, table: &ir::Path, values: Vec) -> Result; +} diff --git a/packages/coln-store/tests/test_path.rs b/packages/coln-store/tests/test_path.rs index 3c2bbb30..7057b2aa 100644 --- a/packages/coln-store/tests/test_path.rs +++ b/packages/coln-store/tests/test_path.rs @@ -10,6 +10,7 @@ use coln_store::{ commit::pst, store::{Store, error::StoreError}, table::{CellValue, RowId}, + txn::rw::StoreWrite, }; use tracing_subscriber::EnvFilter; diff --git a/packages/coln-store/tests/test_subduction.rs b/packages/coln-store/tests/test_subduction.rs index 549fb82c..b86c7152 100644 --- a/packages/coln-store/tests/test_subduction.rs +++ b/packages/coln-store/tests/test_subduction.rs @@ -7,7 +7,7 @@ use std::{collections::BTreeSet, error::Error, net::SocketAddr, sync::Arc, time: use coln_flir_rs::ir::{ BuiltinTy, ColType, ColumnEntry, EntityVariant, FlatRealm, Path, Schema, TableEntry, }; -use coln_store::{commit::hash::CommitHash, store::Store, table::CellValue}; +use coln_store::{commit::hash::CommitHash, store::Store, table::CellValue, txn::rw::StoreWrite}; use future_form::Sendable; use sedimentree_core::{ blob::{Blob, verified::VerifiedBlobMeta},