Operate in exactly one schema, and filter every catalog query to it

Unqualified DDL and DML went wherever search_path happened to point, while
catalog inspection was a mixture of hard-coded 'public' and no filter at all.
With a relation of the same name in two schemas the answers came from both:
_column_types unioned their columns (or raised "Type mismatch"), an index or
constraint in the other schema counted as present, _all_tablenames listed the
name twice, and table_sizes reported only public whatever the session was
using.

PostgresDatabase now takes schema="public", validates it as an identifier once
in the constructor, and pins search_path to it in _configure_session -- which
runs for the first connection and for every replacement, so a reconnect cannot
come back pointing somewhere else.  It is deliberately not part of
_connect_kwargs: psycopg.connect has no such parameter, and passing it there
would reach the driver.

Every catalog query is then filtered to that schema, binding it as a value
rather than interpolating it: _table_exists, _all_tablenames, _index_exists,
_list_indexes, _relation_exists, _constraint_exists, _list_constraints,
_column_types, _relation_columns, refresh_tables' column discovery, the
read-only and knowls capability probes, _grantees, the legacy-extras check,
table_sizes, tablespaces, _check_tmp_leftovers' two probes, the metadata
bootstrap's existing-table set, _approx_most_common and dbdiff's column
reader.  _schema_relations and _approx_most_common previously asked the server
with current_schema(); they now bind the same value as everything else, so
there is one notion of which schema this is.

The userdb.users grant probe keeps its own schema: that one is deliberately
about a different schema, not about this database's.

Author: roed314

CHANGELOG.md

@@ -399,6 +399,19 @@ hardening standalone use; the highlights:
   through the one helper they share; `copy_from` analyzes the live table it
   loaded into.
 
+- **A database operates in exactly one schema.** `PostgresDatabase` takes a
+  `schema=` argument (default `"public"`, so nothing changes for existing
+  deployments), validates it as an identifier, and pins `search_path` to it on
+  the first connection and on every replacement. Catalog inspection was
+  previously a mixture of hard-coded `'public'` and no filter at all -- 30-odd
+  queries across `pg_tables`, `pg_indexes`, `pg_class`, `pg_constraint` and
+  `information_schema` -- so with two schemas holding a relation of the same
+  name, column discovery could mix their columns, an index or constraint in
+  the other schema counted as present, and `_all_tablenames` listed the name
+  twice. Every one is now filtered to the selected schema, which is bound as a
+  value rather than interpolated. *Migration:* none unless you were relying on
+  psycodict seeing relations outside `public`, which it did only by accident.
+
 ### Release candidates
 
 1.0.0 is published as a sequence of release candidates first. `pip` ignores

DataManagement.md

@@ -64,6 +64,14 @@ Copies the column layout, `label_col`, `sort`, `id_ordered`, `id` type and descr
 
 psycodict keeps its own bookkeeping in a handful of tables that live alongside your search tables:
 
