From caaa1312d279f1f830f2109ab73f7a558bd2abe5 Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Tue, 22 Sep 2026 17:25:36 -0400 Subject: [PATCH] feature: Add retention-based partition dropping Adds drop_partitions (and an opt-in maintain --drop-expired) to prune partitions outside a table's retention window. Retention is read from the settings comment as `retention:`: the current period plus the N before it are kept. Tables without a retention are never pruned, and a malformed value is ignored rather than widening what gets dropped. Partitions are selected by bounds (range ends at or before the cutoff), so legacy-named and irregular (e.g. year-resetting weekly) layouts work. DEFAULT, MINVALUE/MAXVALUE and sub-partitioned children are never dropped, and nothing is dropped with CASCADE. Dropping a partition takes ACCESS EXCLUSIVE on the parent, so each partition is dropped oldest-first in its own transaction under lock_timeout (default 5s, shared with --lock-timeout). A timeout stops the run, keeping retained coverage contiguous; DropPartitionsError reports what was already dropped and the rest is retried on the next run. In maintain, tables are extended before they are pruned so a failed drop never costs runway, and the JSONL records gain partitions.dropped. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 31 +- src/cli.ts | 2 + src/commands/drop-partitions.test.ts | 96 +++++ src/commands/drop-partitions.ts | 56 +++ src/commands/maintain.test.ts | 60 ++- src/commands/maintain.ts | 16 +- src/date-ranges.test.ts | 53 +++ src/date-ranges.ts | 28 ++ src/drop-partitions.test.ts | 584 +++++++++++++++++++++++++++ src/index.ts | 2 +- src/pgslice.ts | 167 +++++++- src/table-settings.test.ts | 18 + src/table-settings.ts | 14 +- src/table.ts | 32 +- src/types.ts | 29 ++ 15 files changed, 1170 insertions(+), 18 deletions(-) create mode 100644 src/commands/drop-partitions.test.ts create mode 100644 src/commands/drop-partitions.ts create mode 100644 src/drop-partitions.test.ts diff --git a/README.md b/README.md index 4640897..8f69872 100644 --- a/README.md +++ b/README.md @@ -344,9 +344,38 @@ If the partitioned parent owns a primary key (including a composite key), Postgr By default, pgslice re-issues the parent table's grants on each new partition, because Postgres does not cascade a parent's grants to its partitions. This keeps a replication/CDC role's access intact as new partitions appear. Disable it with `--no-inherit-grants`. +## Dropping Expired Partitions + +Give a table a retention by adding `retention:` to its settings comment. The current period plus the `N` periods before it are kept; older partitions are expired. + +```sql +COMMENT ON TABLE IS 'column:created_at,period:month,cast:date,version:3,retention:3'; +``` + +Preview, then drop, the expired partitions: + +```sh +pgslice drop_partitions
--dry-run +pgslice drop_partitions
+``` + +Or prune every managed table that has a retention as part of `maintain` (off by default): + +```sh +pgslice maintain --drop-expired +``` + +Partitions are selected **by bounds**, not by name: a partition is expired when its range ends at or before the start of the period `N` periods ago. `DEFAULT` and `MINVALUE`/`MAXVALUE` partitions, and sub-partitioned children, are never dropped. Tables without a `retention` are never pruned, and an invalid value (anything but a positive integer) is ignored. + +Dropping a partition takes an `ACCESS EXCLUSIVE` lock on the parent, so queries against the table queue behind a drop that is waiting for its lock. To keep that short, each partition is dropped oldest-first in its own transaction under `--lock-timeout` (default: `5s`). If the lock isn't acquired in time, the run stops without dropping anything further — leaving the retained partitions contiguous — and the rest is retried on the next run. Nothing is dropped with `CASCADE`, so a partition that a view or other object depends on fails loudly instead. + +### Logical replication + +`DROP TABLE` is not decoded by logical replication: no `DELETE`s are emitted, replication slots are unaffected, and changes made to a partition before it was dropped are still delivered. Downstream copies built from logical replication therefore keep the dropped rows. If a downstream system is your archive, confirm it has the data before a partition expires, and note that re-syncing it from the source later won't bring dropped data back. + ## Archiving Partitions -Back up and drop older partitions each day, month, or year. +To keep a copy outside the database, back up older partitions before they expire. ```sh pg_dump -c -Fc -t
_202409 $PGSLICE_URL >
_202409.dump diff --git a/src/cli.ts b/src/cli.ts index 9f0e2a5..f5fcfa8 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -4,6 +4,7 @@ import { AddPartitionsCommand } from "./commands/add-partitions.js"; import { AnalyzeCommand } from "./commands/analyze.js"; import { DisableMirroringCommand } from "./commands/disable-mirroring.js"; import { DisableRetiredMirroringCommand } from "./commands/disable-retired-mirroring.js"; +import { DropPartitionsCommand } from "./commands/drop-partitions.js"; import { EnableMirroringCommand } from "./commands/enable-mirroring.js"; import { EnableRetiredMirroringCommand } from "./commands/enable-retired-mirroring.js"; import { FillCommand } from "./commands/fill.js"; @@ -28,6 +29,7 @@ export function createCli(): Cli { cli.register(AnalyzeCommand); cli.register(DisableMirroringCommand); cli.register(DisableRetiredMirroringCommand); + cli.register(DropPartitionsCommand); cli.register(EnableMirroringCommand); cli.register(EnableRetiredMirroringCommand); cli.register(FillCommand); diff --git a/src/commands/drop-partitions.test.ts b/src/commands/drop-partitions.test.ts new file mode 100644 index 0000000..41813f5 --- /dev/null +++ b/src/commands/drop-partitions.test.ts @@ -0,0 +1,96 @@ +import { describe, expect } from "vitest"; +import { sql, type DatabaseTransactionConnection } from "slonik"; + +import { commandTest as test } from "../testing/index.js"; +import { DropPartitionsCommand } from "./drop-partitions.js"; + +describe("DropPartitionsCommand", () => { + test.scoped({ commandClass: ({}, use) => use(DropPartitionsCommand) }); + + async function createPosts( + transaction: DatabaseTransactionConnection, + comment: string, + ) { + await transaction.query(sql.unsafe` + CREATE TABLE posts ( + id bigint NOT NULL, + created_at timestamp without time zone NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at) + `); + await transaction.query( + sql.unsafe`COMMENT ON TABLE posts IS ${sql.literalValue(comment)}`, + ); + await transaction.query(sql.unsafe` + CREATE TABLE posts_202001 PARTITION OF posts + FOR VALUES FROM ('2020-01-01') TO ('2020-02-01') + `); + } + + test("--dry-run lists expired partitions without dropping them", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts( + transaction, + "column:created_at,period:month,cast:date,version:3,retention:3", + ); + + const exitCode = await cli.run( + ["drop_partitions", "posts", "--dry-run"], + commandContext, + ); + + expect(exitCode).toBe(0); + expect(commandContext.stdout.read()?.toString()).toBe( + "posts: would drop 1 partition(s): posts_202001\n", + ); + const exists = await transaction.oneFirst( + sql.unsafe`SELECT to_regclass('posts_202001') IS NOT NULL`, + ); + expect(exists).toBe(true); + }); + + test("drops expired partitions", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts( + transaction, + "column:created_at,period:month,cast:date,version:3,retention:3", + ); + + const exitCode = await cli.run( + ["drop_partitions", "posts"], + commandContext, + ); + + expect(exitCode).toBe(0); + expect(commandContext.stdout.read()?.toString()).toBe( + "posts: dropped 1 partition(s): posts_202001\n", + ); + }); + + test("fails without a retention setting", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts( + transaction, + "column:created_at,period:month,cast:date,version:3", + ); + + const exitCode = await cli.run( + ["drop_partitions", "posts"], + commandContext, + ); + + expect(exitCode).toBe(1); + expect(commandContext.stderr.read()?.toString()).toContain( + "No retention configured: public.posts", + ); + }); +}); diff --git a/src/commands/drop-partitions.ts b/src/commands/drop-partitions.ts new file mode 100644 index 0000000..35d30bf --- /dev/null +++ b/src/commands/drop-partitions.ts @@ -0,0 +1,56 @@ +import { Command, Option } from "clipanion"; + +import { BaseCommand } from "./base.js"; +import { Pgslice } from "../pgslice.js"; + +export class DropPartitionsCommand extends BaseCommand { + static override paths = [["drop_partitions"]]; + + static override usage = Command.Usage({ + description: "Drop a partitioned table's expired partitions", + details: ` + Drops the partitions that fall entirely outside the table's retention + window. Retention is read from the table's pgslice settings comment as + \`retention:\`: the current period plus the N periods before it are + kept. Tables without a retention setting are never pruned. + + DEFAULT and MINVALUE/MAXVALUE partitions are never dropped, and nothing is + dropped with CASCADE. Each partition is dropped oldest-first in its own + short transaction; if a drop cannot acquire the parent's lock within + --lock-timeout, the command stops and the rest is retried on the next run. + `, + examples: [ + [ + "Preview which partitions would be dropped", + "$0 drop_partitions posts --dry-run", + ], + ["Drop expired partitions", "$0 drop_partitions posts"], + ], + }); + + table = Option.String({ required: true, name: "table" }); + dryRun = Option.Boolean("--dry-run", false, { + description: "List the expired partitions without dropping them", + }); + lockTimeout = Option.String("--lock-timeout", "5s", { + description: + "How long each DROP waits on the parent's lock before backing off (default: 5s)", + }); + + override async perform(pgslice: Pgslice): Promise { + const dropped = await pgslice.start(async (connection) => + pgslice.dropPartitions(connection, { + table: this.table, + dryRun: this.dryRun, + lockTimeout: this.lockTimeout, + }), + ); + + const verb = this.dryRun ? "would drop" : "dropped"; + this.context.stdout.write( + dropped.length > 0 + ? `${this.table}: ${verb} ${dropped.length} partition(s): ${dropped.join(", ")}\n` + : `${this.table}: no expired partitions\n`, + ); + } +} diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts index 5ebd94e..15b57a6 100644 --- a/src/commands/maintain.test.ts +++ b/src/commands/maintain.test.ts @@ -10,7 +10,8 @@ interface LogEntry { level: string; target: { db: string; host: string; schema?: string; table?: string }; future?: { daily: number; weekly: number; monthly: number; yearly: number }; - partitions?: { new: number; total: number }; + dropExpired?: boolean; + partitions?: { new: number; dropped: number; total: number }; success?: number; succeeded?: { count: number; tables: string[] }; failed?: { count: number; tables: string[] }; @@ -90,6 +91,7 @@ describe("MaintainCommand", () => { expect(table?.target.schema).toBe("public"); expect(table?.partitions).toEqual({ new: expect.any(Number), + dropped: 0, total: expect.any(Number), }); @@ -180,10 +182,64 @@ describe("MaintainCommand", () => { expect(table?.msg).toBe("Table already up to date; no extension needed"); expect(table?.level).toBe("info"); expect(table?.success).toBe(1); - expect(table?.partitions).toEqual({ new: 0, total: 1 }); + expect(table?.partitions).toEqual({ new: 0, dropped: 0, total: 1 }); expect(table?.target.host).toBe("localhost"); }); + async function createExpiredPosts( + transaction: DatabaseTransactionConnection, + ) { + await createPosts(transaction); + await transaction.query(sql.unsafe` + COMMENT ON TABLE posts IS 'column:created_at,period:month,cast:date,version:3,retention:1' + `); + await transaction.query(sql.unsafe` + CREATE TABLE posts_y2020m01 PARTITION OF posts + FOR VALUES FROM ('2020-01-01') TO ('2020-02-01') + `); + } + + test("does not drop expired partitions by default", async ({ + cli, + commandContext, + transaction, + }) => { + await createExpiredPosts(transaction); + + expect(await cli.run(["maintain"], commandContext)).toBe(0); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + expect(logs[0].dropExpired).toBe(false); + expect( + logs.find((entry) => entry.target.table === "posts")?.partitions?.dropped, + ).toBe(0); + }); + + test("drops expired partitions with --drop-expired", async ({ + cli, + commandContext, + transaction, + }) => { + await createExpiredPosts(transaction); + + expect(await cli.run(["maintain", "--drop-expired"], commandContext)).toBe( + 0, + ); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + expect(logs[0].dropExpired).toBe(true); + // This suite runs on the real clock, so how many months fall outside the + // one-month window varies; the 2020 partition is always among them. + expect( + logs.find((entry) => entry.target.table === "posts")?.partitions?.dropped, + ).toBeGreaterThanOrEqual(1); + expect( + await transaction.oneFirst( + sql.unsafe`SELECT to_regclass('posts_y2020m01') IS NULL`, + ), + ).toBe(true); + }); + test("reads per-period horizons from PGSLICE_FUTURE_* env vars", async ({ cli, commandContext, diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts index 5f5b5dc..a37c230 100644 --- a/src/commands/maintain.ts +++ b/src/commands/maintain.ts @@ -21,6 +21,11 @@ export class MaintainCommand extends BaseCommand { extending each table this checks that every leaf partition has a replica identity usable for logical replication, and exits non-zero if any does not. + + With --drop-expired, each table whose settings comment carries a + \`retention:\` also has its expired partitions dropped (see + drop_partitions) after it is extended. Tables without a retention setting + are never pruned. `, examples: [ [ @@ -36,6 +41,10 @@ export class MaintainCommand extends BaseCommand { "PGSLICE_FUTURE_MONTHLY=12 $0 maintain", ], ["Restrict to a single schema", "$0 maintain --schema analytics"], + [ + "Also drop partitions outside each table's retention window", + "$0 maintain --drop-expired", + ], ], }); @@ -75,7 +84,11 @@ export class MaintainCommand extends BaseCommand { }); lockTimeout = Option.String("--lock-timeout", "5s", { description: - "How long each partition-creation statement waits on a table's lock before backing off (default: 5s)", + "How long each partition create or drop waits on a table's lock before backing off (default: 5s)", + }); + dropExpired = Option.Boolean("--drop-expired", false, { + description: + "Also drop partitions outside each table's configured retention (default: false)", }); override async perform(pgslice: Pgslice): Promise { @@ -103,6 +116,7 @@ export class MaintainCommand extends BaseCommand { tablespace: this.tablespace, inheritGrants: this.inheritGrants, lockTimeout: this.lockTimeout, + dropExpired: this.dropExpired, }, log, ), diff --git a/src/date-ranges.test.ts b/src/date-ranges.test.ts index 77d846a..62470be 100644 --- a/src/date-ranges.test.ts +++ b/src/date-ranges.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it } from "vitest"; import { + isExpired, + retentionCutoff, DateRanges, advanceDate, extendRanges, @@ -689,3 +691,54 @@ describe("extendRangesBackward", () => { } }); }); + +describe("retentionCutoff", () => { + const now = new Date(Date.UTC(2026, 0, 15, 13, 30)); + + it("keeps the current period plus N prior periods", () => { + expect(retentionCutoff(now, "month", 3).toISOString()).toBe( + "2025-10-01T00:00:00.000Z", + ); + expect(retentionCutoff(now, "day", 90).toISOString()).toBe( + "2025-10-17T00:00:00.000Z", + ); + expect(retentionCutoff(now, "year", 1).toISOString()).toBe( + "2025-01-01T00:00:00.000Z", + ); + }); + + it("anchors weeks on the ISO Monday", () => { + // 2026-01-15 is in the ISO week starting Monday 2026-01-12. + expect(retentionCutoff(now, "week", 2).toISOString()).toBe( + "2025-12-29T00:00:00.000Z", + ); + }); +}); + +describe("isExpired", () => { + const cutoff = new Date(Date.UTC(2025, 11, 1)); + const range = (from: string, to: string): ExistingRange => ({ + lower: new Date(from), + upper: new Date(to), + lowerUnbounded: false, + upperUnbounded: false, + isDefault: false, + }); + + it("is true only when the range ends at or before the cutoff", () => { + expect(isExpired(range("2025-10-01", "2025-11-01"), cutoff)).toBe(true); + expect(isExpired(range("2025-11-01", "2025-12-01"), cutoff)).toBe(true); + expect(isExpired(range("2025-11-15", "2025-12-15"), cutoff)).toBe(false); + expect(isExpired(range("2025-12-01", "2026-01-01"), cutoff)).toBe(false); + }); + + it("never treats DEFAULT or unbounded partitions as expired", () => { + const parsedDefault = parseRangeBound("DEFAULT"); + expect(parsedDefault && isExpired(parsedDefault, cutoff)).toBe(false); + + const minvalue = parseRangeBound( + "FOR VALUES FROM (MINVALUE) TO ('2020-01-01')", + ); + expect(minvalue && isExpired(minvalue, cutoff)).toBe(false); + }); +}); diff --git a/src/date-ranges.ts b/src/date-ranges.ts index 047f5ff..77aa494 100644 --- a/src/date-ranges.ts +++ b/src/date-ranges.ts @@ -622,3 +622,31 @@ export function* extendRangesBackward(options: { end = start; } } + +/** + * The retention cutoff for a table: the start of the period `retention` + * periods before the one containing `now`. A partition whose range ends at or + * before this instant holds only data older than the retained window. + */ +export function retentionCutoff( + now: Date, + period: Period, + retention: number, +): Date { + return advanceDate(roundDate(now, period), period, -retention); +} + +/** + * Whether an existing partition lies entirely before `cutoff`. Only finite, + * bounded ranges qualify: DEFAULT and `MINVALUE`/`MAXVALUE` partitions are + * catch-alls with no finite end, so they are never considered expired. + */ +export function isExpired(range: ExistingRange, cutoff: Date): boolean { + return ( + !range.isDefault && + !range.lowerUnbounded && + !range.upperUnbounded && + range.upper !== null && + range.upper.getTime() <= cutoff.getTime() + ); +} diff --git a/src/drop-partitions.test.ts b/src/drop-partitions.test.ts new file mode 100644 index 0000000..dca2bc9 --- /dev/null +++ b/src/drop-partitions.test.ts @@ -0,0 +1,584 @@ +import { afterEach, beforeEach, describe, expect, vi } from "vitest"; +import { sql, type DatabaseTransactionConnection } from "slonik"; +import { z } from "zod"; + +import { pgsliceTest as test } from "./testing/index.js"; +import { DropPartitionsError } from "./pgslice.js"; +import { Table } from "./table.js"; +import { + addChild, + addDefault, + addMinvalueChild, + childNames, + ident, + nativeParent, + pgsliceParent, + TS, + TSTZ, +} from "./testing/shapes.js"; +import type { Cast, Period } from "./types.js"; + +/** Writes a settings comment that carries a retention. */ +async function setRetention( + tx: DatabaseTransactionConnection, + qualified: string, + column: string, + period: Period, + cast: Cast, + retention: string, +): Promise { + await tx.query( + sql.unsafe`COMMENT ON TABLE ${ident(qualified)} IS ${sql.literalValue( + `column:${column},period:${period},cast:${cast},version:3,retention:${retention}`, + )}`, + ); +} + +/** A native monthly table with partitions for 2025-09 .. 2026-02. */ +async function monthlyPosts( + tx: DatabaseTransactionConnection, + retention = "2", +): Promise { + await nativeParent(tx, "posts", "created_at", TS, "month", "date"); + await setRetention(tx, "posts", "created_at", "month", "date", retention); + const months = [ + ["202509", "2025-09-01", "2025-10-01"], + ["202510", "2025-10-01", "2025-11-01"], + ["202511", "2025-11-01", "2025-12-01"], + ["202512", "2025-12-01", "2026-01-01"], + ["202601", "2026-01-01", "2026-02-01"], + ["202602", "2026-02-01", "2026-03-01"], + ]; + for (const [suffix, from, to] of months) { + await addChild(tx, "posts", `posts_${suffix}`, from, to); + } +} + +describe("Pgslice.dropPartitions", () => { + beforeEach(() => { + vi.useFakeTimers(); + // 2026-01-15 is a Thursday; its ISO week is 2026-W03 (Monday 2026-01-12). + vi.setSystemTime(new Date(Date.UTC(2026, 0, 15))); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test("drops partitions older than the retention window, oldest first", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + await transaction.query( + sql.unsafe`INSERT INTO posts VALUES (1, '2025-09-10', 'old'), (2, '2026-01-10', 'new')`, + ); + + // retention:2 keeps January (current) plus November and December. + const dropped = await pgslice.dropPartitions(transaction, { + table: "posts", + }); + + expect(dropped).toEqual(["posts_202509", "posts_202510"]); + expect(await childNames(transaction, "posts")).toEqual([ + "posts_202511", + "posts_202512", + "posts_202601", + "posts_202602", + ]); + const { count } = await transaction.one( + sql.type(z.object({ count: z.coerce.number() }))` + SELECT count(*) FROM posts + `, + ); + expect(Number(count)).toBe(1); + }); + + test("is idempotent", async ({ pgslice, transaction }) => { + await monthlyPosts(transaction); + + await pgslice.dropPartitions(transaction, { table: "posts" }); + const second = await pgslice.dropPartitions(transaction, { + table: "posts", + }); + + expect(second).toEqual([]); + }); + + test("dryRun reports the expired partitions without dropping them", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + + const wouldDrop = await pgslice.dropPartitions(transaction, { + table: "posts", + dryRun: true, + }); + + expect(wouldDrop).toEqual(["posts_202509", "posts_202510"]); + expect(await childNames(transaction, "posts")).toHaveLength(6); + }); + + test("never drops the current or future partitions, even with retention:1", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction, "1"); + + await pgslice.dropPartitions(transaction, { table: "posts" }); + + expect(await childNames(transaction, "posts")).toEqual([ + "posts_202512", + "posts_202601", + "posts_202602", + ]); + }); + + test("refuses a table without a retention setting", async ({ + pgslice, + transaction, + }) => { + await nativeParent(transaction, "posts", "created_at", TS, "month", "date"); + await addChild( + transaction, + "posts", + "posts_202001", + "2020-01-01", + "2020-02-01", + ); + + await expect( + pgslice.dropPartitions(transaction, { table: "posts" }), + ).rejects.toThrow("No retention configured: public.posts"); + expect(await childNames(transaction, "posts")).toEqual(["posts_202001"]); + }); + + test("never drops DEFAULT or MINVALUE partitions", async ({ + pgslice, + transaction, + }) => { + await nativeParent(transaction, "posts", "created_at", TS, "month", "date"); + await setRetention( + transaction, + "posts", + "created_at", + "month", + "date", + "1", + ); + await addMinvalueChild( + transaction, + "posts", + "posts_historic", + "2025-01-01", + ); + await addChild( + transaction, + "posts", + "posts_202501", + "2025-01-01", + "2025-02-01", + ); + await addChild( + transaction, + "posts", + "posts_202601", + "2026-01-01", + "2026-02-01", + ); + await addDefault(transaction, "posts", "posts_default"); + + const dropped = await pgslice.dropPartitions(transaction, { + table: "posts", + }); + + expect(dropped).toEqual(["posts_202501"]); + expect(await childNames(transaction, "posts")).toEqual([ + "posts_202601", + "posts_default", + "posts_historic", + ]); + }); + + test("never drops a sub-partitioned child", async ({ + pgslice, + transaction, + }) => { + await nativeParent(transaction, "posts", "created_at", TS, "month", "date"); + await setRetention( + transaction, + "posts", + "created_at", + "month", + "date", + "1", + ); + await transaction.query(sql.unsafe` + CREATE TABLE posts_202501 PARTITION OF posts + FOR VALUES FROM ('2025-01-01') TO ('2025-02-01') + PARTITION BY RANGE (created_at) + `); + + const dropped = await pgslice.dropPartitions(transaction, { + table: "posts", + }); + + expect(dropped).toEqual([]); + expect(await childNames(transaction, "posts")).toEqual(["posts_202501"]); + }); + + test("selects by bounds, not names, for legacy-named partitions", async ({ + pgslice, + transaction, + }) => { + // A year-resetting weekly scheme: a one-day stub closes out 2025. + await nativeParent( + transaction, + "activities", + "occurred_at", + TS, + "week", + "date", + ); + await setRetention( + transaction, + "activities", + "occurred_at", + "week", + "date", + "2", + ); + const weeks = [ + ["activities_p2025w52", "2025-12-24", "2025-12-31"], + ["activities_p2025w53", "2025-12-31", "2026-01-01"], + ["activities_p2026w01", "2026-01-01", "2026-01-08"], + ["activities_p2026w02", "2026-01-08", "2026-01-15"], + ["activities_p2026w03", "2026-01-15", "2026-01-22"], + ]; + for (const [name, from, to] of weeks) { + await addChild(transaction, "activities", name, from, to); + } + + // The current ISO week starts 2026-01-12, so retention:2 cuts off at + // 2025-12-29: only partitions ending on or before then are dropped. + const dropped = await pgslice.dropPartitions(transaction, { + table: "activities", + }); + + expect(dropped).toEqual([]); + + vi.setSystemTime(new Date(Date.UTC(2026, 0, 22))); + // Cutoff 2026-01-05: the 2025w52 week and the w53 stub both end before it. + expect( + await pgslice.dropPartitions(transaction, { table: "activities" }), + ).toEqual(["activities_p2025w52", "activities_p2025w53"]); + }); + + test("compares timestamptz bounds in UTC regardless of the session time zone", async ({ + pgslice, + transaction, + }) => { + await nativeParent( + transaction, + "events", + "created_at", + TSTZ, + "month", + "timestamptz", + ); + await setRetention( + transaction, + "events", + "created_at", + "month", + "timestamptz", + "1", + ); + await transaction.query(sql.unsafe`SET LOCAL TIME ZONE 'UTC'`); + await addChild( + transaction, + "events", + "events_202511", + "2025-11-01", + "2025-12-01", + ); + await addChild( + transaction, + "events", + "events_202512", + "2025-12-01", + "2026-01-01", + ); + // Under a +14:00 session the bounds render shifted (e.g. November's upper + // bound reads as 2025-12-01 14:00), so without pinning UTC the partition + // ending exactly at the cutoff would be missed. + await transaction.query( + sql.unsafe`SET LOCAL TIME ZONE 'Pacific/Kiritimati'`, + ); + + const dropped = await pgslice.dropPartitions(transaction, { + table: "events", + }); + + expect(dropped).toEqual(["events_202511"]); + }); + + test("drops classic pgslice partitions and those in another schema", async ({ + pgslice, + transaction, + }) => { + await pgsliceParent( + transaction, + "app.visits", + "created_at", + TS, + "month", + "date", + ); + await setRetention( + transaction, + "app.visits", + "created_at", + "month", + "date", + "1", + ); + await addChild( + transaction, + "app.visits", + "visits_202510", + "2025-10-01", + "2025-11-01", + ["id"], + ); + await transaction.query(sql.unsafe`CREATE SCHEMA IF NOT EXISTS archive`); + await transaction.query(sql.unsafe` + CREATE TABLE archive.visits_202511 PARTITION OF app.visits + FOR VALUES FROM ('2025-11-01') TO ('2025-12-01') + `); + + const dropped = await pgslice.dropPartitions(transaction, { + table: "app.visits", + }); + + expect(dropped).toEqual(["visits_202510", "visits_202511"]); + const leftover = await transaction.one( + sql.type(z.object({ count: z.coerce.number() }))` + SELECT count(*) FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname IN ('app', 'archive') AND c.relname LIKE 'visits_2025%' + `, + ); + expect(Number(leftover.count)).toBe(0); + }); + + test("keeps every surviving leaf CDC-safe", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + + await pgslice.dropPartitions(transaction, { table: "posts" }); + + expect( + await new Table("public", "posts").unsafeReplicaIdentityPartitions( + transaction, + ), + ).toEqual([]); + }); + + test("does not cascade: a dependent object stops the run and survives", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + await transaction.query( + sql.unsafe`CREATE VIEW october_posts AS SELECT * FROM posts_202510`, + ); + + const error = await pgslice + .dropPartitions(transaction, { table: "posts" }) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(DropPartitionsError); + expect(error instanceof DropPartitionsError && error.dropped).toEqual([ + "posts_202509", + ]); + expect(String(error)).toContain("public.posts_202510"); + expect(await childNames(transaction, "posts")).toContain("posts_202510"); + const view = await transaction.maybeOne( + sql.type(z.object({ exists: z.boolean() }))` + SELECT to_regclass('october_posts') IS NOT NULL AS exists + `, + ); + expect(view?.exists).toBe(true); + }); + + test("defaults to a 5s lock_timeout for each drop", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + + await pgslice.dropPartitions(transaction, { table: "posts" }); + + // Each drop runs in a savepoint whose released SET LOCAL propagates to this + // transaction, so reading it here proves the DROP ran under the timeout. + const result = await transaction.one( + sql.type(z.object({ lock_timeout: z.string() }))`SHOW lock_timeout`, + ); + expect(result.lock_timeout).toBe("5s"); + }); + + test("honors a custom lockTimeout", async ({ pgslice, transaction }) => { + await monthlyPosts(transaction); + + await pgslice.dropPartitions(transaction, { + table: "posts", + lockTimeout: "750ms", + }); + + const result = await transaction.one( + sql.type(z.object({ lock_timeout: z.string() }))`SHOW lock_timeout`, + ); + expect(result.lock_timeout).toBe("750ms"); + }); +}); + +describe("Pgslice.maintain with dropExpired", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date(Date.UTC(2026, 0, 15))); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + const horizons = { + futureDaily: 1, + futureWeekly: 1, + futureMonthly: 2, + futureYearly: 1, + }; + + test("does not drop anything unless dropExpired is set", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + + const [result] = await pgslice.maintain(transaction, horizons); + + expect(result.partitionsDropped).toEqual([]); + expect(await childNames(transaction, "posts")).toContain("posts_202509"); + }); + + test("extends and prunes tables with a retention, leaving others alone", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + await nativeParent( + transaction, + "comments", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "comments", + "comments_202001", + "2020-01-01", + "2026-02-01", + ); + + const results = await pgslice.maintain(transaction, { + ...horizons, + dropExpired: true, + }); + + const posts = results.find((r) => r.table === "public.posts"); + expect(posts?.error).toBeNull(); + expect(posts?.partitionsCreated).toEqual(["posts_202603"]); + expect(posts?.partitionsDropped).toEqual(["posts_202509", "posts_202510"]); + expect(posts?.replicaIdentityReady).toBe(true); + + const comments = results.find((r) => r.table === "public.comments"); + expect(comments?.error).toBeNull(); + expect(comments?.partitionsDropped).toEqual([]); + expect(await childNames(transaction, "comments")).toContain( + "comments_202001", + ); + }); + + test("records a failed drop without losing the extension or the fleet", async ({ + pgslice, + transaction, + }) => { + await monthlyPosts(transaction); + await transaction.query( + sql.unsafe`CREATE VIEW october_posts AS SELECT * FROM posts_202510`, + ); + await nativeParent( + transaction, + "reactions", + "created_at", + TS, + "month", + "date", + ); + await setRetention( + transaction, + "reactions", + "created_at", + "month", + "date", + "1", + ); + await addChild( + transaction, + "reactions", + "reactions_202511", + "2025-11-01", + "2025-12-01", + ); + await addChild( + transaction, + "reactions", + "reactions_202601", + "2026-01-01", + "2026-02-01", + ); + + const logs: Array> = []; + const results = await pgslice.maintain( + transaction, + { ...horizons, dropExpired: true }, + (entry) => logs.push(entry), + ); + + const posts = results.find((r) => r.table === "public.posts"); + expect(posts?.error).toContain("public.posts_202510"); + expect(posts?.partitionsCreated).toEqual(["posts_202603"]); + expect(posts?.partitionsDropped).toEqual(["posts_202509"]); + + const reactions = results.find((r) => r.table === "public.reactions"); + expect(reactions?.error).toBeNull(); + expect(reactions?.partitionsDropped).toEqual(["reactions_202511"]); + + const postsLog = logs.find( + (entry) => + typeof entry.target === "object" && + entry.target !== null && + "table" in entry.target && + entry.target.table === "posts", + ); + expect(postsLog?.level).toBe("error"); + expect(postsLog?.partitions).toEqual({ new: 1, dropped: 1, total: 0 }); + expect(postsLog?.droppedPartitions).toEqual(["posts_202509"]); + }); +}); diff --git a/src/index.ts b/src/index.ts index ab77bba..7177a67 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,5 @@ export { createCli } from "./cli.js"; -export { Pgslice } from "./pgslice.js"; +export { Pgslice, DropPartitionsError } from "./pgslice.js"; export { Table } from "./table.js"; export { Filler } from "./filler.js"; export { AdvisoryLockError } from "./advisory-lock.js"; diff --git a/src/pgslice.ts b/src/pgslice.ts index 2139d4e..f97065f 100644 --- a/src/pgslice.ts +++ b/src/pgslice.ts @@ -15,6 +15,7 @@ import type { AnalyzeOptions, ColumnInfo, DisableMirroringOptions, + DropPartitionsOptions, EnableMirroringOptions, FillBatchResult, FillOptions, @@ -40,10 +41,12 @@ import { advanceDate, extendRanges, extendRangesBackward, + isExpired, isUtcMidnight, maxUpperBound, minLowerBound, rangeOverlaps, + retentionCutoff, roundDate, type DateRange, } from "./date-ranges.js"; @@ -69,6 +72,21 @@ const GRANTABLE_PRIVILEGES: Record> = { TRIGGER: sql.fragment`TRIGGER`, }; +/** + * Raised when dropping an expired partition fails part-way through. Earlier + * partitions were each dropped in their own committed transaction, so + * `dropped` records what was already removed before the failure. + */ +export class DropPartitionsError extends Error { + constructor( + message: string, + readonly dropped: string[], + ) { + super(message); + this.name = "DropPartitionsError"; + } +} + interface PgsliceOptions { /** * Whether to use Postgres advisory locks to prevent concurrent operations @@ -514,6 +532,105 @@ export class Pgslice { ); } + /** + * Drops a table's expired partitions: leaf partitions whose range ends at or + * before the retention cutoff (the start of the period `retention` periods + * before the current one, per the table's settings comment). A table without + * a `retention` setting is never pruned. + * + * Only finite, bounded leaf partitions qualify; DEFAULT and + * `MINVALUE`/`MAXVALUE` partitions are never dropped, and nothing is dropped + * with CASCADE, so a partition that another object depends on fails loudly + * instead of taking the dependent with it. + * + * Dropping a partition takes an `ACCESS EXCLUSIVE` lock on the parent, so + * each partition is dropped in its own short transaction under + * `lock_timeout`, oldest first. On a lock timeout the run stops (leaving + * retained coverage contiguous) and a {@link DropPartitionsError} reports + * what was already dropped; the remainder is retried on the next run. + * + * Returns the names of the partitions dropped, or with `dryRun` the names + * that would be. + */ + async dropPartitions( + connection: DatabasePoolConnection, + options: DropPartitionsOptions, + ): Promise { + const table = Table.parse(options.table); + const now = options.now ?? new Date(); + + const { cutoff, expired } = await connection.transaction(async (tx) => { + // Bounds are compared as UTC instants; see rangePartitionBounds. + await tx.query(sql.typeAlias("void")`SET LOCAL TIME ZONE 'UTC'`); + + if (!(await table.exists(tx))) { + throw new Error(`Table not found: ${table.toString()}`); + } + + const settings = await table.fetchSettings(tx); + if (!settings) { + throw new Error(`No settings found: ${table.toString()}`); + } + if (settings.retention === undefined) { + throw new Error(`No retention configured: ${table.toString()}`); + } + + const cutoff = retentionCutoff(now, settings.period, settings.retention); + const expired = (await table.rangePartitionBounds(tx)) + .filter((range) => range.leaf && isExpired(range, cutoff)) + .sort((a, b) => (a.upper?.getTime() ?? 0) - (b.upper?.getTime() ?? 0)); + + return { cutoff, expired }; + }); + + if (options.dryRun) { + return expired.map((range) => range.name); + } + + const dropped: string[] = []; + + for (const range of expired) { + try { + await connection.transaction(async (tx) => + this.#withLock(tx, table, "drop_partitions", async () => { + await tx.query(sql.typeAlias("void")`SET LOCAL TIME ZONE 'UTC'`); + await tx.query( + sql.typeAlias( + "void", + )`SET LOCAL lock_timeout = ${sql.literalValue(options.lockTimeout ?? "5s")}`, + ); + + // Re-read inside this transaction so a partition that was detached, + // re-attached elsewhere, or replaced since the scan is left alone. + const current = (await table.rangePartitionBounds(tx)).find( + (r) => r.schema === range.schema && r.name === range.name, + ); + if (!current?.leaf || !isExpired(current, cutoff)) { + return; + } + + await tx.query( + sql.typeAlias("void")` + DROP TABLE ${sql.identifier([range.schema, range.name])} + `, + ); + dropped.push(range.name); + }), + ); + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + const suffix = + dropped.length > 0 ? ` (already dropped: ${dropped.join(", ")})` : ""; + throw new DropPartitionsError( + `Failed to drop ${range.schema}.${range.name}: ${reason}${suffix}`, + dropped, + ); + } + } + + return dropped; + } + /** * Maintains every managed partitioned table the connection can see: each * partitioned parent (`relkind = 'p'`) carrying a valid pgslice settings @@ -530,6 +647,10 @@ export class Pgslice { * table this verifies every leaf has a usable replica identity, surfacing a * misconfigured table rather than silently shipping a CDC-unsafe partition. * + * With `dropExpired`, each table whose settings comment carries a + * `retention` also has its expired partitions dropped via + * {@link dropPartitions}, after it is extended. + * * Each table is maintained independently: a failure on one table (for example * a non-empty DEFAULT partition that blocks creating the next one) is recorded * on that table's result and does not stop the rest of the fleet. @@ -579,6 +700,7 @@ export class Pgslice { monthly: futureByPeriod.month, yearly: futureByPeriod.year, }, + dropExpired: options.dropExpired ?? false, }); const managed = await this.#discoverManagedTables( @@ -590,19 +712,37 @@ export class Pgslice { const succeeded: string[] = []; const failed: string[] = []; - for (const { table, period } of managed) { + for (const { table, settings } of managed) { let result: MaintainResult; + let partitionsCreated: string[] = []; + let partitionsDropped: string[] = []; try { - const partitionsCreated = await this.addPartitions(connection, { + partitionsCreated = await this.addPartitions(connection, { table: table.toString(), past, - future: futureByPeriod[period], + future: futureByPeriod[settings.period], tablespace: options.tablespace, inheritGrants: options.inheritGrants, lockTimeout: options.lockTimeout, now, }); + // Extend before pruning, so a failed drop never costs the table runway. + if (options.dropExpired && settings.retention !== undefined) { + try { + partitionsDropped = await this.dropPartitions(connection, { + table: table.toString(), + lockTimeout: options.lockTimeout, + now, + }); + } catch (error) { + if (error instanceof DropPartitionsError) { + partitionsDropped = error.dropped; + } + throw error; + } + } + const { model, partitionCount, unsafePartitions } = await connection.transaction(async (tx) => { const parentPrimaryKey = await table.primaryKeyColumns(tx); @@ -621,6 +761,7 @@ export class Pgslice { table: table.toString(), model, partitionsCreated, + partitionsDropped, partitionCount, replicaIdentityReady: unsafePartitions.length === 0, unsafePartitions, @@ -630,7 +771,8 @@ export class Pgslice { result = { table: table.toString(), model: null, - partitionsCreated: [], + partitionsCreated, + partitionsDropped, partitionCount: 0, replicaIdentityReady: false, unsafePartitions: [], @@ -650,10 +792,15 @@ export class Pgslice { msg = result.error; } else if (!result.replicaIdentityReady) { msg = `Partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`; - } else if (result.partitionsCreated.length === 0) { + } else if ( + result.partitionsCreated.length === 0 && + result.partitionsDropped.length === 0 + ) { msg = "Table already up to date; no extension needed"; - } else { + } else if (result.partitionsCreated.length > 0) { msg = "Extended table successfully"; + } else { + msg = "Dropped expired partitions successfully"; } log({ @@ -663,8 +810,12 @@ export class Pgslice { target: { ...target, schema: table.schema, table: table.name }, partitions: { new: result.partitionsCreated.length, + dropped: result.partitionsDropped.length, total: result.partitionCount, }, + // Dropping is irreversible, so name each dropped partition in the log + // rather than only counting them. + droppedPartitions: result.partitionsDropped, success: ok ? 1 : 0, }); } @@ -694,7 +845,7 @@ export class Pgslice { async #discoverManagedTables( connection: DatabasePoolConnection, schema?: string, - ): Promise> { + ): Promise> { const schemaFilter = schema ? sql.fragment`AND n.nspname = ${schema}` : sql.fragment``; @@ -729,7 +880,7 @@ export class Pgslice { } const settings = TableSettings.parseFromComment(row.comment); return settings - ? [{ table: new Table(row.schema, row.name), period: settings.period }] + ? [{ table: new Table(row.schema, row.name), settings }] : []; }); } diff --git a/src/table-settings.test.ts b/src/table-settings.test.ts index b26fa15..875d6c5 100644 --- a/src/table-settings.test.ts +++ b/src/table-settings.test.ts @@ -49,6 +49,24 @@ describe("TableSettings", () => { expect(settings?.format).toBeUndefined(); }); + it("parses an optional retention", () => { + const settings = TableSettings.parseFromComment( + "column:created_at,period:month,cast:date,version:3,retention:3", + ); + expect(settings?.retention).toBe(3); + }); + + it.each(["", "0", "-1", "1.5", "3m", " 3", "abc"])( + "ignores an invalid retention %j rather than pruning", + (value) => { + const settings = TableSettings.parseFromComment( + `column:created_at,period:month,cast:date,retention:${value}`, + ); + expect(settings).not.toBeNull(); + expect(settings?.retention).toBeUndefined(); + }, + ); + describe("periods", () => { it("parses day period", () => { const settings = TableSettings.parseFromComment( diff --git a/src/table-settings.ts b/src/table-settings.ts index 5eed67b..d86502f 100644 --- a/src/table-settings.ts +++ b/src/table-settings.ts @@ -15,6 +15,13 @@ export class TableSettings { * keep its existing naming convention; see {@link formatDateSuffix}. */ readonly format?: string, + /** + * Optional number of past periods to retain before the current one. When + * set, partitions whose range ends at or before the start of the period + * `retention` periods ago are eligible for {@link Pgslice.dropPartitions}. + * Absent means the table is never pruned. + */ + readonly retention?: number, ) {} /** @@ -27,6 +34,7 @@ export class TableSettings { let period: Period | undefined; let cast: Cast | undefined; let format: string | undefined; + let retention: number | undefined; for (const part of parts) { // Split on the first ":" only, so a value that itself contains a colon @@ -45,11 +53,15 @@ export class TableSettings { // Stored verbatim; validated when rendered/parsed (compileFormat), so a // mistyped template fails loudly at add_partitions time rather than here. format = value; + } else if (key === "retention" && /^[1-9]\d*$/.test(value)) { + // Only a positive integer is accepted. Anything else leaves retention + // unset, so a mistyped value can never widen what gets dropped. + retention = Number(value); } } if (column && period && cast) { - return new TableSettings(column, period, cast, format); + return new TableSettings(column, period, cast, format, retention); } return null; diff --git a/src/table.ts b/src/table.ts index 5f16d3b..d264e2c 100644 --- a/src/table.ts +++ b/src/table.ts @@ -109,6 +109,16 @@ function derivePartitionTimeFilter( }; } +/** + * An existing partition's RANGE bounds, together with whether it is a leaf + * (a plain table) rather than a sub-partitioned child. + */ +export interface PartitionBound extends ExistingRange { + schema: string; + name: string; + leaf: boolean; +} + export interface PartitionContext { settings: TableSettings | null; partitions: Table[]; @@ -475,16 +485,28 @@ export class Table { * rendered instant as UTC, so under a non-UTC session every bound would shift * by the session offset. `addPartitions` pins UTC before calling this. */ - async rangePartitionBounds(tx: CommonQueryMethods): Promise { + async rangePartitionBounds( + tx: CommonQueryMethods, + ): Promise { const rows = await tx.any( - sql.type(z.object({ name: z.string(), bound: z.string() }))` + sql.type( + z.object({ + schema: z.string(), + name: z.string(), + bound: z.string(), + leaf: z.boolean(), + }), + )` SELECT + nmsp_child.nspname AS schema, child.relname AS name, - pg_get_expr(child.relpartbound, child.oid) AS bound + pg_get_expr(child.relpartbound, child.oid) AS bound, + child.relkind = 'r' AS leaf FROM pg_inherits JOIN pg_class parent ON pg_inherits.inhparent = parent.oid JOIN pg_class child ON pg_inherits.inhrelid = child.oid JOIN pg_namespace nmsp_parent ON nmsp_parent.oid = parent.relnamespace + JOIN pg_namespace nmsp_child ON nmsp_child.oid = child.relnamespace WHERE nmsp_parent.nspname = ${this.schema} AND parent.relname = ${this.name} @@ -493,7 +515,9 @@ export class Table { return rows.flatMap((r) => { const parsed = parseRangeBound(r.bound); - return parsed ? [{ ...parsed, name: r.name }] : []; + return parsed + ? [{ ...parsed, schema: r.schema, name: r.name, leaf: r.leaf }] + : []; }); } diff --git a/src/types.ts b/src/types.ts index 4f2bf0d..2b6f188 100644 --- a/src/types.ts +++ b/src/types.ts @@ -86,6 +86,27 @@ export interface AddPartitionsOptions { lockTimeout?: string; } +/** + * Options for the `drop_partitions` command. + */ +export interface DropPartitionsOptions { + table: string; + /** + * The reference "now" for computing the retention cutoff. Defaults to the + * current time. + */ + now?: Date; + /** + * How long each `DROP TABLE` may wait on the parent's lock before backing + * off. Dropping a partition takes an `ACCESS EXCLUSIVE` lock on the parent, + * so every query against the table queues behind a waiting drop; keep this + * short. Any Postgres interval literal (e.g. "5s"). Defaults to "5s". + */ + lockTimeout?: string; + /** When true, report the expired partitions without dropping any. */ + dryRun?: boolean; +} + import type { MirroringTargetType } from "./mirroring.js"; // Re-export for convenience @@ -268,6 +289,12 @@ export interface MaintainOptions { * Postgres interval literal (e.g. "5s"). Defaults to "5s". */ lockTimeout?: string; + /** + * When true, also drop each table's expired partitions (see + * {@link DropPartitionsOptions}) after extending it. Only tables whose + * settings comment carries a `retention` are pruned. Defaults to false. + */ + dropExpired?: boolean; /** * The reference "now" for choosing each table's partition horizon. Defaults * to a single instant captured when the run starts, so a fleet run that @@ -308,6 +335,8 @@ export interface MaintainResult { /** The partitioning model, or null if maintenance failed before it was determined. */ model: PartitionModel | null; partitionsCreated: string[]; + /** Expired partitions dropped this run (empty unless `dropExpired`). */ + partitionsDropped: string[]; partitionCount: number; /** * Whether every leaf partition has a replica identity usable for logical