From 65b77df2ab46988c97421f6bcf1c3ee810ce4e59 Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Tue, 18 Aug 2026 22:01:57 +0700 Subject: [PATCH] several statements as one unit of work Every statement already runs in a transaction of its own, so this is not what makes a write atomic. What it holds is the span: two statements are one unit, and the block rolls back when it raises, which is the failure worth writing it for since it is the one nobody wrote a handler for. `conn.transaction()` starts one and hands back a context manager that commits at the end of its block and rolls back when the block raised. It starts at the call rather than at the `with`, so a transaction that cannot start says so at the line that asked for one, and `commit()` and `rollback()` are there for a caller who would rather say when. A block whose transaction has already ended is left alone on the way out, and ending one twice is refused rather than ignored. `conn.in_transaction` answers which side of the block a program is on. The three statements underneath are the engine's own, so a read-only transaction refuses a write at the statement that writes and a second transaction is refused rather than nested. An appender is refused inside one, since its batches are commits of their own and a rollback does not take them back. The wrapper costs 5 microseconds for an empty transaction, so what it costs is what the engine charges, and on this machine that is more rather than less: 200 inserts cost 2.2 seconds each committing on its own and 3.3 seconds inside one transaction. The README says so. --- README.md | 20 +++- python/zudb/__init__.py | 2 + python/zudb/_zudb.pyi | 28 +++++ src/conn.rs | 73 +++++++++++- src/lib.rs | 2 + src/txn.rs | 143 +++++++++++++++++++++++ tests/test_transactions.py | 233 +++++++++++++++++++++++++++++++++++++ 7 files changed, 499 insertions(+), 2 deletions(-) create mode 100644 src/txn.rs create mode 100644 tests/test_transactions.py diff --git a/README.md b/README.md index 5c21c06..d42cbaf 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,24 @@ An appender nobody closed is the one mistake that cannot be reported where it ha On this machine 200,000 rows of an integer and a string take 11 ms to buffer through `append_rows` and 93 ms to flush, which is 1.9 million rows a second including the commit. The same rows one call at a time cost 32 ms of buffering, since a call is a call. Against `INSERT`, 2,000 rows take 32 seconds a row at a time and 28 ms through an appender, and the gap widens with the table because every `INSERT` is a commit and a fold. +## Several statements as one unit of work + +Every statement already runs in a transaction of its own, so this is not what makes a write atomic. What it holds is the span: two statements are one unit, and the block rolls back when it raises, which is the failure worth writing it for since it is the one nobody wrote a handler for. + +```python +with conn.transaction(): + conn.execute("INSERT (a:account {uid: 1, balance: 100})") + conn.execute("INSERT (b:account {uid: 2, balance: 0})") +``` + +It starts at the call rather than at the `with`, so a transaction that cannot start says so at the line that asked for one, and `commit()` and `rollback()` are there for a caller who would rather say when. A block whose transaction has already ended is left alone on the way out, which is what lets a program commit early and carry on, and ending one twice is refused rather than ignored, because the statements between the two are in neither of them. `conn.in_transaction` answers which side of the block a program is on. + +`transaction(read_only=True)` starts one that refuses to write, at the statement that writes rather than at the block that would have written. One transaction runs at a time and a second is refused rather than nested, since a rollback of an inner one would have to invent an answer for what it undoes. The three words underneath are `START TRANSACTION`, `COMMIT` and `ROLLBACK`, and they all still work written out. + +An appender is refused inside a transaction. Its batches are commits of their own and a rollback does not take them back, so an appender opened in a block would promise a span it is not in: load first, then transact. A statement that failed leaves the transaction running, because the engine does not end one on a failed statement and the block that opened it is still the thing that closes it. A connection closed with work uncommitted drops that work, which is the same answer the block would have given. + +The wrapper costs 5 microseconds for an empty transaction, so what it costs is what the engine charges. On this machine that is more rather than less: 200 `INSERT`s cost 2.2 seconds each committing on its own and 3.3 seconds inside one transaction, and reads cost the same either way. A transaction here is worth taking for the span it holds and not for the time it saves, and the v0 write path is where that number has to change. + ## Reading a result as columns A result is rows to iterate and columns to hand to something else. The columns go out over the Arrow C Data Interface, so pyarrow, pandas and polars each read the same buffers and none of them gets a Python object per cell. @@ -117,7 +135,7 @@ The stub is checked against the module it describes in CI: griffe reads the stub ## What works today -The list above is what this client is for. What it does so far is the core of it: `connect`, `execute` and `sql` with named parameters, results that iterate and fetch, values as Python objects both ways including dates, times, datetimes and durations, `Node`, `Rel` and `Path` as classes, `load` for building a graph with edges in it, an appender for growing one, every condition as an exception class carrying its code, its position and its documentation link, results as Arrow columns and as pandas and polars frames, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, and `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it. `register` is next, and each one lands with the tests that say it works. +The list above is what this client is for. What it does so far is the core of it: `connect`, `execute` and `sql` with named parameters, results that iterate and fetch, values as Python objects both ways including dates, times, datetimes and durations, `Node`, `Rel` and `Path` as classes, `load` for building a graph with edges in it, an appender for growing one, transactions as a context manager that commits at the end of a block and rolls back when it raises, every condition as an exception class carrying its code, its position and its documentation link, results as Arrow columns and as pandas and polars frames, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, and `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it. `register` is next, and each one lands with the tests that say it works. ## Wheels diff --git a/python/zudb/__init__.py b/python/zudb/__init__.py index 6b0d606..aec3fbe 100644 --- a/python/zudb/__init__.py +++ b/python/zudb/__init__.py @@ -22,6 +22,7 @@ Path, Rel, Result, + Transaction, __abi_version__, connect, load, @@ -44,6 +45,7 @@ "connect", "load", "Connection", + "Transaction", "Appender", "Result", "Node", diff --git a/python/zudb/_zudb.pyi b/python/zudb/_zudb.pyi index f88953c..03a053a 100644 --- a/python/zudb/_zudb.pyi +++ b/python/zudb/_zudb.pyi @@ -76,6 +76,13 @@ class Connection: def sql(self, statement: str, params: Mapping[str, Value] | None = None) -> Result: """The same call, named for the way it reads in a notebook.""" + def transaction(self, *, read_only: bool = False) -> Transaction: + """Starts a transaction and hands it back for a `with` block.""" + + @property + def in_transaction(self) -> bool: + """Whether an explicit transaction is running on this connection.""" + def appender(self, table: str) -> Appender: """Opens an appender on `table`, for loading rows into a database that already exists.""" @@ -86,6 +93,27 @@ class Connection: def __exit__(self, *_exception: object) -> bool: ... def __repr__(self) -> str: ... +class Transaction: + """A transaction that has been started and not yet ended.""" + + @property + def read_only(self) -> bool: + """Whether it was started `READ ONLY`.""" + + @property + def done(self) -> bool: + """Whether this transaction has already been committed or rolled back.""" + + def commit(self) -> None: + """Ends the transaction and keeps what it wrote.""" + + def rollback(self) -> None: + """Ends the transaction and throws away what it wrote.""" + + def __enter__(self) -> Transaction: ... + def __exit__(self, *_exception: object) -> bool: ... + def __repr__(self) -> str: ... + class Appender: """Rows on their way into a table, buffered until they are flushed.""" diff --git a/src/conn.rs b/src/conn.rs index 4e77a02..1a4a514 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -19,8 +19,9 @@ use zudb::{Config, Database, Interrupt}; use crate::appender::Appender; use crate::columns; -use crate::error::{closed, to_py_err}; +use crate::error::{closed, programming, to_py_err}; use crate::interrupt; +use crate::txn::Transaction; use crate::value::{Names, from_py, to_py}; /// What a capsule holding an Arrow stream is called. The name is part @@ -130,6 +131,52 @@ impl Connection { self.execute(py, statement, params) } + /// Starts a transaction and hands it back for a `with` block. + /// + /// Several statements as one unit of work: the block commits when + /// it ends and rolls back when it raises, which is the failure + /// worth writing this for, since it is the one nobody wrote a + /// handler for. + /// + /// It starts here rather than at the `with`, so a transaction that + /// cannot start says so at the line that asked for one. A + /// connection is inside one transaction at a time, and asking for a + /// second while the first is running is refused by the engine + /// rather than nested, because a transaction inside a transaction + /// would have to invent an answer for what a rollback of the inner + /// one undoes. + /// + /// `read_only=True` starts one that refuses to write, at the + /// statement that writes rather than at the block that would have + /// written. + #[pyo3(signature = (*, read_only = false))] + fn transaction(slf: Py, py: Python<'_>, read_only: bool) -> PyResult { + Transaction::start(py, slf, read_only) + } + + /// Whether an explicit transaction is running on this connection. + /// + /// The statements a program writes outside one are each a + /// transaction of their own, so this is false on a connection that + /// is writing perfectly well. It answers what is true between + /// statements, which is the only moment a caller can ask in: asking + /// while another thread is inside a statement waits for that + /// statement, since the answer belongs to the session and the + /// session is what the statement is holding. + #[getter] + fn in_transaction(&self, py: Python<'_>) -> PyResult { + if !self.alive.load(Ordering::Acquire) { + return Err(closed(py, "this connection")); + } + py.detach(|| { + self.inner.lock().ok().and_then(|mut held| { + held.as_mut() + .map(|conn| conn.session_mut().in_transaction()) + }) + }) + .ok_or_else(|| closed(py, "this connection")) + } + /// Opens an appender on `table`, for loading rows into a database /// that already exists. /// @@ -138,7 +185,20 @@ impl Connection { /// in columns and writes each batch as one commit. The table has to /// be there already, since an appender adds rows to a table and /// does not make one. + /// + /// It is refused inside a transaction. An appender's batches are + /// commits of their own and a rollback does not take them back, so + /// an appender opened inside a `with conn.transaction()` would + /// promise a span it is not in. Load first, then transact, or write + /// the rows as statements. fn appender(slf: Py, py: Python<'_>, table: &str) -> PyResult { + if slf.borrow(py).in_transaction(py)? { + return Err(programming( + py, + "an appender writes its own commits, which a rollback does not take \ + back, so one cannot be opened inside a transaction", + )); + } Appender::open(py, slf, table) } @@ -267,6 +327,17 @@ impl Connection { read_only, }) } + + /// Runs a statement that takes no parameters and gives back + /// nothing, which is what the three transaction words are. + /// + /// It goes through `execute` rather than around it, so a `COMMIT` + /// waits for the connection's lock, releases the GIL and feels a + /// `Ctrl-C` exactly as every other statement does. The result is + /// dropped, since the words return no columns. + pub(crate) fn run(&self, py: Python<'_>, statement: &str) -> PyResult<()> { + self.execute(py, statement, None).map(|_| ()) + } } /// The rows a statement gave back. diff --git a/src/lib.rs b/src/lib.rs index d0f54aa..23066ae 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,6 +19,7 @@ mod conn; mod error; mod interrupt; mod load; +mod txn; mod value; use std::path::PathBuf; @@ -55,6 +56,7 @@ fn _zudb(module: &Bound<'_, PyModule>) -> PyResult<()> { module.add_class::()?; module.add_class::()?; module.add_class::()?; + module.add_class::()?; module.add_class::()?; module.add_class::()?; module.add_class::()?; diff --git a/src/txn.rs b/src/txn.rs new file mode 100644 index 0000000..2649b44 --- /dev/null +++ b/src/txn.rs @@ -0,0 +1,143 @@ +//! Several statements as one unit of work. +//! +//! A statement written on its own already runs in a transaction of its +//! own, so this is not what makes a write atomic. What it holds is the +//! span: two statements are one unit, and the file keeps the state they +//! started from until one word or the other ends them. +//! +//! ```python +//! with conn.transaction(): +//! conn.execute("INSERT (a:account {uid: 1, balance: 100})") +//! conn.execute("INSERT (b:account {uid: 2, balance: 0})") +//! ``` +//! +//! The block commits when it ends and rolls back when it raises, which +//! is the whole reason this is a context manager rather than a pair of +//! methods: the failure that has to roll back is the one nobody wrote a +//! handler for. +//! +//! The three words underneath are `START TRANSACTION`, `COMMIT` and +//! `ROLLBACK`, and a caller who would rather write them can, on this +//! connection, today. What this adds is the rollback nobody remembers +//! to write, and a name for the state, so a program can ask whether it +//! is inside one rather than remembering. + +use pyo3::prelude::*; +use pyo3::types::PyTuple; + +use crate::conn::Connection; +use crate::error::programming; + +/// A transaction that has been started and not yet ended. +/// +/// Take one with `Connection.transaction`. It starts when it is taken, +/// so a transaction that cannot start says so at the line that asked +/// rather than at the `with` below it, and the statements that run +/// inside it are the ones written on the connection it came from. +#[pyclass(module = "zudb")] +pub struct Transaction { + /// The connection this runs on, held rather than borrowed, because + /// a transaction whose connection was collected would be a span + /// with nothing left to commit. + conn: Py, + /// Whether it was started `READ ONLY`, which the engine refuses a + /// write inside of at the statement that writes. + #[pyo3(get)] + read_only: bool, + /// Whether a `COMMIT` or a `ROLLBACK` has already run here. A + /// transaction ended inside its own block is ended, and the block + /// closing over it does nothing, which is what lets a caller commit + /// early and carry on. + done: bool, +} + +#[pymethods] +impl Transaction { + /// Ends the transaction and keeps what it wrote. + /// + /// Doing it twice is refused rather than ignored. A second commit + /// is a program that has lost track of where its transaction ends, + /// and the statements between the two are in neither of them. + fn commit(&mut self, py: Python<'_>) -> PyResult<()> { + self.end(py, "COMMIT") + } + + /// Ends the transaction and throws away what it wrote. + fn rollback(&mut self, py: Python<'_>) -> PyResult<()> { + self.end(py, "ROLLBACK") + } + + /// Whether this transaction has already been committed or rolled + /// back. + #[getter] + fn done(&self) -> bool { + self.done + } + + fn __enter__(slf: Py) -> Py { + slf + } + + /// Commits at the end of the block, and rolls back when the block + /// raised. + /// + /// A rollback that fails raises out of here in place of what was + /// being raised, which reads badly and is still right: the first + /// exception is the `__context__` of the second, and a program told + /// only about the failed statement would go on believing the + /// transaction unwound. + #[pyo3(signature = (*exception))] + fn __exit__(&mut self, py: Python<'_>, exception: &Bound<'_, PyTuple>) -> PyResult { + // The first of the three is the exception's type, and it is + // `None` when the block ended by falling off its own end. That + // is the whole question here, so the other two go unread. + let raised = exception.get_item(0).is_ok_and(|kind| !kind.is_none()); + if !self.done { + self.end(py, if raised { "ROLLBACK" } else { "COMMIT" })?; + } + // False, so an exception raised inside the block carries on out + // of it. A context manager that swallowed one would be a very + // quiet way to lose an error, and a quieter way to lose the + // rollback that answered it. + Ok(false) + } + + fn __repr__(&self) -> String { + let read_only = if self.read_only { " read only" } else { "" }; + let done = if self.done { ", done" } else { "" }; + format!("") + } +} + +impl Transaction { + /// Starts one, which is the statement that starts one. + pub fn start(py: Python<'_>, conn: Py, read_only: bool) -> PyResult { + let statement = if read_only { + "START TRANSACTION READ ONLY" + } else { + "START TRANSACTION" + }; + conn.borrow(py).run(py, statement)?; + Ok(Transaction { + conn, + read_only, + done: false, + }) + } + + fn end(&mut self, py: Python<'_>, statement: &str) -> PyResult<()> { + if self.done { + return Err(programming( + py, + "this transaction has already ended, and the statements after it \ + belong to no transaction of yours", + )); + } + self.conn.borrow(py).run(py, statement)?; + // Written after the statement rather than before it, so a + // commit the engine refused leaves a transaction a caller can + // still roll back. + self.done = true; + Ok(()) + } +} diff --git a/tests/test_transactions.py b/tests/test_transactions.py new file mode 100644 index 0000000..b046c1a --- /dev/null +++ b/tests/test_transactions.py @@ -0,0 +1,233 @@ +"""Several statements as one unit of work. + +A single statement is already atomic, so what these check is the span: +that the work between the two words arrives together, that it goes away +together when the block raises, and that a program can ask which of the +two it is in. + +The rollback tests read the database back through the connection that +wrote it and, where it matters, through one opened afterwards, because +a rollback that only convinced the connection that did it would be a +rollback that had not happened. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest +import zudb + + +def uids(conn: zudb.Connection) -> list[int]: + """The people in the database, in the order they were written.""" + return [uid for (uid,) in conn.execute("MATCH (p:person) RETURN p.uid AS uid")] + + +def test_a_block_that_ends_commits_what_it_wrote(social: zudb.Connection) -> None: + with social.transaction(): + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + social.execute("INSERT (p:person {uid: 50, name: 'barbara', score: 28.5})") + assert uids(social) == [10, 20, 30, 40, 50] + + +def test_a_block_that_raises_rolls_back_and_the_exception_carries_on( + social: zudb.Connection, +) -> None: + with pytest.raises(RuntimeError, match="halfway"): + with social.transaction(): + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + raise RuntimeError("halfway through") + assert uids(social) == [10, 20, 30] + + +def test_the_rollback_is_on_the_file_and_not_just_on_the_connection( + tmp_path: Path, +) -> None: + """Read back through a connection that was not there for it.""" + path = tmp_path / "rolled.zu1" + conn = zudb.connect(path) + conn.execute("INSERT (p:person {uid: 10, name: 'ada'})") + with pytest.raises(RuntimeError): + with conn.transaction(): + conn.execute("INSERT (p:person {uid: 20, name: 'grace'})") + raise RuntimeError("no") + conn.close() + + again = zudb.connect(path) + assert uids(again) == [10] + again.close() + + +def test_commit_and_rollback_can_be_called_rather_than_waited_for( + social: zudb.Connection, +) -> None: + txn = social.transaction() + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + txn.commit() + assert uids(social) == [10, 20, 30, 40] + + txn = social.transaction() + social.execute("INSERT (p:person {uid: 50, name: 'barbara', score: 28.5})") + txn.rollback() + assert uids(social) == [10, 20, 30, 40] + + +def test_a_block_that_committed_early_is_left_alone_on_the_way_out( + social: zudb.Connection, +) -> None: + """The rest of the block runs outside the transaction, deliberately.""" + with social.transaction() as txn: + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + txn.commit() + assert txn.done + assert not social.in_transaction + social.execute("INSERT (p:person {uid: 50, name: 'barbara', score: 28.5})") + assert uids(social) == [10, 20, 30, 40, 50] + + +def test_ending_a_transaction_twice_is_refused(social: zudb.Connection) -> None: + txn = social.transaction() + txn.commit() + with pytest.raises(zudb.ProgrammingError, match="already ended"): + txn.commit() + with pytest.raises(zudb.ProgrammingError, match="already ended"): + txn.rollback() + + +def test_in_transaction_says_which_side_of_the_block_a_program_is_on( + social: zudb.Connection, +) -> None: + assert not social.in_transaction + with social.transaction(): + assert social.in_transaction + assert not social.in_transaction + + +def test_a_statement_on_its_own_is_not_a_transaction_this_can_see( + social: zudb.Connection, +) -> None: + """Every statement runs in one of its own, and none of them is this.""" + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + assert not social.in_transaction + + +def test_a_read_only_transaction_reads_and_refuses_to_write( + social: zudb.Connection, +) -> None: + with social.transaction(read_only=True) as txn: + assert txn.read_only + assert uids(social) == [10, 20, 30] + with pytest.raises(zudb.TransactionError, match="READ ONLY"): + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + + +def test_one_transaction_at_a_time(social: zudb.Connection) -> None: + """Refused rather than nested, because a rollback of an inner one + would have to invent an answer for what it undoes.""" + with social.transaction(): + with pytest.raises(zudb.TransactionError, match="already running"): + social.transaction() + + +def test_the_three_words_still_work_written_out(social: zudb.Connection) -> None: + """This wraps the statements rather than replacing them.""" + social.execute("START TRANSACTION") + assert social.in_transaction + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + social.execute("ROLLBACK") + assert uids(social) == [10, 20, 30] + + +def test_a_commit_with_nothing_to_commit_is_refused(social: zudb.Connection) -> None: + with pytest.raises(zudb.TransactionError, match="no transaction"): + social.execute("COMMIT") + + +def test_a_statement_that_failed_leaves_the_transaction_to_its_owner( + social: zudb.Connection, +) -> None: + """The engine does not end a transaction on a failed statement, so + the block that opened it is still the thing that closes it.""" + with pytest.raises(RuntimeError): + with social.transaction(): + with pytest.raises(zudb.Error): + social.execute("MATCH (p:person) RETURN") + assert social.in_transaction + social.execute("INSERT (p:person {uid: 40, name: 'lynn', score: 31.0})") + raise RuntimeError("and now unwind all of it") + assert uids(social) == [10, 20, 30] + + +def test_an_appender_is_refused_inside_a_transaction(tmp_path: Path) -> None: + """Its batches are commits of their own, which no rollback reaches.""" + path = tmp_path / "graph.zu1" + zudb.load(path, nodes="person", columns={"uid": [10, 20, 30]}) + conn = zudb.connect(path) + with conn.transaction(): + with pytest.raises(zudb.ProgrammingError, match="own commits"): + conn.appender("person") + with conn.appender("person") as appender: + appender.append_row([40]) + assert uids(conn) == [10, 20, 30, 40] + conn.close() + + +def test_closing_the_connection_drops_what_was_uncommitted(tmp_path: Path) -> None: + path = tmp_path / "dropped.zu1" + conn = zudb.connect(path) + conn.execute("INSERT (p:person {uid: 10, name: 'ada'})") + conn.transaction() + conn.execute("INSERT (p:person {uid: 20, name: 'grace'})") + conn.close() + + again = zudb.connect(path) + assert uids(again) == [10] + again.close() + + +def test_a_commit_that_cannot_run_raises_out_of_the_block(tmp_path: Path) -> None: + """Better than a block that ends quietly on a transaction nobody + committed.""" + conn = zudb.connect(tmp_path / "closed.zu1") + conn.execute("INSERT (p:person {uid: 10, name: 'ada'})") + with pytest.raises(zudb.ProgrammingError, match="closed"): + with conn.transaction(): + conn.close() + + +def test_a_closed_connection_has_no_transactions(social: zudb.Connection) -> None: + social.close() + with pytest.raises(zudb.ProgrammingError, match="closed"): + social.transaction() + with pytest.raises(zudb.ProgrammingError, match="closed"): + assert social.in_transaction is None + + +def test_the_wrapper_costs_nothing_worth_measuring(social: zudb.Connection) -> None: + """Two statements and a Python object, which is what it should be. + + The budget is generous by a factor of ten against the 5 microseconds + this measures on a laptop, because it is here to catch a wrapper + that started waiting on something rather than to hold a number. + """ + best = float("inf") + for _ in range(3): + started = time.perf_counter() + for _ in range(200): + with social.transaction(): + pass + best = min(best, (time.perf_counter() - started) / 200) + assert best < 50e-6, f"an empty transaction took {best * 1e6:.0f} us" + + +def test_repr_says_which_kind_and_whether_it_is_over(social: zudb.Connection) -> None: + with social.transaction() as txn: + assert repr(txn) == "" + assert repr(txn) == "" + + txn = social.transaction(read_only=True) + assert repr(txn) == "" + txn.rollback() + assert repr(txn) == ""