+psycodict operates in one PostgreSQL schema, `public` unless the constructor is
+given another (`PostgresDatabase(schema="myschema")`).  Every relation it
+creates goes there, every relation it looks for is looked for there, and every
+catalog query is filtered to it, so a relation of the same name in another
+schema is neither mistaken for one of these nor merged with it.  The schema is
+pinned in each connection's `search_path`, including replacement connections
+after a reconnect.
+
  * **`meta_tables`** — one row per search table, holding `name`, `sort`, `count_cutoff`, `id_ordered`, `out_of_order`, `stats_valid`, `label_col`, `total`, `important` and `include_nones`.  This is the source of truth psycodict reads on connection to reconstruct each table object.
  * **`meta_indexes`** and **`meta_constraints`** — one row per index / constraint, recording how to rebuild it.  `reload` and `restore_indexes` rebuild from these rows, **not** from whatever is physically on the table (see [reload](#reload)).
  * **`meta_tables_hist`**, **`meta_indexes_hist`**, **`meta_constraints_hist`** — versioned history of the three tables above, so that `reload_meta`/`revert_meta` can roll a table's metadata forward and back.

psycodict/base.py

@@ -423,14 +423,25 @@ def _table_exists(self, tablename):
 
         - ``tablename`` -- a string, the name of the table
         """
-        cur = self._execute(SQL("SELECT 1 FROM pg_tables where tablename=%s"), [tablename], silent=True)
+        cur = self._execute(
+            SQL("SELECT 1 FROM pg_tables WHERE schemaname = %s AND tablename = %s"),
+            [self._db.schema, tablename],
+            silent=True,
+        )
         return cur.fetchone() is not None
 
     def _all_tablenames(self):
         """
         Return all (postgres) table names in the database
         """
-        return [rec[0] for rec in self._execute(SQL("SELECT tablename FROM pg_tables ORDER BY tablename"), silent=True)]
+        return [
+            rec[0]
+            for rec in self._execute(
+                SQL("SELECT tablename FROM pg_tables WHERE schemaname = %s ORDER BY tablename"),
+                [self._db.schema],
+                silent=True,
+            )
+        ]
 
     def _get_locks(self):
         return self._execute(SQL(
@@ -535,15 +546,18 @@ def _index_exists(self, indexname, tablename=None):
         """
         if tablename:
             cur = self._execute(
-                SQL("SELECT 1 FROM pg_indexes WHERE indexname = %s AND tablename = %s"),
-                [indexname, tablename],
+                SQL(
+                    "SELECT 1 FROM pg_indexes "
+                    "WHERE schemaname = %s AND indexname = %s AND tablename = %s"
+                ),
+                [self._db.schema, indexname, tablename],
                 silent=True,
             )
             return cur.fetchone() is not None
         else:
             cur = self._execute(
-                SQL("SELECT tablename FROM pg_indexes WHERE indexname=%s"),
-                [indexname],
+                SQL("SELECT tablename FROM pg_indexes WHERE schemaname = %s AND indexname = %s"),
+                [self._db.schema, indexname],
                 silent=True,
             )
             table = cur.fetchone()
@@ -560,7 +574,13 @@ def _relation_exists(self, name):
 
         - ``name`` -- a string, the name of the relation
         """
-        cur = self._execute(SQL("SELECT 1 FROM pg_class where relname = %s"), [name])
+        cur = self._execute(
+            SQL(
+                "SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
+                "WHERE n.nspname = %s AND c.relname = %s"
+            ),
+            [self._db.schema, name],
+        )
         return cur.fetchone() is not None
 
     def _constraint_exists(self, constraintname, tablename=None):
@@ -582,19 +602,19 @@ def _constraint_exists(self, constraintname, tablename=None):
             cur = self._execute(
                 SQL(
                     "SELECT 1 from information_schema.table_constraints "
-                    "WHERE table_name=%s and constraint_name=%s"
+                    "WHERE table_schema = %s AND table_name = %s AND constraint_name = %s"
                 ),
-                [tablename, constraintname],
+                [self._db.schema, tablename, constraintname],
                 silent=True,
             )
             return cur.fetchone() is not None
         else:
             cur = self._execute(
                 SQL(
                     "SELECT table_name from information_schema.table_constraints "
-                    "WHERE constraint_name=%s"
+                    "WHERE table_schema = %s AND constraint_name = %s"
                 ),
-                [constraintname],
+                [self._db.schema, constraintname],
                 silent=True,
             )
             table = cur.fetchone()
@@ -608,8 +628,8 @@ def _list_indexes(self, tablename):
         Lists built index names on the search table ``tablename``
         """
         cur = self._execute(
-            SQL("SELECT indexname FROM pg_indexes WHERE tablename = %s"),
-            [tablename],
+            SQL("SELECT indexname FROM pg_indexes WHERE schemaname = %s AND tablename = %s"),
+            [self._db.schema, tablename],
             silent=True,
         )
         return [elt[0] for elt in cur]
@@ -629,9 +649,9 @@ def _list_constraints(self, tablename):
                 "           ON rel.oid = con.conrelid "
                 "INNER JOIN pg_catalog.pg_namespace nsp "
                 "           ON nsp.oid = connamespace "
-                "WHERE rel.relname = %s"
+                "WHERE nsp.nspname = %s AND rel.relname = %s"
             ),
-            [tablename],
+            [self._db.schema, tablename],
             silent=True,
         )
         return [elt[0] for elt in cur]
