diff --git a/README.md b/README.md index d092fd7..b058846 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ Three ways in, one version number: | | | | |---|---|---| | **Library** | `pip install pg-partsmith` | Python, `asyncio` or sync, on any SQLAlchemy 2 engine. [Getting started](https://bedrock-python.github.io/pg-partsmith/getting-started/installation/) | -| **Command line** | `pip install "pg-partsmith[cli]"` | `pg-partsmith plan`, `apply` and `validate` over a YAML document, with exit codes a CronJob can read. [The CLI](https://bedrock-python.github.io/pg-partsmith/guide/cli/) | +| **Command line** | `pip install "pg-partsmith[cli]"` | `pg-partsmith plan`, `apply`, `validate` and `backfill` over a YAML document, with exit codes a CronJob can read. [The CLI](https://bedrock-python.github.io/pg-partsmith/guide/cli/) | | **Container image** | `ghcr.io/bedrock-python/pg-partsmith:latest` | The command line with no Python of your own: a Job, a CronJob, an init container. [The image](https://bedrock-python.github.io/pg-partsmith/guide/container/) | > [!TIP] diff --git a/docs/agents.md b/docs/agents.md index 969a1de..9f1b8c9 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -55,9 +55,11 @@ A configuration is a **scheme** — the shape of the tree, level by level — an * `PartitionToolkit.from_engine(engine, ...)` builds the repository, the metadata provider, the locks, the service and the maintainer around one engine, giving each setting that belongs to two of them (`marker_prefix`, `ddl_timezone`, `boundary_codec`) exactly once. -* `pg-partsmith` (extra `cli`) runs `inspect` / `plan` / `validate` / `apply` over a - document and a DSN. The first three issue no DDL; `apply` withholds detaches and drops - unless `--allow-destructive`. `plan --save FILE` writes the artifact `apply --plan FILE` +* `pg-partsmith` (extra `cli`) runs `inspect` / `plan` / `validate` / `apply` / `backfill` + over a document and a DSN. The first three issue no DDL; `apply` withholds detaches and + drops unless `--allow-destructive`. `backfill` is `partition_data` from the command line — + the one-off migration of a DEFAULT partition's rows into the partitions they belong in, + `--batch-rows` / `--max-batches`, exiting 2 while anything is left to move. `plan --save FILE` writes the artifact `apply --plan FILE` reads back, and applying it is refused if it was made for another table or under a configuration that has since changed (`--allow-config-drift` overrides). Exit codes: 0 nothing pending, 2 drift under `plan --check`, 3 findings or run issues, 4 configuration, @@ -72,7 +74,7 @@ A configuration is a **scheme** — the shape of the tree, level by level — an `PartitionEvent` as JSON on stdin; a non-zero exit refuses the operation. `PythonHooks` runs a block of Python per phase with `event` and `log` in scope; raising refuses. In a document both are the `hooks` section (a command list, or `{python: ...}` / - `{python_file: ...}`), honoured only under `apply --allow-hooks`; every block is compiled + `{python_file: ...}`), honoured only under `apply --allow-hooks` or `backfill --allow-hooks`; every block is compiled by `validate`. No sandbox is claimed. Hooks never fire during `plan`. See `guide/hooks-in-config.md`. * The same CLI ships as `ghcr.io/bedrock-python/pg-partsmith:` with the command diff --git a/docs/concepts/lifecycle.md b/docs/concepts/lifecycle.md index 48cce17..16ec4ca 100644 --- a/docs/concepts/lifecycle.md +++ b/docs/concepts/lifecycle.md @@ -112,7 +112,7 @@ with the core, which is what keeps a user predicate from turning into an acciden | `DetachMode` | Statement | |---|---| -| `AUTO` (default) | `DETACH … CONCURRENTLY`, falling back to the blocking form when PostgreSQL refuses it — it does when a DEFAULT partition exists | +| `AUTO` (default) | `DETACH … CONCURRENTLY`, falling back to the blocking form when PostgreSQL refuses it | | `CONCURRENT` | the concurrent form only; the refusal propagates | | `BLOCKING` | plain `DETACH`: `ACCESS EXCLUSIVE` on the parent for the duration | @@ -121,6 +121,13 @@ writers through; it cannot run inside a transaction block, so it goes out on an autocommit connection. A detach interrupted mid-way is finished with `DETACH … FINALIZE` on the next attempt. +PostgreSQL refuses the concurrent form outright while the parent holds a DEFAULT +partition. The planner reads that from the catalog, so an `AUTO` detach of a parent with +a DEFAULT partition is planned as the blocking form and says so — `plan --locks` names +the `ACCESS EXCLUSIVE` it will take, and no statement is issued that could only fail. +`CONCURRENT` is left as written: asking for the concurrent form and being refused is the +answer that mode exists to give. + Either form takes `ACCESS EXCLUSIVE` on every table that references the parent through a foreign key, and neither can detach a partition whose rows such a table still references: PostgreSQL refuses with `23503`. The executor records that as an issue and goes on; diff --git a/docs/guide/backfill.md b/docs/guide/backfill.md index 07c30db..2aa142d 100644 --- a/docs/guide/backfill.md +++ b/docs/guide/backfill.md @@ -30,6 +30,15 @@ complete subtree before it is attached, windows that already have a partition sk The return value lists the partitions this call created, in order; `ensure_partition` returns one or `None`. +## From the command line + +`pg-partsmith backfill` is the same migration without Python of your own: it runs +[`partition_data`](partition-existing-table.md) over every table in the document, +resumable through `--max-batches`, exiting `2` while anything is left to move. See +[The command line](cli.md#adopting-a-table-full-of-data). What has no command yet is +`ensure_partitions` for windows that hold no rows — a history imported into a table with +no DEFAULT partition still names its windows in Python. + ## Rows already in a DEFAULT partition If the history sits in the parent's DEFAULT partition, `ensure_partitions` moves each diff --git a/docs/guide/cli.md b/docs/guide/cli.md index e200d9d..0ac826b 100644 --- a/docs/guide/cli.md +++ b/docs/guide/cli.md @@ -13,17 +13,20 @@ pg-partsmith validate -c partitions.yaml # does the document match the datab pg-partsmith inspect -c partitions.yaml # what tree actually exists? pg-partsmith plan -c partitions.yaml # what would maintenance do, and why? pg-partsmith apply -c partitions.yaml # do it — creations only, by default +pg-partsmith backfill -c partitions.yaml # move a DEFAULT partition's rows where they belong ``` The first three issue no DDL, take no lock and fire no hook. `apply` is the one that -acts. +acts. `backfill` is the one an existing installation runs once, before the first `apply` +has anything sensible to do: see [Adopting a table full of data](#adopting-a-table-full-of-data). ## Every flag | Command | Flags | |---|---| -| all four | `-c/--config FILE` (required), `--dsn`, `--table NAME` (repeatable), `-o/--output human\|json\|metrics`, `--write FILE` (the output, into a file, atomically), `-v/--verbose` | +| all five | `-c/--config FILE` (required), `--dsn`, `--table NAME` (repeatable), `-o/--output human\|json\|metrics`, `--write FILE` (the output, into a file, atomically), `-v/--verbose` | | `plan` | `--check` exit `2` on pending operations · `--save FILE` write the plan · `--locks` print what each operation locks | +| `backfill` | `--batch-rows N` rows per statement (default 10000) · `--max-batches N` stop after N statements and exit `2` · `--allow-hooks` · `--ok-if-locked` | | `apply` | `--plan FILE` apply a saved plan · `--allow-destructive` detach and drop too · `--continue-on-error` isolate a failed operation · `--allow-config-drift` apply a plan whose document changed · `--allow-hooks` run the document's hooks · `--ok-if-locked` exit `0` rather than `6` on a held lock | | `schema` | no flags: prints the document's JSON Schema for an editor | | global | `--version` · `--install-completion` · `--show-completion` · `-h/--help` on anything | @@ -288,11 +291,44 @@ fingerprint asks whether the plan is still the same intent: a plan made under `retention_count: 12` names exactly the right partitions to expire, for a reason that stopped being true the moment someone wrote `120`. +## Adopting a table full of data + +Creation walks forward from the cursor, so a table that was partitioned around data +already in it — everything in one DEFAULT partition — is a table `apply` has nothing +useful to say about: the rows it holds are behind the cursor, and no create-ahead will +ever reach them. + +`backfill` is the migration that fixes that, once: + +```bash +pg-partsmith backfill -c partitions.yaml +``` + +Window by window, oldest first, it creates the partition, fills it in batches of +`--batch-rows` and attaches it. Afterwards DEFAULT is empty and the ordinary +`plan`/`apply` cycle takes over. + +It is resumable, which is what makes it usable on a table too large for one maintenance +window. `--max-batches N` stops after N statements per table and exits `2`; every row +already moved stays moved. So a Job can simply be run until it stops saying 2: + +```bash +until pg-partsmith backfill -c partitions.yaml --max-batches 50; do sleep 60; done +``` + +Two things to know before running it on a busy table. It takes the table's maintenance +lock for the duration of each call, so a scheduled `apply` will decline while it runs +(exit `6`), which is the lock doing its job. And a window's rows are invisible through +the parent between the moment they leave DEFAULT and the moment the partition is attached +— PostgreSQL will not attach a partition while DEFAULT still holds rows for it, so there +is no order that keeps them visible throughout. `--batch-rows` bounds how long that is. + ## Commands around the lifecycle A document can name a command, or a block of Python, to run before a drop, after a -create, and at six other moments. They fire during `apply` only, and only with -`--allow-hooks` — see [Commands around the lifecycle](hooks-in-config.md). +create, and at six other moments. They fire during `apply` and `backfill` — the commands +that carry out DDL — and only with `--allow-hooks` — see +[Commands around the lifecycle](hooks-in-config.md). ## In a container, on a schedule, in CI @@ -303,5 +339,9 @@ container, CI, systemd — with a copy-pasteable shape for each. ## What is not here yet -`partition_data` and `unpartition` — the batched row-movement verbs — are library-only for -now, because both want a progress story a one-shot command does not have yet. +`unpartition` — the way back from a partitioned tree to one plain table — is library-only +for now. `partition_data`, its counterpart, is `backfill` above: what a one-shot command +needed first was a progress story, and `--max-batches` with exit `2` and the +`pg_partsmith_backfill_incomplete` gauge is one. `unpartition` also has a destination +table to name and a `drop_emptied` to decide, and neither has an obvious spelling on a +command line yet. diff --git a/docs/index.md b/docs/index.md index 234c947..1c7171e 100644 --- a/docs/index.md +++ b/docs/index.md @@ -11,8 +11,8 @@ Three ways in, one version number: - **Library** — `pip install pg-partsmith`: Python, `asyncio` or sync, on any SQLAlchemy 2 engine. Start with [installation](getting-started/installation.md). -- **Command line** — `pip install "pg-partsmith[cli]"`: `pg-partsmith plan`, `apply` and - `validate` over a YAML document, with exit codes a CronJob can read. See +- **Command line** — `pip install "pg-partsmith[cli]"`: `pg-partsmith plan`, `apply`, + `validate` and `backfill` over a YAML document, with exit codes a CronJob can read. See [the CLI](guide/cli.md). - **Container image** — `ghcr.io/bedrock-python/pg-partsmith:latest`: the command line with no Python of your own, as a Job, a CronJob or an init container. See diff --git a/pg_partsmith/cli/commands.py b/pg_partsmith/cli/commands.py index 7654104..542d67f 100644 --- a/pg_partsmith/cli/commands.py +++ b/pg_partsmith/cli/commands.py @@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any from pg_partsmith.aio import PartitionToolkit, PartitionValidationService +from pg_partsmith.constants import DEFAULT_MOVE_BATCH_ROWS from pg_partsmith.exceptions import InvalidPartitionConfigError from pg_partsmith.plan import MaintenancePlan, OperationKind @@ -24,11 +25,11 @@ if TYPE_CHECKING: from collections.abc import Sequence - from pg_partsmith.entities import MaintenanceResult, TablePartitionConfig + from pg_partsmith.entities import MaintenanceResult, MigrationResult, TablePartitionConfig logger = logging.getLogger("pg_partsmith.cli") -__all__ = ["CommandResult", "run_apply", "run_inspect", "run_plan", "run_validate"] +__all__ = ["CommandResult", "run_apply", "run_backfill", "run_inspect", "run_plan", "run_validate"] @dataclass(frozen=True, slots=True) @@ -215,6 +216,53 @@ async def run_apply( ) +async def run_backfill( + kit: PartitionToolkit, + configs: Sequence[TablePartitionConfig], + *, + batch_rows: int = DEFAULT_MOVE_BATCH_ROWS, + max_batches: int | None = None, +) -> CommandResult: + """Move what a DEFAULT partition holds into the partitions those rows belong in. + + The migration an existing installation needs once, and the half of adoption + create-ahead cannot reach: a table partitioned around data already in it + has every old row in DEFAULT, and creation walks forward from the cursor. + Window by window, oldest first, the partition is created, filled in + batches and attached. + + Exiting DRIFT while anything is left to move is what lets a Job be run + until it exits 0, rather than having to parse the output to learn whether + the batch budget ran out. + + Args: + kit: The wiring. + configs: The tables to migrate, in order. + batch_rows: Rows moved per statement. + max_batches: Stop after this many statements per table and report what is left. + + Returns: + What was moved, and the code to exit with. + """ + blocks: list[str] = [] + entries: list[dict[str, Any]] = [] + incomplete = False + issues = False + for config in configs: + result = await kit.service.partition_data(config, batch_rows=batch_rows, max_batches=max_batches) + incomplete = incomplete or not result.complete + issues = issues or bool(result.issues) + entries.append({"table": config.qualified_name, "result": result.model_dump(mode="json", by_alias=True)}) + blocks.append(_describe_migration(config.qualified_name, result)) + + code = ExitCode.OK + if issues: + code = ExitCode.FINDINGS + elif incomplete: + code = ExitCode.DRIFT + return CommandResult(code=code, lines=blocks, payload=envelope("backfill", entries)) + + def _plan_for(config: TablePartitionConfig, plans: dict[str, MaintenancePlan]) -> MaintenancePlan: """The saved plan for one table, or a refusal naming what the file does hold.""" plan = plans.get(config.qualified_name) @@ -225,6 +273,15 @@ def _plan_for(config: TablePartitionConfig, plans: dict[str, MaintenancePlan]) - return plan +def _describe_migration(table_name: str, result: MigrationResult) -> str: + """One block per table: what moved, into how many partitions, and what is left.""" + into = f" into {len(result.partitions)} partitions" if result.partitions else "" + state = "DEFAULT is drained" if result.complete else "more to move — run it again" + lines = [f"{table_name} — moved {result.rows_moved} rows in {result.batches} batches{into}; {state}"] + lines.extend(f" [{issue.step.value}] {issue.partition_name}: {issue.error}" for issue in result.issues) + return "\n".join(lines) + + def _describe_result(table_name: str, result: MaintenanceResult) -> str: """One block per table: what it did, then anything it wants a human to see.""" counts = ( diff --git a/pg_partsmith/cli/main.py b/pg_partsmith/cli/main.py index 01d5ec9..1c20098 100644 --- a/pg_partsmith/cli/main.py +++ b/pg_partsmith/cli/main.py @@ -31,6 +31,7 @@ from pg_partsmith.__version__ import __version__ from pg_partsmith.aio import CommandHooks, PartitionToolkit, PythonHooks +from pg_partsmith.constants import DEFAULT_MOVE_BATCH_ROWS from pg_partsmith.document import PartitionsDocument from pg_partsmith.exceptions import ( InvalidPartitionConfigError, @@ -42,7 +43,7 @@ from pg_partsmith.python_hooks import PythonHookError from pg_partsmith.utils import pg_sqlstate -from .commands import CommandResult, run_apply, run_inspect, run_plan, run_validate +from .commands import CommandResult, run_apply, run_backfill, run_inspect, run_plan, run_validate from .exit_codes import ExitCode from .loader import ( DSN_ENV_VAR, @@ -164,6 +165,8 @@ class _Invocation: allow_hooks: bool = False write: Path | None = None ok_if_locked: bool = False + batch_rows: int = DEFAULT_MOVE_BATCH_ROWS + max_batches: int | None = None def _version(value: bool) -> None: @@ -330,6 +333,58 @@ def apply( ) +@app.command() +def backfill( + config: ConfigOption, + dsn: DsnOption = None, + table: TableOption = None, + output: OutputOption = Output.human, + verbose: VerboseOption = False, + write: WriteOption = None, + batch_rows: Annotated[ + int, + typer.Option("--batch-rows", metavar="N", min=1, help="rows moved per statement"), + ] = DEFAULT_MOVE_BATCH_ROWS, + max_batches: Annotated[ + int | None, + typer.Option( + "--max-batches", + metavar="N", + min=1, + show_default=False, + help="stop after this many statements per table and exit 2; the next run carries on", + ), + ] = None, + allow_hooks: Annotated[ + bool, + typer.Option( + "--allow-hooks", + help="run what the document's hooks section names; without it, a document declaring any is refused", + ), + ] = False, + ok_if_locked: Annotated[ + bool, + typer.Option("--ok-if-locked", help="exit 0 rather than 6 when another maintainer holds the lock"), + ] = False, +) -> int: + """Move a DEFAULT partition's rows into the partitions they belong in. Exits 2 while more is left.""" + return _execute( + _Invocation( + "backfill", + config=config, + dsn=dsn, + tables=tuple(table or ()), + output=output, + verbose=verbose, + write=write, + batch_rows=batch_rows, + max_batches=max_batches, + allow_hooks=allow_hooks, + ok_if_locked=ok_if_locked, + ) + ) + + @app.command() def schema() -> int: """Print the document's JSON Schema, for an editor to validate against.""" @@ -527,6 +582,10 @@ async def _run(invocation: _Invocation) -> CommandResult: continue_on_error=invocation.continue_on_error, allow_config_drift=invocation.allow_config_drift, ) + if invocation.command == "backfill": + return await run_backfill( + kit, configs, batch_rows=invocation.batch_rows, max_batches=invocation.max_batches + ) return await run_validate(kit, configs) finally: await engine.dispose() @@ -537,7 +596,9 @@ def _hooks( ) -> list[PartitionLifecycleHooks] | None: """The hooks this run honours, and a refusal when it was not told to honour any. - Hooks fire during ``apply`` alone, so nothing else has to decide. Ignoring a + Hooks fire during the commands that carry out DDL -- ``apply`` and + ``backfill``, which creates partitions through the same executor -- so + nothing read-only has to decide. Ignoring a configured ``before_drop`` silently would be the worst outcome available: an operator would read the file, believe their archiver ran, and be wrong. So a document declaring hooks is refused rather than quietly stripped. @@ -548,7 +609,7 @@ def _hooks( hooks = document.hooks if hooks is None or hooks.is_empty: return None - if command != "apply": + if command not in {"apply", "backfill"}: return None if not allow_hooks: named = ", ".join(phase.value for phase in hooks.actions()) diff --git a/pg_partsmith/cli/metrics.py b/pg_partsmith/cli/metrics.py index 8ad05ab..1ca6a35 100644 --- a/pg_partsmith/cli/metrics.py +++ b/pg_partsmith/cli/metrics.py @@ -141,11 +141,29 @@ def _apply_metrics(lines: list[str], tables: list[dict[str, Any]]) -> None: _emit(lines, f"{_PREFIX}_issues", "Problems this run reported and did not fail on.", issues) +def _backfill_metrics(lines: list[str], tables: list[dict[str, Any]]) -> None: + """How far the migration out of DEFAULT has got, and whether it is finished.""" + moved: list[tuple[dict[str, str], float]] = [] + remaining: list[tuple[dict[str, str], float]] = [] + issues: list[tuple[dict[str, str], float]] = [] + for entry in tables: + table = str(entry.get("table", "")) + result = entry.get("result") or {} + moved.append(({"table": table}, result.get("rows_moved", 0))) + remaining.append(({"table": table}, 0.0 if result.get("complete", True) else 1.0)) + issues.append(({"table": table}, len(result.get("issues", [])))) + + _emit(lines, f"{_PREFIX}_backfilled_rows", "Rows this run moved out of a DEFAULT partition.", moved) + _emit(lines, f"{_PREFIX}_backfill_incomplete", "1 while a table still has rows left in DEFAULT.", remaining) + _emit(lines, f"{_PREFIX}_issues", "Problems this run reported and did not fail on.", issues) + + _RENDERERS = { "plan": _plan_metrics, "inspect": _inspect_metrics, "validate": _validate_metrics, "apply": _apply_metrics, + "backfill": _backfill_metrics, } diff --git a/pg_partsmith/plan.py b/pg_partsmith/plan.py index 44be1ff..2174abb 100644 --- a/pg_partsmith/plan.py +++ b/pg_partsmith/plan.py @@ -443,7 +443,7 @@ def capabilities(self) -> OperationCapabilities: transactional=False, lock="SHARE UPDATE EXCLUSIVE on the parent (CONCURRENTLY), ACCESS EXCLUSIVE on the partition and, in " "the second transaction, on every table referencing the parent through a foreign key; ACCESS EXCLUSIVE " - "on the parent when a DEFAULT partition forces the blocking form", + "on the parent should the server refuse the concurrent form", ) @property diff --git a/pg_partsmith/planner.py b/pg_partsmith/planner.py index 7ec8139..f459e6c 100644 --- a/pg_partsmith/planner.py +++ b/pg_partsmith/planner.py @@ -31,7 +31,7 @@ from .constants import MAX_IDENTIFIER_LENGTH from .entities import MaintenanceIssue, MaintenanceIssueStep, TablePartitionConfig from .exceptions import PartitionTopologyError -from .lifecycle import Candidate, DropAfter, LifecyclePolicy +from .lifecycle import Candidate, DetachMode, DropAfter, LifecyclePolicy from .plan import ( AttachPartition, CreatePartition, @@ -546,6 +546,21 @@ def _sequence_member(self, boundaries: RangeBoundaries, child: PartitionNode, bo ) return _Member(child, None, None, None, managed=False, claimed=readable) + def _detach_mode(self, node: PartitionNode) -> DetachMode: + """The mode a detach from ``node`` will really run in. + + PostgreSQL refuses ``DETACH … CONCURRENTLY`` outright while the parent + holds a DEFAULT partition, so AUTO there is the blocking form in + everything but name. Deciding that here rather than on the failure is + what lets ``plan --locks`` name the ``ACCESS EXCLUSIVE`` the operation + is really going to take, and spares the executor a statement that can + only fail. Every other reason to fall back is still discovered by + trying. + """ + if self.policy.detach is not DetachMode.AUTO: + return self.policy.detach + return DetachMode.BLOCKING if any(child.is_default for child in node.children) else DetachMode.AUTO + def _candidate(self, member: _Member, cursor_window: Window, boundaries: RangeBoundaries) -> Candidate: return Candidate( window=member.window, @@ -576,12 +591,15 @@ def _expire( return False detail = f"{boundaries.describe(member.window)} expired under '{self.policy.retention.describe()}'" + mode = self._detach_mode(node) + if mode is not self.policy.detach: + detail += "; the parent has a DEFAULT partition, so the detach is the blocking form" self.detaches.append( DetachPartition( target=member.node.name, oid=member.node.oid, parent_name=node.name, - mode=self.policy.detach, + mode=mode, bounds=member.node.bounds, reason=Reason.RETENTION_EXPIRED, detail=detail, diff --git a/tests/integration/aio/test_cli.py b/tests/integration/aio/test_cli.py index 40dd27d..1f3eabd 100644 --- a/tests/integration/aio/test_cli.py +++ b/tests/integration/aio/test_cli.py @@ -209,6 +209,60 @@ async def test__apply__allow_destructive__retires_what_retention_expired( assert payload["tables"][0]["result"]["detached_count"] == 1 +async def _with_a_default_full_of_history(db_engine: AsyncEngine, table: str) -> None: + """The state of an installation partitioned around data already in it.""" + await exec_sql(db_engine, f'CREATE TABLE "{table}_default" PARTITION OF "{table}" DEFAULT') + await exec_sql( + db_engine, + f'INSERT INTO "{table}" (created_at, payload) ' # noqa: S608 + "SELECT now() - (n * interval '20 days'), 'row' FROM generate_series(1, 9) AS n", + ) + + +async def _rows_left_in_default(db_engine: AsyncEngine, table: str) -> int: + async with db_engine.begin() as conn: + return (await conn.execute(text(f'SELECT count(*) FROM "{table}_default"'))).scalar_one() # noqa: S608 + + +async def test__backfill__rows_in_a_default_partition__are_moved_into_the_windows_they_belong_in( + db_engine: AsyncEngine, table: str, tmp_path: Path +) -> None: + # Arrange: creation walks forward from the cursor, so nothing else reaches these rows + await _with_a_default_full_of_history(db_engine, table) + config = _document(tmp_path, table, _dsn(db_engine)) + + # Act + code = await _run_cli("backfill", "-c", config) + + # Assert + assert code == ExitCode.OK + assert await _rows_left_in_default(db_engine, table) == 0 + async with db_engine.begin() as conn: + total = (await conn.execute(text(f'SELECT count(*) FROM "{table}"'))).scalar_one() # noqa: S608 + assert total == 9 + + +async def test__backfill__the_batch_budget__exits_drift_and_the_next_run_carries_on( + db_engine: AsyncEngine, table: str, tmp_path: Path +) -> None: + # Arrange: exit 2 is what lets a Job be run until it exits 0 + await _with_a_default_full_of_history(db_engine, table) + config = _document(tmp_path, table, _dsn(db_engine)) + + # Act + first = await _run_cli("backfill", "-c", config, "--max-batches", "1") + left_after_the_first = await _rows_left_in_default(db_engine, table) + while await _run_cli("backfill", "-c", config, "--max-batches", "1") == ExitCode.DRIFT: + pass + + # Assert: every row already moved stayed moved, and nothing was lost + assert first == ExitCode.DRIFT + assert 0 < left_after_the_first < 9 + assert await _rows_left_in_default(db_engine, table) == 0 + async with db_engine.begin() as conn: + assert (await conn.execute(text(f'SELECT count(*) FROM "{table}"'))).scalar_one() == 9 # noqa: S608 + + async def test__plan_save__then_apply__is_the_artifact_between_the_two( db_engine: AsyncEngine, table: str, tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: diff --git a/tests/integration/aio/test_lifecycle_policies.py b/tests/integration/aio/test_lifecycle_policies.py index 8180f39..140e405 100644 --- a/tests/integration/aio/test_lifecycle_policies.py +++ b/tests/integration/aio/test_lifecycle_policies.py @@ -404,10 +404,13 @@ async def test__detach_mode_concurrent__without_a_default_partition__detaches( assert result.dropped_count == 1 -async def test__detach_mode_auto__with_a_default_partition__falls_back_to_the_blocking_form( +async def test__detach_mode_auto__with_a_default_partition__is_planned_as_the_blocking_form( db_engine: AsyncEngine, table: str, caplog: pytest.LogCaptureFixture ) -> None: - # Arrange + # Arrange: the server refuses the concurrent form outright while a DEFAULT + # exists, and the planner reads the DEFAULT from the same catalog snapshot + # it plans from -- so the plan says blocking, and no statement is issued + # that could only fail. await _seed_june_and_default(db_engine, table, DetachMode.AUTO, default=True) config = monthly_config(table, lifecycle=_detach_config(table, DetachMode.AUTO)) @@ -415,11 +418,33 @@ async def test__detach_mode_auto__with_a_default_partition__falls_back_to_the_bl with caplog.at_level(logging.WARNING, logger="pg_partsmith.aio.repositories.remover"): result = await run_maintenance(db_engine, config, at_time="2026-08-01") - # Assert + # Assert: the outcome is what the fallback used to reach, without the failure assert result.issues == () assert result.detached_count == 1 assert result.dropped_count == 1 - assert any("falling back to non-concurrent DETACH" in record.getMessage() for record in caplog.records) + assert not any("falling back to non-concurrent DETACH" in record.getMessage() for record in caplog.records) + assert result.maintenance_plan is not None + detach = result.maintenance_plan.detaches[0] + assert detach.mode is DetachMode.BLOCKING + assert "DEFAULT partition" in detach.detail + assert "ACCESS EXCLUSIVE on the parent" in detach.capabilities.lock + + +async def test__detach_mode_auto__without_a_default_partition__stays_concurrent( + db_engine: AsyncEngine, table: str +) -> None: + # Arrange: nothing forces the blocking form here, so AUTO is what it says. + await _seed_june_and_default(db_engine, table, DetachMode.AUTO, default=False) + config = monthly_config(table, lifecycle=_detach_config(table, DetachMode.AUTO)) + + # Act + result = await run_maintenance(db_engine, config, at_time="2026-08-01") + + # Assert + assert result.issues == () + assert result.detached_count == 1 + assert result.maintenance_plan is not None + assert result.maintenance_plan.detaches[0].mode is DetachMode.AUTO # ── Policies over facts ───────────────────────────────────────────────────────── diff --git a/tests/integration/sync/test_lifecycle_policies.py b/tests/integration/sync/test_lifecycle_policies.py index 8243989..1fd8583 100644 --- a/tests/integration/sync/test_lifecycle_policies.py +++ b/tests/integration/sync/test_lifecycle_policies.py @@ -395,10 +395,13 @@ def test__detach_mode_concurrent__without_a_default_partition__detaches(sync_db_ assert result.dropped_count == 1 -def test__detach_mode_auto__with_a_default_partition__falls_back_to_the_blocking_form( +def test__detach_mode_auto__with_a_default_partition__is_planned_as_the_blocking_form( sync_db_engine: Engine, table: str, caplog: pytest.LogCaptureFixture ) -> None: - # Arrange + # Arrange: the server refuses the concurrent form outright while a DEFAULT + # exists, and the planner reads the DEFAULT from the same catalog snapshot + # it plans from -- so the plan says blocking, and no statement is issued + # that could only fail. _seed_june_and_default(sync_db_engine, table, DetachMode.AUTO, default=True) config = monthly_config(table, lifecycle=_detach_config(table, DetachMode.AUTO)) @@ -406,11 +409,30 @@ def test__detach_mode_auto__with_a_default_partition__falls_back_to_the_blocking with caplog.at_level(logging.WARNING, logger="pg_partsmith.sync.repositories.remover"): result = run_maintenance(sync_db_engine, config, at_time="2026-08-01") - # Assert + # Assert: the outcome is what the fallback used to reach, without the failure assert result.issues == () assert result.detached_count == 1 assert result.dropped_count == 1 - assert any("falling back to non-concurrent DETACH" in record.getMessage() for record in caplog.records) + assert not any("falling back to non-concurrent DETACH" in record.getMessage() for record in caplog.records) + assert result.maintenance_plan is not None + detach = result.maintenance_plan.detaches[0] + assert detach.mode is DetachMode.BLOCKING + assert "DEFAULT partition" in detach.detail + + +def test__detach_mode_auto__without_a_default_partition__stays_concurrent(sync_db_engine: Engine, table: str) -> None: + # Arrange: nothing forces the blocking form here, so AUTO is what it says. + _seed_june_and_default(sync_db_engine, table, DetachMode.AUTO, default=False) + config = monthly_config(table, lifecycle=_detach_config(table, DetachMode.AUTO)) + + # Act + result = run_maintenance(sync_db_engine, config, at_time="2026-08-01") + + # Assert + assert result.issues == () + assert result.detached_count == 1 + assert result.maintenance_plan is not None + assert result.maintenance_plan.detaches[0].mode is DetachMode.AUTO # ── Policies over facts ───────────────────────────────────────────────────────── diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 6e0d5e7..95e2314 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -21,7 +21,7 @@ from pg_partsmith.__version__ import __version__ from pg_partsmith.aio.command_hooks import CommandHooks from pg_partsmith.cli import ExitCode, main -from pg_partsmith.cli.commands import CommandResult, run_apply, run_plan, run_validate +from pg_partsmith.cli.commands import CommandResult, run_apply, run_backfill, run_plan, run_validate from pg_partsmith.cli.loader import ( DSN_ENV_VAR, DSN_FILE_ENV_VAR, @@ -36,7 +36,7 @@ from pg_partsmith.cli.main import _hooks as _cli_hooks from pg_partsmith.cli.render import describe_locks, envelope, plan_entry from pg_partsmith.document import PartitionsDocument -from pg_partsmith.entities import MaintenanceIssue, MaintenanceIssueStep, MaintenanceResult +from pg_partsmith.entities import MaintenanceIssue, MaintenanceIssueStep, MaintenanceResult, MigrationResult from pg_partsmith.events import HookPhase from pg_partsmith.exceptions import InvalidPartitionConfigError, LockAcquisitionError from pg_partsmith.hook_commands import CommandHookError @@ -404,6 +404,13 @@ def _applying_kit(result: MaintenanceResult | None = None) -> MagicMock: return kit +def _migrating_kit(result: MigrationResult | None = None) -> MagicMock: + kit = MagicMock() + answer = result if result is not None else MigrationResult(rows_moved=900, batches=1, partitions=("p1", "p2")) + kit.service.partition_data = AsyncMock(return_value=answer) + return kit + + def _destructive_plan() -> MaintenancePlan: return MaintenancePlan( table_name="public.events", @@ -496,6 +503,58 @@ async def test__apply__json__carries_the_plan_beside_the_result() -> None: assert entry["plan"]["operations"][0]["kind"] == "create" +async def test__backfill__a_drained_default__is_exit_ok_and_says_so() -> None: + # Arrange + kit = _migrating_kit() + + # Act + result = await run_backfill(kit, _configs()) + + # Assert + assert result.code is ExitCode.OK + assert kit.service.partition_data.await_args.kwargs == {"batch_rows": 10_000, "max_batches": None} + assert "moved 900 rows in 1 batches into 2 partitions; DEFAULT is drained" in result.render(output="human") + + +async def test__backfill__the_batch_budget_ran_out__is_exit_drift_so_a_job_can_repeat() -> None: + # Arrange: exit 2 is the contract that lets a Job be run until it exits 0, + # rather than having to parse the output to learn there is more left. + kit = _migrating_kit(MigrationResult(rows_moved=10_000, batches=1, complete=False)) + + # Act + result = await run_backfill(kit, _configs(), max_batches=1) + + # Assert + assert result.code is ExitCode.DRIFT + assert "run it again" in result.render(output="human") + assert kit.service.partition_data.await_args.kwargs["max_batches"] == 1 + + +async def test__backfill__a_window_it_could_not_move__outranks_the_drift() -> None: + # Arrange + issue = MaintenanceIssue(step=MaintenanceIssueStep.CREATE, partition_name="public.events_default", error="refused") + kit = _migrating_kit(MigrationResult(rows_moved=0, batches=0, complete=False, issues=(issue,))) + + # Act + result = await run_backfill(kit, _configs()) + + # Assert + assert result.code is ExitCode.FINDINGS + assert "refused" in result.render(output="human") + + +async def test__backfill__metrics__carry_the_rows_and_whether_anything_is_left() -> None: + # Arrange + kit = _migrating_kit(MigrationResult(rows_moved=250, batches=1, complete=False)) + + # Act + text = (await run_backfill(kit, _configs())).render(output="metrics") + + # Assert + assert 'pg_partsmith_backfilled_rows{table="public.events"} 250' in text + assert 'pg_partsmith_backfill_incomplete{table="public.events"} 1' in text + + def test__load_plans__a_file_plan_save_wrote__reads_back_as_the_plan(tmp_path: Path) -> None: # Arrange saved = envelope("plan", [plan_entry(_plan())]) @@ -550,6 +609,15 @@ def test__apply__a_document_that_runs_commands__is_refused_unless_asked_for(tmp_ assert code == ExitCode.CONFIG +def test__backfill__a_document_that_runs_commands__is_refused_unless_asked_for(tmp_path: Path) -> None: + # Arrange: backfill creates partitions through the same executor apply + # uses, so its hooks really do fire; the permission is the same one. + config = _hooks_document(tmp_path) + + # Act / Assert + assert main(["backfill", "-c", config]) == ExitCode.CONFIG + + def test__plan__a_document_that_runs_commands__needs_no_permission(tmp_path: Path) -> None: # Hooks fire during apply alone, so planning one is not running one. This # gets as far as connecting, which is a different failure entirely. diff --git a/tests/unit/test_planner.py b/tests/unit/test_planner.py index 7a9784d..6430c52 100644 --- a/tests/unit/test_planner.py +++ b/tests/unit/test_planner.py @@ -1366,6 +1366,35 @@ def test__plan_maintenance__detach_mode__taken_from_the_policy() -> None: assert plan.detaches[0].mode is DetachMode.BLOCKING +def test__plan_maintenance__detach_mode__auto_is_blocking_when_the_parent_holds_a_default() -> None: + # Arrange: PostgreSQL refuses a concurrent detach while a DEFAULT exists, + # so the plan says so rather than leaving it to the failure. + config = _config(lifecycle=_policy(retention=KeepNewest(count=1))) + root = _root(*_months(7, 8), PartitionNode(name=f"{ROOT}_default", parent_name=ROOT, bounds=DefaultBounds())) + + # Act + plan = _plan(config, root) + + # Assert + assert plan.detaches[0].mode is DetachMode.BLOCKING + assert "DEFAULT partition" in plan.detaches[0].detail + assert "ACCESS EXCLUSIVE on the parent" in plan.detaches[0].capabilities.lock + + +def test__plan_maintenance__detach_mode__concurrent_is_not_downgraded_by_a_default() -> None: + # Arrange: CONCURRENT means "fail when refused", which is a choice the + # planner has no business overruling. + config = _config(lifecycle=_policy(retention=KeepNewest(count=1), detach=DetachMode.CONCURRENT)) + root = _root(*_months(7, 8), PartitionNode(name=f"{ROOT}_default", parent_name=ROOT, bounds=DefaultBounds())) + + # Act + plan = _plan(config, root) + + # Assert + assert plan.detaches[0].mode is DetachMode.CONCURRENT + assert "DEFAULT partition" not in plan.detaches[0].detail + + def test__plan_maintenance__keep_for__expires_windows_over_for_longer_than_the_age() -> None: # Arrange: 90 days before the 28th of August is the 30th of May. config = _config(lifecycle=_policy(retention=KeepFor(age=timedelta(days=90))))