diff --git a/Dockerfile b/Dockerfile index d914c1f..d951ae4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -70,8 +70,6 @@ ENV PEO_DEFAULT_GRAPH=neo4j \ PEO_BOLT_PORT=7687 \ PEO_PG_HOST=127.0.0.1 \ PEO_PG_PORT=5432 \ - PEO_PROPERTY_COLUMNS=true \ - PEO_PROPERTY_COLUMNS_INTERVAL=5 \ PEO_PGADMIN=true \ PEO_PGADMIN_PORT=5050 \ CONFIG_DISTRO_FILE_PATH=/opt/peo/pgadmin/config_distro.py diff --git a/README.md b/README.md index 96e6102..4ba359b 100644 --- a/README.md +++ b/README.md @@ -42,8 +42,7 @@ graph as a schema of the same name, with one table per label. - **Databases are AGE graphs.** Every AGE graph shows up as a Neo4j database, and `:use ` switches between them. A graph named `neo4j` is created on first start (`PEO_DEFAULT_GRAPH`). `CREATE DATABASE x` and `DROP DATABASE x` on the `system` - database call `create_graph` / `drop_graph`; AGE requires names of at least 3 characters, - and `peo` is reserved for the schema holding peo's SQL helpers. + database call `create_graph` / `drop_graph`; AGE requires names of at least 3 characters. - **Queries** are sent to AGE as `SELECT * FROM cypher('', $$ $$) AS (...)`. The bridge works out the result columns from the final `RETURN` clause and inlines query parameters as Cypher literals. @@ -60,31 +59,30 @@ graph as a schema of the same name, with one table per label. ## Querying the graph from SQL AGE stores each node label and relationship type as a table in the graph's schema, -with all properties in one `agtype` column, `properties`. peo adds a generated -column for every property key it finds, so you can use plain SQL: +with all properties in one `agtype` column, `properties`; `properties ->> 'key'::text` +returns a property as text. For typed columns, add them yourself, either as a view or +as a generated column: ```sql -SELECT name, age FROM neo4j."Person" WHERE name LIKE 'A%'; -CREATE INDEX ON neo4j."Person" (name); +CREATE VIEW neo4j.people AS + SELECT id, properties ->> 'name'::text AS name, (properties ->> 'age'::text)::bigint AS age + FROM neo4j."Person"; + +ALTER TABLE neo4j."Person" + ADD COLUMN age bigint GENERATED ALWAYS AS ((properties ->> 'age'::text)::bigint) STORED; ``` -AGE's Cypher writes do not fire PostgreSQL triggers, so new keys cannot be caught -the moment they are written. Instead the bridge calls -`CALL peo.sync_property_columns()` every few seconds -(`PEO_PROPERTY_COLUMNS_INTERVAL`), which you can also run yourself. It only scans -tables that had inserts or updates since its previous run, and adds all new keys of a -table with a single `ALTER TABLE`. Once a column exists, PostgreSQL keeps it in step -with every `CREATE`, `MERGE`, `SET` and `REMOVE`. +PostgreSQL keeps generated columns up to date on every Cypher `CREATE`, `MERGE`, `SET` +and `REMOVE`, and they can be indexed. Some things to keep in mind: -- Columns are `text`, the same as `properties->>'key'`: one key can hold values of - different types in different rows, and a failing cast would make writes fail. Cast - in your query, for example `age::int`. Lists and maps come out as JSON text. -- Keys that clash with an existing column (`id`, `properties`, `start_id`, `end_id`) - get no column. Use `properties->>'id'` for those. -- Adding a column rewrites the table, which locks it for the duration. On a large - table this blocks writes to that label for a while, but only once per new key. -- AGE does not allow dropping columns from label tables, so a column stays after its - key disappears from the data. +- A value that does not fit the column's type makes the Cypher write fail: with the + column above, `CREATE (:Person {age: 'old'})` is rejected. A view only fails when it + is read. +- AGE does not allow dropping columns from its label tables, so a generated column + cannot be removed again. +- Adding a stored generated column rewrites the table and locks it while doing so. +- AGE has no date or time types, so store those as ISO 8601 strings and cast them in + SQL. ## Configuration @@ -96,8 +94,6 @@ with every `CREATE`, `MERGE`, `SET` and `REMOVE`. | `PEO_ADVERTISED_BOLT_PORT` | `7687` | Bolt port advertised to the browser; set it when you publish 7687 on another host port. | | `PEO_ADVERTISED_HOST` | `localhost` | Host reported in `SHOW DATABASES` and to clients that do not send a routing address. | | `PEO_LOG_LEVEL` | `INFO` | `DEBUG` logs every generated SQL statement. | -| `PEO_PROPERTY_COLUMNS` | `true` | Set to `false` to stop adding generated property columns. | -| `PEO_PROPERTY_COLUMNS_INTERVAL` | `5` | Seconds between checks for new property keys. | | `PGADMIN_DEFAULT_EMAIL` | `admin@example.com` | pgAdmin login, created on first start. | | `PGADMIN_DEFAULT_PASSWORD` | `$POSTGRES_PASSWORD` | pgAdmin login password, set on first start. | | `PEO_PGADMIN` | `true` | Set to `false` to not start pgAdmin. | diff --git a/bridge/peo/__main__.py b/bridge/peo/__main__.py index 047ae8b..141bcab 100644 --- a/bridge/peo/__main__.py +++ b/bridge/peo/__main__.py @@ -7,9 +7,6 @@ import functools import logging import os -from psycopg.conninfo import make_conninfo - -from . import property_columns from .backend import Config from .bolt import handle_client from .web import HttpServer @@ -46,19 +43,7 @@ async def main(): servers.append(await asyncio.start_server(http.handle, "0.0.0.0", http_port)) log.info("neo4j-browser on http://localhost:%d/browser/", http_port) - tasks = [s.serve_forever() for s in servers] - if env("PEO_PROPERTY_COLUMNS", "false") == "true": - conninfo = make_conninfo( - host=env("PEO_PROPERTY_COLUMNS_PG_HOST", "/var/run/postgresql"), - port=config.pg_port, - dbname=config.pg_database, - user=env("PEO_PROPERTY_COLUMNS_PG_USER", env("POSTGRES_USER", "postgres")), - application_name="peo property columns", - ) - tasks.append(property_columns.sync_forever( - conninfo, float(env("PEO_PROPERTY_COLUMNS_INTERVAL", "5")))) - - await asyncio.gather(*tasks) + await asyncio.gather(*(s.serve_forever() for s in servers)) if __name__ == "__main__": diff --git a/bridge/peo/backend.py b/bridge/peo/backend.py index 12b9bfb..be7f328 100644 --- a/bridge/peo/backend.py +++ b/bridge/peo/backend.py @@ -337,9 +337,6 @@ class Session: raise BoltFailure("Neo.ClientError.Statement.ArgumentError", f"Invalid graph name '{name}': AGE graph names must be 3-63 characters " "of letters, digits, '_', '.' or '-', starting with a letter or '_'.") - if name == "peo": - raise BoltFailure("Neo.ClientError.Statement.ArgumentError", - "The name 'peo' is reserved: it is the schema of peo's own SQL helpers.") if name in await self.graphs(): if if_not_exists: return Result([], query_type="s") diff --git a/bridge/peo/property_columns.py b/bridge/peo/property_columns.py deleted file mode 100644 index 81a2857..0000000 --- a/bridge/peo/property_columns.py +++ /dev/null @@ -1,47 +0,0 @@ -# SPDX-FileCopyrightText: 2026 Bart Broere -# SPDX-License-Identifier: GPL-3.0-or-later -"""Keeps generated property columns on AGE label tables up to date. - -Installs the procedures in ``property_columns.sql`` and calls -``peo.sync_property_columns()`` every ``interval`` seconds. This needs a -connection that may alter every graph's tables, so it uses its own -(superuser) connection rather than the credentials of Bolt users. -""" - -import asyncio -import logging -from pathlib import Path - -import psycopg - -log = logging.getLogger("peo.property_columns") - -INSTALL_SQL = (Path(__file__).parent / "property_columns.sql").read_text() - - -def log_notice(diag): - level = logging.WARNING if diag.severity_nonlocalized == "WARNING" else logging.INFO - log.log(level, "%s", diag.message_primary) - - -async def sync_forever(conninfo, interval): - error = None - while True: - try: - async with await psycopg.AsyncConnection.connect(conninfo, autocommit=True) as conn: - conn.add_notice_handler(log_notice) - await conn.execute(INSTALL_SQL) - log.info("syncing property columns every %gs", interval) - error = None - while True: - await conn.execute("CALL peo.sync_property_columns()") - await asyncio.sleep(interval) - except psycopg.Error as e: - # PostgreSQL is still starting or restarted, or AGE is not - # installed: report each new problem once and keep retrying. - message = str(e).strip() - if message != error: - level = logging.DEBUG if isinstance(e, psycopg.OperationalError) else logging.WARNING - log.log(level, "property column sync unavailable, retrying: %s", message) - error = message - await asyncio.sleep(interval) diff --git a/bridge/peo/property_columns.sql b/bridge/peo/property_columns.sql deleted file mode 100644 index c8ee77f..0000000 --- a/bridge/peo/property_columns.sql +++ /dev/null @@ -1,104 +0,0 @@ --- SPDX-FileCopyrightText: 2026 Bart Broere --- SPDX-License-Identifier: GPL-3.0-or-later --- --- Generated columns for node and relationship properties. --- --- AGE stores all properties of a node or relationship in one agtype column, --- "properties". For every property key found in a label table, this adds a --- stored generated column with the same name: --- --- name text GENERATED ALWAYS AS (properties ->> 'name'::text) STORED --- --- so the graph can be queried from SQL as SELECT name FROM graph."Person". --- --- AGE's Cypher writes bypass the executor and fire no triggers, row or --- statement level, so new keys cannot be caught by a trigger. Instead --- CALL peo.sync_property_columns() looks for them; the peo bridge calls --- it every few seconds. It only scans label tables that saw inserts or updates --- since its previous run, according to the statistics collector, and commits --- after each table so a table rewrite never holds locks on other tables. --- Generated columns are kept up to date by AGE's CREATE, MERGE, SET and REMOVE. --- --- Columns are text, like ->> on json, because a property can hold values of --- different types in different rows and a failing cast would make writes fail. - -CREATE SCHEMA IF NOT EXISTS peo; - -CREATE TABLE IF NOT EXISTS peo.property_column_state ( - relid oid PRIMARY KEY, - changes bigint NOT NULL -- inserts + updates seen when the table was last scanned -); - --- Adds a generated column for every property key in tbl that has no column of --- that name yet, and returns the keys it added. Keys that are empty or longer --- than PostgreSQL's 63-byte identifier limit are skipped, as are keys that --- clash with an existing column such as id, start_id or end_id. -CREATE OR REPLACE FUNCTION peo.add_property_columns(tbl regclass) -RETURNS SETOF text LANGUAGE plpgsql AS $$ -DECLARE - new_keys text[]; -BEGIN - EXECUTE format($q$ - SELECT array_agg(key ORDER BY key) FROM ( - SELECT DISTINCT ag_catalog.agtype_to_text(k) AS key - FROM %s, ag_catalog.age_unnest(ag_catalog.age_keys(properties)) AS k - ) keys - WHERE key <> '' AND octet_length(key) < 64 - AND NOT EXISTS (SELECT FROM pg_catalog.pg_attribute - WHERE attrelid = $1 AND attname = key AND NOT attisdropped) - $q$, tbl) INTO new_keys USING tbl; - - IF new_keys IS NULL THEN - RETURN; - END IF; - - -- One ALTER TABLE for all keys, so the table is rewritten only once. - EXECUTE format('ALTER TABLE %s %s', tbl, ( - SELECT string_agg(format( - 'ADD COLUMN %I text GENERATED ALWAYS AS (properties OPERATOR(ag_catalog.->>) %L::text) STORED', - key, key), ', ') - FROM unnest(new_keys) AS key)); - - RETURN QUERY SELECT unnest(new_keys); -END -$$; - -CREATE OR REPLACE PROCEDURE peo.sync_property_columns() -LANGUAGE plpgsql AS $$ -DECLARE - t record; - added text[]; -BEGIN - DELETE FROM peo.property_column_state s - WHERE NOT EXISTS (SELECT FROM pg_catalog.pg_class WHERE oid = s.relid); - - FOR t IN - SELECT l.relation AS tbl, c.changes - FROM ag_catalog.ag_label AS l - CROSS JOIN LATERAL (SELECT pg_stat_get_tuples_inserted(l.relation) - + pg_stat_get_tuples_updated(l.relation) AS changes) AS c - LEFT JOIN peo.property_column_state AS s ON s.relid = l.relation - WHERE l.name NOT IN ('_ag_label_vertex', '_ag_label_edge') - AND s.changes IS DISTINCT FROM c.changes - LOOP - BEGIN - -- Adding a column rewrites the table under an exclusive lock; - -- rather than queue behind a long transaction, retry next run. - PERFORM set_config('lock_timeout', '2s', true); - SELECT array_agg(key) INTO added FROM peo.add_property_columns(t.tbl) AS key; - -- The rewrite counts as inserts in this transaction; include them - -- so the table is not scanned again for its own rewrite. - INSERT INTO peo.property_column_state AS s - VALUES (t.tbl, t.changes + pg_stat_get_xact_tuples_inserted(t.tbl) - + pg_stat_get_xact_tuples_updated(t.tbl)) - ON CONFLICT (relid) DO UPDATE SET changes = excluded.changes; - IF added IS NOT NULL THEN - RAISE NOTICE 'added property columns to %: %', t.tbl, array_to_string(added, ', '); - END IF; - EXCEPTION WHEN OTHERS THEN - RAISE WARNING 'could not add property columns to %: %', t.tbl, SQLERRM; - END; - COMMIT; - END LOOP; -END -$$; diff --git a/tests/smoke_test.py b/tests/smoke_test.py index 6f22bae..a178853 100644 --- a/tests/smoke_test.py +++ b/tests/smoke_test.py @@ -4,13 +4,9 @@ Usage: python tests/smoke_test.py [bolt-uri] [user] [password] Requires: pip install neo4j -Set PEO_SMOKE_PG_DSN (e.g. "host=localhost dbname=postgres") to also check the -generated property columns over SQL; that needs psycopg. """ -import os import sys -import time import neo4j from neo4j.graph import Node, Path, Relationship @@ -121,22 +117,6 @@ MATCH ()-[]->() RETURN { name:'relationships', data: count(*)} AS result check("managed write transaction", s.execute_write(work) == 1) - dsn = os.environ.get("PEO_SMOKE_PG_DSN") - if dsn: - import psycopg - - with psycopg.connect(dsn, user=user, password=password, autocommit=True) as pg: - query = f'SELECT name, age FROM "{GRAPH}"."Person" ORDER BY name' - for _ in range(30): - try: - rows = pg.execute(query).fetchall() - break - except psycopg.errors.UndefinedColumn: - time.sleep(0.5) - else: - rows = None - check("generated property columns", rows == [("Ann", "30"), ("Bob's", None)]) - with driver.session(database="system") as s: s.run(f"DROP DATABASE {GRAPH}").consume()