@@ -809,9 +829,9 @@ def _column_types(self, table_name, data_types=None):
                 cur = self._execute(
                     SQL(
                         "SELECT column_name, udt_name::regtype FROM information_schema.columns "
-                        "WHERE table_name = %s ORDER BY ordinal_position"
+                        "WHERE table_schema = %s AND table_name = %s ORDER BY ordinal_position"
                     ),
-                    [tname],
+                    [self._db.schema, tname],
                 )
             else:
                 cur = data_types[tname]
@@ -838,8 +858,11 @@ def _relation_columns(self, table):
         one about columns.
         """
         cur = self._execute(
-            SQL("SELECT column_name FROM information_schema.columns WHERE table_name = %s"),
-            [table],
+            SQL(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = %s AND table_name = %s"
+            ),
+            [self._db.schema, table],
             silent=True,
             commit=False,
         )

psycodict/database.py

@@ -47,6 +47,7 @@
     check_new_table_name,
     derived_identifier,
     physical_table_name,
+    validate_schema_name,
     validate_search_table_name,
     validate_search_table_registry,
 )
@@ -347,6 +348,11 @@ def _configure_session(self, conn):
         # Note that it has some global effects, since register_adapter
         # is not limited to just one connection
         setup_connection(conn)
+        # Pin the schema first: everything below, and every statement this
+        # connection later runs, resolves unqualified names in it.  pg_catalog
+        # is still searched -- PostgreSQL puts it first implicitly when it is
+        # not named -- so the built-in types and functions stay reachable.
+        conn.execute("SELECT set_config('search_path', %s, false)", [self.schema])
         for name, value in self._session_settings.items():
             # set_config takes both as bound values, so nothing is interpolated
             conn.execute("SELECT set_config(%s, %s, false)", [name, str(value)])
@@ -387,7 +393,7 @@ def query(sql, args):
                 "SELECT count(*) FROM information_schema.role_table_grants "
                 "WHERE grantee = %s AND table_schema = %s "
                 "AND privilege_type IN (" + ",".join(["%s"] * len(privileges)) + ")",
-                [user, "public"] + privileges,
+                [user, self.schema] + privileges,
             )
             read_only = rows[0][0] == 0
 
@@ -405,12 +411,12 @@ def query(sql, args):
             rows = sorted(query(
                 "SELECT table_name, privilege_type "
                 "FROM information_schema.role_table_grants "
-                "WHERE grantee = %s AND table_name IN ("
+                "WHERE grantee = %s AND table_schema = %s AND table_name IN ("
                 + ",".join(["%s"] * len(knowls_tables))
                 + ") AND privilege_type IN ("
                 + ",".join(["%s"] * len(privileges))
                 + ")",
-                [user] + knowls_tables + privileges,
+                [user, self.schema] + knowls_tables + privileges,
             ))
             read_and_write_knowls = rows == sorted(
                 [(table, priv) for table in knowls_tables for priv in privileges]
@@ -522,11 +528,19 @@ def _register_object(self, obj):
         self._objects.append(obj)
 
     def __init__(self, config=None, secretsfile=None, create=False, upgrade=False,
-                 session_settings=None, grant_policy=None, **kwargs):
+                 session_settings=None, grant_policy=None, schema="public", **kwargs):
         if config is None:
             from .config import Configuration
             config = Configuration()
         self.config = config
+        # The one schema this database operates in.  Every relation psycodict
+        # creates goes here, every relation it looks for is looked for here,
+        # and every catalog query is filtered to it -- so that a table of the
+        # same name in another schema can neither stand in for one of these nor
+        # be merged with it.  Checked once, here, rather than at each use.
+        # Deliberately not part of _connect_kwargs: it is psycodict's own
+        # setting, and psycopg.connect has no such parameter.
+        self.schema = validate_schema_name(schema)
         self.server_side_counter = 0
         self._nocommit_stack = 0
         self._silenced = False
@@ -576,9 +590,9 @@ def __init__(self, config=None, secretsfile=None, create=False, upgrade=False,
         # Refuse to run against a database that still uses the removed
         # search/extras table split
         legacy = self._execute(SQL(
-            "SELECT 1 FROM information_schema.columns WHERE table_schema = 'public' "
+            "SELECT 1 FROM information_schema.columns WHERE table_schema = %s "
             "AND table_name = 'meta_tables' AND column_name = 'has_extras'"
-        ))
+        ), [self.schema])
         if legacy.rowcount:
             cur = self._execute(SQL("SELECT name FROM meta_tables WHERE has_extras"))
             if cur.rowcount:
@@ -637,8 +651,9 @@ def refresh_tables(self):
         """
         cur = self._execute(SQL(
             "SELECT table_name, column_name, udt_name::regtype "
-            "FROM information_schema.columns ORDER BY table_name, ordinal_position"
-        ))
+            "FROM information_schema.columns WHERE table_schema = %s "
+            "ORDER BY table_name, ordinal_position"
+        ), [self.schema])
         data_types = {}
         for table_name, column_name, regtype in cur:
             if table_name not in data_types:
