diff --git a/CHANGELOG.md b/CHANGELOG.md index 98cafa4..51e67c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -339,6 +339,91 @@ hardening standalone use; the highlights: Connect — no long-lived token is stored anywhere. (#113) - `CITATION.cff`, so GitHub renders a citation for the package. (#124) +### Fixed after the first release candidates + +- **`max_id` and `min_id` compose the table name they are given.** Both took a + `table=` argument and formatted it into the statement as text, so a name + needing quotes was a syntax error and a name carrying its own statement ran + it. Both now use `Identifier`. `max_id` returns -1 for an empty table, which + is the only empty sentinel: 0 is a real id, and `random()` treated a table + whose single row had id 0 as empty. +- **Approximate statistics are scaled by the table they describe.** + `_approx_most_common` took `reltuples` from a hard-coded `public.nf_fields` + while taking frequencies from the real table, so on every other table the + estimate was that table's frequencies multiplied by an unrelated row count. + The column type it interpolates now goes through the validated + `column_type_sql`. +- **`update_from_file` no longer shares log state between calls.** Its + `logging` default was a dictionary literal that the method wrote `logid` and + `aborted` into, so consecutive default calls saw each other's values and a + caller-supplied dictionary came back modified. The default is now `None` and + the mapping is copied per call. +- **Random selection edge cases.** `random(query, pick_first=...)` returned + `None` rather than raising `IndexError` when nothing satisfies the query; a + projected value of `0`, `False`, `""` or `[]` counts as a result instead of + being skipped until `maxtries` ran out; `random_sample` raises `ValueError` + naming the accepted modes instead of silently returning `None` for an + unknown one; and a repeatable `choice` sample uses a local + `random.Random(seed)` rather than reseeding the process-wide generator. + +- **`stats_valid` is enforced, not just recorded.** Write paths cleared the + flag but read paths ignored it, so a count cached before a `restat=False` + write kept being served afterwards -- verified: a query counted at 67, then + every matching row changed, still answered 67. Every lookup that would serve + a cached answer now goes through one predicate and reports a miss while the + flag is false: `quick_count`, `quick_count_distinct` and `_quick_statistic`, + which is what makes `count`, `max`, `min` and `sum` compute the answer + instead of returning a recorded one. The line is whether a miss costs one + bounded query or a rebuild, so these are deliberately not gated: the + empty-query `total`, maintained on every write and so exact; the `_status` / + `status` / `extra_counts` inventory, which is how `refresh_stats` discovers + what to recompute; `_has_stats` / `_has_numstats`, which decide whether a + whole statistics family needs computing; and `null_counts`, whose fallback + is one full count *per search column*. Gating that last group made + `column_counts`, `numstats` and `null_counts` rebuild on every call with + nothing to converge on, since only `refresh_stats` restores the flag -- + measured on the LMFDB, four minutes of downstream suite became over + forty-five. **The gap that leaves:** `column_counts`, `numstats` and + `null_counts` can still report a value recorded before an unrefreshed + write. + Closing it needs freshness per statistic rather than one flag per table, + which is a metadata format change; `refresh_stats()` is the remedy + meanwhile. A suffixed (`_tmp`, `_oldN`) table is not gated by the live + table's flag, since it carries its own caches. The flag is restored only by + `refresh_stats()`, inside the transaction that rebuilt the caches, so a + failed refresh leaves the table invalid; refreshing a `_tmp` copy does not + validate the live table. +- **Bulk paths run `ANALYZE`.** A relation that has just been bulk loaded has + no planner statistics until autovacuum reaches it, so queries against it are + costed as though it were tiny. Replacement tables are analyzed while still + named `_tmp` -- before the swap, and outside its transaction, since the + catalog entry follows the relation through the rename -- which covers + `reload`, `rewrite`, non-inplace `update_from_file` and staged commits + 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, `_all_tablenames` listed the name + twice, and a lock on the other schema's copy refused a write to this one. + Every one is now filtered to the selected schema, which is bound as a value + rather than interpolated. The schema goes into `search_path` as one quoted + identifier rather than as text (a `search_path` is a comma-separated list, + in which a name needing quotes is parsed rather than used), `pg_temp` is + named after it so a temporary relation cannot shadow a real one, and + `current_schema()` is checked afterwards, since PostgreSQL silently ignores + an entry naming a schema that is missing or not accessible. *Migration:* + none unless you were relying on psycodict seeing relations outside `public`, + which it did only by accident. The schema must exist and the role needs + `USAGE` on it; the name `$user` is refused, PostgreSQL substituting the + role's name for it even when quoted. + ### Release candidates 1.0.0 is published as a sequence of release candidates first. `pip` ignores diff --git a/DataManagement.md b/DataManagement.md index 0554b22..3b9028d 100644 --- a/DataManagement.md +++ b/DataManagement.md @@ -64,6 +64,29 @@ 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, as one quoted identifier (a `search_path` is a +comma-separated *list*, so a name that needs quoting must not be written to it +as text). psycodict names `pg_temp` explicitly *after* it, so that a temporary +relation cannot shadow `meta_tables` or a search table, and leaves `pg_catalog` +unnamed, which is what keeps it ahead of both so the built-in types and +functions stay reachable. The effective order is therefore `pg_catalog`, the +selected schema, `pg_temp`. + +The schema must already exist: psycodict never creates one, and `create=True` +bootstraps the meta tables *within* it. The connecting role needs `USAGE` on +it, and `CREATE` to make any relation there. PostgreSQL silently ignores a +`search_path` entry naming a schema that is missing or not accessible, so +psycodict checks `current_schema()` after setting it and refuses to connect +rather than operate somewhere nobody chose. One name cannot be selected at all: +`$user`, which PostgreSQL substitutes with the connecting role's name even when +quoted, is rejected by the constructor. + * **`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. @@ -108,7 +131,32 @@ These mutate the live table directly. They are convenient for small edits; for * **`update(query, changes, resort=False, restat=True)`** — a plain SQL `UPDATE` of every row matching `query`; `changes` maps column names to constants. * **`delete(query, restat=True)`** — deletes every row matching `query` and decrements `total`. -**Statistics invalidation.** Any write that can change the data calls `_break_stats`, which sets `meta_tables.stats_valid = false` so that cached statistics are known to be stale. If the table has `saving` on and you left `restat=True`, statistics are refreshed at the end of the call; otherwise they are simply marked invalid. Inserting rows (and updating a sort-key column) also calls `_break_order`, setting `out_of_order = true` to record that the `id` order no longer matches `sort`; `delete` leaves the order flag alone. +**Statistics invalidation.** Any write that can change the data calls `_break_stats`, which sets `meta_tables.stats_valid = false` so that cached statistics are known to be stale. If the table has `saving` on and you left `restat=True`, statistics are refreshed at the end of the call; otherwise they are simply marked invalid. + +`stats_valid` is enforced rather than merely recorded: while it is false, a +cached nonempty-query count, distinct count, minimum, maximum or sum reports a +miss, and the method computes the answer instead of returning the stored one. +The empty-query `total` is the exception, since it is maintained on every write +and so stays exact. + +`column_counts`, `numstats` and `null_counts` are the other exception, and a +caveat worth knowing. The line is what a cache miss costs: the counts above +fall back to a single statement about the rows in question, while these fall +back to rebuilding a whole statistics family, or to one full count per search +column. Making them miss while the table is invalid would rebuild on every +call and never converge, since only `refresh_stats()` restores the flag, so +they read what is recorded. A value recorded before an unrefreshed write is +therefore still reported by them; run `refresh_stats()` after a write you did +not `restat`. The flag goes back to true only in `refresh_stats()`, inside +the transaction that rebuilt the caches, so a refresh that fails part-way leaves +the table marked invalid rather than claiming a cache it does not have. +Refreshing a `_tmp` copy does not validate the live table. + +Bulk paths also run PostgreSQL's own `ANALYZE`, which is a different thing from +psycodict's statistics: a freshly loaded relation has no planner statistics +until autovacuum reaches it. Replacement tables are analyzed while still named +`_tmp`, before the swap, since the catalog entry follows the relation through +the rename; `copy_from` analyzes the live table it loaded into. Inserting rows (and updating a sort-key column) also calls `_break_order`, setting `out_of_order = true` to record that the `id` order no longer matches `sort`; `delete` leaves the order flag alone. ### Resorting is disabled @@ -212,7 +260,7 @@ Every search table has two companion tables: ## Operational tips - * **`db.show_locks()`** prints every lock currently held on any table — name, mode, pid and age. This is how you find the process id to kill when a write raised `LockError`. + * **`db.show_locks()`** prints every lock currently held on a table in this database's schema — name, mode, pid and age. This is how you find the process id to kill when a write raised `LockError`. A lock on a same-named table in another schema is not shown, and does not block anything done here. * **`table.analyze(query, projection=1, limit=1000, sort=None, explain_only=False)`** prints the SQL it would run and its `EXPLAIN ANALYZE` plan (or just `EXPLAIN`, with `explain_only=True`). Use it when tuning indexes for a slow search. * **Slow-query logging** is configured in `config.ini`: `[logging] slowcutoff` (seconds, default `0.1`) and `slowlogfile` (default `slow_queries.log`). Any statement slower than `slowcutoff` is logged with its interpolated SQL; slow searches additionally log a `Replicate with db.table.analyze(...)` hint so you can reproduce them. * On **PostgreSQL 18+** a `reload` or `update_from_file` prints a harmless warning like `Constraint of ... with name ..._id_not_null does not end with the suffix _tmp` during the swap — Postgres now catalogs `NOT NULL` as a named constraint that the swap logic does not recognize. The constraint is preserved; the warning can be ignored. diff --git a/psycodict/base.py b/psycodict/base.py index ae61f04..4613e6e 100644 --- a/psycodict/base.py +++ b/psycodict/base.py @@ -423,22 +423,47 @@ 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( - "SELECT t.relname, l.mode, l.pid, age(clock_timestamp(), a.backend_start) " - "FROM pg_locks l " - "JOIN pg_stat_all_tables t ON l.relation = t.relid JOIN pg_stat_activity a ON l.pid = a.pid " - "WHERE l.granted AND t.schemaname <> 'pg_toast'::name AND t.schemaname <> 'pg_catalog'::name" - )) + """ + The granted locks on relations in this database's schema. + + Filtered to that schema in SQL, because the caller has only the + relation name to go on: a lock on ``other_schema.same_name`` would + otherwise be reported as a lock on this schema's ``same_name``, and + ``_check_locks`` would refuse a write that nothing is blocking. Another + session's temporary table of the same name does the same thing. + Equality with the schema also subsumes the old exclusions of + ``pg_toast`` and ``pg_catalog``. + """ + return self._execute( + SQL( + "SELECT t.relname, l.mode, l.pid, age(clock_timestamp(), a.backend_start) " + "FROM pg_locks l " + "JOIN pg_stat_all_tables t ON l.relation = t.relid JOIN pg_stat_activity a ON l.pid = a.pid " + "WHERE l.granted AND t.schemaname = %s" + ), + [self._db.schema], + ) def _table_locked(self, tablename, types="all"): """ @@ -535,15 +560,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 +588,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,9 +616,9 @@ 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 @@ -592,9 +626,9 @@ def _constraint_exists(self, constraintname, tablename=None): 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 +642,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 +663,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 +843,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 +872,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, ) diff --git a/psycodict/database.py b/psycodict/database.py index 7cd95f9..791e3cc 100644 --- a/psycodict/database.py +++ b/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, ) @@ -182,6 +183,15 @@ class PostgresDatabase(PostgresBase): MetadataFormats.md). Without it, a database using an older but compatible metadata format connects with a warning and operates at the older format. + - ``schema`` -- the one PostgreSQL schema this database operates in, + ``public`` by default. Every relation psycodict creates goes there, + every relation it looks for is looked for there, and every catalog query + is filtered to it. It is pinned in each connection's ``search_path``, + ahead of ``pg_temp`` and behind the implicit ``pg_catalog``. The schema + must already exist -- psycodict does not create one, and ``create=True`` + bootstraps the metadata tables *within* it -- and the connecting role + must have ``USAGE`` on it, plus ``CREATE`` to make any relation there. + Connecting raises ``ValueError`` when the session does not end up in it. - ``session_settings`` -- a dictionary of PostgreSQL session settings (``statement_timeout``, ``lock_timeout``, ``idle_in_transaction_session_timeout``, ``application_name``) applied to @@ -203,6 +213,7 @@ class PostgresDatabase(PostgresBase): - ``server_side_counter`` -- an integer tracking how many buffered connections have been created - ``conn`` -- the psycopg connection object + - ``schema`` -- the schema this database operates in - ``tables`` -- the search tables, by name - ``tablenames`` -- the names of the search tables, sorted - ``meta_format`` -- the metadata format this connection operates at (see @@ -330,16 +341,20 @@ def _configure_session(self, conn): Apply psycodict's per-session setup to a fresh connection. Everything a session needs beyond the connection itself: the type - adapters, and the session settings (statement, lock and - idle-in-transaction timeouts) that protect the server from this - connection. A replacement connection that skipped this would be a - connection without those protections. + adapters, the ``search_path`` that pins this database's schema, and the + session settings (statement, lock and idle-in-transaction timeouts) + that protect the server from this connection. A replacement connection + that skipped this would be a connection without those protections, and + one pointing at whatever schema the role's default happens to be. This is for the main connection and its replacements. A notification listener opens its own connection and does not come through here. Runs directly on ``conn`` rather than through ``_execute``, because it is called while recovering from a failure in ``_execute``. + + Raises ``ValueError`` if the session does not end up in + ``self.schema``. """ # The following function controls how Python classes are converted to # strings for passing to Postgres, and how the results are decoded upon @@ -347,6 +362,45 @@ 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. The + # effective order is pg_catalog, this schema, pg_temp. + # + # The schema is composed in as one Identifier rather than bound as a + # value, because search_path is not an identifier: it is a + # comma-separated *list*, in which unquoted names are case-folded and + # ``$user`` is a substitution. A schema legitimately named + # ``a, b``, or one whose spelling needs quoting, would otherwise be + # parsed into something other than the schema every catalog query here + # binds by name -- and unqualified DDL would then create relations + # somewhere this database does not look. + # + # pg_temp is named last on purpose. Left unnamed it is searched + # *first*, so a temporary relation in this session would shadow the + # real meta_tables or search table of the same name. pg_catalog is + # left unnamed, which puts it first, so built-in types and functions + # stay reachable and cannot be shadowed either. + conn.execute( + SQL("SET search_path TO {schema}, pg_temp").format( + schema=Identifier(self.schema) + ) + ) + # PostgreSQL accepts a search_path naming a schema that does not exist, + # or one this role has no USAGE on: it silently drops the entry and + # resolves names further down the path -- pg_temp, here, which would + # make an unqualified CREATE TABLE produce a *temporary* table. Ask + # what the setting actually selected, rather than let a session run on + # in a schema nobody chose. + current = conn.execute("SELECT current_schema()").fetchone()[0] + if current != self.schema: + raise ValueError( + "Could not operate in schema %r: the session resolves " + "unqualified names in %s instead. The schema must exist, and " + "the role %r must have USAGE on it." + % (self.schema, + "no schema at all" if current is None else repr(current), + conn.info.user) + ) 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 +441,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 +459,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 +576,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 +638,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 +699,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 +857,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 +988,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 +1096,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 +1400,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) @@ -1395,19 +1458,28 @@ def _clone_storage_settings(self, new_name, table): ).fetchone()[0] # Column level compression settings (attcompression) appeared in # postgres 14; on older servers there is only attstorage to copy. - if server_version >= 140000: - selecter = SQL( - "SELECT attname, attstorage, attcompression FROM pg_attribute " - "WHERE attrelid = %s::regclass AND attnum > 0 AND NOT attisdropped" - ) - else: - selecter = SQL( - "SELECT attname, attstorage, '' FROM pg_attribute " - "WHERE attrelid = %s::regclass AND attnum > 0 AND NOT attisdropped" - ) + # + # The relation is reached by joining pg_class to pg_namespace and + # binding this database's schema, rather than by casting the name to + # regclass: a regclass cast resolves through the search path, so it + # would read a temporary relation of the same name in preference to the + # table being copied, and would silently follow the session's schema + # rather than this database's. + attcompression = "a.attcompression" if server_version >= 140000 else "''" + selecter = SQL( + "SELECT a.attname, a.attstorage, " + attcompression + " " + "FROM pg_catalog.pg_attribute a " + "JOIN pg_catalog.pg_class c ON c.oid = a.attrelid " + "JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = %s AND c.relname = %s " + "AND a.attnum > 0 AND NOT a.attisdropped" + ) def settings(tname): - return {rec[0]: (rec[1], rec[2]) for rec in self._execute(selecter, [tname])} + return { + rec[0]: (rec[1], rec[2]) + for rec in self._execute(selecter, [self.schema, tname]) + } source = settings(table.search_table) target = settings(new_name) @@ -2387,7 +2459,13 @@ def show_blocked(self): def show_locks(self): """ - Prints information on all locks currently held on any table. + Prints information on the locks currently held on tables in this + database's schema. + + Confined to that schema, like every other catalog query here and like + the ``_check_locks`` that consults the same result: a lock on a + same-named table elsewhere in the cluster is somebody else's business + and does not block anything done here. """ locks = sorted(self._get_locks()) if locks: @@ -2411,7 +2489,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): diff --git a/psycodict/dbdiff.py b/psycodict/dbdiff.py index 56680e3..dc1cdf4 100644 --- a/psycodict/dbdiff.py +++ b/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: diff --git a/psycodict/searchtable.py b/psycodict/searchtable.py index 4e55dea..d0a6716 100644 --- a/psycodict/searchtable.py +++ b/psycodict/searchtable.py @@ -25,6 +25,10 @@ # (psycopg2 had a single cursor class, which this name used to alias) pg_cursor = (Cursor, ServerCursor) +# The sampling strategies random_sample accepts, upper-cased because SYSTEM and +# BERNOULLI go into the TABLESAMPLE clause verbatim. +_RANDOM_SAMPLE_MODES = ("SYSTEM", "BERNOULLI", "CHOICE") + def _qualify(frag, tablename): """ @@ -1642,6 +1646,11 @@ def random(self, query={}, projection=0, pick_first=None): """ if pick_first: colvals = self.distinct(pick_first, query) + if not colvals: + # No row satisfies the query, so there is no value to pick; + # random.choice([]) would raise IndexError instead of + # returning the documented None. + return None query = dict(query) query[pick_first] = random.choice(colvals) return self.random(query, projection) @@ -1680,10 +1689,10 @@ def random(self, query={}, projection=0, pick_first=None): # a temporary hack FIXME # maxid = self.max('id') maxid = self.max_id() - # max_id returns -1 on an empty table (MAX(id) is NULL), so - # testing for 0 sent an empty table into randint(0, -1); - # anything below 1 means there are no rows. - if maxid < 1: + # max_id returns -1 on an empty table (MAX(id) is NULL). That is + # the only empty sentinel: 0 is a legitimate id, so a table whose + # single row has id 0 must not be reported as empty. + if maxid < 0: return None # a temporary hack FIXME minid = self.min_id() @@ -1693,7 +1702,11 @@ def random(self, query={}, projection=0, pick_first=None): # rid = random.randint(1, maxid) rid = random.randint(minid, maxid) res = self.lucky({"id": rid}, projection=projection) - if res: + # lucky returns None when no row has that id. Anything else is + # a hit, including a projection whose value is 0, False, "" or + # an empty list -- testing truthiness discarded those rows and + # could exhaust maxtries on a table full of them. + if res is not None: return res raise RuntimeError("Random selection failed!") @@ -1719,7 +1732,21 @@ def random_sample(self, ratio, query={}, projection=1, mode=None, repeatable=Non mode = "bernoulli" else: mode = "choice" + if not isinstance(mode, str): + raise ValueError( + "mode must be one of %s or None, not %s" + % (", ".join(map(repr, _RANDOM_SAMPLE_MODES)), type(mode).__name__) + ) mode = mode.upper() + # Checked before any work is done: an unrecognized mode used to fall + # through every branch below and return None, which reads like an empty + # result rather than a mistake. + if mode not in _RANDOM_SAMPLE_MODES: + raise ValueError( + "%r is not a valid mode; use one of %s, or None to choose " + "between 'bernoulli' and 'choice' by result count" + % (mode.lower(), ", ".join(map(repr, _RANDOM_SAMPLE_MODES))) + ) search_cols = self._parse_projection(projection) if ratio > 1 or ratio <= 0: raise ValueError("Ratio must be a positive number between 0 and 1") @@ -1728,9 +1755,11 @@ def random_sample(self, ratio, query={}, projection=1, mode=None, repeatable=Non elif mode == "CHOICE": results = list(self.search(query, projection, sort=[])) count = int(len(results) * ratio) - if repeatable is not None: - random.seed(repeatable) - return random.sample(results, count) + # A local generator, so asking for a repeatable sample does not + # reseed the process-wide random module and make every other + # caller's sequence repeat with it. + rng = random if repeatable is None else random.Random(repeatable) + return rng.sample(results, count) elif mode in ["SYSTEM", "BERNOULLI"]: cols = SQL(", ").join(self._column_composable(c) for c in search_cols) if repeatable is None: diff --git a/psycodict/statstable.py b/psycodict/statstable.py index 92d575e..8540f4b 100644 --- a/psycodict/statstable.py +++ b/psycodict/statstable.py @@ -21,7 +21,7 @@ from psycopg.sql import SQL, Identifier, Literal from .base import PostgresBase -from .validation import physical_table_name +from .validation import column_type_sql, physical_table_name from .encoding import Json, numeric_converter from .utils import DelayCommit, KeyedDefaultDict, make_tuple @@ -269,6 +269,54 @@ def _get_tablespace(self): # We use the same tablespace for stats and counts tables as for the main search table return self.table._get_tablespace() + def _may_use_cache(self, suffix=""): + """ + Whether a cached count or statistic may be used as an answer. + + ``stats_valid`` is an assertion that every cached nonempty-query count, + distinct count and custom statistic for the live table agrees with the + live data. A write that does not refresh them clears it, and until a + refresh restores it a cached row is a stale answer, not an answer -- so + every lookup that would serve one is routed through here and reports a + miss instead, leaving the caller to compute or recompute. + + Two things are deliberately outside this rule: + + - the empty-query ``total``, which is maintained on every write and + stays usable regardless (see :meth:`quick_count`); + - a suffixed table. A ``_tmp`` or ``_oldN`` copy carries its own + counts and stats, built or loaded together with its data, and + ``stats_valid`` says nothing about them. + + Reads that report what the cache *contains*, rather than answering a + question about the data -- ``_status``, ``status``, ``extra_counts`` -- + are also not gated: ``refresh_stats`` uses them to discover what to + recompute, so gating them would make an invalid table forget what + statistics it is supposed to have. + + The line this draws is whether a miss costs one bounded query or a + rebuild. ``quick_count``, ``quick_count_distinct`` and + ``_quick_statistic`` each fall back to a single statement about the + rows in question, so a miss is affordable and they are gated. + ``_has_stats`` and ``_has_numstats`` decide whether a whole statistics + family needs computing, and ``null_counts`` falls back to one full + count *per search column*; gating those makes ``column_counts``, + ``numstats`` and ``null_counts`` rebuild on every call with nothing to + converge on, since only ``refresh_stats`` restores the flag. Measured + on the LMFDB, that took a four-minute downstream suite past + forty-five, mostly inside ``null_counts`` over ``nf_fields`` and + friends. + + What that leaves is a real gap: a value recorded before an unrefreshed + write is still reported by ``column_counts``, ``numstats`` and + ``null_counts``. Closing it needs freshness per statistic rather than + one flag per table, which is a metadata format change; + ``refresh_stats()`` is the remedy meanwhile. + """ + if suffix: + return True + return self.table._stats_valid + def _has_stats(self, jcols, ccols, cvals, threshold, split_list=False, threshold_inequality=False, suffix=""): """ Checks whether statistics have been recorded for a given set of columns. @@ -284,6 +332,15 @@ def _has_stats(self, jcols, ccols, cvals, threshold, split_list=False, threshold rows are thrown away. - ``split_list`` -- whether entries of lists should be counted once for each entry. - ``threshold_inequality`` -- if true, then any lower threshold will still count for having stats. + + Deliberately *not* gated on ``stats_valid``: this answers "is this + statistic recorded", which is what ``add_stats`` and ``column_counts`` + use to decide whether to compute it. Reporting False while the table + is invalid makes them recompute the whole family on every call, and + since nothing but ``refresh_stats`` restores the flag, they never stop + -- measured on the LMFDB, that turned a four-minute test suite into one + still running after forty-five. See :meth:`_may_use_cache` for what + that costs in staleness. """ if split_list: values = [jcols, "split_total"] @@ -322,7 +379,11 @@ def quick_count(self, query, split_list=False, suffix="", startup=False): Either an integer giving the number of results, or None if not cached. """ if not query and not startup: + # The empty-query total is maintained on every write, so it is + # exact even when the rest of the cache is not. return self.total + if not self._may_use_cache(suffix): + return None cols, vals = self._split_dict(query) selecter = SQL( "SELECT count FROM {0} WHERE cols = %s AND values = %s AND split = %s" @@ -542,8 +603,11 @@ def quick_count_distinct(self, cols, query={}, suffix=""): OUTPUT: - Either an integer giving the number of distinct values, or None if not cached. + Either an integer giving the number of distinct values, or None if not + cached or if the cache may not be used. """ + if not self._may_use_cache(suffix): + return None ccols, cvals = self._split_dict(query) selecter = SQL("SELECT value FROM {0} WHERE stat = %s AND cols = %s AND constraint_cols = %s AND constraint_values = %s").format(Identifier(self.stats + suffix)) cur = self._execute(selecter, ["distinct", Json(cols), ccols, cvals]) @@ -736,6 +800,8 @@ def _quick_statistic(self, col, ccols, cvals, kind="max"): the constraint columns take on these values. - ``kind`` -- either "min" or "max" or "sum" """ + if not self._may_use_cache(): + return None constraint = SQL("constraint_cols = %s AND constraint_values = %s") values = [kind, Json([col]), ccols, cvals] selecter = SQL( @@ -1290,6 +1356,9 @@ def _has_numstats(self, jcol, cgcols, cvals, threshold, suffix=""): - ``threshold`` -- an integer: if the number of rows with a given tuple of values for the grouping columns is less than this threshold, those rows are thrown away. + + Not gated on ``stats_valid``, for the reason given in + :meth:`_has_stats`. """ values = [jcol, "ntotal", cgcols, cvals] if threshold is None: @@ -1651,21 +1720,27 @@ def _approx_most_common(self, col, n): """ if col not in self.table.search_cols: raise ValueError("Column %s not a search column for %s" % (col, self.search_table)) + # reltuples has to come from the table these frequencies are about. It + # used to be read from a hard-coded public.nf_fields, so on any other + # table the estimate was that table's frequencies scaled by an + # unrelated row count. selecter = SQL( """SELECT v.{0}, (c.reltuples * freq)::int as estimate_ct FROM pg_stats s CROSS JOIN LATERAL - unnest(s.most_common_vals::text::""" - + self.table.col_type[col] - + """[] + unnest(s.most_common_vals::text::{1}[] , s.most_common_freqs) WITH ORDINALITY v ({0}, freq, ord) CROSS JOIN ( - SELECT reltuples FROM pg_class - WHERE oid = regclass 'public.nf_fields') c -WHERE schemaname = 'public' AND tablename = %s AND attname = %s + SELECT c.reltuples FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + 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)) - cur = self._execute(selecter, [self.search_table, col, n]) + ).format(Identifier(col), column_type_sql(self.table.col_type[col])) + 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): @@ -1844,6 +1919,12 @@ def refresh_stats(self, total=True, reset_None_to_1=False, suffix=""): # Refresh total in meta_tables self._set_total(self._slow_count({}, suffix=suffix, extra=False), suffix=suffix) self.refresh_null_counts(suffix=suffix) + if not suffix: + # Everything above ran in this transaction, so the caches now + # agree with the data and the table can be marked valid with + # them. A suffixed refresh is rebuilding some other relation's + # caches and says nothing about the live table. + self.table._restore_stats() self._logger.info("Refreshed statistics in %.3f secs" % (time.time() - t0)) def status(self, reset_None_to_1=False): diff --git a/psycodict/table.py b/psycodict/table.py index a7a24a5..4c1fa0e 100644 --- a/psycodict/table.py +++ b/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): @@ -1182,6 +1185,20 @@ def _break_stats(self): self._execute(updater, [self.search_table], silent=True) self._stats_valid = False + def _restore_stats(self): + """ + Record that the cached counts and statistics agree with the live data. + + The counterpart of :meth:`_break_stats`, and the only way the flag goes + back to true. Call it from inside the transaction that rebuilt or + loaded the caches, so that a failure part-way through leaves the table + marked invalid rather than claiming a cache it does not have. + """ + if not self._stats_valid: + updater = SQL("UPDATE meta_tables SET stats_valid = true WHERE name = %s") + self._execute(updater, [self.search_table], silent=True) + self._stats_valid = True + def _break_order(self): """ This function should be called when the id ordering is invalidated by an insertion or update. @@ -1350,7 +1367,7 @@ def update_from_file( resort=None, reindex=None, restat=True, - logging={"operation":"file_update"}, + logging=None, **kwds ): """ @@ -1373,7 +1390,9 @@ def update_from_file( - ``resort`` -- whether this table should be resorted after updating (default is to resort when the sort columns intersect the updated columns) - ``reindex`` -- only meaningful when ``inplace`` is set: whether to drop the indexes touching the updated columns before the update and recreate them afterward, which is faster when many rows change (by default this is done when more than 1000 rows are updated). Without ``inplace``, all indexes are necessarily recreated on the replacement table, so ``reindex=True`` is redundant and ``reindex=False`` raises an error. - ``restat`` -- whether to recompute stats for the table - - ``logging`` -- a dictionary of keyword arguments for _log_db_change + - ``logging`` -- a dictionary of keyword arguments for _log_db_change. + A copy is taken, so the caller's dictionary is not modified and two + calls sharing one dictionary do not see each other's ``logid``. - ``kwds`` -- passed on to the ``COPY`` command. Cannot include "columns". """ self._forbid_reindex_false(reindex, inplace) @@ -1384,8 +1403,12 @@ def update_from_file( # The counts and stats tables are not checked: this method # deliberately reuses their _tmp versions when they exist. self._check_tmp_leftovers([self.search_table]) - logid = self._check_locks(logging["operation"], datafile=datafile) - logging["aborted"] = True + # Copied rather than used directly: this dictionary is mutated below, + # and the default used to be a shared literal, so consecutive default + # calls carried the previous call's logid and aborted flag. + log_data = {"operation": "file_update"} if logging is None else dict(logging) + logid = self._check_locks(log_data["operation"], datafile=datafile) + log_data["aborted"] = True try: sep = kwds.get("sep", "|") print("Updating %s from %s..." % (self.search_table, datafile)) @@ -1505,11 +1528,11 @@ def drop_tmp(): self._set_ordered() # Delete the temporary table used to load the data drop_tmp() - logging["logid"] = logid - logging["aborted"] = False + log_data["logid"] = logid + log_data["aborted"] = False print("Updated %s in %.3f secs" % (self.search_table, time.time() - now)) finally: - self._log_db_change(**logging) + self._log_db_change(**log_data) def delete(self, query, restat=True): """ @@ -1937,6 +1960,24 @@ def _next_backup_number(self): ) return backup_number + def _analyze(self, tables, suffix=""): + """ + Refresh PostgreSQL's planner statistics for the given relations. + + These are the server's own statistics, not the counts and stats + psycodict maintains: a relation that has just been bulk loaded has none + until autovacuum reaches it, and until then the planner costs queries + against it as though it were tiny. + + Run on a ``_tmp`` copy before the swap rather than on the live table + after it, so that no query is served by an unanalyzed relation; the + catalog entry follows the relation through the rename. + """ + for table in tables: + self._execute( + SQL("ANALYZE {0}").format(Identifier(table + suffix)), silent=True + ) + def _swap_in_tmp(self, tables): """ Helper function for ``reload``: appends _old{n} to the names of tables/indexes/pkeys @@ -1947,6 +1988,10 @@ def _swap_in_tmp(self, tables): - ``tables`` -- a list of tables to rename (e.g. self.search_table, self.stats.counts, self.stats.stats) """ now = time.time() + # Before the swap, and outside its transaction: the _tmp relations are + # complete by now, and analyzing them here keeps the window in which + # the live names are locked as short as it was. + self._analyze(tables, "_tmp") backup_number = self._next_backup_number() with DelayCommit(self, silence=True): self._swap(tables, "", "_old" + str(backup_number)) @@ -2021,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, ) ] @@ -2036,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 @@ -2848,10 +2895,16 @@ def _staged_abort(self, logid): def max_id(self, table=None): """ The largest id occurring in the given table. Used in the random method. + + Returns -1 for a table with no rows, which is below every id psycodict + generates; callers distinguishing "empty" from "has rows" must test + ``< 0`` rather than ``< 1``, since 0 is a legitimate id. """ if table is None: table = self.search_table - res = self._execute(SQL("SELECT MAX(id) FROM {}".format(table))).fetchone()[0] + res = self._execute( + SQL("SELECT MAX(id) FROM {0}").format(Identifier(table)) + ).fetchone()[0] if res is None: res = -1 return res @@ -2860,10 +2913,16 @@ def max_id(self, table=None): def min_id(self, table=None): """ The smallest id occurring in the given table. Used in the random method. + + Returns 0 for a table with no rows. Unlike :meth:`max_id` that is not a + sentinel a caller can test for, since 0 is also a real id; pair it with + ``max_id() < 0`` to detect an empty table. """ if table is None: table = self.search_table - res = self._execute(SQL("SELECT MIN(id) FROM {}".format(table))).fetchone()[0] + res = self._execute( + SQL("SELECT MIN(id) FROM {0}").format(Identifier(table)) + ).fetchone()[0] if res is None: res = 0 return res @@ -2919,6 +2978,10 @@ def copy_from( if reindex: self.restore_indexes() self._break_stats() + # A bulk COPY can change the table's size and distribution + # enough that the planner's statistics no longer describe it, + # and the stats refresh below plans against them. + self._analyze([self.search_table]) if self.stats.saving and restat: self.stats.refresh_stats(total=False) self.stats._update_total(search_count) diff --git a/psycodict/validation.py b/psycodict/validation.py index edbede7..553e93c 100644 --- a/psycodict/validation.py +++ b/psycodict/validation.py @@ -574,6 +574,43 @@ def utf8_prefix(name, max_bytes): return encoded[:max_bytes].decode("utf-8", "ignore") +def validate_schema_name(name): + """ + Check a PostgreSQL schema name psycodict will operate in. + + A schema name is an identifier like any other, and it is quoted wherever + psycodict emits it -- including in ``search_path``, which it is composed + into as one ``Identifier`` rather than written as text -- so the rules are + the identifier rules: not empty, no control characters, and short enough + that the server will not truncate it into a different schema. It is + checked once, in the constructor, rather than at each use. + + The one name a quoted identifier cannot express is ``$user``: PostgreSQL + unquotes each ``search_path`` element before looking for that token, so + ``"$user"`` is replaced by the connecting role's name and never selects a + schema actually called ``$user``. psycodict refuses it here rather than + silently operating somewhere else. Any other spelling, ``$USER`` + included, is an ordinary identifier. + + EXAMPLES:: + + >>> from psycodict.validation import validate_schema_name + >>> validate_schema_name("my schema, other") + 'my schema, other' + >>> validate_schema_name("$user") + Traceback (most recent call last): + ... + psycodict.validation.InvalidDefinitionError: Schema name '$user' ... + """ + if name == "$user": + raise InvalidDefinitionError( + "Schema name '$user' cannot be selected: PostgreSQL substitutes " + "the connecting role's name for it in search_path, even quoted. " + "Name the schema you mean." + ) + return validate_relation_name(name, kind="Schema", max_length=MAX_IDENTIFIER_LENGTH) + + def catalog_identifier(name): """ The spelling the catalog holds ``name`` under. diff --git a/tests/test_correctness.py b/tests/test_correctness.py new file mode 100644 index 0000000..6f98d8c --- /dev/null +++ b/tests/test_correctness.py @@ -0,0 +1,257 @@ +# -*- coding: utf-8 -*- +""" +Regression tests for the correctness fixes in the rc3 review round. + +Each test here fails on the code as it stood at v1.0.0rc2. They are grouped by +the thing that was wrong rather than by the method, since several of them are +the same mistake made in two places: a value formatted into SQL text instead of +composed, and a result tested for truth instead of for existence. +""" +import random + +import pytest + +from psycopg.sql import SQL, Identifier + +import conftest + + +# --------------------------------------------------------------------------- +# identifiers in max_id / min_id +# --------------------------------------------------------------------------- + +# Names that are legal PostgreSQL identifiers once quoted, and that a bare +# "SELECT MAX(id) FROM %s" would either mis-parse or execute as extra SQL. +AWKWARD_NAMES = [ + "plain_name_9", + "has space", + 'has"quote', + "semi;colon", + "dash--dash", + "slash/*star", + "Ünïcødé", +] + + +@pytest.mark.parametrize("suffix", AWKWARD_NAMES) +def test_max_id_and_min_id_quote_the_table_they_are_given(db, empty_table, suffix): + """ + max_id/min_id take a table name as an argument and used to format it into + the statement as text. A name needing quotes was a syntax error, and one + containing a statement terminator was an injection. + """ + scratch = "t_%s_%s" % (suffix, empty_table.search_table[-8:]) + db._execute( + SQL("CREATE TABLE {0} (id bigint)").format(Identifier(scratch)) + ) + try: + db._execute( + SQL("INSERT INTO {0} (id) VALUES (3), (11)").format(Identifier(scratch)) + ) + assert empty_table.max_id(scratch) == 11 + assert empty_table.min_id(scratch) == 3 + finally: + db._execute(SQL("DROP TABLE {0}").format(Identifier(scratch))) + + +def test_max_id_does_not_execute_an_injected_statement(db, empty_table): + """ + The marker table must not exist afterwards: a name carrying its own + statement has to fail to resolve as a relation, not run. + """ + marker = "marker_%s" % empty_table.search_table[-8:] + injected = 'nonexistent"; CREATE TABLE %s (x int); --' % marker + with pytest.raises(Exception): + empty_table.max_id(injected) + db.conn.rollback() + assert not db._table_exists(marker) + + +def test_max_id_reports_empty_as_minus_one(empty_table): + assert empty_table.max_id() == -1 + + +# --------------------------------------------------------------------------- +# approximate statistics use the owning table +# --------------------------------------------------------------------------- + +def test_approx_most_common_scales_by_the_owning_table(db, table_factory): + """ + Frequencies came from the right table but reltuples came from a hard-coded + public.nf_fields, so on every other table the estimate was that table's + frequencies scaled by an unrelated row count. + + Two tables with the same value distribution and very different row counts + must therefore get very different estimates. + """ + small = table_factory() + big = table_factory() + small.insert_many([conftest.sample_row(i) for i in range(50)]) + big.insert_many([conftest.sample_row(i) for i in range(1000)]) + for table in (small, big): + db._execute(SQL("ANALYZE {0}").format(Identifier(table.search_table))) + + small_est = dict(small.stats._approx_most_common("flag", 2)) + big_est = dict(big.stats._approx_most_common("flag", 2)) + assert small_est and big_est + + # every row has flag set, so the estimates must bracket the real counts + assert sum(small_est.values()) == pytest.approx(50, rel=0.25) + assert sum(big_est.values()) == pytest.approx(1000, rel=0.25) + assert sum(big_est.values()) > 5 * sum(small_est.values()) + + +# --------------------------------------------------------------------------- +# update_from_file does not share log state between calls +# --------------------------------------------------------------------------- + +def _write_update(path, table, rows): + """ + A minimal update file: the label column first, as update_from_file requires, + then one column to change. + """ + cols = ["label", "num"] + with open(path, "w") as F: + F.write("|".join(cols) + "\n") + F.write("|".join(table.col_type[c] for c in cols) + "\n\n") + for label, num in rows: + F.write("%s|%s\n" % (label, num)) + + +def test_update_from_file_does_not_carry_log_state_between_calls(filled_table, tmp_path): + """ + The default was a shared dictionary literal that the method wrote logid and + aborted into, so the second default call started out holding the first + call's values -- and a caller who passed a dictionary got it modified. + """ + first = tmp_path / "first.txt" + second = tmp_path / "second.txt" + _write_update(first, filled_table, [("l0", 111)]) + _write_update(second, filled_table, [("l1", 222)]) + + filled_table.update_from_file(str(first), inplace=True, restat=False) + filled_table.update_from_file(str(second), inplace=True, restat=False) + + # Both updates landed, and the second call logged its own operation. + assert filled_table.lucky({"label": "l0"}, "num") == 111 + assert filled_table.lucky({"label": "l1"}, "num") == 222 + + # The default really is rebuilt per call. + import inspect + + from psycodict.table import PostgresTable + + default = inspect.signature(PostgresTable.update_from_file).parameters["logging"].default + assert default is None + + +def test_update_from_file_leaves_a_supplied_dictionary_alone(filled_table, tmp_path): + datafile = tmp_path / "u.txt" + _write_update(datafile, filled_table, [("l0", 333)]) + + supplied = {"operation": "caller_owned"} + filled_table.update_from_file( + str(datafile), inplace=True, restat=False, logging=supplied + ) + assert supplied == {"operation": "caller_owned"} + + +def test_update_from_file_leaves_a_supplied_dictionary_alone_on_failure( + filled_table, tmp_path +): + bad = tmp_path / "bad.txt" + bad.write_text("label|nosuchcolumn\ntext|text\n\nl0|x\n") + supplied = {"operation": "caller_owned"} + with pytest.raises(Exception): + filled_table.update_from_file( + str(bad), inplace=True, restat=False, logging=supplied + ) + filled_table._db.conn.rollback() + assert supplied == {"operation": "caller_owned"} + + +# --------------------------------------------------------------------------- +# random() edge cases +# --------------------------------------------------------------------------- + +def test_random_with_pick_first_returns_none_when_nothing_matches(filled_table): + """ + distinct() over a query nothing satisfies is empty, and random.choice([]) + raised IndexError where the documented behavior is None. + """ + assert filled_table.random({"n": -1}, pick_first="label") is None + + +def test_random_finds_the_only_row_when_its_id_is_zero(db, table_factory): + """ + -1 is the empty sentinel from max_id, so a table whose single row has id 0 + is not empty. Testing `maxid < 1` reported it as such. + """ + table = table_factory() + table.insert_many([conftest.sample_row(1)]) + db._execute( + SQL("UPDATE {0} SET id = 0").format(Identifier(table.search_table)) + ) + assert table.max_id() == 0 + assert table.random() == "l1" + + +def test_random_returns_a_false_valued_projection(db, table_factory): + """ + `if res:` discarded a row whose projected value was 0, False or "", so a + table of them exhausted maxtries and raised "Random selection failed!". + """ + table = table_factory() + table.insert_many([dict(conftest.sample_row(i), num=0) for i in range(20)]) + for _ in range(10): + assert table.random({}, "num") == 0 + + +def test_random_returns_none_for_an_empty_table(empty_table): + assert empty_table.random() is None + + +# --------------------------------------------------------------------------- +# random_sample() mode handling and RNG isolation +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("mode", ["nonsense", "SYSTEMATIC", "", "choise"]) +def test_random_sample_rejects_an_unknown_mode(filled_table, mode): + """ + An unrecognized mode matched no branch and the method returned None, which + is indistinguishable from an empty result. + """ + with pytest.raises(ValueError, match="mode"): + filled_table.random_sample(0.5, mode=mode) + + +def test_random_sample_rejects_a_non_string_mode(filled_table): + with pytest.raises(ValueError, match="mode"): + filled_table.random_sample(0.5, mode=17) + + +@pytest.mark.parametrize("mode", ["system", "bernoulli", "choice", "CHOICE"]) +def test_random_sample_accepts_every_documented_mode(filled_table, mode): + result = filled_table.random_sample(0.5, mode=mode) + assert list(result) is not None + + +def test_repeatable_choice_sampling_leaves_the_global_rng_alone(filled_table): + """ + random.seed(repeatable) reseeded the process-wide generator, so asking for + a reproducible sample made every later random number in the program repeat. + """ + random.seed(12345) + baseline = [random.random() for _ in range(5)] + + random.seed(12345) + filled_table.random_sample(0.5, mode="choice", repeatable=99) + after = [random.random() for _ in range(5)] + + assert baseline == after + + +def test_repeatable_choice_sampling_is_still_repeatable(filled_table): + first = filled_table.random_sample(0.5, mode="choice", repeatable=7) + second = filled_table.random_sample(0.5, mode="choice", repeatable=7) + assert first == second diff --git a/tests/test_doctests.py b/tests/test_doctests.py index c50cfcd..e0850a6 100644 --- a/tests/test_doctests.py +++ b/tests/test_doctests.py @@ -171,6 +171,13 @@ def doc_tables(db): sort=["conductor_norm", "label"], ) db.test_curves.insert_many(_rows(CURVE_COLUMNS, CURVES)) + # insert_many invalidates the statistics, and a cached count may not be + # served while they are invalid. Nothing is cached yet and the totals are + # maintained by the inserts, so the caches do agree with the data; saying + # so puts both tables in the state a freshly loaded table is in, which is + # what the statistics examples assume. + db.test_fields._restore_stats() + db.test_curves._restore_stats() yield db # The namespace was verified empty above and the suite runs serially, # so everything in it now is ours: the two tables, their stats/counts diff --git a/tests/test_locks.py b/tests/test_locks.py index 5977103..1c264bb 100644 --- a/tests/test_locks.py +++ b/tests/test_locks.py @@ -12,10 +12,16 @@ import psycopg import pytest +from psycopg.sql import SQL, Identifier from psycodict.utils import LockError +# A schema of its own, so that the same table name can exist twice in the +# cluster while these tests run. +LOCK_OTHER = "psycodict_lock_other" + + def raw_connection(): """ A second connection to the test database, configured from the same @@ -151,6 +157,53 @@ def test_check_locks_warns_for_reload_when_table_locked(db, empty_table, capsys) conn.close() +def test_locks_in_another_schema_are_not_this_schema_s(db, empty_table): + """ + Lock discovery reports relation names, so it has to ask the server for the + locks in this database's schema. Unfiltered, a lock on a same-named table + anywhere else in the cluster -- another schema, or another session's + temporary table -- was reported as a lock on this one, and refused a write + that nothing was blocking. + """ + name = empty_table.search_table + db._execute(SQL("CREATE SCHEMA IF NOT EXISTS {0}").format(Identifier(LOCK_OTHER))) + db._execute( + SQL("CREATE TABLE IF NOT EXISTS {0}.{1} (id bigint)").format( + Identifier(LOCK_OTHER), Identifier(name) + ) + ) + holder = raw_connection() + try: + holder.execute( + SQL("LOCK TABLE {0}.{1} IN ACCESS EXCLUSIVE MODE").format( + Identifier(LOCK_OTHER), Identifier(name) + ) + ) + # the lock is real, and held by somebody else ... + assert any( + row[1] == "AccessExclusiveLock" and row[2] == holder.info.backend_pid + for row in db._execute( + SQL( + "SELECT t.relname, l.mode, l.pid FROM pg_locks l " + "JOIN pg_stat_all_tables t ON l.relation = t.relid " + "WHERE l.granted AND t.schemaname = %s AND t.relname = %s" + ), + [LOCK_OTHER, name], + ) + ) + # ... but it is not a lock on this schema's table of that name + assert db._table_locked(name, "all") == [] + assert empty_table._table_locked(name, "all") == [] + # so a write that would conflict with it goes ahead + empty_table._check_locks("insert_many") + finally: + holder.rollback() + holder.close() + db._execute( + SQL("DROP SCHEMA IF EXISTS {0} CASCADE").format(Identifier(LOCK_OTHER)) + ) + + def test_check_locks_reload_is_quiet_without_locks(empty_table, capsys): assert empty_table._check_locks("reload") is None assert "Warning" not in capsys.readouterr().out diff --git a/tests/test_schema_contract.py b/tests/test_schema_contract.py new file mode 100644 index 0000000..9879385 --- /dev/null +++ b/tests/test_schema_contract.py @@ -0,0 +1,326 @@ +# -*- coding: utf-8 -*- +""" +A database operates in exactly one schema. + +Every relation psycodict 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 can neither stand in for one of these nor be +merged with it. Before this, catalog queries were a mixture of hard-coded +``'public'`` and no filter at all, so two schemas holding ``same_name`` gave +answers assembled from both. +""" +import pytest + +from psycopg.sql import SQL, Identifier + +from psycodict.database import PostgresDatabase +from psycodict.validation import InvalidDefinitionError + + +OTHER = "psycodict_other" + +# A schema whose name is a valid identifier but *not* a valid search_path, and +# the two schemas its text parses into if it is written to the setting as text. +SPLIT = "psycodict_path_a, psycodict_path_b" +DECOYS = ["psycodict_path_a", "psycodict_path_b"] + + +def relations_in(db, schema): + """ + The names of the relations in ``schema``, asked of the catalog rather than + of the search path. + """ + return { + rec[0] + for rec in db._execute( + SQL( + "SELECT c.relname FROM pg_catalog.pg_class c " + "JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = %s" + ), + [schema], + ) + } + + +@pytest.fixture +def two_schemas(db): + """ + ``.same_name`` in the configured schema and in a second one, with + different columns, plus the metadata tables the second schema needs to be + connectable in its own right. + """ + db._execute(SQL("CREATE SCHEMA IF NOT EXISTS {0}").format(Identifier(OTHER))) + db._execute( + SQL("CREATE TABLE IF NOT EXISTS {0}.{1} (id bigint, only_here text)").format( + Identifier(OTHER), Identifier("same_name") + ) + ) + db._execute( + SQL("CREATE TABLE IF NOT EXISTS {0} (id bigint, label text, n integer)").format( + Identifier("same_name") + ) + ) + yield db + db._execute(SQL("DROP TABLE IF EXISTS {0}").format(Identifier("same_name"))) + db._execute(SQL("DROP SCHEMA IF EXISTS {0} CASCADE").format(Identifier(OTHER))) + + +def test_the_default_schema_is_public(db): + assert db.schema == "public" + + +def test_an_invalid_schema_name_is_refused(config): + for bad in ["", "a" * 64, "with\x00nul"]: + with pytest.raises((InvalidDefinitionError, ValueError)): + PostgresDatabase(config=config, schema=bad) + + +def test_schema_is_not_passed_to_the_driver(db): + """ + psycopg.connect has no ``schema`` parameter; it is psycodict's own setting + and must not end up among the connection overrides. + """ + assert "schema" not in db._connect_kwargs + assert "schema" not in db._connection_options() + + +def test_the_session_search_path_is_the_selected_schema(db): + cur = db._execute(SQL("SELECT current_schema()")) + assert cur.fetchone()[0] == db.schema + + +def test_a_reconnect_keeps_the_schema(db): + db.reset_connection() + cur = db._execute(SQL("SELECT current_schema()")) + assert cur.fetchone()[0] == db.schema + + +# --------------------------------------------------------------------------- +# the schema is one search_path entry, and pg_temp comes after it +# --------------------------------------------------------------------------- + +@pytest.fixture +def split_schemas(db): + """ + A schema whose name contains a comma, alongside the two schemas that name's + text would parse into. All three exist, so that a search_path written as + text selects a real schema rather than failing. + """ + for name in [SPLIT] + DECOYS: + db._execute(SQL("CREATE SCHEMA IF NOT EXISTS {0}").format(Identifier(name))) + yield db + for name in [SPLIT] + DECOYS: + db._execute(SQL("DROP SCHEMA IF EXISTS {0} CASCADE").format(Identifier(name))) + + +def test_a_comma_in_the_schema_name_is_not_a_path_separator(split_schemas, config): + """ + search_path is a comma-separated *list*, so a schema name written into it + as text is parsed rather than used. Passed as text, this name selects + ``psycodict_path_a`` -- and ``create=True`` then bootstraps the metadata + tables there, while every catalog query keeps binding the literal name and + reports a schema the session is not actually using. + """ + other = PostgresDatabase(config=config, schema=SPLIT, create=True) + try: + assert other.schema == SPLIT + cur = other._execute(SQL("SELECT current_schema()")) + assert cur.fetchone()[0] == SPLIT + + # the metadata went to the schema that was asked for ... + created = relations_in(other, SPLIT) + assert {"meta_tables", "meta_indexes", "meta_constraints"} <= created + # ... and to neither of the schemas its text parses into + for decoy in DECOYS: + assert relations_in(other, decoy) == set() + + # a replacement connection is pinned the same way + other.reset_connection() + cur = other._execute(SQL("SELECT current_schema()")) + assert cur.fetchone()[0] == SPLIT + for decoy in DECOYS: + assert relations_in(other, decoy) == set() + finally: + other.conn.close() + + +def test_a_schema_needing_quotes_is_one_path_entry(db, config): + """ + Mixed case and spaces: legal in a quoted identifier, and (unlike the comma) + rejected outright by the server when written to the setting as text. + """ + name = "Psycodict Mixed Case" + db._execute(SQL("CREATE SCHEMA IF NOT EXISTS {0}").format(Identifier(name))) + try: + other = PostgresDatabase(config=config, schema=name, create=True) + try: + cur = other._execute(SQL("SELECT current_schema()")) + assert cur.fetchone()[0] == name + assert "meta_tables" in relations_in(other, name) + finally: + other.conn.close() + finally: + db._execute(SQL("DROP SCHEMA IF EXISTS {0} CASCADE").format(Identifier(name))) + + +def test_the_dollar_user_token_is_refused(config): + """ + PostgreSQL unquotes each search_path element before looking for ``$user``, + so ``"$user"`` is replaced by the connecting role's name and can never + select a schema of that name. Any other spelling is an ordinary identifier. + """ + with pytest.raises(InvalidDefinitionError, match=r"\$user"): + PostgresDatabase(config=config, schema="$user") + + +def test_a_missing_schema_is_refused_rather_than_fallen_back_from(config): + """ + PostgreSQL drops a search_path entry naming a schema that does not exist + (or one the role has no USAGE on) and resolves names further down the path. + With pg_temp there, an unqualified CREATE TABLE would make a temporary one. + """ + with pytest.raises(ValueError, match="Could not operate in schema"): + PostgresDatabase(config=config, schema="psycodict_no_such_schema") + + +def test_a_temporary_relation_does_not_shadow_the_selected_schema(db): + """ + pg_temp is searched first unless it is named, so a temporary relation would + otherwise stand in for meta_tables or for a search table. + """ + persistent = db._execute( + SQL( + "SELECT c.oid FROM pg_catalog.pg_class c " + "JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = %s AND c.relname = 'meta_tables'" + ), + [db.schema], + ).fetchone()[0] + db._execute(SQL("CREATE TEMP TABLE meta_tables (not_the_real_one text)")) + try: + resolved = db._execute(SQL("SELECT %s::regclass::oid"), ["meta_tables"]) + assert resolved.fetchone()[0] == persistent + # and the same thing from the other side: an ordinary metadata read + # reaches the persistent table. The temporary one has no ``name`` + # column, so this would raise UndefinedColumn if it were resolved. + cur = db._execute(SQL("SELECT name FROM meta_tables")) + assert [col.name for col in cur.description] == ["name"] + finally: + db._execute(SQL("DROP TABLE IF EXISTS pg_temp.meta_tables")) + + +# --------------------------------------------------------------------------- +# catalog queries see one schema +# --------------------------------------------------------------------------- + +def test_table_exists_does_not_see_the_other_schema(two_schemas): + db = two_schemas + assert db._table_exists("same_name") + db._execute(SQL("DROP TABLE {0}").format(Identifier("same_name"))) + # still present in the other schema, but not in ours + assert not db._table_exists("same_name") + db._execute( + SQL("CREATE TABLE {0} (id bigint, label text, n integer)").format( + Identifier("same_name") + ) + ) + + +def test_all_tablenames_does_not_merge_schemas(two_schemas): + db = two_schemas + names = db._all_tablenames() + assert names.count("same_name") == 1 + + +def test_schema_relations_is_confined(two_schemas): + db = two_schemas + relations = db._schema_relations() + assert "same_name" in relations + db._execute(SQL("DROP TABLE {0}").format(Identifier("same_name"))) + assert "same_name" not in db._schema_relations() + db._execute( + SQL("CREATE TABLE {0} (id bigint, label text, n integer)").format( + Identifier("same_name") + ) + ) + + +def test_column_discovery_does_not_mix_the_two(two_schemas): + """ + The columns of ``same_name`` differ between the schemas. Reading them + unfiltered used to raise "Type mismatch" or silently union them. + """ + db = two_schemas + cols, col_type, has_id = db._column_types("same_name") + assert sorted(cols) == ["label", "n"] + assert "only_here" not in col_type + + +def test_relation_columns_is_confined(two_schemas): + db = two_schemas + assert db._relation_columns("same_name") == {"id", "label", "n"} + + +def test_index_lookups_are_confined(two_schemas): + db = two_schemas + db._execute( + SQL("CREATE INDEX {0} ON {1}.{2} (id)").format( + Identifier("same_name_idx"), Identifier(OTHER), Identifier("same_name") + ) + ) + # the index exists, but in the other schema + assert not db._index_exists("same_name_idx", "same_name") + assert db._index_exists("same_name_idx", "same_name") is False + assert "same_name_idx" not in db._list_indexes("same_name") + + +def test_constraint_lookups_are_confined(two_schemas): + db = two_schemas + db._execute( + SQL("ALTER TABLE {0}.{1} ADD CONSTRAINT {2} CHECK (id > 0)").format( + Identifier(OTHER), Identifier("same_name"), Identifier("same_name_chk") + ) + ) + assert not db._constraint_exists("same_name_chk", "same_name") + assert "same_name_chk" not in db._list_constraints("same_name") + + +def test_table_sizes_report_one_schema(two_schemas): + db = two_schemas + sizes = db.table_sizes() + assert sizes.get("same_name") is not None + + +# --------------------------------------------------------------------------- +# a database pointed at the other schema sees only that one +# --------------------------------------------------------------------------- + +def test_another_schema_does_not_inherit_the_metadata_tables(two_schemas, config): + """ + The clearest demonstration of the confinement: meta_tables exists, but not + in the other schema, so connecting there without create=True is refused + rather than quietly operating on the configured schema's metadata. + """ + with pytest.raises(ValueError, match="metadata tables"): + PostgresDatabase(config=config, schema=OTHER) + + +def test_a_database_in_another_schema_sees_only_its_own(two_schemas, config): + db = two_schemas + other = PostgresDatabase(config=config, schema=OTHER, create=True) + try: + assert other.schema == OTHER + cur = other._execute(SQL("SELECT current_schema()")) + assert cur.fetchone()[0] == OTHER + + # same relation name, and each database sees its own columns + assert other._relation_columns("same_name") == {"id", "only_here"} + assert db._relation_columns("same_name") == {"id", "label", "n"} + + # each schema has its own metadata, and the configured schema's + # search tables are not visible from the other one + assert "same_name" in other._all_tablenames() + assert set(db.tablenames) - set(other.tablenames) == set(db.tablenames) + finally: + other.conn.close() diff --git a/tests/test_stats.py b/tests/test_stats.py index f64c8cb..f31482b 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -31,6 +31,11 @@ def saving_table(table_factory): table = table_factory() table.insert_many([sample_row(i) for i in range(200)]) table.stats.saving = True + # insert_many invalidates the statistics, and a cached count may not be + # used while they are invalid. Nothing is cached yet and the total is + # maintained by the insert, so the caches do agree with the data: say so, + # which is the state these tests are about. + table._restore_stats() return table diff --git a/tests/test_stats_duplicates.py b/tests/test_stats_duplicates.py index 26141a9..94d0acf 100644 --- a/tests/test_stats_duplicates.py +++ b/tests/test_stats_duplicates.py @@ -39,6 +39,11 @@ def saving_table(table_factory): table = table_factory() table.insert_many([sample_row(i) for i in range(200)]) table.stats.saving = True + # insert_many invalidates the statistics, and a cached count may not be + # used while they are invalid. Nothing is cached yet and the total is + # maintained by the insert, so the caches do agree with the data: say so, + # which is the state these tests are about. + table._restore_stats() return table @@ -265,6 +270,10 @@ def constrained_table(table_factory): [{"n": i, "a": i % 3, "z": i % 2, "label": "l%d" % i} for i in range(30)] ) table.stats.saving = True + # As in ``saving_table``: nothing is cached yet, so the (empty) caches do + # agree with the data, and these tests are about what add_numstats writes + # rather than about invalidation. + table._restore_stats() return table diff --git a/tests/test_stats_validity.py b/tests/test_stats_validity.py new file mode 100644 index 0000000..a19b9c4 --- /dev/null +++ b/tests/test_stats_validity.py @@ -0,0 +1,325 @@ +# -*- coding: utf-8 -*- +""" +``stats_valid`` means what it says: no cached answer survives it being false. + +Before this, write paths cleared the flag but read paths ignored it, so a +count cached before a ``restat=False`` write kept being served afterwards. +These tests pin the whole contract -- which lookups are gated, which two are +deliberately not, and how the flag is restored. +""" +import pytest + +from psycopg.sql import SQL, Identifier + +from conftest import sample_row + + +@pytest.fixture +def cached_table(table_factory): + """ + A saving table with statistics computed and recorded, and the flag true. + """ + table = table_factory() + table.insert_many([sample_row(i) for i in range(200)]) + table.stats.saving = True + table.stats.refresh_stats() + table._restore_stats() if hasattr(table, '_restore_stats') else None + return table + + +def stats_valid_in_meta(table): + """ + The flag as stored, rather than as cached on the Python object. + """ + cur = table._execute( + SQL("SELECT stats_valid FROM meta_tables WHERE name = %s"), + [table.search_table], + ) + return cur.fetchone()[0] + + +# --------------------------------------------------------------------------- +# every gated lookup reports a miss while the flag is false +# --------------------------------------------------------------------------- + +def test_a_stale_count_is_not_served_after_an_unrestatted_write(cached_table): + """ + The case from the review: cache a nonempty query, change the rows it + matches without refreshing, and the old number kept coming back. + """ + query = {"flag": True} + before = cached_table.stats.count(query, record=True) + assert cached_table.stats.quick_count(query) == before + + cached_table.update(query, {"flag": False}, restat=False) + assert not cached_table._stats_valid + + # the cached row is still physically there ... + cur = cached_table._execute( + SQL("SELECT count FROM {0} WHERE cols = %s").format( + Identifier(cached_table.stats.counts) + ), + [cached_table.stats._split_dict(query)[0]], + ) + assert cur.rowcount + + # ... but it is not an answer any more, and count() computes the truth + assert cached_table.stats.quick_count(query) is None + assert cached_table.stats.count(query) == 0 + + +def test_quick_count_distinct_is_gated(cached_table): + cols = ["flag"] + cached_table.stats._slow_count_distinct(cols, record=True) + assert cached_table.stats.quick_count_distinct(cols) is not None + cached_table._break_stats() + assert cached_table.stats.quick_count_distinct(cols) is None + + +def test_quick_statistic_is_gated(cached_table): + from psycodict.encoding import Json + + assert cached_table.stats.max("n") == 199 + ccols, cvals = Json([]), Json([]) + assert cached_table.stats._quick_statistic("n", ccols, cvals, "max") is not None + cached_table._break_stats() + assert cached_table.stats._quick_statistic("n", ccols, cvals, "max") is None + # and the public method still returns the right answer, the slow way + assert cached_table.stats.max("n") == 199 + + +def test_the_recompute_predicates_are_not_gated(cached_table): + """ + _has_stats and _has_numstats answer "is this recorded", which is what + add_stats and column_counts use to decide whether to compute. Gating them + makes those recompute the whole family on every call and never converge, + because only refresh_stats restores the flag: measured on the LMFDB, that + turned a four-minute downstream suite into one still running after + forty-five. They stay ungated, and the staleness that leaves is recorded + in the test below. + """ + from psycodict.encoding import Json + + cached_table.stats.add_stats(["flag"]) + cached_table.stats.add_numstats("num", ["flag"]) + jcols, empty = Json(["flag"]), Json([]) + assert cached_table.stats._has_stats(jcols, empty, empty, None) + assert cached_table.stats._has_numstats(Json(["num"]), Json(["flag"]), empty, None) + + cached_table._break_stats() + assert cached_table.stats._has_stats(jcols, empty, empty, None) + assert cached_table.stats._has_numstats(Json(["num"]), Json(["flag"]), empty, None) + + +@pytest.mark.xfail( + reason="column_counts can still report a value recorded before an " + "unrefreshed write; closing this needs freshness per statistic " + "rather than one flag per table", + strict=True, +) +def test_column_counts_can_still_be_stale(cached_table): + """ + The gap left by the paragraph above, pinned so that it is a known quantity + rather than a surprise, and so that a future per-statistic freshness change + turns this green. + """ + cached_table.stats.add_stats(["flag"]) + flagged = cached_table.stats.column_counts("flag")[True] + assert flagged > 0 + cached_table.update({"flag": True}, {"flag": False}, restat=False) + assert cached_table.stats.column_counts("flag").get(True, 0) == 0 + + +def test_null_counts_is_not_gated(cached_table): + """ + A miss here costs one full count per search column, not one bounded query, + so null_counts reads what is recorded like the other bulk paths. This is + the call LMFDB's results_complete makes for every query it checks, and + gating it is what took the downstream suite past forty-five minutes. + """ + cached_table.stats.refresh_null_counts() + before = cached_table.stats.null_counts() + cached_table._break_stats() + assert cached_table.stats.null_counts() == before + + +# --------------------------------------------------------------------------- +# what is deliberately not gated +# --------------------------------------------------------------------------- + +def test_the_empty_query_total_survives_invalidation(cached_table): + """ + total is maintained on every write, so it is exact regardless of the flag; + gating it would make count() do a full scan after every insert. + """ + cached_table.insert_many([sample_row(1000)], restat=False) + assert not cached_table._stats_valid + assert cached_table.stats.quick_count({}) == 201 + assert cached_table.count() == 201 + + +def test_status_still_reports_what_the_cache_holds(cached_table): + """ + refresh_stats learns which statistics to recompute from _status, so gating + it would make an invalid table forget what it is supposed to have. + """ + cached_table.stats.add_stats(["flag"]) + before = cached_table.stats._status() + cached_table._break_stats() + assert cached_table.stats._status() == before + + +def test_a_suffixed_table_is_not_gated_by_the_live_flag(cached_table): + """ + A _tmp copy carries its own caches; stats_valid describes the live table. + """ + table = cached_table + assert table.stats.count({"flag": True}, record=True) > 0 + tmp = table.search_table + "_tmp" + table._db._execute( + SQL("CREATE TABLE {0} AS TABLE {1}").format( + Identifier(tmp), Identifier(table.search_table) + ) + ) + table._db._execute( + SQL("CREATE TABLE {0} AS TABLE {1}").format( + Identifier(table.stats.counts + "_tmp"), Identifier(table.stats.counts) + ) + ) + try: + table._break_stats() + assert table.stats.quick_count({"flag": True}) is None + assert table.stats.quick_count({"flag": True}, suffix="_tmp") is not None + finally: + for name in (tmp, table.stats.counts + "_tmp"): + table._db._execute(SQL("DROP TABLE IF EXISTS {0}").format(Identifier(name))) + + +# --------------------------------------------------------------------------- +# restoring the flag +# --------------------------------------------------------------------------- + +def test_refresh_stats_restores_the_flag_and_the_cache(cached_table): + cached_table.update({"flag": True}, {"flag": False}, restat=False) + assert not cached_table._stats_valid + assert stats_valid_in_meta(cached_table) is False + + cached_table.stats.refresh_stats() + assert cached_table._stats_valid + assert stats_valid_in_meta(cached_table) is True + assert cached_table.stats.count({"flag": False}, record=True) == 200 + assert cached_table.stats.quick_count({"flag": False}) == 200 + + +def test_a_failed_refresh_leaves_the_table_invalid(cached_table, monkeypatch): + """ + The flag is set inside the refresh transaction, so a failure part-way + cannot leave a table claiming a cache it does not have. + """ + cached_table._break_stats() + + def boom(*args, **kwargs): + raise RuntimeError("refresh blew up") + + monkeypatch.setattr(cached_table.stats, "refresh_null_counts", boom) + with pytest.raises(RuntimeError): + cached_table.stats.refresh_stats() + cached_table._db.conn.rollback() + + cached_table._refresh() + assert not cached_table._stats_valid + assert stats_valid_in_meta(cached_table) is False + + +def test_refreshing_a_tmp_copy_does_not_validate_the_live_table(cached_table): + table = cached_table + for base in (table.search_table, table.stats.counts, table.stats.stats): + table._db._execute( + SQL("CREATE TABLE {0} AS TABLE {1}").format( + Identifier(base + "_tmp"), Identifier(base) + ) + ) + try: + table._break_stats() + table.stats.refresh_stats(suffix="_tmp") + assert not table._stats_valid + assert stats_valid_in_meta(table) is False + finally: + for base in (table.search_table, table.stats.counts, table.stats.stats): + table._db._execute( + SQL("DROP TABLE IF EXISTS {0}").format(Identifier(base + "_tmp")) + ) + + +# --------------------------------------------------------------------------- +# planner statistics +# --------------------------------------------------------------------------- + +def analyzed(table, name=None): + """ + Whether PostgreSQL holds planner statistics for a relation. + """ + cur = table._execute( + SQL( + "SELECT c.reltuples >= 0 AND s.last_analyze IS NOT NULL " + "OR s.last_analyze IS NOT NULL " + "FROM pg_class c " + "JOIN pg_namespace n ON n.oid = c.relnamespace " + "LEFT JOIN pg_stat_all_tables s ON s.relid = c.oid " + "WHERE n.nspname = current_schema() AND c.relname = %s" + ), + [name or table.search_table], + ) + row = cur.fetchone() + return bool(row and row[0]) + + +def test_a_reload_analyzes_before_the_swap(cached_table, tmp_path): + """ + A bulk-loaded relation has no planner statistics until autovacuum reaches + it, and a rename carries the catalog entry along, so the _tmp copy is + analyzed while it is still _tmp. + """ + searchfile = tmp_path / "data.txt" + cached_table.copy_to(str(searchfile)) + + seen = [] + original = type(cached_table)._analyze + + def record(self, tables, suffix=""): + seen.append((list(tables), suffix)) + return original(self, tables, suffix) + + type(cached_table)._analyze = record + try: + cached_table.reload(str(searchfile)) + finally: + type(cached_table)._analyze = original + + assert seen, "the reload did not analyze anything" + tables, suffix = seen[0] + assert suffix == "_tmp" + assert cached_table.search_table in tables + assert analyzed(cached_table) + + +def test_copy_from_analyzes_the_live_table(cached_table, table_factory, tmp_path): + searchfile = tmp_path / "more.txt" + cached_table.copy_to(str(searchfile)) + target = table_factory() + + seen = [] + original = type(target)._analyze + + def record(self, tables, suffix=""): + seen.append((list(tables), suffix)) + return original(self, tables, suffix) + + type(target)._analyze = record + try: + target.copy_from(str(searchfile), restat=False) + finally: + type(target)._analyze = original + + assert seen == [([target.search_table], "")] + assert target.count() == 200 diff --git a/tests/test_table_like_size.py b/tests/test_table_like_size.py index 65be642..b3ce039 100644 --- a/tests/test_table_like_size.py +++ b/tests/test_table_like_size.py @@ -19,6 +19,10 @@ from psycopg.sql import SQL, Identifier +# A schema for holding a same-named decoy of a table under test. +DECOY = "psycodict_storage_decoy" + + def fresh_name(): """ A table name that no other test (or run) will collide with. @@ -137,6 +141,51 @@ def test_create_table_like_preserves_column_storage_settings(db, empty_table, tr assert settings["label"][1] == "" +def test_column_storage_settings_are_read_from_the_selected_schema( + db, empty_table, transient +): + """ + The storage settings are read with a bound schema predicate rather than by + casting the table name to ``regclass``, which resolves through the search + path. A same-named relation the session would reach first -- another + schema put ahead of this one, or a temporary table -- must not be the one + whose settings are copied. + """ + source = empty_table.search_table + db._execute( + SQL("ALTER TABLE {0} ALTER COLUMN {1} SET STORAGE EXTERNAL").format( + Identifier(source), Identifier("mat") + ) + ) + target = fresh_name() + db.create_table_like(target, empty_table) + transient.append(target) + assert _column_settings(db, target, ["mat"])["mat"][0] == "e" + + db._execute(SQL("CREATE SCHEMA IF NOT EXISTS {0}").format(Identifier(DECOY))) + try: + # a decoy of the same name, whose mat column keeps the type default + db._execute( + SQL("CREATE TABLE {0}.{1} (mat numeric[])").format( + Identifier(DECOY), Identifier(source) + ) + ) + # ... reachable before the real one, for the rest of this session + db._execute( + SQL("SET search_path TO {0}, {1}, pg_temp").format( + Identifier(DECOY), Identifier(db.schema) + ) + ) + db._clone_storage_settings(target, empty_table) + finally: + # a fresh connection, pinned the way this database pins every one of + # them, whatever the assertions above did + db.reset_connection() + db._execute(SQL("DROP SCHEMA IF EXISTS {0} CASCADE").format(Identifier(DECOY))) + # the decoy's default storage did not overwrite the source's EXTERNAL + assert _column_settings(db, target, ["mat"])["mat"][0] == "e" + + ################################################################## # On-disk size parity # ##################################################################