diff --git a/Cargo.lock b/Cargo.lock index 65042b7..8a04620 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1647,7 +1647,7 @@ dependencies = [ [[package]] name = "zu" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "zu-common", "zu-encoding", @@ -1663,7 +1663,7 @@ dependencies = [ [[package]] name = "zu-common" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "thiserror", ] @@ -1671,7 +1671,7 @@ dependencies = [ [[package]] name = "zu-encoding" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "ruzstd", "zu-common", @@ -1680,7 +1680,7 @@ dependencies = [ [[package]] name = "zu-exec" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "zu-common", "zu-query", @@ -1690,7 +1690,7 @@ dependencies = [ [[package]] name = "zu-query" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "crossbeam-deque", "zu-common", @@ -1701,7 +1701,7 @@ dependencies = [ [[package]] name = "zu-s3" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "crc32c", "object_store", @@ -1712,7 +1712,7 @@ dependencies = [ [[package]] name = "zu-sqlite" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "rusqlite", "zu-common", @@ -1722,7 +1722,7 @@ dependencies = [ [[package]] name = "zu-storage" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "zu-common", "zu-encoding", @@ -1731,7 +1731,7 @@ dependencies = [ [[package]] name = "zu-vector" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "zu-common", ] @@ -1739,7 +1739,7 @@ dependencies = [ [[package]] name = "zu-zu1" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "crc32c", "loom", diff --git a/Cargo.toml b/Cargo.toml index 051f3bc..7a8715a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,8 +18,8 @@ crate-type = ["cdylib"] # with (ADR 0002), so a revision is the honest way to say which one. # A local checkout is used instead with a `paths` override in # `.cargo/config.toml`, which is untracked on purpose. -zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "95c7c9909f3a2624515d27eb436da52936016960" } -zu-common = { git = "https://github.com/tamnd/zu", rev = "95c7c9909f3a2624515d27eb436da52936016960" } +zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "130f67db924bcd0f766ee814b0da2edae32150d4" } +zu-common = { git = "https://github.com/tamnd/zu", rev = "130f67db924bcd0f766ee814b0da2edae32150d4" } # `extension-module` is asked for by maturin, in pyproject.toml, and # not here. Only the build backend knows how an extension is linked on # the platform it is building for, and a crate that turns the feature diff --git a/README.md b/README.md index ddced15..da6625f 100644 --- a/README.md +++ b/README.md @@ -250,6 +250,24 @@ The markup is a table, a stylesheet and no script, so it survives `nbconvert`, a IPython is not a dependency. It is what a notebook already has, and nothing here imports it until `%load_ext` does. +## A second connection, made from the first + +`cursor()` is another connection to the same database, made from a connection rather than from a path. It is how a pool is written. + +```python +import zudb + +with zudb.connect("social.zu1") as conn: + with conn.cursor() as other: + other.execute("MATCH (p:person) RETURN p.name AS name").fetchall() +``` + +`duplicate()` is the same call under the name that says what it does; `cursor()` is what every other embedded database calls it and a caller who learned the word elsewhere should not have to learn another one here. It forks off the database the connection already holds rather than opening the file again, so it costs a schema load and no path lookup, and it works on a database in memory, where there is no path to open a second time. + +The two are connections in every sense rather than two names for one. Each has its own prepared statements, its own caches and its own transaction, so a thread taking one from a pool is not in whatever transaction the last borrower left open, and closing one does not close the other. What they share is the write side: they queue behind each other to write and each sees what the other has committed, which is what two connections to one file have always done. + +On an event loop it is awaited, because forking reaches the engine, and the new connection gets a thread of its own, which is the point: one connection runs one statement at a time, so two results in flight means two connections. In `zudb.dbapi` the name is `duplicate()` alone, since `cursor()` there is the thing PEP 249 means by it, which shares its connection and its transaction rather than making new ones. + ## Stopping a statement A statement that is running can be stopped two ways, and neither of them closes the connection: the session, its plans and its warm readers are all there afterwards, which is the whole difference between stopping a statement and starting again. @@ -333,7 +351,7 @@ Half of this package is compiled, which is the one thing an inspection cannot se ## 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, 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, `register` for putting a frame under a name a statement can match on and reading it where it lies, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it, `zudb.aio` for the same calls awaited on an event loop, results, nodes, rels and paths that draw themselves in a notebook with `%gql` and `%%gql` to run statements in one, `zudb.dbapi` for code written against PEP 249, `prepare`, `explain` and `profile` for a statement compiled once, the plan it would run and the plan it did, and `stream` for a result read as the engine makes it rather than after it has made all of it. Each one landed 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, `register` for putting a frame under a name a statement can match on and reading it where it lies, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it, `zudb.aio` for the same calls awaited on an event loop, results, nodes, rels and paths that draw themselves in a notebook with `%gql` and `%%gql` to run statements in one, `zudb.dbapi` for code written against PEP 249, `prepare`, `explain` and `profile` for a statement compiled once, the plan it would run and the plan it did, `stream` for a result read as the engine makes it rather than after it has made all of it, and `cursor()` for a second connection made from the first, which is how a pool is written. Each one landed with the tests that say it works. ## Wheels diff --git a/python/zudb/_zudb.pyi b/python/zudb/_zudb.pyi index 2ed6b77..b152a51 100644 --- a/python/zudb/_zudb.pyi +++ b/python/zudb/_zudb.pyi @@ -71,6 +71,12 @@ class Connection: def rows_read(self) -> int: """How many rows the statement running on this connection has read out of storage.""" + def cursor(self) -> Connection: + """Another connection to the same database, made from this one.""" + + def duplicate(self) -> Connection: + """The same call as `cursor()`, under the name that says what it does.""" + def interrupt(self) -> None: """Asks the statement running on this connection to stop.""" diff --git a/python/zudb/aio.py b/python/zudb/aio.py index 87c3f8d..468f73a 100644 --- a/python/zudb/aio.py +++ b/python/zudb/aio.py @@ -373,6 +373,43 @@ async def registered(self) -> list[str]: """ return await self._call(lambda: self._conn.registered) + def cursor(self) -> _Opening[AsyncConnection]: + """Another connection to the same database, made from this one. + + async with conn.cursor() as other: + rows = await other.execute("MATCH (p:person) RETURN p.name AS name") + + This is how a pool is written, and on an event loop it is the + only way to have two statements in flight at once, since one + connection runs one statement at a time by design. It forks off + the database this connection already holds rather than opening + the file again, so it works on a database in memory, where + there is nothing to open a second time. + + The new connection gets a thread of its own, which is the whole + point: two of them run at once. It gets its own prepared + statements, its own caches and its own transaction, and shares + only the write side, so the two queue behind each other to + write and each sees what the other has committed. + + Closing this one does not close that one. Forking reaches the + engine, so it is awaited like everything else here. + """ + return _Opening(self._cursor) + + def duplicate(self) -> _Opening[AsyncConnection]: + """The same call as `cursor()`, under the name that says what + it does. + """ + return _Opening(self._cursor) + + async def _cursor(self) -> AsyncConnection: + made = await self._call(self._conn.cursor) + # A thread of its own, started here rather than by the engine + # call, so that a fork that failed leaves no thread behind. + pool = ThreadPoolExecutor(max_workers=1, thread_name_prefix="zudb aio") + return AsyncConnection(made, pool) + async def close(self) -> None: """Closes the connection, frees what it held, and ends the thread it ran on. diff --git a/python/zudb/dbapi.py b/python/zudb/dbapi.py index 3664c3c..e33118d 100644 --- a/python/zudb/dbapi.py +++ b/python/zudb/dbapi.py @@ -711,11 +711,38 @@ def closed(self) -> bool: return self._conn.closed def cursor(self) -> Cursor: - """A new cursor on this connection.""" + """A new cursor on this connection. + + PEP 249's cursor, which shares this connection and this + transaction with every other cursor taken from it. The call + that gives back a second connection instead is `duplicate`, + and the two are worth telling apart: cursors here are a way of + reading several results at once, not a way of running several + statements at once. + """ if self.closed: raise InterfaceError("this connection is closed") return Cursor(self) + def duplicate(self) -> Connection: + """Another connection to the same database, made from this one. + + This is how a pool is written, and it is what the native + client spells `cursor()`, after the way every other embedded + database spells it. That name is taken here by the thing PEP + 249 means by it, so this one says what it does. + + The new connection has a transaction of its own and starts + outside one, whatever this connection is in the middle of. It + carries this connection's `autocommit`, since a pool handing + out connections that behaved differently from the one it was + seeded with would be a trap. + """ + if self.closed: + raise InterfaceError("this connection is closed") + with _translating(): + return Connection(self._conn.cursor(), autocommit=self._autocommit) + def commit(self) -> None: """Keeps what the transaction wrote and ends it. diff --git a/src/conn.rs b/src/conn.rs index b60ad8d..910b88a 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -356,6 +356,48 @@ impl Connection { register::registered(py, self) } + /// Another connection to the same database, made from this one. + /// + /// This is how a pool is written. `zudb.connect()` opens the file + /// again and looks the database up by path; this forks off the one + /// this connection already holds, which costs a schema load and no + /// lookup, and works on a database in memory, where there is no + /// path to open a second time. + /// + /// The two are connections in every sense rather than two names + /// for one. Each has its own prepared statements, its own caches + /// and its own transaction, so a thread that takes one from a pool + /// is not in whatever transaction the last borrower left open. What + /// they share is the write side: they queue behind each other to + /// write and each sees what the other has committed, which is what + /// two connections to one file have always done. + /// + /// `duplicate()` is the same call under the name that says what it + /// does. `cursor()` is what every other embedded database calls + /// it, and a caller who learned the word somewhere else should not + /// have to learn another one here. + fn cursor(&self, py: Python<'_>) -> PyResult { + let made = self + .engine(py, |conn| conn.duplicate())? + .map_err(|err| to_py_err(py, err))?; + Ok(Connection { + stop: made.interrupt(), + inner: Arc::new(Mutex::new(Some(made))), + runner: OnceLock::new(), + alive: AtomicBool::new(true), + feeding: Arc::new(stream::Feeding::new()), + path: self.path.clone(), + read_only: self.read_only, + memory: self.memory, + }) + } + + /// The same call as `cursor()`, under the name that says what it + /// does rather than the name every embedded database uses for it. + fn duplicate(&self, py: Python<'_>) -> PyResult { + self.cursor(py) + } + /// Asks the statement running on this connection to stop. /// /// The one call meant to be made from another thread while the diff --git a/tests/test_aio.py b/tests/test_aio.py index 99513ce..567443b 100644 --- a/tests/test_aio.py +++ b/tests/test_aio.py @@ -507,3 +507,44 @@ async def test_connect_with_no_path_is_a_database_in_memory( rows = await conn.execute("MATCH (p:person) RETURN p.name AS n") assert rows.fetchall() == [("ada",)] assert list(tmp_path.iterdir()) == [] + + +@run +async def test_a_cursor_is_another_connection_with_a_thread_of_its_own() -> None: + async with zudb.aio.connect() as conn: + await conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + async with conn.cursor() as other: + rows = await other.execute("MATCH (p:person) RETURN p.name AS n") + assert rows.fetchall() == [("ada",)] + await other.execute("INSERT (p:person {uid: 2, name: 'grace'})") + rows = await conn.execute("MATCH (p:person) RETURN count(*) AS n") + assert rows.fetchall() == [(2,)] + + +@run +async def test_two_connections_read_at_once_where_one_would_queue() -> None: + """The reason this call exists on an event loop: one connection + runs one statement at a time, so two results in flight means two + connections.""" + async with zudb.aio.connect() as conn: + await conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + other = await conn.duplicate() + try: + both = await asyncio.gather( + conn.execute("MATCH (p:person) RETURN p.name AS n"), + other.execute("MATCH (p:person) RETURN p.name AS n"), + ) + assert [rows.fetchall() for rows in both] == [[("ada",)], [("ada",)]] + finally: + await other.close() + + +@run +async def test_closing_one_leaves_the_other_open() -> None: + async with zudb.aio.connect() as conn: + await conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + other = await conn.cursor() + await conn.close() + rows = await other.execute("MATCH (p:person) RETURN p.name AS n") + assert rows.fetchall() == [("ada",)] + await other.close() diff --git a/tests/test_connection.py b/tests/test_connection.py index 33d6968..14d8dea 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -138,3 +138,58 @@ def test_a_database_on_disk_is_not_one_in_memory(tmp_path: Path) -> None: def test_a_database_in_memory_cannot_be_read_only() -> None: with pytest.raises(zudb.Error): zudb.connect(read_only=True) + + +def test_a_cursor_is_another_connection_to_the_same_database() -> None: + """The pool case, on a database in memory because that is the one + with no path to reopen: nothing but the shared write side could be + carrying the rows.""" + with zudb.connect() as conn: + conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + with conn.cursor() as other: + assert other.execute("MATCH (p:person) RETURN p.name AS n").fetchall() == [("ada",)] + other.execute("INSERT (p:person {uid: 2, name: 'grace'})") + assert conn.execute("MATCH (p:person) RETURN count(*) AS n").fetchall() == [(2,)] + + +def test_duplicate_is_the_same_call_under_the_other_name(tmp_path: Path) -> None: + with zudb.connect(tmp_path / "pooled.zu1") as conn: + conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + with conn.duplicate() as other: + assert other.execute("MATCH (p:person) RETURN p.name AS n").fetchall() == [("ada",)] + + +def test_a_cursor_says_what_the_connection_it_came_from_says(tmp_path: Path) -> None: + path = tmp_path / "said.zu1" + zudb.connect(path).close() + with zudb.connect(path, read_only=True) as conn, conn.cursor() as other: + assert other.path == conn.path + assert other.read_only is True + assert other.memory is False + assert other.closed is False + + +def test_a_cursor_outlives_the_connection_it_was_made_from() -> None: + """It is a connection and not a view of one, which is what makes a + pool able to hand its seed back.""" + conn = zudb.connect() + conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + other = conn.cursor() + conn.close() + assert other.execute("MATCH (p:person) RETURN p.name AS n").fetchall() == [("ada",)] + other.close() + + +def test_a_cursor_has_a_transaction_of_its_own() -> None: + with zudb.connect() as conn, conn.cursor() as other: + conn.execute("INSERT (p:person {uid: 1, name: 'ada'})") + with conn.transaction(): + assert conn.in_transaction is True + assert other.in_transaction is False + + +def test_a_closed_connection_makes_no_cursor() -> None: + conn = zudb.connect() + conn.close() + with pytest.raises(zudb.ProgrammingError, match="closed"): + conn.cursor() diff --git a/tests/test_dbapi.py b/tests/test_dbapi.py index 21a23c6..a913577 100644 --- a/tests/test_dbapi.py +++ b/tests/test_dbapi.py @@ -530,3 +530,35 @@ def test_connect_with_no_path_is_a_database_in_memory( cur.execute("MATCH (p:person) RETURN p.name AS n") assert cur.fetchall() == [("ada",)] assert list(tmp_path.iterdir()) == [] + + +def test_duplicate_is_another_connection_and_cursor_is_not() -> None: + """The two words this layer has to keep apart: a cursor shares the + connection it came from, and a duplicate is a connection of its + own.""" + with dbapi.connect() as conn: + cur = conn.cursor() + cur.execute("INSERT (p:person {uid: 1, name: 'ada'})") + conn.commit() + other = conn.duplicate() + assert other is not conn + assert other.zu is not conn.zu + theirs = other.cursor() + theirs.execute("MATCH (p:person) RETURN p.name AS n") + assert theirs.fetchall() == [("ada",)] + other.close() + assert conn.closed is False + + +def test_a_duplicate_carries_autocommit_and_not_the_transaction() -> None: + with dbapi.connect(autocommit=True) as conn: + other = conn.duplicate() + assert other.autocommit is True + other.close() + + +def test_a_closed_connection_duplicates_nothing() -> None: + conn = dbapi.connect() + conn.close() + with pytest.raises(dbapi.InterfaceError, match="closed"): + conn.duplicate() diff --git a/tests/test_readme.py b/tests/test_readme.py index 2c8a716..e07e51e 100644 --- a/tests/test_readme.py +++ b/tests/test_readme.py @@ -71,7 +71,7 @@ def run(program: str, where: Path) -> subprocess.CompletedProcess[str]: def test_the_readme_prints_programs_and_not_fragments() -> None: """The rule above holds: the page has both kinds and knows which.""" - assert len(programs()) == 5, "the README's whole programs" + assert len(programs()) == 6, "the README's whole programs" assert len(blocks("python")) > len(programs()), "and its fragments" @@ -120,3 +120,12 @@ def test_the_snippet_with_no_path_leaves_the_directory_empty(tmp_path: Path) -> assert done.returncode == 0, done.stderr assert done.stdout.strip() == "[('ada',)]" assert list(tmp_path.iterdir()) == [] + + +def test_the_cursor_snippet_runs_as_printed(tmp_path: Path) -> None: + """A second connection made from the first, on a database the + snippet creates as it goes.""" + snippet = program_with("conn.cursor()", "as other") + done = run(snippet, tmp_path) + assert done.returncode == 0, done.stderr + assert (tmp_path / "social.zu1").is_file()