@@ -794,9 +809,9 @@ def _grantees(self, table_name):
         cur = self._execute(
             SQL(
                 "SELECT DISTINCT grantee FROM information_schema.role_table_grants "
-                "WHERE table_name = %s AND grantee <> grantor"
+                "WHERE table_schema = %s AND table_name = %s AND grantee <> grantor"
             ),
-            [table_name],
+            [self.schema, table_name],
             silent=True,
         )
         return {rec[0] for rec in cur}
@@ -925,12 +940,12 @@ def _schema_relations(self, relkinds=_TABLE_RELKINDS):
         query = (
             "SELECT c.relname FROM pg_class c "
             "JOIN pg_namespace n ON n.oid = c.relnamespace "
-            "WHERE n.nspname = current_schema()"
+            "WHERE n.nspname = %s"
         )
-        values = None
+        values = [self._db.schema]
         if relkinds is not None:
             query += " AND c.relkind = ANY(%s)"
-            values = [list(relkinds)]
+            values.append(list(relkinds))
         cur = self._execute(SQL(query), values, silent=True)
         return {rec[0] for rec in cur}
 
@@ -1033,10 +1048,10 @@ def table_sizes(self):
          pg_total_relation_size(reltoastrelid) AS toast_bytes
   FROM pg_class c
   LEFT JOIN pg_namespace n ON n.oid = c.relnamespace
