pgcolumnar.expire drops fully expired row groups (#403 item 5a)

CHANGELOG.md

@@ -16,6 +16,67 @@ true until the next version shipped.
 
 ## [Unreleased]
 
+### Added
+
+- `pgcolumnar.expire` drops row groups whose rows are all older than a declared
+  retention, without reading or rewriting them (#403 item 5a). Declare the
+  retention with `pgcolumnar.set_options(..., ttl_column => 'ts',
+  ttl_interval => '90 days')`; both halves are needed, and either alone means no
+  retention.
+
+  This is the tractable half of the paper's "merge-time data transformation".
+  The rewrites already retire whole row groups: `pgcolumnar.compact` drops every
+  group that is fully deleted. Retention is the same operation with a different
+  predicate, and the zone map already holds what decides it, so the decision is a
+  catalog read. Nothing is decoded and nothing is rewritten, and it holds only
+  `ShareUpdateExclusiveLock`.
+
+  **It is called by name and never runs on its own.** It deletes rows, and an
+  operation a user runs for maintenance must not do that silently, so it is not
+  wired into `VACUUM`, `compact` or autovacuum. A table with no declared
+  retention raises an error rather than reporting that it did nothing.
+
+  A group is kept whole or dropped whole. A group holding rows on both sides of
+  the cutoff is kept, so retention is approximate at the group boundary and errs
+  toward keeping data. Measured on 5,000 rows in 5 groups with 1,440 rows past a
+  three-day retention: one group dropped, 1,000 rows removed, and all 3,560 rows
+  still inside the retention kept, including the 440 expired rows sharing the
+  straddling group.
+
+  The retention column must be `timestamp` or `timestamptz`.
+
+- `pgcolumnar.parallel_copy` can refuse a load it has already taken, with
+  `dedup => true` (#403 item 7). A load that commits, whose acknowledgement the
+  client never receives, is retried and the rows go in twice: measured, the same
+  file loaded twice gives 100,000 rows and then 200,000.
+
+  With `dedup`, the SHA-256 of the file is recorded in the new
+  `pgcolumnar.load_fingerprint` catalog after a successful load. A later load of
+  the same contents into the same table stores nothing, returns 0, and raises a
+  `NOTICE` rather than failing. A file whose contents changed is a different load
+  and is stored, even at the same path.
+
+  It is off by default, because discarding rows a caller asked to store is not
+  ordinary `INSERT` behavior.
+
+  The unit is the whole load rather than a "part". `parallel_copy` is atomic
+  through 2PC, proved by a malformed row at line 50,001 leaving 0 rows and 0
+  prepared transactions, so parts never commit independently and a part hash
+  would deduplicate nothing that is not already all-or-nothing.
+
+  The check runs after every worker has prepared and before anything commits,
+  which is the only point at which a repeat can be refused without charging every
+  load for it. A refused load therefore does its work and discards it. The
+  alternative, hashing before dispatch, costs 42% of a 264 MiB load; as
+  implemented the fingerprint has no measurable cost, because the coordinator
+  computes it while the loaders are already reading the same file.
+
+  Three limits, all stated in `docs/sql-reference.md`: the fingerprint is
+  recorded after the data commits, so a crash between them leaves data that a
+  retry will store again; two identical loads running at once both store, because
+  each checks the record before either writes it; and a refused load still reads
+  and parses the file.
+
 ### Fixed
 
 - The cost model reads the row-group geometry a table was **written** with,

docs/sql-reference.md

@@ -33,9 +33,18 @@ append in insertion order, so re-run `vacuum_sorted` to re-establish it, like
 PostgreSQL `CLUSTER`. Column names must exist and cannot be virtual generated
 columns.
 
+`ttl_column name` and `ttl_interval interval` declare a retention. They are read
+only by [`pgcolumnar.expire`](#pgcolumnarexpiretablename-regclass-returns-bigint),
+which you run yourself. Declaring a retention does not delete anything on its
+own. Both are needed: either one alone means no retention.
+
 ```sql
 SELECT pgcolumnar.set_options('events', sort_by => ARRAY['customer_id','ts']);
 SELECT pgcolumnar.reset_options('events', sort_by => true);   -- clear it
+
+-- declare a retention; nothing is deleted until you call expire
+SELECT pgcolumnar.set_options('events', ttl_column => 'ts',
+                              ttl_interval => '90 days');
 ```
 
 ### pgcolumnar.get_storage_id(rel regclass) returns bigint
@@ -156,6 +165,34 @@ Returns the number of groups retired.
 SELECT pgcolumnar.compact('events');
 ```
 
+### pgcolumnar.expire(tablename regclass) returns bigint
+
+Drops row groups whose rows are all older than the retention declared by
+`set_options`. Returns the number of groups dropped. Holds only
+`ShareUpdateExclusiveLock`, so it runs against a live table.
+
+**This deletes rows.** It runs only when you call it. No other operation applies
+a retention, and `VACUUM` never does.
+
+It reads no data. A row group records the maximum value of each of its columns.
+A group whose maximum is older than the cutoff holds no row still inside the
+retention. Its metadata is dropped without decoding anything.
+
+A group is kept whole or dropped whole. A group holding rows on both sides of the
+cutoff is kept, and its expired rows stay until every row in that group has
+expired. Retention is therefore approximate at the group boundary, and it errs
+toward keeping data. A smaller `stripe_row_limit` narrows the boundary.
+
+The retention column must be `timestamp` or `timestamptz`. The table must have
+both `ttl_column` and `ttl_interval` declared, or the function raises an error
+rather than reporting that it did nothing.
+
+```sql
+SELECT pgcolumnar.set_options('events', ttl_column => 'ts',
+                              ttl_interval => '90 days');
+SELECT pgcolumnar.expire('events');    -- returns groups dropped
+```
+
 ### pgcolumnar.compact_rewrite(tablename regclass, min_deleted_fraction float8 DEFAULT 0.2, max_groups int DEFAULT 0) returns bigint
 
 Rewrites partially-deleted row groups, those whose deleted fraction is at least
@@ -457,7 +494,7 @@ SELECT pgcolumnar.import_parquet('events_copy', '/tmp/events.parquet');
 SELECT pgcolumnar.import_parquet('events_copy', '/data/events/');
 ```
 
-### pgcolumnar.parallel_copy(target regclass, filename text, workers int DEFAULT NULL) returns bigint
+### pgcolumnar.parallel_copy(target regclass, filename text, workers int DEFAULT NULL, dedup boolean DEFAULT false) returns bigint
 
 Loads a text file into a columnar table with several background workers at once,
 as one atomic operation. Returns the number of rows loaded. The caller needs
@@ -486,6 +523,42 @@ When `workers` is omitted the function derives a value from the target. For a
 partitioned target it lowers `workers` to the partition count when the count is
 smaller.
 
+#### Refusing a load this table has already taken
+
+A load that commits, whose acknowledgement the client never receives, is retried,
+and the rows go in twice. Pass `dedup => true` to refuse the repeat.
+
+With `dedup`, the function records the SHA-256 of the file after a successful
+load, in `pgcolumnar.load_fingerprint`. A later load of a file with the same
+contents into the same table stores nothing, returns 0, and raises a `NOTICE`
+saying why. It does not fail. A file whose contents changed is a different load
+and is stored, even at the same path.
+
+`dedup` is off by default. Discarding rows a caller asked to store is not
+ordinary `INSERT` behavior, so it happens only when asked for, and only on this
+function.
+
+Three limits apply.
+
+The fingerprint is recorded after the data commits. A crash between the two
+leaves the data stored and unrecorded, so a later retry stores it again. That is
+the behavior without `dedup` and is the safe direction.
+
+Two identical loads running at the same time both store their rows. Each checks
+the record before either writes it. `dedup` refuses a load that follows a
+completed one; it does not serialize concurrent loads.
+
+A refused load still does the work. The rows are read, parsed and written, and
+then discarded without being made visible, because the check happens after every
+worker has prepared. The alternative costs every load a serial pass over the file
+before any worker starts. Measured on a 264 MiB file, that pass takes 42% of the
+load, while the fingerprint as implemented has no measurable cost.
+
+```sql
+-- refuse a repeat of a load this table has already taken
+SELECT pgcolumnar.parallel_copy('events', '/data/events.txt', 8, true);
+```
+
 ```sql
 -- single columnar table, any row order
 CREATE TABLE events (id bigint, ts timestamptz, val double precision) USING pgcolumnar;

pgcolumnar--1.0-alpha2--1.0-alpha3.sql

@@ -9,9 +9,216 @@
  */
 \echo Use "ALTER EXTENSION pgcolumnar UPDATE" to load this file. \quit
 
--- sort_status gains an OUT parameter (#761), which changes its signature, so it
--- cannot be a CREATE OR REPLACE.
+-- New catalog for pgcolumnar.parallel_copy's opt-in load dedup (#403 item 7).
+-- Loads pgcolumnar.parallel_copy has already performed, for its opt-in dedup
+-- (#403 item 7). One row per (table, file fingerprint) that committed.
+--
+-- The fingerprint is the SHA-256 of the loaded file's bytes, so a file that
+-- changed at the same path is a different load. Path, size and mtime would all
+-- call that the same file.
+--
+-- The row is written AFTER the data commits, never before. A crash between the
+-- two leaves data with no fingerprint, so a retry loads again, which is the
+-- behaviour without this feature and is the safe direction. The reverse order
+-- would leave a fingerprint with no data and refuse rows that were never stored.
+CREATE TABLE pgcolumnar.load_fingerprint (
+	relation_oid oid NOT NULL,
+	fingerprint bytea NOT NULL,       -- SHA-256 of the file's bytes
+	rows bigint NOT NULL,
+	loaded_at timestamptz NOT NULL DEFAULT now()
+);
+-- NOT unique, deliberately. The lookup is an existence test, and a unique
+-- index would turn the one case that can produce a second row into an ERROR
+-- raised AFTER the data committed: two concurrent loads of the same file both
+-- check before either records, both commit, and the loser's record insert would
+-- fail, reporting failure for a load that succeeded. A duplicate record is
+-- harmless; a false failure is not.
+CREATE INDEX load_fingerprint_idx
+	ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint);
+
+-- Declared retention for pgcolumnar.expire (#403 item 5a). Nothing drops rows on
+-- its own; expire is called by name.
+ALTER TABLE pgcolumnar.options ADD COLUMN IF NOT EXISTS ttl_column name;
+ALTER TABLE pgcolumnar.options ADD COLUMN IF NOT EXISTS ttl_interval interval;
+
+-- Three functions gain parameters, which changes their signatures, so none can
+-- be a CREATE OR REPLACE: sort_status (#761), parallel_copy (#403 item 7) and
+-- set_options (#403 item 5a).
 DROP FUNCTION IF EXISTS pgcolumnar.sort_status(regclass);
+DROP FUNCTION IF EXISTS pgcolumnar.parallel_copy(regclass, text, int);
+DROP FUNCTION IF EXISTS pgcolumnar.set_options(regclass, int, int, name, int, name, name[]);
+
+CREATE FUNCTION pgcolumnar.set_options(
+	table_name regclass,
+	chunk_group_row_limit int DEFAULT NULL,
+	stripe_row_limit int DEFAULT NULL,
+	compression name DEFAULT NULL,
+	compression_level int DEFAULT NULL,
+	encode_effort name DEFAULT NULL,
+	sort_by name[] DEFAULT NULL,
+	ttl_column name DEFAULT NULL,
+	ttl_interval interval DEFAULT NULL)
+	RETURNS void
+	LANGUAGE plpgsql
+	AS $set_options$
+DECLARE
+	col name;
+BEGIN
+	/*
+	 * The options are per-relation and are read by the columnar writer, so a row
+	 * recorded for a relation that is not columnar can never be used. Storing one
+	 * is not merely useless: the drop hook that clears pgcolumnar.options fires
+	 * only for columnar relations, so the row outlives the table and is left
+	 * keyed to a dangling oid that a later relation reusing that oid inherits.
+	 * Measured before this guard, on the same cluster: set_options on a heap
+	 * table stored a row, DROP TABLE left it behind, and regclass then rendered
+	 * as the bare oid; the identical sequence on a columnar table cleaned up.
+	 *
+	 * Rejecting is safe for the one workflow that could want the other order:
+	 * ALTER TABLE ... SET ACCESS METHOD pgcolumnar keeps the relation's oid
+	 * (measured), so options set after the conversion apply to the same relation
+	 * a caller would have been trying to name before it.
+	 *
+	 * The ERRCODE is explicit. plpgsql's RAISE EXCEPTION defaults to P0001, and
+	 * the C paths raise this same sentence with ERRCODE_WRONG_OBJECT_TYPE
+	 * (42809). Without it the identical message carried two different SQLSTATEs
+	 * depending on which path refused the caller, in a tree whose own privilege
+	 * suites deliberately assert SQLSTATE rather than message text.
+	 *
+	 * relkind is part of the test, and it is what makes the guard match the
+	 * cleanup rather than merely look strict. The drop hook returns before it
+	 * examines the access method for anything that is not an ordinary table
+	 * (columnar_tableam.c: `if (get_rel_relkind(objectId) != RELKIND_RELATION)
+	 * return;`), so 'r' is exactly the set of relations whose options row can
+	 * ever be cleaned up. From PG17 a PARTITIONED table may carry an access
+	 * method, so `relam = pgcolumnar` alone admits a parent that has no storage,
+	 * that the writer never writes, and whose row the hook will never clear.
+	 * Measured on 17.6 with the amname-only test: accepted, one row recorded,
+	 * and the row still there after DROP TABLE keyed to the dropped oid, while
+	 * an ordinary columnar table in the same run cleaned up. PG16 and earlier
+	 * cannot reach it -- they refuse `PARTITION BY ... USING pgcolumnar`
+	 * outright, checked on 16.14 -- so this is PG17, 18 and 19.
+	 */
+	IF NOT EXISTS (SELECT 1 FROM pg_class c
+					 JOIN pg_am a ON a.oid = c.relam
+					WHERE c.oid = table_name
+					  AND a.amname = 'pgcolumnar'
+					  AND c.relkind = 'r') THEN
+		RAISE EXCEPTION 'relation "%" is not a columnar table', table_name
+			USING ERRCODE = 'wrong_object_type',
+				HINT = 'Per-table options are read by the columnar writer and '
+				'apply only to an ordinary table using the pgcolumnar access '
+				'method. A partitioned table has no storage of its own: set the '
+				'options on each partition. Otherwise convert the table first '
+				'with ALTER TABLE ... SET ACCESS METHOD pgcolumnar, then set '
+				'the options.';
+	END IF;
+
+	IF encode_effort IS NOT NULL AND
+	   encode_effort NOT IN ('full', 'fast') THEN
+		RAISE EXCEPTION 'unknown columnar encode_effort "%"', encode_effort
+			USING HINT = 'Valid values are "full" and "fast".';
+	END IF;
+
+	IF compression IS NOT NULL AND
+	   compression NOT IN ('none', 'pglz', 'lz4', 'zstd') THEN
+		RAISE EXCEPTION 'unknown columnar compression "%"', compression;
+	END IF;
+
+	/*
+	 * Bound the integer limits to the same valid ranges as the instance-wide
+	 * GUCs (pgcolumnar.chunk_group_row_limit, pgcolumnar.stripe_row_limit,
+	 * pgcolumnar.compression_level). A per-table value outside these ranges is
+	 * rejected here rather than stored: a limit of zero or below would produce
+	 * a stripe whose recorded chunk_row_count is zero and make the row-number
+	 * arithmetic (chunk id = offset / chunk_row_count) divide by zero on
+	 * delete, update, and index fetch.
+	 */
+	IF chunk_group_row_limit IS NOT NULL AND chunk_group_row_limit < 100 THEN
+		RAISE EXCEPTION 'chunk_group_row_limit must be at least 100';
+	END IF;
+	IF stripe_row_limit IS NOT NULL AND stripe_row_limit < 1000 THEN
+		RAISE EXCEPTION 'stripe_row_limit must be at least 1000';
+	END IF;
+	IF compression_level IS NOT NULL AND
+	   (compression_level < 1 OR compression_level > 22) THEN
+		RAISE EXCEPTION 'compression_level must be between 1 and 22';
+	END IF;
+
+	/*
+	 * sort_by declares the physical sort key applied by vacuum_sorted() with no
+	 * explicit columns (#288). This is a cheap early check only: each named
+	 * column must exist, not be dropped, and not be a VIRTUAL generated column
+	 * (its value is not stored, so it cannot be sorted on). Orderability
+	 * (a default btree ordering operator) is NOT checked here -- the C apply
+	 * path is authoritative and re-resolves and re-validates the names every
+	 * run, because a column can be dropped or altered after it is declared.
+	 * attgenerated is '' or 's' before PG18; 'v' only exists from PG18, so the
+	 * "<> 'v'" test is correct and inert on older majors.
+	 */
+	IF sort_by IS NOT NULL THEN
+		FOREACH col IN ARRAY sort_by LOOP
+			IF NOT EXISTS (SELECT 1 FROM pg_attribute a
+						   WHERE a.attrelid = table_name
+							 AND a.attname = col
+							 AND a.attnum > 0
+							 AND NOT a.attisdropped
+							 AND a.attgenerated <> 'v') THEN
+				RAISE EXCEPTION 'column "%" cannot be used in sort_by for table %',
+					col, table_name
+					USING HINT = 'The column must exist, must not be dropped, '
+						'and must not be a VIRTUAL generated column.';
+			END IF;
+		END LOOP;
+	END IF;
+
+	INSERT INTO pgcolumnar.options AS o
+		(regclass, chunk_group_row_limit, stripe_row_limit,
+		 compression, compression_level, encode_effort, sort_by,
+		 ttl_column, ttl_interval)
+	VALUES (table_name, chunk_group_row_limit, stripe_row_limit,
+			compression, compression_level, encode_effort, sort_by,
+			ttl_column, ttl_interval)
+	ON CONFLICT (regclass) DO UPDATE SET
+		chunk_group_row_limit =
+			COALESCE(EXCLUDED.chunk_group_row_limit, o.chunk_group_row_limit),
+		stripe_row_limit =
+			COALESCE(EXCLUDED.stripe_row_limit, o.stripe_row_limit),
+		compression =
+			COALESCE(EXCLUDED.compression, o.compression),
+		compression_level =
+			COALESCE(EXCLUDED.compression_level, o.compression_level),
+		encode_effort =
+			COALESCE(EXCLUDED.encode_effort, o.encode_effort),
+		sort_by =
+			COALESCE(EXCLUDED.sort_by, o.sort_by),
+		ttl_column =
+			COALESCE(EXCLUDED.ttl_column, o.ttl_column),
+		ttl_interval =
+			COALESCE(EXCLUDED.ttl_interval, o.ttl_interval);
+END;
+$set_options$;
+
+COMMENT ON FUNCTION pgcolumnar.set_options(regclass, int, int, name, int, name, name[], name, interval)
+	IS 'set per-table columnar options; NULL leaves a value unchanged. sort_by declares the physical sort key applied by vacuum_sorted() with no explicit columns (#288); it is NOT auto-maintained -- rows inserted after a sort append in insert order, so re-run vacuum_sorted() to re-establish it, like PostgreSQL CLUSTER';
+
+CREATE FUNCTION pgcolumnar.expire(tablename regclass)
+	RETURNS bigint
+	LANGUAGE C STRICT
+	AS 'MODULE_PATHNAME', 'pgcolumnar_expire';
+
+COMMENT ON FUNCTION pgcolumnar.expire(regclass)
+	IS 'drop row groups whose rows are all older than the retention declared by set_options(ttl_column, ttl_interval), without reading or rewriting them (#403)';
+
+CREATE FUNCTION pgcolumnar.parallel_copy(target regclass, filename text,
+										 workers int DEFAULT NULL,
+										 dedup boolean DEFAULT false)
+	RETURNS bigint
+	LANGUAGE C
+	AS 'MODULE_PATHNAME', 'pgcolumnar_parallel_copy';
+
+COMMENT ON FUNCTION pgcolumnar.parallel_copy(regclass, text, int, boolean)
+	IS 'atomic parallel bulk load of a COPY text file into a columnar table using background workers: a single columnar table (any row order), or a RANGE-partitioned columnar table sorted by the partition key with one distinct partition set per worker (#300). With dedup, a file already loaded into this table is refused rather than loaded twice (#403)';
 
 CREATE FUNCTION pgcolumnar.sort_status(
 	rel regclass,