diff --git a/README.md b/README.md index 1eea0d1..0d8f36c 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,9 @@ variables](https://www.postgresql.org/docs/current/libpq-envars.html), such as (`CREATE`, `MERGE`, `SET`, `DELETE`, `REMOVE`, or `CALL` of a procedure) run to completion first and are buffered, because AGE performs writes while their rows are fetched. +- **Write statistics** come from PostgreSQL's per-transaction table statistics, read + before and after each query that can write, so such a query runs in a transaction + of its own. - **Neo4j procedures** that neo4j-browser relies on (`dbms.components`, `SHOW DATABASES`, `db.labels`, `db.schema.visualization`, `SHOW PROCEDURES`, ...) are emulated from the AGE catalog. @@ -519,8 +522,10 @@ addressing it in [`plans/`](plans/README.md): 1. `RETURN *`, `EXPLAIN` and `PROFILE` ([plan](plans/01-return-star-explain-profile.md)). -2. Write statistics: write queries report "(no changes, no records)" even when they - changed the graph ([plan](plans/02-write-statistics.md)). +2. Write statistics are only approximate for properties: `properties-set` counts the + nodes and relationships changed by `SET` and `REMOVE`, not their properties, and + leaves out the properties of `CREATE` and `MERGE` + ([plan](plans/02-write-statistics.md)). 3. Anything AGE does not support, for example multiple labels per node, pattern comprehensions, temporal and spatial types, indexes and constraints through Cypher ([plan](plans/03-age-unsupported-features.md)). diff --git a/boltage/backend.py b/boltage/backend.py index be8360c..2237239 100644 --- a/boltage/backend.py +++ b/boltage/backend.py @@ -29,6 +29,31 @@ NEO4J_VERSION = "5.26.0" SYSTEM_DB = "system" SYSTEM_LABELS = ("_ag_label_vertex", "_ag_label_edge") FETCH_SIZE = 1000 # rows fetched from a PostgreSQL cursor at a time +SYSTEM_UPDATE = {"system-updates": 1, "contains-system-updates": True} + +# Rows the current transaction has written to a graph's label tables, as PostgreSQL's +# statistics count them. AGE reports no counts of its own and fires no triggers. Neo4j's +# counters, in the order of the query below. AGE rewrites a whole row for a change to its +# properties, so properties-set is the number of nodes and relationships changed. +WRITE_COUNTERS = ("nodes-created", "nodes-deleted", "relationships-created", "relationships-deleted", + "labels-added", "properties-set") +WRITE_COUNTS = """SELECT + coalesce(sum(pg_stat_get_xact_tuples_inserted(id)) FILTER (WHERE base = '_ag_label_vertex'), 0)::bigint, + coalesce(sum(pg_stat_get_xact_tuples_deleted(id)) FILTER (WHERE base = '_ag_label_vertex'), 0)::bigint, + coalesce(sum(pg_stat_get_xact_tuples_inserted(id)) FILTER (WHERE base = '_ag_label_edge'), 0)::bigint, + coalesce(sum(pg_stat_get_xact_tuples_deleted(id)) FILTER (WHERE base = '_ag_label_edge'), 0)::bigint, + coalesce(sum(pg_stat_get_xact_tuples_inserted(id)) + FILTER (WHERE base = '_ag_label_vertex' AND name <> base), 0)::bigint, + coalesce(sum(pg_stat_get_xact_tuples_updated(id)), 0)::bigint +FROM ( + -- Not from ag_label: a role that may write to a graph need not be able to read that. + SELECT c.oid AS id, c.relname AS name, coalesce(p.relname, c.relname) AS base + FROM pg_catalog.pg_class c + JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace + LEFT JOIN pg_catalog.pg_inherits i ON i.inhrelid = c.oid + LEFT JOIN pg_catalog.pg_class p ON p.oid = i.inhparent + WHERE n.nspname = {graph} AND c.relkind = 'r' +) labels""" GRAPH_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.\-]{2,62}$") # Procedures emulated by boltage, reported through SHOW PROCEDURES. @@ -96,6 +121,7 @@ class Result: t_first: int = 0 cursor: Cursor | None = None position: int = 0 # the next row to send + stats: dict = field(default_factory=dict) # Neo4j's update counters, when there are any @property def buffered(self): @@ -165,6 +191,14 @@ def failure_from_pg(e: Exception) -> BoltFailure: return BoltFailure(code, message) +def write_stats(before, after) -> dict: + """Neo4j's update counters from two rows of ``WRITE_COUNTS``.""" + stats = {name: a - b for name, a, b in zip(WRITE_COUNTERS, after, before) if a != b} + if stats: + stats["contains-updates"] = True + return stats + + def _norm(query: str) -> str: return re.sub(r"\s+", " ", strip_statement(query)).strip() @@ -238,6 +272,24 @@ class Session: raise failure_from_pg(e) from e return await cur.fetchall() if cur.description else [] + async def _script(self, script, own_transaction) -> list: + """Run statements in one round trip; return the rows of those that return rows. + + ``own_transaction`` says that the script starts a transaction, to end if it fails. + """ + try: + cur = await self.conn.execute(script) + results = [] + while True: + if cur.description: + results.append(await cur.fetchall()) + if not cur.nextset(): + return results + except psycopg.Error as e: + if own_transaction: + await self.rollback() + raise failure_from_pg(e) from e + async def graphs(self): rows = await self._sql("SELECT name FROM ag_catalog.ag_graph ORDER BY name") return [r[0] for r in rows] @@ -291,12 +343,18 @@ class Session: # stops reading. A procedure called with CALL may write too. if columns and not writes and not has_keyword(text, "CALL"): return await self._stream(stmt, columns) - rows = await self._sql(stmt) - if columns: - rows = [[agtype.parse(v) for v in row] for row in rows] - else: - rows = [] - return Result(columns, rows, query_type="rw" if writes else "r") + # The counters only stay comparable within a transaction: outside of one, PostgreSQL + # can reset them between the two readings. + own_transaction = not self.in_transaction + counts = WRITE_COUNTS.format(graph=sql.Literal(graph).as_string(self.conn)) + script = f"{counts}; {stmt}; {counts}" + if own_transaction: + script = f"BEGIN; {script}; COMMIT" + before, rows, after = await self._script(script, own_transaction) + stats = write_stats(before[0], after[0]) + rows = [[agtype.parse(v) for v in row] for row in rows] if columns else [] + query_type = ("rw" if columns else "w") if writes or stats else "r" + return Result(columns, rows, query_type=query_type, stats=stats) async def _stream(self, stmt, columns) -> Result: """Open a cursor on a read query and fetch the first batch, which reports most errors.""" @@ -306,15 +364,7 @@ class Session: # usually read everything, so plan like for a plain query. script = (f"{'BEGIN; ' if own_transaction else ''}SET LOCAL cursor_tuple_fraction = 1; " f"DECLARE {name} NO SCROLL CURSOR FOR {stmt}; FETCH FORWARD {FETCH_SIZE} FROM {name}") - try: - cur = await self.conn.execute(script) - while cur.nextset(): - pass - batch = await cur.fetchall() - except psycopg.Error as e: - if own_transaction: - await self.rollback() - raise failure_from_pg(e) from e + batch = (await self._script(script, own_transaction))[-1] result = Result(columns, cursor=Cursor(self, name, own_transaction)) await result.add_batch(batch) return result @@ -430,7 +480,7 @@ class Session: raise BoltFailure("Neo.ClientError.Database.ExistingDatabaseFound", f"Failed to create the specified database '{name}': Database already exists.") await self._sql("SELECT ag_catalog.create_graph(%s)", [name]) - return Result([], query_type="s") + return Result([], query_type="s", stats=SYSTEM_UPDATE) async def _drop_graph(self, name, if_exists): graphs = {g.lower(): g for g in await self.graphs()} @@ -440,7 +490,7 @@ class Session: raise BoltFailure("Neo.ClientError.Database.DatabaseNotFound", f"Failed to delete the specified database '{name}': Database does not exist.") await self._sql("SELECT ag_catalog.drop_graph(%s, true)", [graphs[name.lower()]]) - return Result([], query_type="s") + return Result([], query_type="s", stats=SYSTEM_UPDATE) async def _labels(self, graph): rows = await self._sql( diff --git a/boltage/bolt.py b/boltage/bolt.py index ebebb10..46d6478 100644 --- a/boltage/bolt.py +++ b/boltage/bolt.py @@ -351,6 +351,8 @@ class BoltConnection: return del self.results[qid] metadata = {"type": result.query_type, "t_last": 0, "db": result.db} + if result.stats: + metadata["stats"] = result.stats if not self.session.in_transaction: metadata["bookmark"] = "boltage:bookmark" await self.success(metadata) diff --git a/plans/02-write-statistics.md b/plans/02-write-statistics.md index f29ef50..5f21e1f 100644 --- a/plans/02-write-statistics.md +++ b/plans/02-write-statistics.md @@ -1,5 +1,30 @@ # Write statistics +## Status + +Done. Queries that can write report `stats` in the final `PULL` summary, with the +node, relationship and label counters exact, and `properties-set` approximate as +described below. `CREATE DATABASE` and `DROP DATABASE` report system updates. + +The spike and testing changed some details below: + +- The counters do **not** reset at commit: they stay until PostgreSQL flushes them, + some time later. That makes the difference only reliable within one transaction, + which is still why each write runs in its own. Before, query and after are sent as + one script, so this costs no extra round trips. +- The counters are read with `pg_stat_get_xact_tuples_*`, not the + `pg_stat_xact_user_tables` view. +- Label tables are found through `pg_class` and `pg_inherits`, not `ag_label`: reading + `ag_label` made writes fail for roles without `SELECT` on it. +- `labels-removed` is not reported. Deleting a node does not count in Neo4j, and AGE + cannot remove a label. +- `SET n.a = 1, n.b = 2` is one update per node, so `properties-set` is the number of + nodes and relationships changed, not of properties. + +The price is about 1 ms per query that can write (2.1 ms to 3.0 ms for a single-row +`CREATE` on a local database). A statement that writes many rows pays this once. +Caching a graph's label tables could lower it. + ## Problem Neo4j reports what a write query changed in the `stats` entry of the final `PULL` diff --git a/plans/README.md b/plans/README.md index 5131880..2267ad5 100644 --- a/plans/README.md +++ b/plans/README.md @@ -28,9 +28,9 @@ plans small: transaction-local counters, streaming (06) needs a transaction to keep a server-side cursor open, and `PROFILE` (01) needs a savepoint. All three want `Session.run` to wrap auto-commit queries in `BEGIN ... COMMIT` instead of relying on the - connection's autocommit mode. Plan 06 only did this for read queries, which run - in a cursor's transaction; queries that can write still use autocommit, so 02 still - has to add it for them. + connection's autocommit mode. This is done: reads run in a cursor's transaction + (06), and queries that can write in one around the statistics (02). `Session._script` + runs such a transaction in one round trip. 3. **Unit tests.** The only test is the end-to-end `tests/smoke_test.py`. Most of this work is text rewriting in `cypher.py`, which is cheap to unit test without a database. Add `tests/test_cypher.py` (pytest) alongside the first plan that diff --git a/tests/smoke_test.py b/tests/smoke_test.py index 0205cf7..10ca3ce 100644 --- a/tests/smoke_test.py +++ b/tests/smoke_test.py @@ -137,13 +137,48 @@ MATCH ()-[]->() RETURN { name:'relationships', data: count(*)} AS result check("managed write transaction", s.execute_write(work) == 1) + def counts(tx): + first = tx.run("CREATE (:Temp {n: 4})").consume().counters + second = tx.run("MATCH (n:Temp {n: 4}) SET n.m = 1").consume().counters + return first, second + + first, second = s.execute_write(counts) + check("each query in a transaction reports its own counts", + first.nodes_created == 1 and first.properties_set == 0 + and second.nodes_created == 0 and second.properties_set == 1) + def read_your_writes(tx): tx.run("CREATE (:Temp {n: 3}) RETURN 1") return tx.run("UNWIND range(1, 1500) AS i MATCH (n:Temp {n: 3}) RETURN i").value() check("query sees an earlier write in its transaction", len(s.execute_write(read_your_writes)) == 1500) + with driver.session(database=GRAPH) as s: + c = s.run("CREATE (a:Counted {i: 1})-[:LINK {w: 1}]->(b:Counted {i: 2})").consume().counters + check("CREATE counts", c.nodes_created == 2 and c.relationships_created == 1 and c.labels_added == 2 + and c.contains_updates) + c = s.run("CREATE ({marker: 1})").consume().counters + check("a node without a label adds no label", c.nodes_created == 1 and c.labels_added == 0) + c = s.run("MATCH (n {marker: 1}) DELETE n").consume().counters + check("DELETE counts", c.nodes_deleted == 1) + summary = s.run("MATCH (n:Counted) SET n.j = 1, n.k = 2 RETURN n.j").consume() + check("SET counts changed nodes", summary.counters.properties_set == 2 and summary.query_type == "rw") + check("REMOVE counts", s.run("MATCH (n:Counted) REMOVE n.j").consume().counters.properties_set == 2) + check("MATCH of nothing with SET changes nothing", + not s.run("MATCH (n:Nothing) SET n.j = 1").consume().counters.contains_updates) + c = s.run("MERGE (n:Counted {i: 1})").consume().counters + check("MERGE that matches changes nothing", not c.contains_updates) + c = s.run("MERGE (n:Counted {i: 3})").consume().counters + check("MERGE that creates", c.nodes_created == 1) + summary = s.run("CREATE (:Counted {i: 4})").consume() + check("query without RETURN is a write", summary.query_type == "w") + c = s.run("MATCH (n:Counted) DETACH DELETE n").consume().counters + check("DETACH DELETE counts", c.nodes_deleted == 4 and c.relationships_deleted == 1) + summary = s.run("MATCH (n) RETURN count(n) AS c").consume() + check("a read changes nothing", not summary.counters.contains_updates and summary.query_type == "r") + with driver.session(database="system") as s: - s.run(f"DROP DATABASE {GRAPH}").consume() + c = s.run(f"DROP DATABASE {GRAPH}").consume().counters + check("system updates", c.system_updates == 1 and c.contains_system_updates) sys.exit(1 if check.failed else 0)