-  WHERE n.nspname = 'public' AND relkind = 'r'
+  WHERE n.nspname = %s AND relkind = 'r'
 ) a"""
         sizes = defaultdict(lambda: defaultdict(int))
-        cur = self._execute(SQL(query))
+        cur = self._execute(SQL(query), [self.schema])
         for (
             table_name,
             row_estimate,
@@ -1337,8 +1352,8 @@ def _bootstrap_meta(self):
         existing = {
             rec[0] for rec in self._execute(SQL(
                 "SELECT table_name FROM information_schema.tables "
-                "WHERE table_schema = 'public'"
-            ))
+                "WHERE table_schema = %s"
+            ), [self.schema])
         }
         stored, _ = self._stored_meta_format()
         fmt = META_FORMAT if stored is None else min(stored, META_FORMAT)
@@ -2411,7 +2426,13 @@ def tablespaces(self):
         """
         Returns a dictionary giving giving the tablespace for all tables
         """
-        D = {rec[0]: rec[1] for rec in self._execute(SQL("SELECT tablename, tablespace FROM pg_tables"))}
+        D = {
+            rec[0]: rec[1]
+            for rec in self._execute(
+                SQL("SELECT tablename, tablespace FROM pg_tables WHERE schemaname = %s"),
+                [self.schema],
+            )
+        }
         return {name: space if space else "" for (name, space) in D.items()}
 
     def compare(self, other, tables=None, row_counts=True, null_counts=False, exact=False):

psycodict/dbdiff.py

@@ -88,9 +88,9 @@ def _column_types(db, names):
         "FROM pg_attribute a "
         "JOIN pg_class c ON a.attrelid = c.oid "
         "JOIN pg_namespace n ON c.relnamespace = n.oid "
-        "WHERE n.nspname = 'public' AND c.relkind = 'r' "
+        "WHERE n.nspname = %s AND c.relkind = 'r' "
         "AND a.attnum > 0 AND NOT a.attisdropped"
-    ))
+    ), [db.schema])
     columns = {}
     for table_name, column_name, typ in cur:
         if table_name in names:

psycodict/statstable.py

@@ -1728,11 +1728,14 @@ def _approx_most_common(self, col, n):
 CROSS  JOIN (
    SELECT c.reltuples FROM pg_class c
    JOIN pg_namespace n ON n.oid = c.relnamespace
-   WHERE n.nspname = current_schema() AND c.relname = %s) c
-WHERE schemaname = current_schema() AND tablename = %s AND attname = %s
+   WHERE n.nspname = %s AND c.relname = %s) c
+WHERE schemaname = %s AND tablename = %s AND attname = %s
 ORDER BY v.ord LIMIT %s"""
         ).format(Identifier(col), column_type_sql(self.table.col_type[col]))
-        cur = self._execute(selecter, [self.search_table, self.search_table, col, n])
+        schema = self._db.schema
+        cur = self._execute(
+            selecter, [schema, self.search_table, schema, self.search_table, col, n]
+        )
         return [tuple(x) for x in cur]
 
     def _common_cols(self, threshold=700):

psycodict/table.py

@@ -328,7 +328,10 @@ def _get_tablespace(self):
         """
         Determine the tablespace hosting this table (which is then used for indexes and constraints)
         """
-        cur = self._execute(SQL("SELECT tablespace FROM pg_tables WHERE tablename=%s"), [self.search_table])
+        cur = self._execute(
+            SQL("SELECT tablespace FROM pg_tables WHERE schemaname = %s AND tablename = %s"),
+            [self._db.schema, self.search_table],
+        )
         return cur.fetchone()[0]
 
     def _create_index_statement(self, name, table, type, columns, modifiers, storage_params, whereclause=None):
@@ -2063,9 +2066,11 @@ def _check_tmp_leftovers(self, clone_tables=None):
                 SQL(
                     "SELECT rel.relname, con.conname FROM pg_constraint con "
                     "JOIN pg_class rel ON rel.oid = con.conrelid "
-                    "WHERE rel.relname = ANY(%s) AND con.conname ~ %s"
+                    "JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace "
+                    "WHERE nsp.nspname = %s AND rel.relname = ANY(%s) "
+                    "AND con.conname ~ %s"
                 ),
-                [tables, pattern],
+                [self._db.schema, tables, pattern],
                 silent=True,
             )
         ]
@@ -2078,9 +2083,9 @@ def _check_tmp_leftovers(self, clone_tables=None):
             for tbl, name in self._execute(
                 SQL(
                     "SELECT tablename, indexname FROM pg_indexes "
-                    "WHERE tablename = ANY(%s) AND indexname ~ %s"
+                    "WHERE schemaname = %s AND tablename = ANY(%s) AND indexname ~ %s"
                 ),
-                [tables, pattern],
+                [self._db.schema, tables, pattern],
                 silent=True,
             )
             if (tbl, name) not in found