diff --git a/docs/providers/aws.md b/docs/providers/aws.md index ebb4407..da5c411 100644 --- a/docs/providers/aws.md +++ b/docs/providers/aws.md @@ -668,10 +668,10 @@ database: capture_log_seconds: 600 ``` -Each `collect` then watches the log for `capture_log_seconds` and reads that -window over its own connection. Keep the schedule interval longer than -`capture_log_seconds`. The same connection reads a pgAudit csv log when -`capture_log_type` is `pgaudit` instead of `statement`. +Each `collect` then watches the log for `capture_log_seconds`, takes its +snapshot, and reads that window over its own connection. Keep the schedule +interval longer than `capture_log_seconds`. The same connection reads a pgAudit +csv log when `capture_log_type` is `pgaudit` instead of `statement`. **Turn it off afterwards:** return `log_min_duration_statement` to the value it had. RDS re-adds `stderr` alongside `csvlog`, so every event is written twice diff --git a/docs/workload-bundle.md b/docs/workload-bundle.md index e1980b0..21f5c18 100644 --- a/docs/workload-bundle.md +++ b/docs/workload-bundle.md @@ -50,6 +50,14 @@ Two more files appear only when the query log was captured: | `burst.csv` | the query log: every statement the server logged, **with literal values in the `parameters` column**, which is real data, by design. `session_id` and `command_tag` carry the transaction framing: which statements shared a session, and which rows opened or closed a transaction | | `coverage.json` | how much of the aggregate workload the log windows observed, and whether they landed on ordinary traffic | +Both checks in `coverage.json` compare the log with the calls that a client +sent. A call that a function, procedure or trigger runs never reaches the log, +so the checks leave out the `nested_calls` of each statement. +`nested_calls_left_out` counts those calls, and `statements.only_nested` counts +the statements that ran only inside a function, procedure or trigger. When the +server cannot separate the two (PostgreSQL 13 and earlier), `nesting_unknown` is +`true` and the checks count every call. + Every capture declares the columns your migration engineer names, and the bundle then holds one more file: @@ -106,12 +114,23 @@ Each record in `workload.sql` carries its counters as comments, so the file holds the traffic weight as well as the statement: -- neki:stmt id= - -- neki:metrics calls=12904 total_exec_time_ms=88213.4 rows=1032320 + -- neki:metrics calls=12904 total_exec_time_ms=88213.4 rows=1032320 nested_calls=0 SELECT ... ; A query log alone weights every pattern equally, so the weight travels with the statement. +`nested_calls` counts how many of the `calls` ran inside a function, procedure +or trigger. It comes from the `toplevel` column of `pg_stat_statements`. A +statement whose `nested_calls` equals `calls` ran only inside one, so it can +name `NEW`, `OLD` or a function parameter, which exist only there. PostgreSQL +records a nested statement only when `pg_stat_statements.track` is `all`. + +The header leaves `nested_calls` out when the server cannot say. +`pg_stat_statements` has no `toplevel` column before PostgreSQL 14, and there +one entry holds both kinds of call. The README's caveats name this case, and +count the statements that ran nested. + The planning tools do not read these headers yet. Their SQL loader supplies no count, so every pattern arrives weighted 1 and a rare statement carries the weight of a hot one. This is being addressed on the planning side. Until it is, @@ -191,6 +210,8 @@ when the server cannot say: `pg_stat_statements_info` arrived in PostgreSQL 14, so 12 and 13 report unknown rather than a `false` that would claim nothing was lost. Treat `null` as unknown, not as falsy. +When `pgss.evicted` is `true`, the README's caveats say so too. + ## Column statistics Row counts say how big a table is. They cannot say whether a column can spread @@ -224,6 +245,14 @@ you declare are in [`distributions.json`](#value-distributions). The figures come from the last `ANALYZE`, so they are a sample and an estimate, and they describe one column at a time rather than a composite key. +PostgreSQL shows a column in `pg_stats` only to a role that may `SELECT` the +column. A role that holds only `pg_monitor`, as the capture guide sets up, sees +only the columns you grant it. When no table in the capture has a row in +`pg_stats`, the README's caveats say so and name the cause: no `ANALYZE` has +run, the analyzed tables hold no rows, the role may not `SELECT` the columns, +or no snapshot could read `pg_stats`. A snapshot that cannot read `pg_stats` +does not replace an earlier reading, and the caveats count those snapshots. + ## Value distributions `column_stats.json` says how skewed a column is. It cannot say which values diff --git a/docs/workload_capture.md b/docs/workload_capture.md index ffba13a..607bbdb 100644 --- a/docs/workload_capture.md +++ b/docs/workload_capture.md @@ -699,9 +699,10 @@ the file's encoding itself: plain text, csvlog, or a Cloud SQL or AlloyDB pgAudit JSON export. With `capture_log_source: log_fdw`, the tool reads the server's log over its own -connection instead, and each `collect` watches the log for -`capture_log_seconds` after its snapshot, so the log window always falls inside -the snapshot window around it. **Keep the schedule interval longer than +connection instead. Each `collect` watches the log for `capture_log_seconds`, +takes its snapshot, and then reads the log. The log window therefore falls +between the previous snapshot and this one, so the counters in `workload.sql` +cover the same traffic as `burst.csv`. **Keep the schedule interval longer than `capture_log_seconds`**, or the next `collect` starts before this one returns. With the hourly crontab line `init` prints, anything up to about 45 minutes is safe. @@ -1156,7 +1157,10 @@ function or procedure is attributed to the call rather than recorded on its own. If your application's work happens inside stored procedures, the query log will describe the calls rather than the queries. Set `pg_stat_statements.track = all` to record both. It costs more shared memory and more entries, so raise -`pg_stat_statements.max` with it. +`pg_stat_statements.max` with it. With `all`, `workload.sql` gives each +statement a `nested_calls` count: the calls that ran inside a function, +procedure or trigger. `coverage.json` leaves those calls out, because the query +log never holds them. ### W202 statement_eviction_possible diff --git a/planetscale_discovery/workload/bundle.py b/planetscale_discovery/workload/bundle.py index d8b6ebb..eda07bc 100644 --- a/planetscale_discovery/workload/bundle.py +++ b/planetscale_discovery/workload/bundle.py @@ -169,6 +169,7 @@ def write_bundle( outside_window, unreadable_bursts, distributions_summary, + plannable, ), } _write_text(out_dir / "README.md", _render_readme(summary, out_dir)) @@ -542,12 +543,15 @@ def render_workload_sql( for statement in sorted(statements, key=lambda s: s["id"]): counters = statement.get("counters") or {} + # Absent, not 0, when the server cannot say which calls were nested. + nested = statement.get("nested_calls") lines.append(f"-- neki:stmt id={statement['id']}") lines.append( "-- neki:metrics" f" calls={_num(counters.get('calls'))}" f" total_exec_time_ms={_ms(counters.get('total_exec_time'))}" f" rows={_num(counters.get('rows'))}" + + ("" if nested is None else f" nested_calls={_num(nested)}") ) text = (statement.get("query") or "").rstrip().rstrip(";") lines.append(f"{text};") @@ -697,6 +701,7 @@ def _caveats( outside_window=None, unreadable_bursts=None, distributions_summary=None, + plannable=None, ) -> List[str]: """Notes that change how the counts in this bundle should be read.""" caveats: List[str] = [] @@ -738,6 +743,31 @@ def _caveats( "A growth rate needs a size reading at each end of the window." ) + empty_stats = _empty_column_stats(merged) + if empty_stats: + caveats.append(empty_stats) + failed_reads = merged.get("pg_stats_failed_reads") or 0 + if failed_reads and merged.get("pg_stats_read"): + # Only the collect console showed these, and the bundle outlives it. + caveats.append( + f"{failed_reads} snapshot(s) could not read pg_stats " + f"({label('column_stats_read_failed')}). column_stats.json holds " + "the last reading that succeeded, taken at " + f"{merged.get('pg_stats_read_at') or 'an unknown time'}." + ) + + pgss = merged.get("pgss") or {} + if pgss.get("evicted"): + # The window's own evictions. W202 at init only says they are possible. + limit = f" (now {pgss['max']})" if pgss.get("max") else "" + caveats.append( + "pg_stat_statements was full during the window, so it discarded " + "entries. It discards the least-executed statements first. A " + "statement that ran rarely can be missing from workload.sql, or " + "its counts can be low. Raise pg_stat_statements.max" + f"{limit} before the next capture." + ) + skipped = (merged.get("coverage") or {}).get("intervals_skipped") or [] if skipped: caveats.append( @@ -801,6 +831,29 @@ def _caveats( "the system catalogs rather than your application's traffic." ) + nested = [s for s in plannable or [] if s.get("nested_calls")] + if nested: + only = sum( + 1 + for s in nested + if s["nested_calls"] >= ((s.get("counters") or {}).get("calls") or 0) + ) + caveats.append( + f"{len(nested)} statement(s) in workload.sql ran inside a function, " + f"procedure or trigger: {only} only there, and {len(nested) - only} " + "also as statements of their own. The nested_calls value on each " + "neki:metrics line counts those calls." + ) + if merged.get("toplevel_known") is False: + caveats.append( + "This server's pg_stat_statements has no toplevel column, which " + "arrived in PostgreSQL 14. So workload.sql cannot show which " + "statements ran inside a function, procedure or trigger, and its " + "neki:metrics lines carry no nested_calls. With " + "pg_stat_statements.track set to 'all', those statements look like " + "statements your application sent." + ) + if parse.get("available") is False: caveats.append( "The schema file was not re-parsed before being written, because the " @@ -866,6 +919,57 @@ def _caveats( return caveats +def _empty_column_stats(merged: Dict[str, Any]) -> Optional[str]: + """A caveat when no table in the capture has column statistics, or None. + + An empty column_stats.json looks the same whatever the cause, so the caveat + names the cause that the capture can see. The analyze timestamps come from + pg_stat_user_tables, which pg_monitor can read. pg_stats shows a column only + to a role that may SELECT it, so the two sources can disagree. + """ + tables = list((merged.get("tables") or {}).values()) + if not tables or merged.get("columns"): + return None + effect = ( + "column_stats.json is empty, and plantest_counts.json holds no NDV or " + "null fraction for any column." + ) + if merged.get("pg_stats_read") is False: + return ( + f"{effect} No snapshot could read pg_stats " + f"({label('column_stats_read_failed')})." + ) + analyzed = [t for t in tables if t.get("last_analyze") or t.get("last_autoanalyze")] + # ANALYZE records no statistics for an empty table. + with_rows = sum(1 for t in analyzed if (t.get("estimated_rows") or 0) > 0) + seen = f"pg_stats showed no row for any of the {len(tables)} table(s)." + if with_rows: + return ( + f"{effect} {seen} {with_rows} of those tables hold rows and record " + "an ANALYZE, so PostgreSQL holds statistics for them. pg_stats " + "hides a column from a role that may not SELECT the column, and " + "from a role that row-level security applies to. See 'Grant the " + "capture role each column' in the workload capture guide." + ) + # No table both holds rows and records an ANALYZE. Name each group, so a + # capture with some of each gets advice for both. + causes = [] + never = len(tables) - len(analyzed) + if never: + causes.append( + f"{never} table(s) record no ANALYZE, and PostgreSQL fills pg_stats " + "only when ANALYZE runs. Run ANALYZE, then run 'collect' again " + "before 'finalize'. pg_stats also hides a column from a role that " + "may not SELECT the column." + ) + if analyzed: + causes.append( + f"{len(analyzed)} table(s) record an ANALYZE but hold no rows, and " + "ANALYZE records no statistics for an empty table." + ) + return f"{effect} {seen} {' '.join(causes)}" + + def _render_readme(summary: Dict[str, Any], out_dir: Path) -> str: covered = summary.get("covered_seconds") window = ( diff --git a/planetscale_discovery/workload/bundle_burst.py b/planetscale_discovery/workload/bundle_burst.py index a1fdc5e..9f5e67f 100644 --- a/planetscale_discovery/workload/bundle_burst.py +++ b/planetscale_discovery/workload/bundle_burst.py @@ -86,24 +86,13 @@ def _bursts_in_window(bursts, merged): if start is None or end is None: undated.append(burst) included.append(burst) - elif end < window_start: - excluded.append(burst) - elif start > window_end and not _read_during_the_capture(burst): + elif end < window_start or start > window_end: excluded.append(burst) else: included.append(burst) return included, excluded, unreadable, undated -def _read_during_the_capture(burst: Dict[str, Any]) -> bool: - """A live read watches the log after its own snapshot, so the last window - of a capture always starts after the last snapshot it is measured against. - """ - from planetscale_discovery.workload.burst.collector import SOURCE_LIVE - - return burst.get("source") == SOURCE_LIVE - - def _scope_burst(burst: Dict[str, Any], known: Set[str]) -> Dict[str, Any]: if not known: return burst diff --git a/planetscale_discovery/workload/burst/coverage.py b/planetscale_discovery/workload/burst/coverage.py index ce6a8af..0659b54 100644 --- a/planetscale_discovery/workload/burst/coverage.py +++ b/planetscale_discovery/workload/burst/coverage.py @@ -1,7 +1,13 @@ -from typing import Any, Dict, Iterable, List, Optional +from typing import Any, Dict, Iterable, List, Optional, Tuple MISSED_REPORTED = 20 +NESTING_UNKNOWN = ( + ". This server's pg_stat_statements cannot separate the calls that ran " + "inside a function, procedure or trigger. With pg_stat_statements.track " + "set to 'all', the window rate includes them, so the ratio reads low" +) + STATUS_OK = "ok" STATUS_NO_JOIN_KEY = "no_join_key" STATUS_NO_AGGREGATE = "no_aggregate" @@ -47,16 +53,22 @@ def coverage( total_time = 0.0 seen_calls = 0.0 seen_time = 0.0 + matched = 0 + only_nested = 0 missed: List[Dict[str, Any]] = [] for record in aggregate: - counters = record.get("counters") or {} - calls = float(counters.get("calls") or 0) - exec_time = float(counters.get("total_exec_time") or 0) + calls, exec_time, nested = _top_level(record) + if nested and not calls: + # The log never holds a statement that only a function, + # procedure or trigger ran. + only_nested += 1 + continue total_calls += calls total_time += exec_time if _observed(record, observed_ids): + matched += 1 seen_calls += calls seen_time += exec_time else: @@ -70,7 +82,7 @@ def coverage( ) missed.sort(key=lambda item: item["calls"], reverse=True) - matched = len(aggregate) - len(missed) + nested_calls, nesting_unknown = _nested_split(aggregate) return { "status": STATUS_OK, @@ -79,6 +91,7 @@ def coverage( "in_aggregate": len(aggregate), "observed_in_burst": matched, "not_observed": len(missed), + "only_nested": only_nested, }, "calls": { "total": total_calls, @@ -92,6 +105,8 @@ def coverage( }, "heaviest_not_observed": missed[:MISSED_REPORTED], "not_observed_calls": sum(item["calls"] for item in missed), + "nested_calls_left_out": nested_calls, + "nesting_unknown": nesting_unknown, } @@ -103,7 +118,23 @@ def representativeness( statements = list(burst_statements) window_seconds = merged.get("covered_seconds") totals = merged.get("totals") or {} - window_calls = float(totals.get("calls") or 0) + nested_calls, nesting_unknown = _nested_split( + (merged.get("statements") or {}).values() + ) + # totals.calls also counts the calls that ran nested, which the log never + # holds. + all_calls = float(totals.get("calls") or 0) + window_calls = max(all_calls - nested_calls, 0.0) + + if all_calls and not window_calls: + return { + "known": False, + "detail": ( + "every call in the aggregate window ran inside a function, " + "procedure or trigger, and the log holds only the statements " + "that a client sent" + ), + } if not (burst_seconds and window_seconds and window_calls): return { @@ -123,10 +154,48 @@ def representativeness( "burst_statements_per_second": burst_rate, "window_calls_per_second": window_rate, "ratio": ratio, - "detail": _representativeness_detail(ratio), + "nested_calls_left_out": nested_calls, + "nesting_unknown": nesting_unknown, + "detail": _representativeness_detail(ratio) + + (NESTING_UNKNOWN if nesting_unknown else ""), } +def _top_level(record: Dict[str, Any]) -> Tuple[float, float, Optional[float]]: + """The calls and execution time of the statement as a client sent it. + + The query log holds only the statements that a client sends. A call that a + function, procedure or trigger runs never reaches the log. The record sums + its top-level and nested rows, so the nested part is subtracted. The third + value is the nested calls, or None when the server cannot separate them. + """ + counters = record.get("counters") or {} + calls = float(counters.get("calls") or 0) + exec_time = float(counters.get("total_exec_time") or 0) + nested = record.get("nested_calls") + if nested is None: + return calls, exec_time, None + nested_time = (record.get("nested_counters") or {}).get("total_exec_time") + return ( + max(calls - float(nested), 0.0), + max(exec_time - float(nested_time or 0), 0.0), + float(nested), + ) + + +def _nested_split(records: Iterable[Dict[str, Any]]) -> Tuple[float, bool]: + """The nested calls, and whether a statement with calls could not say.""" + nested_calls = 0.0 + nesting_unknown = False + for record in records: + calls, _, nested = _top_level(record) + if nested is None: + nesting_unknown = nesting_unknown or calls > 0 + else: + nested_calls += nested + return nested_calls, nesting_unknown + + def _representativeness_detail(ratio: Optional[float]) -> str: if ratio is None: return "the aggregate window recorded no calls per second" diff --git a/planetscale_discovery/workload/cli_workload.py b/planetscale_discovery/workload/cli_workload.py index da5e684..c0d8c53 100644 --- a/planetscale_discovery/workload/cli_workload.py +++ b/planetscale_discovery/workload/cli_workload.py @@ -540,14 +540,16 @@ def _collect(args, config, logger) -> int: ) return EXIT_CAP_REACHED - from_file = workload.capture_log and workload.capture_log_source != SOURCE_LIVE + if workload.capture_log and workload.capture_log_source == SOURCE_LIVE: + _capture_log_window(store, config, workload, logger) + return EXIT_OK + + from_file = workload.capture_log _snapshot(store, config, logger, optional=from_file) # The snapshot is stored, so an unreadable log is never a non-zero exit. if from_file: return _collect_exported_log(store, workload, logger) - if workload.capture_log: - _capture_log_window(store, config, workload, logger) return EXIT_OK @@ -660,7 +662,7 @@ def _collect_distributions(store, connection, config, logger) -> None: def _capture_log_window(store, config, workload, logger) -> None: - """Watch the server's log for capture_log_seconds, then read that window.""" + """Watch the log for capture_log_seconds, snapshot, then read the window.""" from planetscale_discovery.workload.burst import STATUS_FAILED, BurstCollector seconds = workload.capture_log_seconds @@ -673,6 +675,9 @@ def _capture_log_window(store, config, workload, logger) -> None: ) until = _utc_now() + # After the watch so its traffic is counted; before the read so a failed read keeps it. + _snapshot(store, config, logger, optional=False) + try: connection = _connect(config, logger) except Exception as exc: @@ -1092,8 +1097,9 @@ def _print_cron_hint(session: str, workload: WorkloadConfig) -> None: print("") if watching: print( - f"Each 'collect' takes one snapshot, then watches the query log for " - f"{workload.capture_log_seconds}s. To collect hourly, add" + f"Each 'collect' watches the query log for " + f"{workload.capture_log_seconds}s, then takes one snapshot. To collect " + "hourly, add" ) elif workload.capture_log: print( diff --git a/planetscale_discovery/workload/collect.py b/planetscale_discovery/workload/collect.py index e54a5dc..7b6b2a5 100644 --- a/planetscale_discovery/workload/collect.py +++ b/planetscale_discovery/workload/collect.py @@ -549,7 +549,9 @@ def _statement_row(self, row: Dict[str, Any]) -> Dict[str, Any]: "userid": _int(row.get("userid")), "dbid": _int(row.get("dbid")), "queryid": _text(row.get("queryid")), - "toplevel": bool(row.get("toplevel", True)), + # None before PostgreSQL 14, which has no toplevel column. A + # missing value is unknown, not a top-level or a nested call. + "toplevel": _bool(row.get("toplevel")), "query": redacted, "query_kind": None if masked else statement_kind(raw), "text_unavailable": masked, @@ -696,6 +698,10 @@ def _number(value): return None +def _bool(value): + return None if value is None else bool(value) + + def _int(value): try: return int(value) diff --git a/planetscale_discovery/workload/merge.py b/planetscale_discovery/workload/merge.py index 9ccc174..85289de 100644 --- a/planetscale_discovery/workload/merge.py +++ b/planetscale_discovery/workload/merge.py @@ -45,10 +45,17 @@ def _column_stats(usable: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: evenly at the start of a capture and piles onto one value by the end is a different risk from one that was always lopsided, and a single reading cannot tell them apart. + + A snapshot that could not read pg_stats holds no columns, which is not a + reading. The readings come from the snapshots that read it, so one failed + tick, the last one included, does not empty the file. """ - first = {(c["table"], c["column"]): c for c in (usable[0].get("columns") or [])} + readings = [s for s in usable if _read_pg_stats(s)] + if not readings: + return [] + first = {(c["table"], c["column"]): c for c in (readings[0].get("columns") or [])} out: List[Dict[str, Any]] = [] - for current in usable[-1].get("columns") or []: + for current in readings[-1].get("columns") or []: earlier = first.get((current["table"], current["column"])) or {} top = current.get("top_frequencies") or [] top_first = (earlier.get("top_frequencies") or [None])[0] @@ -76,6 +83,25 @@ def _column_stats(usable: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: return out +def _read_pg_stats(snapshot: Dict[str, Any]) -> bool: + """Whether the snapshot read pg_stats, whatever the read returned.""" + return not any( + w.get("code") == "column_stats_read_failed" + for w in (snapshot.get("warnings") or []) + ) + + +def _pg_stats_reads(usable: Sequence[Dict[str, Any]]) -> Dict[str, Any]: + """Whether any snapshot read pg_stats, how many could not, and when the + reading that column_stats.json keeps was taken.""" + read = [s for s in usable if _read_pg_stats(s)] + return { + "pg_stats_read": bool(read), + "pg_stats_failed_reads": len(usable) - len(read), + "pg_stats_read_at": read[-1].get("captured_at_server") if read else None, + } + + def _row_identity(entry: Dict[str, Any]) -> tuple: """What makes one ``pg_stat_statements`` row the same row as before. @@ -180,6 +206,7 @@ def merge_snapshots( "window_end": usable[-1].get("captured_at_server"), }, "pgss": _pgss_window(usable), + "toplevel_known": _toplevel_known(usable), "row_cap_reached": any( w.get("code") == "row_cap_reached" for s in usable @@ -195,6 +222,8 @@ def merge_snapshots( # Carried so a reader can tell which database a window came from. "server": dict(usable[-1].get("server") or {}), "columns": _column_stats(usable), + # Separates "pg_stats returned no rows" from "pg_stats was never read". + **_pg_stats_reads(usable), } @@ -254,6 +283,8 @@ def _cumulative(usable, snapshots, allow_partial) -> Dict[str, Any]: "text_unavailable": entry.get("text_unavailable", False), "queryids": [], "counters": {}, + "nested_counters": {}, + "nesting_unknown": False, "max_exec_time": None, "min_exec_time": None, }, @@ -264,6 +295,7 @@ def _cumulative(usable, snapshots, allow_partial) -> Dict[str, Any]: for name, value in (entry.get("counters") or {}).items(): if value is not None: record["counters"][name] = record["counters"].get(name, 0.0) + value + _add_nested(record, snapshot, entry, {}) record["max_exec_time"] = _max( record["max_exec_time"], _lifetime_value(entry, "max_exec_time") ) @@ -285,12 +317,14 @@ def _cumulative(usable, snapshots, allow_partial) -> Dict[str, Any]: "window_end": snapshot.get("captured_at_server"), }, "resets_observed": 0, + "toplevel_known": _toplevel_known(usable), "statements": statements, "tables": {t["table"]: dict(t) for t in snapshot.get("tables") or []}, "indexes": {_index_key(i): dict(i) for i in snapshot.get("indexes") or []}, "lifetime": _lifetime(snapshot), "server": dict(snapshot.get("server") or {}), "columns": _column_stats(usable), + **_pg_stats_reads(usable), "totals": _totals(statements, None), } @@ -349,6 +383,8 @@ def _accumulate_statements(previous, current, out) -> int: "text_unavailable": entry.get("text_unavailable", False), "queryids": [], "counters": {}, + "nested_counters": {}, + "nesting_unknown": False, "max_exec_time": None, "min_exec_time": None, }, @@ -360,6 +396,7 @@ def _accumulate_statements(previous, current, out) -> int: prior = (before.get(_row_identity(entry)) or {}).get("counters") or {} if _add_counters(record["counters"], prior, entry.get("counters") or {}): resets += 1 + _add_nested(record, current, entry, prior) # Lifetime values: carried, never differenced. The collector keeps them # beside `counters` for that reason, so read the record itself first. @@ -372,6 +409,41 @@ def _accumulate_statements(previous, current, out) -> int: return resets +def _add_nested(record, snapshot, entry, prior) -> None: + """Add the row's calls from inside a function, procedure or trigger. + + The output keys each statement by its text, so a top-level row and a nested + row with the same text become one record. The nested part is kept beside + the record's counters, so the split survives the merge. + """ + nested = _nested(snapshot, entry) + if nested is None: + record["nesting_unknown"] = True + elif nested: + # A reset is already counted once, with the record's own counters. + _add_counters(record["nested_counters"], prior, entry.get("counters") or {}) + + +def _nested(snapshot, entry) -> Optional[bool]: + """Whether the row counts calls from inside a function, procedure or trigger. + + None when the server cannot say. Before PostgreSQL 14 pg_stat_statements + has no toplevel column, and one row holds both kinds of call. Snapshots + from older versions of this tool stored that missing value as False, so + the snapshot's own list of missing columns decides, not the row. + """ + if "toplevel" in (snapshot.get("statement_columns_missing") or []): + return None + toplevel = entry.get("toplevel") + return None if toplevel is None else not toplevel + + +def _toplevel_known(usable: Sequence[Dict[str, Any]]) -> bool: + return not any( + "toplevel" in (s.get("statement_columns_missing") or []) for s in usable + ) + + def _accumulate_relations(previous, current, tables, indexes) -> int: resets = 0 resets += _accumulate_group( @@ -536,6 +608,13 @@ def _finish_statements(statements, covered_seconds) -> None: record["calls_per_second"] = ( (calls / covered_seconds) if covered_seconds and calls else None ) + # The calls that ran inside a function, procedure or trigger, out of + # calls. None when the server cannot separate the two. + record["nested_calls"] = ( + None + if record.get("nesting_unknown") + else (record.get("nested_counters") or {}).get("calls") or 0.0 + ) def _totals(statements, covered_seconds) -> Dict[str, Any]: diff --git a/tests/unit/test_workload_bundle.py b/tests/unit/test_workload_bundle.py index 533e333..336493d 100644 --- a/tests/unit/test_workload_bundle.py +++ b/tests/unit/test_workload_bundle.py @@ -440,6 +440,27 @@ def test_the_transaction_blind_spot_is_always_stated(self, tmp_path): "records statements, not transactions" in c for c in manifest["caveats"] ) + def test_evictions_during_the_window_are_disclosed(self, tmp_path): + """manifest.json records the eviction, but no reader of it sees why.""" + manifest = write_bundle( + tmp_path, + merged([stmt("a")], pgss={"max": 5000, "evicted": True}), + SCHEMA, + "t", + ) + caveat = next(c for c in manifest["caveats"] if "was full" in c) + assert "pg_stat_statements.max (now 5000)" in caveat + + @pytest.mark.parametrize("evicted", [False, None]) + def test_no_eviction_or_an_unknown_one_is_not_a_caveat(self, tmp_path, evicted): + manifest = write_bundle( + tmp_path, + merged([stmt("a")], pgss={"max": 5000, "evicted": evicted}), + SCHEMA, + "t", + ) + assert not any("was full" in c for c in manifest["caveats"]) + def test_never_analyzed_tables_are_disclosed(self, tmp_path): schema = { "table_analysis": [ @@ -645,3 +666,40 @@ def test_a_capture_with_no_calls_is_called_out(self, tmp_path): def test_a_capture_with_traffic_is_not(self, tmp_path): summary = write_bundle(tmp_path, merged([stmt("a")]), SCHEMA, "t") assert not any("No queries ran" in c for c in summary["caveats"]) + + +class TestNestedCalls: + """A statement that a trigger runs is not one the application sent.""" + + def _nested(self, sid, calls, nested_calls): + query = "UPDATE orders SET id = $1 WHERE id = NEW.id" + return dict(stmt(sid, "UPDATE", query, calls), nested_calls=nested_calls) + + def test_the_metrics_line_carries_the_nested_calls(self): + sql = render_workload_sql([self._nested("a", 10, 4)], 60.0) + assert ( + "-- neki:metrics calls=10 total_exec_time_ms=5.000 rows=10 " + "nested_calls=4\n" + ) in sql + + def test_an_unknown_split_is_left_off_the_metrics_line(self): + """0 would claim that no call was nested.""" + sql = render_workload_sql([self._nested("a", 10, None)], 60.0) + assert "nested_calls" not in sql + + def test_nested_statements_are_counted_in_the_caveats(self, tmp_path): + statements = [ + self._nested("a", 10, 10), + self._nested("b", 10, 4), + self._nested("c", 10, 0), + ] + summary = write_bundle(tmp_path, merged(statements), SCHEMA, "t") + caveat = next(c for c in summary["caveats"] if "nested_calls" in c) + assert caveat.startswith("2 statement(s) in workload.sql ran inside") + assert "1 only there, and 1 also" in caveat + + def test_a_server_with_no_toplevel_column_is_named(self, tmp_path): + summary = write_bundle( + tmp_path, merged([stmt("a")], toplevel_known=False), SCHEMA, "t" + ) + assert any("no toplevel column" in c for c in summary["caveats"]) diff --git a/tests/unit/test_workload_bundle_burst.py b/tests/unit/test_workload_bundle_burst.py index 1541b2c..d557bf4 100644 --- a/tests/unit/test_workload_bundle_burst.py +++ b/tests/unit/test_workload_bundle_burst.py @@ -120,15 +120,14 @@ def test_a_burst_entirely_after_the_window_is_excluded(self): assert included == [] assert excluded == [b] - def test_a_live_read_after_the_window_is_kept(self): - """The bug: log_fdw watches the log after its own snapshot, so the - last window of every capture fell outside and was dropped.""" + def test_a_live_read_after_the_window_is_excluded(self): + """A live window after the last snapshot has no counters, like any other.""" b = burst( "2026-08-25 12:00:05+00:00", "2026-08-25 12:10:00+00:00", source="log_fdw" ) included, excluded, unreadable, undated = _bursts_in_window([b], self.merged()) - assert included == [b] - assert excluded == [] + assert included == [] + assert excluded == [b] def test_a_live_read_before_the_window_is_still_excluded(self): b = burst( diff --git a/tests/unit/test_workload_burst_coverage.py b/tests/unit/test_workload_burst_coverage.py index 3943770..83a3f36 100644 --- a/tests/unit/test_workload_burst_coverage.py +++ b/tests/unit/test_workload_burst_coverage.py @@ -94,3 +94,128 @@ def test_caught_a_peak(self): result = representativeness([{}] * 1000, merged, 100) assert result["ratio"] > 2.0 assert "peak" in result["detail"] + + +def _record(sid, queryid, calls, exec_time, nested_calls=0.0, nested_time=0.0): + """A merged statement, shaped as merge.py leaves it.""" + return { + "id": sid, + "query": f"select {sid}", + "queryids": [queryid], + "counters": {"calls": calls, "total_exec_time": exec_time}, + "nested_counters": ( + {"calls": nested_calls, "total_exec_time": nested_time} + if nested_calls + else {} + ), + "nested_calls": nested_calls, + } + + +class TestNestedCallsInCoverage: + """The log holds only the statements that a client sends. A call that a + function, procedure or trigger runs never reaches it.""" + + def test_a_statement_only_a_trigger_runs_is_left_out(self): + """It was reported as the heaviest statement the burst missed.""" + merged = { + "statements": { + "app": _record("app", 1, 10, 100), + "trigger": _record("trigger", 2, 50, 500, 50, 500), + } + } + result = coverage([{"query_id": 1}], merged) + + assert result["calls"]["total"] == 10.0 + assert result["calls"]["share"] == 1.0 + assert result["exec_time_ms"]["share"] == 1.0 + assert result["heaviest_not_observed"] == [] + assert result["statements"]["only_nested"] == 1 + assert result["statements"]["observed_in_burst"] == 1 + assert result["nested_calls_left_out"] == 50.0 + assert result["nesting_unknown"] is False + + def test_a_statement_also_run_nested_counts_its_top_level_part(self): + """Its nested calls and nested time lowered the observed share.""" + merged = { + "statements": { + "app": _record("app", 1, 10, 100), + "both": _record("both", 2, 30, 300, 20, 250), + } + } + result = coverage([{"query_id": 2}], merged) + + assert result["calls"]["total"] == 20.0 + assert result["calls"]["observed"] == 10.0 + assert result["exec_time_ms"]["total"] == 150.0 + assert result["exec_time_ms"]["observed"] == 50.0 + missed = result["heaviest_not_observed"] + assert [(m["id"], m["calls"]) for m in missed] == [("app", 10.0)] + + def test_a_missed_statement_reports_its_top_level_calls(self): + merged = { + "statements": { + "app": _record("app", 1, 10, 100), + "both": _record("both", 2, 30, 300, 20, 250), + } + } + missed = coverage([{"query_id": 1}], merged)["heaviest_not_observed"] + assert missed == [ + { + "id": "both", + "query": "select both", + "calls": 10.0, + "total_exec_time_ms": 50.0, + } + ] + + def test_an_unknown_split_keeps_every_call_and_says_so(self): + """PostgreSQL 13 and earlier have no toplevel column.""" + unknown = dict(_record("old", 2, 30, 300), nested_calls=None) + merged = {"statements": {"app": _record("app", 1, 10, 100), "old": unknown}} + result = coverage([{"query_id": 1}], merged) + + assert result["calls"]["total"] == 40.0 + assert result["statements"]["only_nested"] == 0 + assert result["nesting_unknown"] is True + + +class TestNestedCallsInRepresentativeness: + def _merged(self, *records): + statements = {r["id"]: r for r in records} + calls = sum(r["counters"]["calls"] for r in records) + return { + "covered_seconds": 100, + "statements": statements, + "totals": {"calls": calls}, + } + + def test_the_window_rate_counts_top_level_calls(self): + """With track = all, the trigger's calls made an ordinary burst look + like a quiet one.""" + merged = self._merged( + _record("app", 1, 100, 100), _record("trigger", 2, 900, 900, 900, 900) + ) + result = representativeness([{}] * 100, merged, 100) + + assert result["window_calls_per_second"] == 1.0 + assert result["ratio"] == 1.0 + assert "quieter" not in result["detail"] + assert result["nested_calls_left_out"] == 900.0 + assert result["nesting_unknown"] is False + + def test_an_unknown_split_keeps_every_call_and_says_so(self): + unknown = dict(_record("old", 1, 1000, 1000), nested_calls=None) + result = representativeness([{}] * 100, self._merged(unknown), 100) + + assert result["window_calls_per_second"] == 10.0 + assert result["nesting_unknown"] is True + assert "cannot separate the calls" in result["detail"] + + def test_a_window_of_nested_calls_only_is_unknown(self): + """The ratio would divide by zero top-level calls.""" + merged = self._merged(_record("trigger", 1, 50, 50, 50, 50)) + result = representativeness([{}] * 10, merged, 100) + + assert result["known"] is False + assert "every call in the aggregate window ran inside" in result["detail"] diff --git a/tests/unit/test_workload_cli_log_capture.py b/tests/unit/test_workload_cli_log_capture.py index 7f15f8d..7e19949 100644 --- a/tests/unit/test_workload_cli_log_capture.py +++ b/tests/unit/test_workload_cli_log_capture.py @@ -418,6 +418,52 @@ def test_the_burst_is_read_over_the_window_the_wait_held(self, session, mocker): assert burst.call_args.kwargs["until"] == "END" assert len(session.burst_paths()) == 1 + def test_the_snapshot_closes_the_window_before_the_log_is_read( + self, session, mocker + ): + """The bug: a snapshot before the watch left the last window uncounted.""" + order = [] + mocker.patch.object(cli_workload, "_connect", return_value=mocker.Mock()) + snapshotter = mocker.Mock() + snapshotter.collect.side_effect = lambda: ( + order.append("snapshot") or dict(SNAPSHOT) + ) + mocker.patch.object(cli_workload, "_new_collector", return_value=snapshotter) + mocker.patch.object( + cli_workload, + "_wait", + side_effect=lambda seconds: order.append("wait") or False, + ) + reader = mocker.Mock() + reader.collect.side_effect = lambda: ( + order.append("read") + or { + "status": "ok", + "window_start": "2026-09-14T10:00:00Z", + "window_end": "2026-09-14T10:10:00Z", + "statements": [], + "sessions": {}, + "files_read": ["a"], + } + ) + mocker.patch( + "planetscale_discovery.workload.burst.BurstCollector", + return_value=reader, + ) + workload = WorkloadConfig( + capture_log=True, + capture_log_type="statement", + capture_log_source="log_fdw", + capture_log_seconds=10, + ) + + assert ( + _collect(_Args(session.directory), _Config(workload), mocker.Mock()) + == EXIT_OK + ) + assert order == ["wait", "snapshot", "read"] + assert len(session.snapshot_paths()) == 1 + class TestASuccessfulBurstReportsItsLoss: def test_warnings_are_logged_and_the_burst_is_stored(self, session, mocker): diff --git a/tests/unit/test_workload_collect.py b/tests/unit/test_workload_collect.py index 9b40b28..2b231fc 100644 --- a/tests/unit/test_workload_collect.py +++ b/tests/unit/test_workload_collect.py @@ -152,6 +152,20 @@ def test_masked_text_is_flagged_rather_than_redacted(self): assert row["query"] is None assert row["query_kind"] is None + def test_a_server_with_no_toplevel_column_reports_it_as_unknown(self): + """PostgreSQL 13 and earlier select NULL AS toplevel. bool(None) read + every statement there as nested.""" + row = self._collector()._statement_row( + {"query": "SELECT 1", "calls": 1, "toplevel": None} + ) + assert row["toplevel"] is None + + def test_a_nested_row_is_recorded_as_nested(self): + row = self._collector()._statement_row( + {"query": "SELECT 1", "calls": 1, "toplevel": False} + ) + assert row["toplevel"] is False + def test_long_text_is_truncated_and_flagged(self): from unittest.mock import MagicMock diff --git a/tests/unit/test_workload_column_stats.py b/tests/unit/test_workload_column_stats.py index cb311a3..8f12f55 100644 --- a/tests/unit/test_workload_column_stats.py +++ b/tests/unit/test_workload_column_stats.py @@ -146,3 +146,135 @@ def test_it_carries_the_capture_id(self, tmp_path): def test_it_counts_what_it_measured(self, tmp_path): assert self._write(tmp_path)["columns_measured"] == 1 + + +class TestAnEmptyReadingIsNamed: + """An empty column_stats.json must say why it is empty. + + The capture role from the guide holds pg_monitor only, and pg_stats shows a + column only to a role that may SELECT it. A database that ANALYZE never + reached looks the same in the file, so the caveat names the cause the + capture can see. + """ + + def _caveats(self, tmp_path, **extra): + payload = merged([stmt("a")], **extra) + payload["server"] = SERVER + return write_bundle(tmp_path, payload, SCHEMA, "t")["caveats"] + + def _tables(self, last_autoanalyze=None, estimated_rows=1000.0): + return { + "public.orders": { + "table": "public.orders", + "last_analyze": None, + "last_autoanalyze": last_autoanalyze, + "estimated_rows": estimated_rows, + } + } + + def test_analyzed_tables_with_no_visible_statistics_name_the_grant(self, tmp_path): + caveats = self._caveats( + tmp_path, + tables=self._tables("2026-10-01 00:00:00+00"), + columns=[], + pg_stats_read=True, + ) + caveat = next(c for c in caveats if "column_stats.json is empty" in c) + assert "may not SELECT" in caveat + assert "Run ANALYZE" not in caveat + + def test_tables_that_were_never_analyzed_ask_for_analyze(self, tmp_path): + caveats = self._caveats( + tmp_path, tables=self._tables(None), columns=[], pg_stats_read=True + ) + caveat = next(c for c in caveats if "column_stats.json is empty" in c) + assert "1 table(s) record no ANALYZE" in caveat + assert "Run ANALYZE" in caveat + + def test_a_mix_of_empty_and_unanalyzed_tables_names_both(self, tmp_path): + tables = { + **self._tables("2026-10-01 00:00:00+00", estimated_rows=0.0), + "public.items": { + "table": "public.items", + "last_analyze": None, + "last_autoanalyze": None, + "estimated_rows": None, + }, + } + caveats = self._caveats(tmp_path, tables=tables, columns=[], pg_stats_read=True) + caveat = next(c for c in caveats if "column_stats.json is empty" in c) + assert "1 table(s) record no ANALYZE" in caveat + assert "1 table(s) record an ANALYZE but hold no rows" in caveat + + def test_an_analyzed_empty_table_is_not_blamed_on_the_grant(self, tmp_path): + caveats = self._caveats( + tmp_path, + tables=self._tables("2026-10-01 00:00:00+00", estimated_rows=0.0), + columns=[], + pg_stats_read=True, + ) + caveat = next(c for c in caveats if "column_stats.json is empty" in c) + assert "hold no rows" in caveat + assert "SELECT" not in caveat + + def test_a_failed_read_is_not_reported_as_an_empty_one(self, tmp_path): + caveats = self._caveats( + tmp_path, tables=self._tables(None), columns=[], pg_stats_read=False + ) + caveat = next(c for c in caveats if "column_stats.json is empty" in c) + assert "W214" in caveat + assert "ANALYZE" not in caveat + + def test_a_capture_with_column_statistics_has_no_such_caveat(self, tmp_path): + caveats = self._caveats( + tmp_path, + tables=self._tables(None), + columns=[dict(column("tenant_id"), first_seen={}, top_share=0.02)], + pg_stats_read=True, + ) + assert not any("column_stats.json is empty" in c for c in caveats) + + def test_a_failed_last_read_keeps_the_earlier_reading(self): + failed = snap("2026-08-25 11:00:00+00:00", []) + failed["warnings"] = [{"code": "column_stats_read_failed", "detail": "x"}] + result = merge_snapshots( + [snap("2026-08-25 10:00:00+00:00", [column("tenant_id")]), failed] + ) + assert [c["column"] for c in result["columns"]] == ["tenant_id"] + assert result["pg_stats_read"] is True + + def test_a_failed_read_after_a_good_one_is_named(self, tmp_path): + """A lost tick showed only on the collect console.""" + caveats = self._caveats( + tmp_path, + tables=self._tables("2026-10-01 00:00:00+00"), + columns=[dict(column("tenant_id"), first_seen={}, top_share=0.02)], + pg_stats_read=True, + pg_stats_failed_reads=2, + pg_stats_read_at="2026-08-25 11:00:00+00:00", + ) + caveat = next(c for c in caveats if "could not read pg_stats" in c) + assert caveat.startswith("2 snapshot(s)") + assert "taken at 2026-08-25 11:00:00+00:00" in caveat + + def test_the_merge_counts_the_failed_reads(self): + failed = snap("2026-08-25 12:00:00+00:00", []) + failed["warnings"] = [{"code": "column_stats_read_failed", "detail": "x"}] + result = merge_snapshots( + [ + snap("2026-08-25 10:00:00+00:00", [column("tenant_id")]), + snap("2026-08-25 11:00:00+00:00", [column("tenant_id")]), + failed, + ] + ) + assert result["pg_stats_failed_reads"] == 1 + assert result["pg_stats_read_at"] == "2026-08-25 11:00:00+00:00" + + def test_no_reading_at_all_is_recorded_as_unread(self): + failed = [ + dict(snap(at, []), warnings=[{"code": "column_stats_read_failed"}]) + for at in ("2026-08-25 10:00:00+00:00", "2026-08-25 11:00:00+00:00") + ] + result = merge_snapshots(failed) + assert result["columns"] == [] + assert result["pg_stats_read"] is False diff --git a/tests/unit/test_workload_duplicate_rows.py b/tests/unit/test_workload_duplicate_rows.py index 869974f..ecad0bd 100644 --- a/tests/unit/test_workload_duplicate_rows.py +++ b/tests/unit/test_workload_duplicate_rows.py @@ -11,6 +11,8 @@ duplicated texts. """ +import pytest + from planetscale_discovery.workload.merge import merge_snapshots, statement_id SERVER = {"database": "app", "database_oid": 1, "version_num": 170000} @@ -140,3 +142,77 @@ def test_one_snapshot_sums_the_rows_rather_than_keeping_the_last(self): record = merged["statements"][statement_id(COMMIT)] assert merged["basis"] == "cumulative" assert record["counters"]["calls"] == 1055 + + +TRIGGER = "UPDATE counts SET n = n + $1 WHERE org_id = NEW.org_id" + + +class TestNestedCallsSurviveTheMerge: + """A top-level row and a nested row with the same text become one record. + + With pg_stat_statements.track = all, a statement that a trigger runs is + recorded beside the statements the application sends. The record keeps + the nested part, so a reader can tell the two apart. + """ + + def _record(self, first_rows, second_rows, missing=()): + first, second = snap(first_rows, T0), snap(second_rows, T1) + for snapshot in (first, second): + snapshot["statement_columns_missing"] = list(missing) + merged = merge_snapshots([first, second]) + return merged, merged["statements"][statement_id(TRIGGER)] + + def test_the_nested_calls_are_counted_beside_the_total(self): + _, record = self._record( + [row(TRIGGER, 900, toplevel=True), row(TRIGGER, 40, toplevel=False)], + [row(TRIGGER, 950, toplevel=True), row(TRIGGER, 70, toplevel=False)], + ) + assert record["counters"]["calls"] == 80 + assert record["nested_calls"] == 30 + + def test_a_statement_only_a_trigger_runs_is_all_nested(self): + _, record = self._record( + [row(TRIGGER, 40, toplevel=False)], [row(TRIGGER, 70, toplevel=False)] + ) + assert record["nested_calls"] == record["counters"]["calls"] == 30 + + def test_a_top_level_statement_has_no_nested_calls(self): + merged, record = self._record([row(TRIGGER, 900)], [row(TRIGGER, 950)]) + assert record["nested_calls"] == 0 + assert merged["toplevel_known"] is True + + def test_a_server_with_no_toplevel_column_cannot_say(self): + """Snapshots from older versions of this tool stored False there.""" + merged, record = self._record( + [row(TRIGGER, 40, toplevel=False)], + [row(TRIGGER, 70, toplevel=False)], + missing=("toplevel",), + ) + assert record["nested_calls"] is None + assert merged["toplevel_known"] is False + + @pytest.mark.parametrize("unknown", [1, 2]) + def test_one_unknown_interval_makes_the_window_unknown(self, unknown): + """A nested count from the known interval alone would understate the + window, in either order. A flag reset per interval passed the suite.""" + snapshots = [ + snap([row(TRIGGER, calls, toplevel=False)], at) + for calls, at in ((40, T0), (70, T1), (90, "2026-08-25 10:20:00+00:00")) + ] + snapshots[unknown]["statement_columns_missing"] = ["toplevel"] + merged = merge_snapshots(snapshots) + record = merged["statements"][statement_id(TRIGGER)] + assert record["nested_calls"] is None + assert merged["toplevel_known"] is False + + def test_one_snapshot_counts_the_nested_calls_as_well(self): + merged = merge_snapshots( + [ + snap( + [row(TRIGGER, 900, toplevel=True), row(TRIGGER, 4, toplevel=False)], + T0, + ) + ], + allow_partial=True, + ) + assert merged["statements"][statement_id(TRIGGER)]["nested_calls"] == 4