diff --git a/CHANGELOG.md b/CHANGELOG.md index 98cafa4..57eb5af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -339,6 +339,119 @@ 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 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 the database's, and it is tested in the same statement as the + cache.** Each gated lookup carries `AND EXISTS (SELECT 1 FROM meta_tables + WHERE name = %s AND stats_valid)` into the `SELECT` that reads the cached + row, rather than consulting the `_stats_valid` attribute a table object was + built with. That attribute is a copy: another process's `restat=False` write + moves the row and not the copy, so a second webserver process would have gone + on serving the counts it had cached, and a rolled-back transaction moves the + copy and not the row. Testing the flag in one statement and reading the cache + in the next would leave a window between two snapshots for a write to commit + in, so both readings go into one statement. `count()` on an empty query + likewise answers from `meta_tables.total` rather than from `self.total` -- + a single-row metadata lookup, not a scan -- so a total another process + changed is the one served. `_break_stats` and `_restore_stats` now issue + their `UPDATE` unconditionally, since a transition skipped because the local + copy already said so is a transition skipped on stale information; the row + lock that `UPDATE` takes is also what serializes a refresh against concurrent + writers, a refresh now claiming the row before it rebuilds anything rather + than only restoring it at the end. +- **Every replacement and bulk path makes its validity transition, in the + transaction that does the work.** `update_from_file` (in place or not), + `rewrite`, `reload`, `reload_all`, `reload_revert` and the staged swaps could + all change the live data while leaving `stats_valid = true`, so the new gate + went on serving the old counts. `_swap_in_tmp` and `reload_final_swap` now + take the intended state as an argument and write it with the renames: + true when the counts and stats arriving at the live names were rebuilt or + loaded for the data arriving with them, false when the old cache companions + are kept. A `metafile`'s own `stats_valid` is overruled by what the swap + actually did, `reload_revert` invalidates (a backup carries no validity bit + of its own) and also recounts the total, and nothing inherits the old live + table's flag. +- **`reload` returns what it prepared, and `reload_all` finalizes that.** + `reload_all` runs every reload before any swap, and built each deferred + swap's list of relations from the files in the input folder. That is not the + list `reload` prepared: a saving table's reload always readies *both* cache + companions, refreshing them when a file is missing, so a folder with no + `_counts.txt` produced a swap naming only the search table — stranding the + refreshed companions under `_tmp` and leaving the old live ones, describing + data that had just been replaced, paired with the new data and (once the + transitions above were added) marked valid. `reload` now returns a + `ReloadPlan` with the exact swap list, the intended `stats_valid` and the + resolved `ordered`, and `reload_all` passes it straight through, so + preparation and finalization cannot drift apart. `stats_valid` is true only + when both cache companions are in the list actually being swapped. +- **Cache maintenance asks whether the row exists, not whether it may be + served.** `_record_count`, `_record_count_distinct` and `_record_statistic` + chose between `INSERT` and `UPDATE`/`DELETE` by calling the public lookups, + which under the new gate answer "missing" for a row that is physically there + -- on every invalid table, which is every table they run on. Each write left + a second row behind under a key the rest of the code takes to identify at + most one; the plainest case was a write clearing the flag and then + duplicating the `{}` total row it maintains. They now use private physical + lookups that ignore the flag. This is also the first thing ever to reach + `_record_count_distinct`'s update statement, which named a column `stats` + that has always been called `stat`. +- **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. + ### 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..cb59570 100644 --- a/DataManagement.md +++ b/DataManagement.md @@ -108,7 +108,100 @@ 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. + +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. + +### What `stats_valid` promises + +`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. + +The promise is made by the database, not by the process making it. Each of +those lookups tests `meta_tables.stats_valid` **in the same `SELECT`** that +reads the cached row, so a cached answer can only be served under a snapshot +that says the cache is valid; a table object's `_stats_valid` attribute is +advisory, and correctness never rests on it. That is what makes the flag hold +in a deployment running several webserver processes: one process's +`restat=False` write stops every other process from serving the affected +counts, without any of them being told. For the same reason `count()` on an +empty query answers from `meta_tables.total` rather than from the copy its +table object was built with, which is a single-row metadata lookup and not a +scan. + +`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; a rollback +likewise leaves the stored flag false whatever the Python object was left +saying. Refreshing a `_tmp` copy does not validate the live table. + +A live `refresh_stats()` claims the table's `meta_tables` row at the start, by +marking it invalid, and holds that row lock for the whole rebuild. Every +library write marks the same row, so a write that overlaps a refresh waits for +it, and the result is one of the two orderings rather than a race: either the +write went first and its rows are in the caches the refresh commits, or the +refresh went first and the write's invalidation lands after it, leaving the +table invalid. + +### Which operations set it, and to what + +Every write and swap makes its validity transition **in the same transaction as +the data change or rename**, so the flag and the relations cannot come apart: + +| operation | leaves `stats_valid` | +| --- | --- | +| `insert_many`, `upsert`, `update`, `delete`, `copy_from`, in-place `update_from_file` | false, then true if `saving` and `restat` refreshed the caches | +| non-inplace `update_from_file`, `rewrite` | true iff `saving` and `restat` (which is exactly when the rebuilt `_tmp` counts and stats are swapped in with the data) | +| `reload`, `reload_all` | true iff `saving` and either `restat`, or both a `countsfile` and a `statsfile` were supplied | +| staged commit, `staged_force_swap` | false — the staged counts and stats tables are empty, not refreshed | +| `reload_revert` | false | +| `refresh_stats()` on the live table | true | + +`reload_final_swap` and `_swap_in_tmp` take the intended state as a +`stats_valid=` argument, defaulting to false. Nothing inherits the old live +table's flag, which was a fact about data that is no longer there. + +A deferred final swap (`reload` with `final_swap=False`, which is how +`reload_all` runs every reload before any swap) must finalize what its reload +prepared rather than working it out again afterwards: `reload` returns a +`ReloadPlan` giving the exact list of base names whose `_tmp` copies belong at +the live names, the `stats_valid` the swap should write, and whether the ids +were resorted. Pass those three to `reload_final_swap`. The distinction is not +academic — a saving table's `reload` prepares *both* cache companions whatever +files it was given, refreshing them when one is missing, so a swap list +reconstructed from the input filenames can name fewer relations than were +prepared, stranding a refreshed companion under `_tmp` and leaving the old live +one paired with new data. `stats_valid` is true only when both companions are +in the list actually being swapped and were either loaded for this search file +or rebuilt from its `_tmp` data. + +Two consequences worth spelling out. A `metafile` carries a `stats_valid` +column, and the swap's own answer overrules it: the file records what was true +of the table it was exported from, at export time, and cannot know whether the +relations being swapped in were rebuilt. And `reload_revert` clears the flag +because a backup carries no validity bit of its own — `meta_tables` has one row +and it stayed with the live name — so a backup taken while the table was +invalid could otherwise come back under a flag that had since been set true; it +also recounts the total for the same reason. + +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. ### Resorting is disabled diff --git a/psycodict/database.py b/psycodict/database.py index 7cd95f9..cfcf76e 100644 --- a/psycodict/database.py +++ b/psycodict/database.py @@ -2191,12 +2191,9 @@ def reload_all( self.create_table(tablename, search_columns, None, force_description=False) for tablename in self.tablenames: - included = [] - searchfile = data_folder / (tablename + ".txt") if not searchfile.exists(): continue - included.append(tablename) table = self[tablename] @@ -2208,15 +2205,11 @@ def reload_all( ) countsfile = data_folder / (tablename + "_counts.txt") - if countsfile.exists(): - included.append(tablename + "_counts") - else: + if not countsfile.exists(): countsfile = None statsfile = data_folder / (tablename + "_stats.txt") - if statsfile.exists(): - included.append(tablename + "_stats") - else: + if not statsfile.exists(): statsfile = None indexesfile = data_folder / (tablename + "_indexes.txt") @@ -2242,15 +2235,24 @@ def reload_all( constraintsfile, metafile, ), - included, ) ) tablenames.append(tablename) print("Reloading {0}".format(", ".join(tablenames))) failures = [] - for table, filedata, included in file_list: + # Every reload happens before any swap, so what each one prepared + # has to be carried to the second pass. Deriving it there instead + # -- from which files the folder happened to contain -- is how the + # two came apart: a saving table's reload always prepares both + # cache companions, refreshing them when a file is missing, so a + # folder with no _counts.txt yielded a swap list naming only the + # search table. The refreshed companions then stayed under _tmp + # while the old live ones, describing data that had just been + # replaced, were marked valid. + plans = {} + for table, filedata in file_list: try: - table.reload( + plans[table.search_table] = table.reload( *filedata, resort=resort, restat=restat, @@ -2265,10 +2267,17 @@ def reload_all( else: traceback.print_exc() failures.append(table) - for table, filedata, included in file_list: + for table, filedata in file_list: if table in failures: continue - table.reload_final_swap(tables=included, metafile=filedata[-1], sep=sep) + plan = plans[table.search_table] + table.reload_final_swap( + tables=plan.tables, + metafile=filedata[-1], + sep=sep, + ordered=plan.ordered, + stats_valid=plan.stats_valid, + ) if failures: print("Reloaded %s" % (", ".join(tablenames))) 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..65f3ec6 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 @@ -260,7 +260,9 @@ def _init_total(self, total=None): ``_set_total``, which stores the result of a recount. """ if total is None: - total = self.quick_count({}, startup=True) + # The physical row, not the maintained total: this runs precisely + # when there is no maintained total to read. + total = self._cached_count({}) if total is None: total = self._slow_count({}, extra=False) self.total = total @@ -269,6 +271,136 @@ 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 gated on it and reports a miss + instead, leaving the caller to compute or recompute. + + The flag is read from ``meta_tables`` rather than from + ``self.table._stats_valid``, which is a copy taken when the table + object was built: another process invalidating or refreshing the table + moves the row and not the copy, and the copy is what a second + webserver process would otherwise be serving stale counts on the + strength of. The copy is refreshed from the row here, and remains + advisory. + + The gated lookups do not call this. A separate "is it valid" query + followed by a cache query is still two snapshots, and another + connection's write can commit between them; they inline + :meth:`_cache_gate` instead, so that one statement reads both. This + is the readable form of the same question, for callers -- and tests -- + that want the flag itself. + + Two things are deliberately outside the 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. Neither are the *physical* lookups + (``_cached_count`` and friends), which ask whether a row with a given + cache key exists rather than whether its value is an answer; cache + maintenance runs on invalid tables by definition. + + 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 + cur = self._execute( + SQL("SELECT stats_valid FROM meta_tables WHERE name = %s"), + [self.search_table], + ) + row = cur.fetchone() + # A table with no meta_tables row has no assertion to go on, and the + # conservative reading of a missing assertion is that it does not hold. + valid = bool(row) and bool(row[0]) + self.table._stats_valid = valid + return valid + + def _cache_gate(self, suffix=""): + """ + A ``WHERE``-clause fragment, and its parameters, restricting a cache + read to the case where the live table's statistics are valid. + + Returned as SQL rather than evaluated, so that the caller can put it in + the statement that reads the cached row. Testing the flag in one + statement and reading the cache in the next leaves a window between two + snapshots, and a ``restat=False`` write committing inside it is exactly + the case this is here to stop: the answer served would be one the + database had already disowned by the time it was served. One statement + has one snapshot, and no window. + + A suffixed relation is not gated; see :meth:`_may_use_cache` for that + and for the rest of the contract. + + OUTPUT: + + A pair ``(sql, values)``, to be appended to the ``WHERE`` clause and to + the parameter list respectively. + """ + if suffix: + return SQL(""), [] + return ( + SQL(" AND EXISTS (SELECT 1 FROM meta_tables WHERE name = %s AND stats_valid)"), + [self.search_table], + ) + + def _live_total(self): + """ + The live table's row count, as recorded in ``meta_tables``. + + ``self.total`` is a copy taken when this object was built; a write from + another process updates the row rather than the copy, so an old table + object answering ``count()`` from the copy reports a total that is + simply out of date. The row is what every writer maintains, so the row + is what is answered from, and the copy is refreshed from it here. + + This is a single-row lookup on ``meta_tables``, not a scan of the + search table, which is why ``count()`` can afford to make it on every + empty query -- the reason the total is exempt from the validity gate in + the first place is that it is maintained rather than cached. + """ + cur = self._execute( + SQL("SELECT total FROM meta_tables WHERE name = %s"), + [self.search_table], + ) + row = cur.fetchone() + if row is None or row[0] is None: + # No meta_tables row, or none recorded in it: nothing better than + # the value this object was built with. + return self.total + self.total = row[0] + return self.total + 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 +416,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"] @@ -305,6 +446,32 @@ def _has_stats(self, jcols, ccols, cvals, threshold, split_list=False, threshold cur = self._execute(selecter, values) return cur.rowcount > 0 + def _cached_count(self, query, split_list=False, suffix=""): + """ + The count stored under this cache key, or None if there is no such row. + + The physical counterpart of :meth:`quick_count`: it answers "is there a + row here", not "may this value be served", so it does not consult + ``stats_valid``. Cache maintenance needs precisely that question -- + ``_record_count`` chooses between ``INSERT`` and ``UPDATE``/``DELETE`` + by it, and a gated lookup calling an existing row missing would insert + a second row under a key the rest of the code takes to identify at most + one. See :meth:`_may_use_cache` for the distinction. + + INPUT: + + - ``query`` -- a mongo-style dictionary, as in the ``search`` method. + - ``split_list`` -- see the ``add_stats`` method + - ``suffix`` -- if provided, the table with that suffix added is read + """ + cols, vals = self._split_dict(query) + selecter = SQL( + "SELECT count FROM {0} WHERE cols = %s AND values = %s AND split = %s" + ).format(Identifier(self.counts + suffix)) + cur = self._execute(selecter, [cols, vals, split_list]) + if cur.rowcount: + return int(cur.fetchone()[0]) + def quick_count(self, query, split_list=False, suffix="", startup=False): """ Tries to quickly determine the number of results for a given query @@ -316,18 +483,29 @@ def quick_count(self, query, split_list=False, suffix="", startup=False): - ``split_list`` -- see the ``add_stats`` method - ``suffix`` -- if provided, the table with that suffix added will be used to perform the count + - ``startup`` -- for an empty query, read the counts table rather than + the maintained total. Used when ``meta_tables`` has no total yet. OUTPUT: - Either an integer giving the number of results, or None if not cached. - """ - if not query and not startup: - return self.total + Either an integer giving the number of results, or None if not cached + or if the cache may not be used. + """ + if not query: + # The empty-query total is maintained on every write rather than + # cached, so it is exact even when the rest of the cache is not, + # and it is not gated. A suffixed relation has no maintained + # total -- the live one describes the live table -- so there, as + # at startup, the counts row is what there is to read. + if startup or suffix: + return self._cached_count(query, split_list, suffix) + return self._live_total() + gate, gate_values = self._cache_gate(suffix) cols, vals = self._split_dict(query) selecter = SQL( - "SELECT count FROM {0} WHERE cols = %s AND values = %s AND split = %s" - ).format(Identifier(self.counts + suffix)) - cur = self._execute(selecter, [cols, vals, split_list]) + "SELECT count FROM {0} WHERE cols = %s AND values = %s AND split = %s{1}" + ).format(Identifier(self.counts + suffix), gate) + cur = self._execute(selecter, [cols, vals, split_list] + gate_values) if cur.rowcount: return int(cur.fetchone()[0]) @@ -449,11 +627,12 @@ def _record_count(self, query, count, split_list=False, suffix="", extra=True): data = [count, cols, vals, split_list] # Consult the counts table itself to decide between INSERT and UPDATE, # passing split_list so that we test the same key we are about to - # write. startup=True skips the in-memory total shortcut for an empty - # query: that total says nothing about whether the row is actually - # present (in particular in a suffixed counts table during a reload), - # and updating a missing row would silently record nothing. - if self.quick_count(query, split_list, suffix=suffix, startup=True) is None: + # write. This is the *physical* lookup, not quick_count: the question + # is whether a row is there to update, and while stats_valid is false + # -- which, on every path that gets here after a write, it is -- the + # gated lookup would call an existing row missing and this would insert + # a second row under the same key. + if self._cached_count(query, split_list, suffix=suffix) is None: if count == 0 and not nullrec: return # we don't want to store 0 counts since it can break stats updater = SQL("INSERT INTO {0} (count, cols, values, split, extra) VALUES (%s, %s, %s, %s, %s)") @@ -542,10 +721,31 @@ 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. + """ + gate, gate_values = self._cache_gate(suffix) + 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{1}" + ).format(Identifier(self.stats + suffix), gate) + cur = self._execute(selecter, ["distinct", Json(cols), ccols, cvals] + gate_values) + if cur.rowcount: + return int(cur.fetchone()[0]) + + def _cached_count_distinct(self, cols, query={}, suffix=""): + """ + Whether a distinct count is stored under this cache key, and its value. + + The physical counterpart of :meth:`quick_count_distinct`, ungated for + the reason given in :meth:`_cached_count`. """ 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)) + 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]) if cur.rowcount: return int(cur.fetchone()[0]) @@ -592,10 +792,15 @@ def _record_count_distinct(self, cols, query, count, suffix=""): """ ccols, cvals = self._split_dict(query) data = [count, Json(cols), "distinct", ccols, cvals] - if self.quick_count_distinct(cols, query, suffix=suffix) is None: + # The physical lookup, for the reason given in ``_record_count``. + if self._cached_count_distinct(cols, query, suffix=suffix) is None: updater = SQL("INSERT INTO {0} (value, cols, stat, constraint_cols, constraint_values) VALUES (%s, %s, %s, %s, %s)") else: - updater = SQL("UPDATE {0} SET value = %s WHERE cols = %s AND stats = %s AND constraint_cols = %s AND constraint_values = %s") + # ``stat``, not ``stats``: the column is singular, and with the + # gated lookup here this branch was never reached on a table whose + # flag a write had just cleared, so the misspelling went unnoticed + # until the physical lookup started sending rows down it. + updater = SQL("UPDATE {0} SET value = %s WHERE cols = %s AND stat = %s AND constraint_cols = %s AND constraint_values = %s") try: # This will fail if we don't have write permission, # for example, if we're running as the lmfdb user @@ -736,15 +941,39 @@ def _quick_statistic(self, col, ccols, cvals, kind="max"): the constraint columns take on these values. - ``kind`` -- either "min" or "max" or "sum" """ + gate, gate_values = self._cache_gate() constraint = SQL("constraint_cols = %s AND constraint_values = %s") - values = [kind, Json([col]), ccols, cvals] + values = [kind, Json([col]), ccols, cvals] + gate_values selecter = SQL( - "SELECT value FROM {0} WHERE stat = %s AND cols = %s AND threshold IS NULL AND {1}" - ).format(Identifier(self.stats), constraint) + "SELECT value FROM {0} WHERE stat = %s AND cols = %s AND threshold IS NULL AND {1}{2}" + ).format(Identifier(self.stats), constraint, gate) cur = self._execute(selecter, values) if cur.rowcount: return cur.fetchone()[0] + def _cached_statistic(self, col, ccols, cvals, kind="max"): + """ + Whether a statistic is stored under this cache key, and its value. + + The physical counterpart of :meth:`_quick_statistic`, ungated for the + reason given in :meth:`_cached_count`. ``None`` here means no row: a + row holding a NULL value is reported as ``(True, None)``, since + ``_record_statistic`` must still update it rather than insert beside + it. + + OUTPUT: + + A pair ``(present, value)``. + """ + constraint = SQL("constraint_cols = %s AND constraint_values = %s") + selecter = SQL( + "SELECT value FROM {0} WHERE stat = %s AND cols = %s AND threshold IS NULL AND {1}" + ).format(Identifier(self.stats), constraint) + cur = self._execute(selecter, [kind, Json([col]), ccols, cvals]) + if cur.rowcount: + return True, cur.fetchone()[0] + return False, None + def _slow_statistic(self, col, constraint, kind="max"): """ Compute the minimum/maximum value achieved by the column. @@ -801,15 +1030,28 @@ def _record_statistic(self, col, ccols, cvals, m, kind="max"): - ``kind`` -- the kind of statistic. Usually ``min`` or ``max`` or ``sum`` """ try: - inserter = SQL( - "INSERT INTO {0} " - "(cols, stat, value, constraint_cols, constraint_values) " - "VALUES (%s, %s, %s, %s, %s)" - ) - self._execute( - inserter.format(Identifier(self.stats)), - [Json([col]), kind, m, ccols, cvals], - ) + # Replace the row under this key if there is one. ``max`` and + # friends only get here when the gated lookup missed, which + # happens both when nothing is recorded and when the table is + # invalid -- so the physical lookup is what distinguishes "record + # this" from "correct what is recorded", and without it every + # max/min/sum taken on an invalid table left another row behind. + present, _ = self._cached_statistic(col, ccols, cvals, kind) + if present: + updater = SQL( + "UPDATE {0} SET value = %s " + "WHERE cols = %s AND stat = %s AND threshold IS NULL " + "AND constraint_cols = %s AND constraint_values = %s" + ) + values = [m, Json([col]), kind, ccols, cvals] + else: + updater = SQL( + "INSERT INTO {0} " + "(cols, stat, value, constraint_cols, constraint_values) " + "VALUES (%s, %s, %s, %s, %s)" + ) + values = [Json([col]), kind, m, ccols, cvals] + self._execute(updater.format(Identifier(self.stats)), values) except Exception: pass @@ -1290,6 +1532,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 +1896,24 @@ 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 = current_schema() AND c.relname = %s) c +WHERE schemaname = current_schema() 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])) + cur = self._execute(selecter, [self.search_table, self.search_table, col, n]) return [tuple(x) for x in cur] def _common_cols(self, threshold=700): @@ -1805,10 +2053,31 @@ def refresh_stats(self, total=True, reset_None_to_1=False, suffix=""): - ``reset_None_to_1`` -- change threshold None to 1 in stored statistics - ``suffix`` -- appended to the table name when computing and storing stats. Used when reloading a table. + + A live refresh claims the table's ``meta_tables`` row before it starts, + by marking it invalid, and marks it valid again only at the end -- so + the flag is false for the whole rebuild, and the row lock that claim + takes is held until this transaction ends. Every library write marks + the same row too, so a writer and a refresh that overlap cannot + interleave; one of them waits, and the two possible orders are the two + acceptable outcomes: + + - the writer went first, so its rows are in the data this rebuild + reads, and the caches this commits describe them; or + - the refresh went first, and the writer's ``_break_stats`` lands + afterwards, leaving the table invalid -- correctly, since its rows + are not in the caches just built. + + Restoring the flag at the end without claiming it at the start would + admit a third: a write committing mid-rebuild, its invalidation then + overwritten by a refresh that had already counted some families + without it. """ self._logger.info("Refreshing statistics on %s" % self.search_table) t0 = time.time() with DelayCommit(self, silence=True): + if not suffix: + self.table._break_stats() # Determine the stats and counts currently recorded stat_cmds, split_cmds, nstat_cmds = self._status(reset_None_to_1) col_value_dict = self.extra_counts(include_counts=False, suffix=suffix) @@ -1844,6 +2113,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): @@ -1923,7 +2198,12 @@ def _add_extra_counts(self, col_value_dict, suffix=""): continue for values in values_list: query = self._join_dict(cols, values) - if self.quick_count(query, suffix=suffix) is None: + # "is this one already recorded here", not "may it be served": + # the caller is filling a counts table, and on the live table + # it is doing so inside a refresh, which has marked the table + # invalid. The gated lookup would report every one of these + # missing and recount them all. + if self._cached_count(query, suffix=suffix) is None: self._slow_count(query, record=True, suffix=suffix) def extra_counts(self, include_counts=True, suffix=""): diff --git a/psycodict/table.py b/psycodict/table.py index a7a24a5..39433d8 100644 --- a/psycodict/table.py +++ b/psycodict/table.py @@ -17,6 +17,7 @@ import time import re from bisect import bisect +from collections import namedtuple from functools import partial from psycopg.sql import SQL, Identifier, Placeholder, Literal @@ -56,6 +57,17 @@ # counts and stats columns and their types # ################################################################## +# What a reload prepared, and therefore what its final swap must do: the exact +# base names whose _tmp copies are to be renamed into place, whether the counts +# and stats arriving with them describe the data arriving with them, and whether +# the ids were successfully resorted. Returned by ``PostgresTable.reload`` so +# that a deferred swap (``PostgresDatabase.reload_all``) finalizes what was +# prepared rather than re-deriving it from the arguments -- the two drifted +# apart, and a swap list assembled from the input filenames could leave a +# refreshed cache companion stranded under _tmp while marking the table valid. +ReloadPlan = namedtuple("ReloadPlan", ["tables", "stats_valid", "ordered"]) + + _counts_cols = ("cols", "values", "count", "extra", "split") _counts_types = dict(zip(_counts_cols, ("jsonb", "jsonb", "bigint", "boolean", "boolean"))) _counts_jsonb_idx = jsonb_idx(_counts_cols, _counts_types) @@ -1172,15 +1184,49 @@ def _check_locks(self, changetype, datafile=None, suffix=""): print(locktype + " " * (typelen - len(locktype)) + str(pid)) raise LockError("Table is locked. Please resolve the lock by killing the above processes and try again") + def _set_stats_valid(self, valid): + """ + Write this table's ``stats_valid`` flag, unconditionally. + + The ``UPDATE`` is issued whatever ``self._stats_valid`` says, and that + is the point of the method. The attribute is a copy taken when this + object was built: another process's refresh or ``restat=False`` write + moves the row without moving the copy, and a rolled-back transaction + moves the copy without moving the row, so a transition skipped because + "it is already in that state" is a transition skipped on the strength + of a value that may be describing neither the database nor the present. + + Issuing it always is also what serializes refreshes against writers: + the ``UPDATE`` takes a row lock on this table's ``meta_tables`` row + that is held to the end of the transaction, so a writer and a refresh + that overlap are ordered by it rather than racing (see + :meth:`PostgresStatsTable.refresh_stats`). + + INPUT: + + - ``valid`` -- whether the cached counts and statistics agree with the + live data. + """ + updater = SQL("UPDATE meta_tables SET stats_valid = %s WHERE name = %s") + self._execute(updater, [valid, self.search_table], silent=True) + self._stats_valid = valid + def _break_stats(self): """ This function should be called when the statistics are invalidated by an insertion or update. """ - if self._stats_valid: - # Only need to interact with database in this case. - updater = SQL("UPDATE meta_tables SET stats_valid = false WHERE name = %s") - self._execute(updater, [self.search_table], silent=True) - self._stats_valid = False + self._set_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. + """ + self._set_stats_valid(True) def _break_order(self): """ @@ -1350,7 +1396,7 @@ def update_from_file( resort=None, reindex=None, restat=True, - logging={"operation":"file_update"}, + logging=None, **kwds ): """ @@ -1373,7 +1419,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 +1432,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)) @@ -1464,6 +1516,14 @@ def drop_tmp(): Identifier(tmp_table), Identifier(self.search_table), Identifier(label_col))) + # The rows this table serves have changed, whatever happens to + # the caches below, so say so now: the invalidation rides the + # same transaction as the change, and a refresh that overlaps + # this one is serialized against it by the row lock this takes + # (see PostgresStatsTable.refresh_stats). If the caches are + # rebuilt or replaced further down, the flag goes back to true + # there, in this transaction, once they have been. + self._break_stats() if reindex and inplace: # also restores constraints self.restore_indexes(columns[1:]) @@ -1491,7 +1551,13 @@ def drop_tmp(): # and the _tmp tables are left orphaned. reload builds its # swap list the same way. tables = [self.search_table] - if restat and self.stats.saving: + # The same condition decides both what is swapped in and + # what the swap leaves true: with it, the counts and stats + # arriving at the live names were rebuilt from the _tmp + # data just above; without it, the live counts and stats + # are the old ones and the data underneath them is new. + swapped_caches = bool(restat and self.stats.saving) + if swapped_caches: tables += [self.stats.counts, self.stats.stats] if self.stats.counts in tables: # _clone built the _tmp counts table with a bare LIKE, @@ -1500,16 +1566,16 @@ def drop_tmp(): # table keeps them (otherwise cached-count lookups # degrade to sequential scans). reload does the same. self._create_counts_indexes(suffix=suffix) - self._swap_in_tmp(tables) + self._swap_in_tmp(tables, stats_valid=swapped_caches) if ordered: 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,7 +2003,68 @@ def _next_backup_number(self): ) return backup_number - def _swap_in_tmp(self, tables): + 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 _reload_stats_valid(self, tables, countsfile, statsfile, restat): + """ + Whether a reload leaves the live counts and stats describing the live + data. + + A reload replaces the search table wholesale, so the old cached answers + are about data that is no longer there; whether the new ones are about + the data that is comes down to two things. + + They have to *arrive*: both cache companions must be in ``tables``, the + list this reload is actually going to swap. A swap list holding the + search table and one companion, or none, leaves the live table paired + with a cache relation describing data that has just been replaced -- + and a refreshed ``_tmp`` companion stranded under its temporary name. + Nothing in a reload builds such a list on purpose, but the deferred + swap in ``PostgresDatabase.reload_all`` used to assemble one from the + input filenames, so the check is made against the list rather than + against the arguments that suggested it. + + And they have to *describe the new data*: either the caller supplied + both files alongside the search file, which gets the same trust the + search file itself gets, or ``restat`` rebuilt them from the loaded + ``_tmp`` data. Failing both, what is swapped in is an empty clone + (nothing lost, but nothing vouched for either) or, if some earlier + operation left a ``_tmp`` copy behind, a relation this reload never + wrote. Without ``saving``, no companions are swapped at all. + + INPUT: + + - ``tables`` -- the exact list of base table names this reload will + hand to :meth:`reload_final_swap` + - ``countsfile``, ``statsfile`` -- as passed to :meth:`reload` + - ``restat`` -- as passed to :meth:`reload`, before or after its + ``None`` is resolved; both give the same answer + """ + if not self.stats.saving: + return False + if self.stats.counts not in tables or self.stats.stats not in tables: + return False + if restat is None: + restat = countsfile is None or statsfile is None + return bool(restat or (countsfile is not None and statsfile is not None)) + + def _swap_in_tmp(self, tables, stats_valid=False): """ Helper function for ``reload``: appends _old{n} to the names of tables/indexes/pkeys and renames the _tmp versions to the live versions. @@ -1945,12 +2072,26 @@ def _swap_in_tmp(self, tables): INPUT: - ``tables`` -- a list of tables to rename (e.g. self.search_table, self.stats.counts, self.stats.stats) + - ``stats_valid`` -- whether the counts and statistics the live names + are left addressing describe the data the live names are left + addressing. Written in this method's transaction, with the renames, + so that the two cannot come apart: a caller that swaps in a search + table while keeping the old cache companions has just made every + cached answer stale, and one that swaps in companions rebuilt or + loaded for the new data has just made them all correct. The default + is the conservative answer, since a caller that has not thought about + it has usually done the former. """ 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)) self._swap(tables, "_tmp", "") + self._set_stats_valid(stats_valid) # Which relation is which is known here, so the policy is looked up # by it rather than guessed from the name: a search table may end # in _stats (the LMFDB has one), and granting it a counts table's @@ -2137,13 +2278,25 @@ def reload( - ``restat`` -- whether to refresh statistics afterward. Default behavior is to refresh stats if either countsfile or statsfile is missing. - ``final_swap`` -- whether to perform the final swap exchanging the - temporary table with the live one. + temporary table with the live one. Deferring it defers the + validity transition too, so pass the returned plan's ``tables``, + ``ordered`` and ``stats_valid`` to ``reload_final_swap`` when the + time comes; it invalidates by default, and reassembling the swap + list by hand is how a refreshed cache companion gets stranded. - ``silence_meta`` -- suppress the warning message when using a metafile - ``adjust_schema`` -- If True, it will create the new tables using the header columns, otherwise expects the schema specified by the files to match the current one - ``kwds`` -- passed on to the ``COPY`` command. Cannot include "columns". + OUTPUT: + + A :data:`ReloadPlan` describing the swap: the exact list of base table + names whose ``_tmp`` copies belong at the live names, whether the + resulting cache companions describe the resulting data, and whether the + ids were resorted. With ``final_swap`` it says what the swap did; with + ``final_swap=False`` it is what the deferred swap must be told. + .. NOTE: If the search file contains ids, they should be contiguous, @@ -2276,15 +2429,28 @@ def reload( # create index on counts table self._create_counts_indexes(suffix=suffix) + # What the final swap has to be told, decided here, where the + # relations this reload prepared are known. A deferred swap + # gets it from the return value rather than reconstructing it, + # so that it cannot swap a different set than was prepared. + plan = ReloadPlan( + tables=list(tables), + stats_valid=self._reload_stats_valid(tables, countsfile, statsfile, restat), + ordered=bool(ordered), + ) if final_swap: - self.reload_final_swap(tables=tables, - metafile=metafile, - ordered=ordered) + self.reload_final_swap( + tables=plan.tables, + metafile=metafile, + ordered=plan.ordered, + stats_valid=plan.stats_valid, + ) elif metafile is not None and not silence_meta: print("Warning: since the final swap was not requested, we have not updated meta_tables") print("when performing the final swap with reload_final_swap, pass the metafile as an argument to update the meta_tables") print("Reloaded %s in %.3f secs" % (self.search_table, time.time() - now_overall)) aborted = False + return plan finally: self._log_db_change( "reload", @@ -2294,7 +2460,8 @@ def reload( stats=(statsfile is not None), ) - def reload_final_swap(self, tables=None, metafile=None, ordered=False, sep="|"): + def reload_final_swap(self, tables=None, metafile=None, ordered=False, sep="|", + stats_valid=False): """ Renames the ``_tmp`` versions of ``tables`` to the live versions, and updates the corresponding meta_tables row if ``metafile`` is provided. @@ -2304,6 +2471,20 @@ def reload_final_swap(self, tables=None, metafile=None, ordered=False, sep="|"): - ``tables`` -- list of strings (optional), of the tables to be renamed. If None is provided, renames all the tables ending in ``_tmp`` - ``metafile`` -- a string (optional), giving a file containing the meta information for the table. - ``sep`` -- a character (default ``|``) to separate columns + - ``stats_valid`` -- whether the counts and statistics that end up at + the live names describe the search table that ends up at the live + name. Passed to :meth:`_swap_in_tmp`, which writes it with the + renames; the caller states it rather than the flag being inherited, + since what the old live table's flag said is a fact about data that + is no longer there. The default is the conservative answer. + + A ``metafile`` carries a ``stats_valid`` column of its own, and it is + deliberately overruled: it records what was true of the table the file + was exported from, at export time, and cannot know whether the counts + and stats relations swapped in here were rebuilt. It is applied first + so that the rest of the row (the sort order, ``id_ordered``, and the + rest) takes effect, and ``stats_valid`` is then set to what this swap + actually did. """ with DelayCommit(self, silence=True): if tables is None: @@ -2316,9 +2497,12 @@ def reload_final_swap(self, tables=None, metafile=None, ordered=False, sep="|"): if self._table_exists(tablename + "_tmp"): tables.append(tablename) - self._swap_in_tmp(tables) + self._swap_in_tmp(tables, stats_valid=stats_valid) if metafile is not None: self.reload_meta(metafile, sep=sep) + # reload_meta rewrote the whole row, including the flag the + # swap had just set; restate it (see the docstring). + self._set_stats_valid(stats_valid) if ordered: self._set_ordered() # The swapped-in table's row count bears no relation to the old @@ -2391,6 +2575,18 @@ def reload_revert(self, backup_number=None): self._swap(tables, "", "_tmp") self._swap(tables, old, "") self._swap(tables, "_tmp", old) + # A backup carries no validity bit of its own -- meta_tables has + # one row, for the live name, and it followed the live table + # rather than the relations just restored. It may therefore say + # true of caches that were invalid when they were backed up, so + # the flag is cleared and a refresh is what earns it back. Marked + # here, with the renames, so the two commit together. + self._set_stats_valid(False) + # The total is maintained rather than cached, so the gate above + # does not cover it and it has to be made true: what meta_tables + # holds counts the table that is now the backup. reload_final_swap + # recounts for the same reason. + self.stats._set_total(self.stats._slow_count({}, record=False)) # The exchange moved each relation's privileges to the other one, # so both are told again what they are -- inside this DelayCommit, # with the renames. @@ -2777,16 +2973,18 @@ def _staged_swap_in(self, out_of_order): may have broken) the id ordering. """ # The staged writes could not update the meta_tables row (it is - # keyed on the live name), so transfer the order flag and invalidate - # the statistics: the swapped-in counts and stats tables are empty, - # not refreshed + # keyed on the live name), so transfer the order flag here. if out_of_order: self._break_order() - self._break_stats() # reload_final_swap backs the live tables up under _old, renames # the _tmp ones into place, recounts the total and replaces the - # table object held by the database - self.reload_final_swap(tables=self._staged_tables(), ordered=False) + # table object held by the database. stats_valid=False because the + # counts and stats tables it swaps in are the staged ones, which are + # empty rather than refreshed; it is written with the renames, so the + # invalidation and the swap commit together. + self.reload_final_swap( + tables=self._staged_tables(), ordered=False, stats_valid=False + ) def staged_force_swap(self): """ @@ -2848,10 +3046,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 +3064,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 +3129,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/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_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..19d8ce9 --- /dev/null +++ b/tests/test_stats_validity.py @@ -0,0 +1,1057 @@ +# -*- 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 are +deliberately not, when the flag is cleared and restored, and that all of it is +decided by the database rather than by a Boolean on a Python object, which is +what makes it hold across processes and across a rollback. +""" +import threading +import time + +import pytest + +from psycopg.sql import SQL, Identifier + +from conftest import sample_row +from psycodict.utils import DelayCommit + +# Long enough that a loaded machine does not fail the concurrency tests +# spuriously, short enough that a genuine hang is reported rather than waited +# out. +TIMEOUT = 30 + +# Of sample_row(0), ..., sample_row(199), the rows with flag = True are those +# with i % 3 == 0. +NFLAGGED = len([i for i in range(200) if i % 3 == 0]) # 67 + + +@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() + return table + + +@pytest.fixture +def other_handle(config): + """ + A second ``PostgresDatabase``, on its own connection. + + The point of most of what follows: a deployment runs several webserver + processes, each holding its own table objects, and an invalidation from + one of them has to be visible to the others. A fixture rather than a + second cursor because the object-level caching is what is under test. + """ + from psycodict.database import PostgresDatabase + + other = PostgresDatabase(config=config) + yield other + other.conn.close() + + +def twin(table, other_handle): + """ + The same search table, as seen by a second database handle. + + Its ``_stats_valid`` and ``total`` are read when this object is built and + then left alone, exactly like those of a table object held by another + webserver process. + """ + other_handle.refresh_tables() + return other_handle[table.search_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 + + +# --------------------------------------------------------------------------- +# the flag is the database's, not the Python object's +# +# A deployment runs several webserver processes, and a rollback undoes SQL and +# not Python, so ``self.table._stats_valid`` answers neither "is this table +# valid" nor "was this transition already made". These are the cases where it +# disagrees with the row. +# --------------------------------------------------------------------------- + +def test_a_second_handle_does_not_serve_a_count_invalidated_elsewhere( + cached_table, other_handle +): + """ + The headline case in a deployment: one process writes with restat=False, + and every other process must stop serving the count it had cached. + """ + query = {"flag": True} + before = cached_table.stats.count(query, record=True) + assert before == NFLAGGED + + reader = twin(cached_table, other_handle) + assert reader.stats.quick_count(query) == before + + cached_table.update(query, {"flag": False}, restat=False) + + # The reader's copy of the flag still says true -- nothing told it + # otherwise, and nothing ever would ... + assert reader._stats_valid is True + # ... but the row it would have answered from is not an answer any more. + assert reader.stats.quick_count(query) is None + assert reader.stats.count(query) == 0 + + +def test_a_second_handle_does_not_serve_a_distinct_count_invalidated_elsewhere( + cached_table, other_handle +): + cols = ["flag"] + assert cached_table.stats._slow_count_distinct(cols, record=True) == 2 + + reader = twin(cached_table, other_handle) + assert reader.stats.quick_count_distinct(cols) == 2 + + cached_table.update({"flag": True}, {"flag": False}, restat=False) + + assert reader._stats_valid is True + assert reader.stats.quick_count_distinct(cols) is None + assert reader.stats.count_distinct("flag") == 1 + + +@pytest.mark.parametrize("kind,before,after", [("max", 199, 99), ("min", 0, 0), ("sum", 19900, 4950)]) +def test_a_second_handle_does_not_serve_a_statistic_invalidated_elsewhere( + cached_table, other_handle, kind, before, after +): + from psycodict.encoding import Json + + assert getattr(cached_table.stats, kind)("n") == before + + reader = twin(cached_table, other_handle) + empty = Json([]) + assert reader.stats._quick_statistic("n", empty, empty, kind) is not None + + cached_table.delete({"n": {"$gte": 100}}, restat=False) + + assert reader._stats_valid is True + assert reader.stats._quick_statistic("n", empty, empty, kind) is None + assert getattr(reader.stats, kind)("n", record=False) == after + + +def test_a_second_handle_sees_a_total_another_handle_changed(cached_table, other_handle): + """ + ``total`` is maintained rather than cached, so it is exempt from the gate -- + which makes reading it from meta_tables rather than from ``self.total`` the + only thing keeping it honest across processes. + """ + reader = twin(cached_table, other_handle) + + cached_table.insert_many([sample_row(1000 + i) for i in range(5)], restat=False) + cached_table.delete({"n": 0}, restat=False) + + # The value this reader was built with, untouched by either write ... + assert reader.stats.total == 200 + # ... and the value it answers with. + assert reader.stats.quick_count({}) == 204 + assert reader.count() == 204 + + +def test_a_stale_local_false_does_not_skip_the_invalidation(cached_table, other_handle): + """ + The inverse failure: a table object whose copy of the flag is already false + would, if the transition were conditional on it, leave the row true while + changing the data. + """ + query = {"n": {"$lt": 100}} + assert cached_table.stats.count(query, record=True) == 100 + + cached_table.update({"label": "l0"}, {"x": 9.5}, restat=False) + assert cached_table._stats_valid is False + + refresher = twin(cached_table, other_handle) + refresher.stats.saving = True + refresher.stats.refresh_stats() + assert stats_valid_in_meta(cached_table) is True + # ... which the first handle has no way of hearing about. + assert cached_table._stats_valid is False + + cached_table.update({"label": "l1"}, {"x": 8.5}, restat=False) + assert stats_valid_in_meta(cached_table) is False + assert cached_table.stats.quick_count(query) is None + + +def test_a_rolled_back_restore_does_not_make_the_cache_usable(cached_table): + """ + A rollback undoes the ``UPDATE`` and not the attribute, so after one the + Python object claims a validity the database denies. The database wins. + """ + query = {"flag": True} + assert cached_table.stats.count(query, record=True) == NFLAGGED + cached_table.update(query, {"flag": False}, restat=False) + assert cached_table.stats.quick_count(query) is None + + with pytest.raises(RuntimeError): + with DelayCommit(cached_table): + cached_table.stats.refresh_stats() + assert cached_table._stats_valid is True + raise RuntimeError("the caller changed its mind") + + # The object was left saying true and the row says false ... + assert cached_table._stats_valid is True + assert stats_valid_in_meta(cached_table) is False + # ... so the stale row the rollback restored is still not an answer. + assert cached_table.stats.quick_count(query) is None + assert cached_table.stats.count(query) == 0 + + +def test_a_refresh_and_a_concurrent_write_are_serialized(cached_table, other_handle): + """ + A refresh claims the meta_tables row before it rebuilds anything, so a + write that overlaps it waits, and the outcome is one of the two the + ordering allows rather than a lost invalidation. + + Here the refresh goes first: the writer's rows are not in the caches it + builds, so the writer's ``_break_stats`` lands afterwards and the table + ends invalid. Without the claim, the refresh's restore would have + overwritten an invalidation that had already committed. + """ + query = {"flag": True} + assert cached_table.stats.count(query, record=True) == NFLAGGED + cached_table.update({"label": "l0"}, {"x": 7.5}, restat=False) + + writer = twin(cached_table, other_handle) + refresh_started = threading.Event() + write_finished = threading.Event() + failures = [] + + def write(): + try: + refresh_started.wait(TIMEOUT) + # Blocks on the row lock the refresh took, until it commits. + writer.update(query, {"flag": False}, restat=False) + except Exception as err: # pragma: no cover - reported below + failures.append(err) + finally: + write_finished.set() + + thread = threading.Thread(target=write) + thread.start() + try: + # Hooked at the very start of the rebuild, before refresh_stats writes + # anything of its own to meta_tables: later than this and the total's + # own UPDATE has already taken the row lock, and the test would pass + # whether or not the refresh claimed the row deliberately. + original = cached_table.stats._status + + def midway(*args, **kwds): + refresh_started.set() + # Long enough for the writer to reach the lock and block on it. + time.sleep(1) + assert not write_finished.is_set(), ( + "the write committed in the middle of the refresh" + ) + return original(*args, **kwds) + + cached_table.stats._status = midway + try: + cached_table.stats.refresh_stats() + finally: + del cached_table.stats._status + finally: + refresh_started.set() + thread.join(TIMEOUT) + + assert not thread.is_alive(), "the writer never unblocked" + assert not failures, failures + # The write landed after the refresh, so the caches it built do not + # describe it, and the table says so. + assert stats_valid_in_meta(cached_table) is False + assert cached_table.stats.quick_count(query) is None + assert cached_table.stats.count(query) == 0 + + +# --------------------------------------------------------------------------- +# every write and swap path makes the transition, in its own transaction +# +# The row-level paths always cleared the flag; the bulk and replacement paths +# did not, so a swap could leave new data under old caches and a true flag. +# Each of these starts from a valid cached count of the rows it is about to +# change. +# --------------------------------------------------------------------------- + +def _write_flag_file(path, labels, value): + """An update file setting ``flag`` to ``value`` for the given labels.""" + with open(path, "w") as F: + F.write("label|flag\ntext|boolean\n\n") + for label in labels: + F.write("%s|%s\n" % (label, "t" if value else "f")) + + +@pytest.fixture +def flagged(cached_table): + """ + A cached, valid count of the flagged rows: what every path below is about + to make wrong. + """ + query = {"flag": True} + assert cached_table.stats.count(query, record=True) == NFLAGGED + assert cached_table.stats.quick_count(query) == NFLAGGED + assert stats_valid_in_meta(cached_table) is True + return query + + +def assert_disowned(table, query, after=0): + """ + The persisted flag and the public answer, after a path that changed the + data without rebuilding the caches. + """ + assert stats_valid_in_meta(table) is False + assert table.stats.quick_count(query) is None + assert table.stats.count(query) == after + + +def test_in_place_update_from_file_without_restat_invalidates(cached_table, flagged, tmp_path): + datafile = str(tmp_path / "flags.txt") + _write_flag_file(datafile, ["l%d" % i for i in range(200)], value=False) + cached_table.update_from_file(datafile, "label", inplace=True, restat=False) + assert_disowned(cached_table, flagged) + + +def test_a_swapping_update_from_file_without_restat_invalidates(cached_table, flagged, tmp_path): + """ + The worst of the untransitioned paths: the search table is replaced and the + old counts and stats relations are kept, so every cached answer is now + about data that is no longer there. + """ + datafile = str(tmp_path / "flags.txt") + _write_flag_file(datafile, ["l%d" % i for i in range(200)], value=False) + cached_table.update_from_file(datafile, "label", inplace=False, restat=False) + # The old counts row really did survive the swap ... + assert cached_table.stats._cached_count({"flag": True}) == NFLAGGED + # ... and is not served. + assert_disowned(cached_table, flagged) + + +def test_rewrite_without_restat_invalidates(cached_table, flagged): + cached_table.rewrite(lambda rec: dict(rec, flag=False), restat=False) + assert_disowned(cached_table, flagged) + + +def test_reload_without_restat_invalidates(cached_table, flagged, tmp_path): + searchfile = str(tmp_path / "data.txt") + cached_table.copy_to(searchfile) + cached_table.update({"flag": True}, {"flag": False}, restat=False) + cached_table.stats.refresh_stats() + assert stats_valid_in_meta(cached_table) is True + + # Reload the file dumped before that change: the data goes back to having + # flagged rows, and the counts and stats swapped in are empty clones this + # reload never wrote, so it does not vouch for them. + cached_table.reload(searchfile, restat=False) + table = cached_table._db[cached_table.search_table] + assert stats_valid_in_meta(table) is False + assert table.stats.count({"flag": True}) == NFLAGGED + + +def test_a_swapping_update_from_file_with_restat_ends_valid(cached_table, tmp_path): + """ + The other half of the contract: when the counts and stats reaching the live + names were rebuilt for the data reaching the live name, the swap says so, + and the rebuilt value is served. + """ + cached_table.stats.add_stats(["flag"]) + assert cached_table.stats.quick_count({"flag": True}) == NFLAGGED + + # Start invalid, so that ending valid is a transition this swap made + # rather than one it inherited from the table it replaced. + cached_table.update({"label": "l0"}, {"x": 6.5}, restat=False) + assert stats_valid_in_meta(cached_table) is False + + datafile = str(tmp_path / "flags.txt") + _write_flag_file(datafile, ["l%d" % i for i in range(200)], value=True) + cached_table.update_from_file(datafile, "label", inplace=False, restat=True) + + assert stats_valid_in_meta(cached_table) is True + assert cached_table.stats.quick_count({"flag": True}) == 200 + + +def test_a_reload_with_restat_ends_valid(cached_table, tmp_path): + cached_table.stats.add_stats(["flag"]) + searchfile = str(tmp_path / "data.txt") + cached_table.copy_to(searchfile) + + cached_table.update({"flag": True}, {"flag": False}, restat=False) + assert stats_valid_in_meta(cached_table) is False + + cached_table.reload(searchfile, restat=True) + table = cached_table._db[cached_table.search_table] + assert stats_valid_in_meta(table) is True + assert table.stats.quick_count({"flag": True}) == NFLAGGED + + +def test_a_reload_with_a_metafile_does_not_take_its_validity_from_the_file( + cached_table, tmp_path +): + """ + A metafile's ``stats_valid`` describes the table it was exported from at + export time, and cannot know whether this reload rebuilt anything. The + swap's own answer wins, and the rest of the row still lands. + """ + searchfile = str(tmp_path / "data.txt") + metafile = str(tmp_path / "meta.txt") + cached_table.copy_to(searchfile, metafile=metafile) + with open(metafile) as F: + assert "|t|" in F.read() # exported while valid + + cached_table.reload(searchfile, metafile=metafile, restat=False, silence_meta=True) + table = cached_table._db[cached_table.search_table] + assert stats_valid_in_meta(table) is False + + +def test_reload_revert_invalidates(cached_table, tmp_path): + """ + A backup carries no validity bit of its own: meta_tables has one row, and + it stayed with the live name. So a backup taken while the table was + invalid can be swapped back under a flag that has since been set true, and + the only safe answer is to clear it. + """ + table = cached_table + table.stats.add_stats(["flag"]) + searchfile = str(tmp_path / "data.txt") + table.copy_to(searchfile) + + # Back up an invalid state: change the data, do not restat, then reload + # (which files the current relations away as _old1 and ends valid). + table.update({"flag": True}, {"flag": False}, restat=False) + assert stats_valid_in_meta(table) is False + table.reload(searchfile, restat=True) + table = table._db[table.search_table] + assert stats_valid_in_meta(table) is True + + table.reload_revert() + assert stats_valid_in_meta(table) is False + # and the total describes what is now live, not what was + assert table.stats.quick_count({}) == table.stats._slow_count({}, record=False) + + +def test_the_flag_and_the_swap_roll_back_together(cached_table, tmp_path, monkeypatch): + """ + The transition is written in the transaction that does the renames, so a + failure between them cannot commit one without the other. + """ + table = cached_table + searchfile = str(tmp_path / "data.txt") + table.copy_to(searchfile) + table.update({"label": "l0"}, {"x": 4.5}, restat=False) + assert stats_valid_in_meta(table) is False + + def boom(self, *args, **kwds): + raise RuntimeError("the swap blew up") + + # The last thing reload_final_swap does inside its transaction, after both + # the renames and the flag. + monkeypatch.setattr(type(table._db), "_notify_schema_change", boom) + with pytest.raises(RuntimeError): + table.reload(searchfile, restat=True) + table._db.conn.rollback() + + # Neither half committed: the flag is still what it was, and there is no + # backup, so the renames did not happen either. (DDL is transactional, so + # the rollback took the _tmp relations the reload had built with it.) + assert stats_valid_in_meta(table) is False + assert not table._table_exists(table.search_table + "_old1") + assert not table._table_exists(table.search_table + "_tmp") + assert table.stats.count({}) == 200 + + +@pytest.mark.parametrize( + "path,after", + [ + ("update", 0), + ("delete", 0), + ("upsert", 0), + ("insert_many", 0), + ("copy_from", 2 * NFLAGGED), + ], +) +def test_the_row_level_paths_still_invalidate(cached_table, flagged, tmp_path, path, after): + """ + The paths that always did, re-checked now that the transition is + unconditional -- a table object whose copy of the flag has drifted must not + skip it. + """ + table = cached_table + if path == "update": + table.update({"flag": True}, {"flag": False}, restat=False) + elif path == "delete": + table.delete({"flag": True}, restat=False) + elif path == "upsert": + for i in range(0, 200, 3): + table.upsert({"label": "l%d" % i}, {"flag": False}) + elif path == "insert_many": + table.delete({"flag": True}, restat=False) + table._restore_stats() + table.insert_many([dict(sample_row(1000), flag=False)], restat=False) + else: + # include_id=False so the appended rows get fresh ids after the + # existing ones rather than colliding with them. + searchfile = str(tmp_path / "more.txt") + table.copy_to(searchfile, query={"flag": True}, include_id=False) + table.copy_from(searchfile, restat=False) + assert_disowned(table, flagged, after) + + +def test_a_staged_commit_invalidates(cached_table, flagged): + """ + A staged commit swaps in empty counts and stats tables, so whatever the + live ones held is gone and nothing has been rebuilt. + """ + with cached_table.staged() as staged: + staged.update({"flag": True}, {"flag": False}, restat=False) + table = cached_table._db[cached_table.search_table] + assert_disowned(table, flagged) + + +# --------------------------------------------------------------------------- +# "may this be served" is not "is this row here" +# +# Cache maintenance chooses between INSERT and UPDATE by asking whether the row +# it is about to write already exists. Routed through the gated lookups, that +# question is answered "no" on every invalid table -- which is every table the +# maintenance runs on -- and each recorder leaves a second row behind under a +# key the rest of the code takes to identify at most one. +# --------------------------------------------------------------------------- + +def stored_counts_for(table, query, split=False): + """ + Every count physically stored under this cache key, gate or no gate. + """ + cols, vals = table.stats._split_dict(query) + cur = table._execute( + SQL( + "SELECT count FROM {0} WHERE cols = %s AND values = %s AND split = %s" + ).format(Identifier(table.stats.counts)), + [cols, vals, split], + ) + return [row[0] for row in cur] + + +def stored_stat_values(table, cols, kind, query={}): + """ + Every statistic physically stored under this cache key. + """ + from psycodict.encoding import Json + + ccols, cvals = table.stats._split_dict(query) + cur = table._execute( + SQL( + "SELECT value FROM {0} WHERE stat = %s AND cols = %s " + "AND constraint_cols = %s AND constraint_values = %s" + ).format(Identifier(table.stats.stats)), + [kind, Json(cols), ccols, cvals], + ) + return [int(row[0]) for row in cur] + + +def test_updating_the_total_while_invalid_leaves_one_row(cached_table): + """ + The reproducible case: a write clears the flag and then maintains the + total, and the maintenance asked the gated lookup whether the ``{}`` row + was there. It said no, because the flag was false, so a second ``{}`` row + was inserted -- by the write that had just cleared the flag itself. + """ + assert stored_counts_for(cached_table, {}) == [200] + + cached_table.insert_many([sample_row(1000)], restat=False) + assert stats_valid_in_meta(cached_table) is False + + assert stored_counts_for(cached_table, {}) == [201] + assert cached_table.stats.quick_count({}) == 201 + + +def test_recording_a_count_again_while_invalid_leaves_one_row(cached_table): + query = {"n": {"$lt": 100}} + assert cached_table.stats.count(query, record=True) == 100 + assert stored_counts_for(cached_table, query) == [100] + + cached_table.delete({"n": {"$lt": 50}}, restat=False) + assert stats_valid_in_meta(cached_table) is False + + assert cached_table.stats.count(query, record=True) == 50 + # One row, holding the recomputed value -- not two that happen to agree. + assert stored_counts_for(cached_table, query) == [50] + + +def test_recording_a_distinct_count_again_while_invalid_leaves_one_row(cached_table): + """ + Also the first thing ever to run ``_record_count_distinct``'s update + statement, which named a column ``stats`` that has always been called + ``stat``: while the gated lookup answered this question, the update branch + was unreachable on exactly the tables that needed it. + """ + cols = ["flag"] + assert cached_table.stats._slow_count_distinct(cols, record=True) == 2 + assert stored_stat_values(cached_table, cols, "distinct") == [2] + + cached_table.update({"flag": True}, {"flag": False}, restat=False) + assert stats_valid_in_meta(cached_table) is False + + assert cached_table.stats._slow_count_distinct(cols, record=True) == 1 + assert stored_stat_values(cached_table, cols, "distinct") == [1] + + +@pytest.mark.parametrize("kind,before,after", [("max", 199, 99), ("min", 0, 0), ("sum", 19900, 4950)]) +def test_recording_a_statistic_again_while_invalid_leaves_one_row( + cached_table, kind, before, after +): + assert getattr(cached_table.stats, kind)("n") == before + assert stored_stat_values(cached_table, ["n"], kind) == [before] + + cached_table.delete({"n": {"$gte": 100}}, restat=False) + assert stats_valid_in_meta(cached_table) is False + + assert getattr(cached_table.stats, kind)("n") == after + assert stored_stat_values(cached_table, ["n"], kind) == [after] + + +def test_refreshing_an_invalid_table_leaves_every_key_unique(cached_table): + """ + A refresh records the total through the same recorder, on a table it has + just marked invalid, so it could duplicate the very keys it was rebuilding. + """ + from test_stats_duplicates import duplicated_count_keys, duplicated_stat_keys + + table = cached_table + table.stats.add_stats(["flag"]) + assert table.stats.count({"n": {"$lt": 100}}, record=True) == 100 # an extra count + table.stats.max("n") + + table.delete({"n": {"$gte": 150}}, restat=False) + assert stats_valid_in_meta(table) is False + + table.stats.refresh_stats() + assert stats_valid_in_meta(table) is True + + assert duplicated_count_keys(table) == [] + assert duplicated_stat_keys(table) == [] + assert stored_counts_for(table, {}) == [150] + assert stored_counts_for(table, {"n": {"$lt": 100}}) == [100] + assert stored_counts_for(table, {"flag": True}) == [len([i for i in range(150) if i % 3 == 0])] + + +# --------------------------------------------------------------------------- +# reload_all finalizes what each reload prepared +# +# reload_all runs every reload first and swaps afterwards, so the second pass +# has to be handed the swap list the first pass built. Reconstructing it there +# from the files in the folder is not the same list: a saving table's reload +# always prepares both cache companions, refreshing them when a file is +# missing, so a folder without a _counts.txt yielded a swap list naming only +# the search table -- leaving the refreshed companions under _tmp and the old +# live ones, describing replaced data, marked valid. +# --------------------------------------------------------------------------- + +def reload_folder(table, tmp_path, counts=True, stats=True): + """ + A ``reload_all`` input folder for one table, optionally without its cache + files, exported before whatever change the test is about. + """ + folder = tmp_path / ("data_%d_%d" % (counts, stats)) + table._db.copy_to([table.search_table], str(folder)) + if not counts: + (folder / (table.search_table + "_counts.txt")).unlink() + if not stats: + (folder / (table.search_table + "_stats.txt")).unlink() + return folder + + +def assert_coherent_after_reload_all(table, expect_valid): + """ + Nothing stranded under _tmp, and the flag saying what the swap did. + """ + live = table._db[table.search_table] + for name in (live.search_table, live.stats.counts, live.stats.stats): + assert not live._table_exists(name + "_tmp"), "%s_tmp was left behind" % name + assert stats_valid_in_meta(live) is expect_valid + return live + + +@pytest.mark.parametrize("counts,stats", [(False, False), (True, False), (False, True)]) +@pytest.mark.parametrize("sequential_swap", [False, True]) +def test_reload_all_with_a_cache_file_missing_swaps_the_whole_family( + cached_table, tmp_path, counts, stats, sequential_swap +): + """ + The refreshed companions must reach the live names, and the flag may only + be true because they did. + """ + cached_table.stats.add_stats(["flag"]) + assert cached_table.stats.quick_count({"flag": True}) == NFLAGGED + folder = reload_folder(cached_table, tmp_path, counts=counts, stats=stats) + + # Make the live caches disagree with what the folder holds, so that + # serving them after the reload is visibly wrong. + cached_table.update({"flag": True}, {"flag": False}, restat=False) + cached_table.stats.refresh_stats() + assert cached_table.stats.quick_count({"flag": True}) is None + + cached_table._db.reload_all( + str(folder), restat=True, sequential_swap=sequential_swap + ) + + live = assert_coherent_after_reload_all(cached_table, expect_valid=True) + # the rebuilt value, from the reloaded data -- not the pre-reload cache + assert live.stats.quick_count({"flag": True}) == NFLAGGED + assert live.stats.count({"flag": True}) == NFLAGGED + + +@pytest.mark.parametrize("counts,stats", [(False, False), (True, False)]) +def test_reload_all_without_restat_leaves_the_table_invalid( + cached_table, tmp_path, counts, stats +): + """ + Retaining old caches is allowed; claiming they are valid is not. + """ + cached_table.stats.add_stats(["flag"]) + folder = reload_folder(cached_table, tmp_path, counts=counts, stats=stats) + cached_table.update({"flag": True}, {"flag": False}, restat=False) + cached_table.stats.refresh_stats() + + cached_table._db.reload_all(str(folder), restat=False) + + live = assert_coherent_after_reload_all(cached_table, expect_valid=False) + assert live.stats.quick_count({"flag": True}) is None + assert live.stats.count({"flag": True}) == NFLAGGED + + +def test_reload_all_with_both_cache_files_ends_valid(cached_table, tmp_path): + """ + The unchanged case, kept honest: both companions come from the folder, so + the swap has all three relations and may say so. + """ + cached_table.stats.add_stats(["flag"]) + folder = reload_folder(cached_table, tmp_path) + cached_table.update({"flag": True}, {"flag": False}, restat=False) + + cached_table._db.reload_all(str(folder)) + + live = assert_coherent_after_reload_all(cached_table, expect_valid=True) + assert live.stats.quick_count({"flag": True}) == NFLAGGED + + +def test_a_deferred_swap_is_told_what_the_reload_prepared(cached_table, tmp_path): + """ + The plan itself: a saving table's reload prepares both cache companions + whatever files it was given, and says so, so a caller deferring the swap + has no reason to reconstruct the list. + """ + searchfile = str(tmp_path / "data.txt") + cached_table.copy_to(searchfile) + plan = cached_table.reload(searchfile, restat=True, final_swap=False) + try: + assert set(plan.tables) == { + cached_table.search_table, + cached_table.stats.counts, + cached_table.stats.stats, + } + assert plan.stats_valid is True + finally: + cached_table.drop_tmp()