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.

Author: dimitri-yatsenko

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:

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]