feat(cli): backfill from the command line, and a plan that names the blocking detach

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]

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:<version>` with the command

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;

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

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.

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

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 = (

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())