diff --git a/Dockerfile b/Dockerfile index 30240c5..9504ad3 100644 --- a/Dockerfile +++ b/Dockerfile @@ -70,6 +70,8 @@ ENV PEO4C_DEFAULT_GRAPH=neo4j \ PEO4C_BOLT_PORT=7687 \ PEO4C_PG_HOST=127.0.0.1 \ PEO4C_PG_PORT=5432 \ + PEO4C_PROPERTY_COLUMNS=true \ + PEO4C_PROPERTY_COLUMNS_INTERVAL=5 \ PEO4C_PGADMIN=true \ PEO4C_PGADMIN_PORT=5050 \ CONFIG_DISTRO_FILE_PATH=/opt/peo4c/pgadmin/config_distro.py diff --git a/README.md b/README.md index 4e3e249..bf8ed1f 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,8 @@ 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 (`PEO4C_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. + database call `create_graph` / `drop_graph`; AGE requires names of at least 3 characters, + and `peo4c` is reserved for the schema holding peo4c's SQL helpers. - **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. @@ -56,6 +57,35 @@ graph as a schema of the same name, with one table per label. - **Authentication** is passed straight through: every Bolt connection opens its own PostgreSQL connection with the credentials you log in with. +## 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`. peo4c adds a generated +column for every property key it finds, so you can use plain SQL: + +```sql +SELECT name, age FROM neo4j."Person" WHERE name LIKE 'A%'; +CREATE INDEX ON neo4j."Person" (name); +``` + +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 peo4c.sync_property_columns()` every few seconds +(`PEO4C_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`. + +- 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. + ## Configuration | Variable | Default | Purpose | @@ -66,6 +96,8 @@ graph as a schema of the same name, with one table per label. | `PEO4C_ADVERTISED_BOLT_PORT` | `7687` | Bolt port advertised to the browser; set it when you publish 7687 on another host port. | | `PEO4C_ADVERTISED_HOST` | `localhost` | Host reported in `SHOW DATABASES` and to clients that do not send a routing address. | | `PEO4C_LOG_LEVEL` | `INFO` | `DEBUG` logs every generated SQL statement. | +| `PEO4C_PROPERTY_COLUMNS` | `true` | Set to `false` to stop adding generated property columns. | +| `PEO4C_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. | | `PEO4C_PGADMIN` | `true` | Set to `false` to not start pgAdmin. | diff --git a/bridge/peo4c/__main__.py b/bridge/peo4c/__main__.py index e9fe5ab..8f5bd0a 100644 --- a/bridge/peo4c/__main__.py +++ b/bridge/peo4c/__main__.py @@ -7,6 +7,9 @@ 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 @@ -43,7 +46,19 @@ 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) - await asyncio.gather(*(s.serve_forever() for s in servers)) + tasks = [s.serve_forever() for s in servers] + if env("PEO4C_PROPERTY_COLUMNS", "false") == "true": + conninfo = make_conninfo( + host=env("PEO4C_PROPERTY_COLUMNS_PG_HOST", "/var/run/postgresql"), + port=config.pg_port, + dbname=config.pg_database, + user=env("PEO4C_PROPERTY_COLUMNS_PG_USER", env("POSTGRES_USER", "postgres")), + application_name="peo4c property columns", + ) + tasks.append(property_columns.sync_forever( + conninfo, float(env("PEO4C_PROPERTY_COLUMNS_INTERVAL", "5")))) + + await asyncio.gather(*tasks) if __name__ == "__main__": diff --git a/bridge/peo4c/backend.py b/bridge/peo4c/backend.py index 4ac938f..d1f82a0 100644 --- a/bridge/peo4c/backend.py +++ b/bridge/peo4c/backend.py @@ -337,6 +337,9 @@ 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 == "peo4c": + raise BoltFailure("Neo.ClientError.Statement.ArgumentError", + "The name 'peo4c' is reserved: it is the schema of peo4c's own SQL helpers.") if name in await self.graphs(): if if_not_exists: return Result([], query_type="s") diff --git a/bridge/peo4c/property_columns.py b/bridge/peo4c/property_columns.py new file mode 100644 index 0000000..d65098e --- /dev/null +++ b/bridge/peo4c/property_columns.py @@ -0,0 +1,47 @@ +# 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 +``peo4c.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("peo4c.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 peo4c.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/peo4c/property_columns.sql b/bridge/peo4c/property_columns.sql new file mode 100644 index 0000000..512643e --- /dev/null +++ b/bridge/peo4c/property_columns.sql @@ -0,0 +1,104 @@ +-- 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 peo4c.sync_property_columns() looks for them; the peo4c 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 peo4c; + +CREATE TABLE IF NOT EXISTS peo4c.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 peo4c.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 peo4c.sync_property_columns() +LANGUAGE plpgsql AS $$ +DECLARE + t record; + added text[]; +BEGIN + DELETE FROM peo4c.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 peo4c.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 peo4c.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 peo4c.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 7755598..191ae80 100644 --- a/tests/smoke_test.py +++ b/tests/smoke_test.py @@ -4,9 +4,13 @@ Usage: python tests/smoke_test.py [bolt-uri] [user] [password] Requires: pip install neo4j +Set PEO4C_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 @@ -117,6 +121,22 @@ MATCH ()-[]->() RETURN { name:'relationships', data: count(*)} AS result check("managed write transaction", s.execute_write(work) == 1) + dsn = os.environ.get("PEO4C_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()