From 727f19303190f30babae89e1af49fa8a7ee1b217 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C2=A0Alex=20Shalaev?= Date: Fri, 4 Sep 2026 08:22:55 +0300 Subject: [PATCH 1/2] fix(planner): an AUTO detach under a DEFAULT partition is planned as blocking PostgreSQL refuses DETACH ... CONCURRENTLY while the parent holds a DEFAULT partition, so AUTO there was the blocking form discovered on the failure: one statement per partition that could only fail, and a WARNING whose reason never reached the operator. The planner reads the DEFAULT from the same catalog snapshot it plans from, so it decides this up front -- plan --locks now names the ACCESS EXCLUSIVE the run will really take, and the detail on the operation says why. CONCURRENT is untouched: being refused is the answer it asks for. --- docs/concepts/lifecycle.md | 9 ++++- pg_partsmith/plan.py | 2 +- pg_partsmith/planner.py | 22 +++++++++++-- .../aio/test_lifecycle_policies.py | 33 ++++++++++++++++--- .../sync/test_lifecycle_policies.py | 30 ++++++++++++++--- tests/unit/test_planner.py | 29 ++++++++++++++++ 6 files changed, 113 insertions(+), 12 deletions(-) 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/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_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_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)))) From 11ca77afc314a1064c76be7f950ebd3d3f0eb07f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C2=A0Alex=20Shalaev?= Date: Fri, 4 Sep 2026 08:32:58 +0300 Subject: [PATCH 2/2] feat(cli): backfill moves a DEFAULT partition's rows from the command line partition_data was library-only because a one-shot command had no progress story. --max-batches with exit 2 is one: the run stops after N statements per table, every row already moved stays moved, and a Job can be repeated until it exits 0. --output metrics carries pg_partsmith_backfilled_rows and pg_partsmith_backfill_incomplete for the same reason. This is the half of adoption creation cannot reach. A table partitioned around data already in it holds that data in DEFAULT, behind the cursor, where no create-ahead will ever arrive -- so for an installation that is not new, apply had nothing useful to do until somebody wrote Python. It no longer needs any. Hooks fire during backfill as they do during apply, because it creates partitions through the same executor: the --allow-hooks gate covers both rather than letting a declared after_create pass unnoticed. --- README.md | 2 +- docs/agents.md | 10 +++-- docs/guide/backfill.md | 9 ++++ docs/guide/cli.md | 52 +++++++++++++++++++--- docs/index.md | 4 +- pg_partsmith/cli/commands.py | 61 +++++++++++++++++++++++++- pg_partsmith/cli/main.py | 67 ++++++++++++++++++++++++++-- pg_partsmith/cli/metrics.py | 18 ++++++++ tests/integration/aio/test_cli.py | 54 +++++++++++++++++++++++ tests/unit/test_cli.py | 72 ++++++++++++++++++++++++++++++- 10 files changed, 329 insertions(+), 20 deletions(-) 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/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/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/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.