From ed89a30bcbf264b5d62e526adcbf00410d525730 Mon Sep 17 00:00:00 2001 From: Dimitri Yatsenko Date: Tue, 13 Oct 2026 15:37:32 -0500 Subject: [PATCH] fix(postgres): read primary and foreign keys from pg_catalog PostgreSQL's `information_schema.table_constraints` lists only constraints on tables the current user owns or holds a privilege other than SELECT on. A login with only SELECT therefore saw no primary key on any table: `Table.primary_key` was empty, `parents(primary=True)` returned nothing so `key_source` raised, and `len()` built `count(DISTINCT )`. MySQL was unaffected. The keys are now read from `pg_constraint` and `pg_attribute`, which every login can read, the same way `load_foreign_keys_sql` already did: - `get_primary_key_sql` (the heading, `Table.primary_key`) and `load_primary_keys_sql` (the dependency graph) select `contype = 'p'`, ordered by position within the constraint -- the same column order `information_schema` reported. - `get_foreign_keys_sql` had the same problem, worse: `constraint_column_usage` shows only tables the user owns. It also joined the two column lists on constraint name alone, so a multi-column foreign key came back as a cross product of its columns. It now returns one row per foreign-key column, pairing each with the column it references, as the MySQL adapter does. Nothing in the library calls it today; output column names are unchanged. The new integration test creates a SELECT-only login on each backend and checks that primary keys, headings, parents and children, the dependency graph, `key_source`, and `get_foreign_keys_sql` all match what the owner sees. Against the previous code all five PostgreSQL cases fail; MySQL passed before and after. Closes #1554. --- src/datajoint/adapters/postgres.py | 75 +++++----- tests/integration/test_read_only_login.py | 167 ++++++++++++++++++++++ 2 files changed, 208 insertions(+), 34 deletions(-) create mode 100644 tests/integration/test_read_only_login.py diff --git a/src/datajoint/adapters/postgres.py b/src/datajoint/adapters/postgres.py index 3a55d8eb8..36edd5056 100644 --- a/src/datajoint/adapters/postgres.py +++ b/src/datajoint/adapters/postgres.py @@ -791,52 +791,59 @@ def get_columns_sql(self, schema_name: str, table_name: str) -> str: f"ORDER BY c.ordinal_position" ) + # Key and constraint queries read the system catalogs (pg_constraint, + # pg_attribute) rather than information_schema. information_schema hides + # constraints on tables where the current user holds only SELECT, so a + # read-only login would see no primary keys and no foreign keys. The + # catalogs are readable by every login. + def get_primary_key_sql(self, schema_name: str, table_name: str) -> str: - """Query to get primary key columns.""" + """Query to get primary key columns, in key order.""" return ( - f"SELECT column_name FROM information_schema.key_column_usage " - f"WHERE table_schema = {self.quote_string(schema_name)} " - f"AND table_name = {self.quote_string(table_name)} " - f"AND constraint_name IN (" - f" SELECT constraint_name FROM information_schema.table_constraints " - f" WHERE table_schema = {self.quote_string(schema_name)} " - f" AND table_name = {self.quote_string(table_name)} " - f" AND constraint_type = 'PRIMARY KEY'" - f") " - f"ORDER BY ordinal_position" + f"SELECT a.attname AS column_name " + f"FROM pg_constraint c " + f"JOIN pg_class cl ON c.conrelid = cl.oid " + f"JOIN pg_namespace ns ON cl.relnamespace = ns.oid " + f"CROSS JOIN LATERAL unnest(c.conkey) WITH ORDINALITY AS cols(conkey, ord) " + f"JOIN pg_attribute a ON a.attrelid = cl.oid AND a.attnum = cols.conkey " + f"WHERE c.contype = 'p' " + f"AND ns.nspname = {self.quote_string(schema_name)} " + f"AND cl.relname = {self.quote_string(table_name)} " + f"ORDER BY cols.ord" ) def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str: """Query to get foreign key constraints.""" return ( - f"SELECT kcu.constraint_name, kcu.column_name, " - f"ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name " - f"FROM information_schema.key_column_usage AS kcu " - f"JOIN information_schema.constraint_column_usage AS ccu " - f" ON kcu.constraint_name = ccu.constraint_name " - f"WHERE kcu.table_schema = {self.quote_string(schema_name)} " - f"AND kcu.table_name = {self.quote_string(table_name)} " - f"AND kcu.constraint_name IN (" - f" SELECT constraint_name FROM information_schema.table_constraints " - f" WHERE table_schema = {self.quote_string(schema_name)} " - f" AND table_name = {self.quote_string(table_name)} " - f" AND constraint_type = 'FOREIGN KEY'" - f") " - f"ORDER BY kcu.constraint_name, kcu.ordinal_position" + f"SELECT c.conname AS constraint_name, a1.attname AS column_name, " + f"cl2.relname AS foreign_table_name, a2.attname AS foreign_column_name " + f"FROM pg_constraint c " + f"JOIN pg_class cl1 ON c.conrelid = cl1.oid " + f"JOIN pg_namespace ns1 ON cl1.relnamespace = ns1.oid " + f"JOIN pg_class cl2 ON c.confrelid = cl2.oid " + f"CROSS JOIN LATERAL unnest(c.conkey, c.confkey) WITH ORDINALITY AS cols(conkey, confkey, ord) " + f"JOIN pg_attribute a1 ON a1.attrelid = cl1.oid AND a1.attnum = cols.conkey " + f"JOIN pg_attribute a2 ON a2.attrelid = cl2.oid AND a2.attnum = cols.confkey " + f"WHERE c.contype = 'f' " + f"AND ns1.nspname = {self.quote_string(schema_name)} " + f"AND cl1.relname = {self.quote_string(table_name)} " + f"ORDER BY c.conname, cols.ord" ) def load_primary_keys_sql(self, schemas_list: str, like_pattern: str) -> str: """Query to load all primary key columns across schemas.""" - tab_expr = "'\"' || kcu.table_schema || '\".\"' || kcu.table_name || '\"'" + tab_expr = "'\"' || ns.nspname || '\".\"' || cl.relname || '\"'" return ( - f"SELECT {tab_expr} as tab, kcu.column_name " - f"FROM information_schema.key_column_usage kcu " - f"JOIN information_schema.table_constraints tc " - f"ON kcu.constraint_name = tc.constraint_name " - f"AND kcu.table_schema = tc.table_schema " - f"WHERE kcu.table_name NOT LIKE {like_pattern} " - f"AND kcu.table_schema in ({schemas_list}) " - f"AND tc.constraint_type = 'PRIMARY KEY'" + f"SELECT {tab_expr} as tab, a.attname as column_name " + f"FROM pg_constraint c " + f"JOIN pg_class cl ON c.conrelid = cl.oid " + f"JOIN pg_namespace ns ON cl.relnamespace = ns.oid " + f"CROSS JOIN LATERAL unnest(c.conkey) WITH ORDINALITY AS cols(conkey, ord) " + f"JOIN pg_attribute a ON a.attrelid = cl.oid AND a.attnum = cols.conkey " + f"WHERE c.contype = 'p' " + f"AND cl.relname NOT LIKE {like_pattern} " + f"AND ns.nspname in ({schemas_list}) " + f"ORDER BY ns.nspname, cl.relname, cols.ord" ) def load_foreign_keys_sql(self, schemas_list: str, like_pattern: str) -> str: diff --git a/tests/integration/test_read_only_login.py b/tests/integration/test_read_only_login.py new file mode 100644 index 000000000..d19126496 --- /dev/null +++ b/tests/integration/test_read_only_login.py @@ -0,0 +1,167 @@ +""" +A login holding only SELECT on a schema must see the same table structure as +the schema's owner: primary keys, headings, and the dependency graph (#1554). + +On PostgreSQL, information_schema hides constraints on tables where the current +user holds only SELECT, so key and foreign-key metadata is read from the system +catalogs instead. The same checks run on MySQL, where a SELECT-only login has +always seen this metadata. +""" + +import pytest + +import datajoint as dj + +READER = "djreadonly" +TABLES = ("Rig", "Subject", "Session", "Pairing", "Analysis") + + +@pytest.fixture +def owner_schema(connection_by_backend, backend, prefix): + """A small pipeline declared and populated by the owner.""" + schema = dj.Schema(f"{prefix}_read_only_{backend}", connection=connection_by_backend) + + @schema + class Rig(dj.Lookup): + definition = """ + rig : varchar(16) + """ + contents = [("A",), ("B",)] + + @schema + class Subject(dj.Manual): + definition = """ + subject_id : int32 + --- + species : varchar(32) + """ + + @schema + class Session(dj.Manual): + # Primary key order is not alphabetical, so a reordered key would show. + definition = """ + -> Subject + session_idx : int16 + --- + -> Rig + note = '' : varchar(255) + """ + + @schema + class Pairing(dj.Manual): + definition = """ + -> Subject + -> Subject.proj(partner_id='subject_id') + """ + + @schema + class Analysis(dj.Computed): + definition = """ + -> Session + --- + result : float64 + """ + + def make(self, key): + self.insert1(dict(key, result=1.0)) + + Subject.insert([(1, "mouse"), (2, "rat")], skip_duplicates=True) + Session.insert([(1, 1, "A", ""), (1, 2, "B", ""), (2, 1, "A", "")], skip_duplicates=True) + Pairing.insert([(1, 2)], skip_duplicates=True) + Analysis.populate(Session & "session_idx = 1") + + yield schema + schema.drop() + + +def _drop_reader(conn, backend): + if backend == "postgresql": + if conn.query("SELECT 1 FROM pg_roles WHERE rolname = %s", args=(READER,)).fetchone(): + conn.query(f"DROP OWNED BY {READER}") + conn.query(f"DROP ROLE {READER}") + else: + conn.query(f"DROP USER IF EXISTS '{READER}'@'%%'") + + +@pytest.fixture +def reader_connection(owner_schema, connection_by_backend, db_creds_by_backend, backend): + """A connection whose login holds only SELECT on the owner's schema.""" + owner = connection_by_backend + database = owner.adapter.quote_identifier(owner_schema.database) + _drop_reader(owner, backend) + if backend == "postgresql": + owner.query(f"CREATE ROLE {READER} LOGIN PASSWORD '{READER}'") + owner.query(f"GRANT USAGE ON SCHEMA {database} TO {READER}") + owner.query(f"GRANT SELECT ON ALL TABLES IN SCHEMA {database} TO {READER}") + else: + owner.query(f"CREATE USER '{READER}'@'%%' IDENTIFIED BY '{READER}'") + owner.query(f"GRANT SELECT ON {database}.* TO '{READER}'@'%%'") + + reader = dj.Connection(host=db_creds_by_backend["host"], user=READER, password=READER) + yield reader + reader.close() + _drop_reader(owner, backend) + + +@pytest.fixture +def owner_and_reader(owner_schema, connection_by_backend, reader_connection): + """The same schema loaded as virtual modules by the owner and by the reader.""" + owner = dj.VirtualModule("owner", owner_schema.database, connection=connection_by_backend) + reader = dj.VirtualModule("reader", owner_schema.database, connection=reader_connection) + return owner, reader + + +def test_primary_keys(owner_and_reader): + owner, reader = owner_and_reader + for name in TABLES: + assert getattr(owner, name).primary_key, name + assert getattr(reader, name).primary_key == getattr(owner, name).primary_key, name + assert reader.Session.primary_key == ["subject_id", "session_idx"] + + +def test_headings(owner_and_reader): + owner, reader = owner_and_reader + for name in TABLES: + assert getattr(reader, name).heading.attributes == getattr(owner, name).heading.attributes, name + + +def _graph(conn): + conn.dependencies.load() + nodes = dict(conn.dependencies.nodes(data=True)) + edges = {(u, v, k): d for u, v, k, d in conn.dependencies.edges(keys=True, data=True)} + return nodes, edges + + +def test_dependencies(owner_and_reader, connection_by_backend, reader_connection): + owner, reader = owner_and_reader + for name in TABLES: + owner_table, reader_table = getattr(owner, name), getattr(reader, name) + for primary in (None, True, False): + assert reader_table.parents(primary=primary) == owner_table.parents(primary=primary), (name, primary) + assert reader_table.children(primary=primary) == owner_table.children(primary=primary), (name, primary) + + nodes, edges = _graph(connection_by_backend) + assert _graph(reader_connection) == (nodes, edges) + assert len(edges) == 5 # Subject->Session, Rig->Session, Subject->Pairing (x2), Session->Analysis + + +def test_queries_that_need_keys(owner_and_reader): + owner, reader = owner_and_reader + assert len(reader.Session()) == len(owner.Session()) == 3 + assert len(reader.Analysis.key_source) == 3 + assert len(reader.Analysis.key_source - reader.Analysis) == 1 + assert reader.Session.keys(order_by="KEY") == owner.Session.keys(order_by="KEY") + assert reader.Session.describe() == owner.Session.describe() + + +def test_foreign_keys_sql(owner_and_reader, connection_by_backend, reader_connection): + """``get_foreign_keys_sql`` returns one row per foreign-key column, for any login.""" + owner, _ = owner_and_reader + adapter = connection_by_backend.adapter + rows = {} + for name in TABLES: + table = getattr(owner, name) + sql = adapter.get_foreign_keys_sql(table.database, table.table_name) + rows[name] = connection_by_backend.query(sql).fetchall() + assert reader_connection.query(sql).fetchall() == rows[name], name + assert [len(rows[name]) for name in TABLES] == [0, 0, 2, 2, 2]