From 2f9ffb1e4bccd417645e87064326da8c5c0a3519 Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Fri, 28 Aug 2026 11:49:44 -0600 Subject: [PATCH 1/4] feat: parallel_copy can refuse a load it has already taken (#403 item 7) A load that commits, whose acknowledgement the client never receives, is retried and the rows go in twice. Measured before anything was written: the same file loaded twice into the same table gives 100,000 rows and then 200,000. pgcolumnar.parallel_copy(target, filename, workers, dedup boolean DEFAULT false) records the SHA-256 of the file in the new pgcolumnar.load_fingerprint catalog after a successful load, and refuses a later load of the same contents into the same table: nothing stored, 0 returned, and a NOTICE saying why rather than a silent 0 or an error. Off by default. Discarding rows a caller asked to store is not ordinary INSERT behaviour, so it happens only when asked for and only on this function. ## The unit is the whole load, not a "part" The paper this comes from keeps hashes of the last N inserted PARTS, because there each part commits on its own. parallel_copy is atomic through 2PC. Proved rather than assumed, with a malformed integer at line 50,001: 0 rows after, 0 prepared transactions left. Parts never commit independently here, so a part hash would deduplicate nothing that is not already all-or-nothing. My first attempt to prove that used pgcolumnar.sink_fail_after and the load succeeded anyway, because the injection GUC never reaches the background workers. That would have been a clean-looking pass proving nothing. ## Where the check goes, and what it costs After every loader has PREPAREd and before anything commits. That is the only point at which a repeat can be refused without charging every load for it, and 2PC is what makes it possible: the work can be thrown away after it is done. A refused load therefore does its work and discards it. The alternative is hashing before dispatch, which costs 42% of the load on a 264 MiB file (1.13 s across four workers against 0.48 s single-threaded SHA-256). As implemented the coordinator hashes while the loaders are already reading the same file: min-of-8, 1.09 s without dedup and 1.08 s with, each rep asserting 5,000,000 rows returned. No measurable cost. ## The order of the two writes is the safety argument Data commits first; the fingerprint is recorded after. A crash between them leaves data with no fingerprint, so a retry loads again -- the behaviour without this feature, and the safe direction. The reverse would leave a fingerprint with no data and refuse rows that were never stored. ## A defect my own removal proof found With the check bypassed, the record insert hit "duplicate key value violates unique constraint" AFTER the data had committed, reporting failure for a load that succeeded. Two concurrent loads of one file reach that state without any mutation: both check before either records. The index is therefore NOT unique -- the lookup is an existence test, a duplicate record is harmless, and a false failure is not. Concurrent identical loads are not serialized, and both store. That is the behaviour without dedup and is the safe direction; what this refuses is the case it was built for, a completed load retried afterwards. Serializing the concurrent case needs a lock held across the check, the COMMIT PREPAREDs and the record, and COMMIT PREPARED cannot run inside a transaction block. Stated in the code and in docs/sql-reference.md rather than left to be discovered. ## Gate test/parallel_copy_dedup.sh, 14 checks, registered in run_all_versions.sh. Its two premises measure the defect itself, so the suite fails if the double-insert ever stops happening for some unrelated reason. Removal proof: with the dedup check forced false, the mutation asserted applied, the repeat stores 40,000 rows and three checks go red. native_upgrade_converge passes, which is what proves the new catalog and the changed function signature reach an upgraded catalog identically to a fresh 1.0-alpha3 install, in definition, ACL and comment. parallel_copy, entry_point_privilege, harness_selftest, docs_style and eager_ordering_record pass on PG 18.4. Built with COPT=-Werror, 0 warnings. Refs #403. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UaQ4vThXcmTCf3KRQpUDrE --- CHANGELOG.md | 34 +++++ docs/sql-reference.md | 38 ++++- pgcolumnar--1.0-alpha2--1.0-alpha3.sql | 42 +++++- pgcolumnar--1.0-alpha3.sql | 33 ++++- src/columnar.h | 9 ++ src/columnar_metadata.c | 85 +++++++++++ src/columnar_parallel_copy.c | 198 +++++++++++++++++++++++++ test/parallel_copy_dedup.sh | 110 ++++++++++++++ test/run_all_versions.sh | 1 + 9 files changed, 544 insertions(+), 6 deletions(-) create mode 100755 test/parallel_copy_dedup.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 42b3631f..1ef21342 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,40 @@ true until the next version shipped. ## [Unreleased] +### Added + +- `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. + ### Changed - Chunk-group skip predicates are now evaluated most-selective-first, so a group diff --git a/docs/sql-reference.md b/docs/sql-reference.md index a54ef823..458f45c0 100644 --- a/docs/sql-reference.md +++ b/docs/sql-reference.md @@ -457,7 +457,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 +486,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; diff --git a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql index 78277346..dd98fbbb 100644 --- a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql +++ b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql @@ -9,9 +9,47 @@ */ \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_pkey + ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint); + +-- Two functions gain a parameter, which changes their signatures, so neither can +-- be a CREATE OR REPLACE: sort_status (#761) and parallel_copy (#403 item 7). DROP FUNCTION IF EXISTS pgcolumnar.sort_status(regclass); +DROP FUNCTION IF EXISTS pgcolumnar.parallel_copy(regclass, text, int); + +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, diff --git a/pgcolumnar--1.0-alpha3.sql b/pgcolumnar--1.0-alpha3.sql index 758de65f..d9759506 100644 --- a/pgcolumnar--1.0-alpha3.sql +++ b/pgcolumnar--1.0-alpha3.sql @@ -258,6 +258,32 @@ CREATE TABLE pgcolumnar.bloom ( CREATE UNIQUE INDEX bloom_pkey ON pgcolumnar.bloom USING btree (storage_id, group_number, column_index); +-- 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_pkey + ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint); + /* --------------------------------------------------------------------------- * pgcolumnar.free_space (Phase F physical reclaim) * @@ -1205,13 +1231,14 @@ COMMENT ON FUNCTION pgcolumnar.file_split_offsets(text, int) -- the loaders would block on that lock and the wait is invisible to the deadlock -- detector. See design/PARALLEL_COPY_PLAN.md. CREATE FUNCTION pgcolumnar.parallel_copy(target regclass, filename text, - workers int DEFAULT NULL) + 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) - 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)'; +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)'; /* * Per-column statistics without reading the whole table (#414). diff --git a/src/columnar.h b/src/columnar.h index a1bb5b2d..4893586e 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -644,6 +644,15 @@ extern uint64 PgColumnarVectorDecodes(PgColumnarReadState *readState); extern uint64 PgColumnarVectorsRuledOutByValue(PgColumnarReadState *readState); extern uint64 PgColumnarZoneMapProbes(PgColumnarReadState *readState); +/* pgcolumnar.parallel_copy load dedup (#403 item 7) */ +extern bool PgColumnarLoadFingerprintSeen(Oid relationOid, + const uint8 *fingerprint, + int fingerprintLen, + Snapshot snapshot); +extern void PgColumnarRecordLoadFingerprint(Oid relationOid, + const uint8 *fingerprint, + int fingerprintLen, int64 rows); + /* * How many of the scan keys the reader was handed became skip predicates it can * actually exclude a chunk group with (#479). Never larger than the scan-key diff --git a/src/columnar_metadata.c b/src/columnar_metadata.c index ca547967..2db59d3a 100644 --- a/src/columnar_metadata.c +++ b/src/columnar_metadata.c @@ -109,6 +109,13 @@ #define Anum_bloom_filter 4 #define Natts_bloom 4 +/* pgcolumnar.load_fingerprint (#403 item 7) */ +#define Anum_load_fingerprint_relation_oid 1 +#define Anum_load_fingerprint_fingerprint 2 +#define Anum_load_fingerprint_rows 3 +#define Anum_load_fingerprint_loaded_at 4 +#define Natts_load_fingerprint 4 + /* attribute numbers for columnar.free_space (Phase F physical reclaim) */ #define Anum_free_space_storage_id 1 #define Anum_free_space_file_offset 2 @@ -2415,6 +2422,84 @@ PgColumnarInsertBloomRow(const NativeBloomMetadata *b) metadata_flush_close(rel, RowExclusiveLock); } +/* + * PgColumnarLoadFingerprintSeen + * Has this exact file already been loaded into this table by a committed + * pgcolumnar.parallel_copy (#403 item 7)? + * + * Keyed by (relation_oid, fingerprint), which is load_fingerprint_pkey, so + * this is an exact index lookup. The fingerprint is the SHA-256 of the + * file's bytes: a file that changed at the same path is a different load. + */ +bool +PgColumnarLoadFingerprintSeen(Oid relationOid, const uint8 *fingerprint, + int fingerprintLen, Snapshot snapshot) +{ + Relation rel = open_columnar_table("load_fingerprint", AccessShareLock); + ScanKeyData key[2]; + SysScanDesc scan; + Oid idxOid; + HeapTuple tuple; + bytea *fp; + bool found; + + fp = (bytea *) palloc(VARHDRSZ + fingerprintLen); + SET_VARSIZE(fp, VARHDRSZ + fingerprintLen); + memcpy(VARDATA(fp), fingerprint, fingerprintLen); + + ScanKeyInit(&key[0], Anum_load_fingerprint_relation_oid, BTEqualStrategyNumber, + F_OIDEQ, ObjectIdGetDatum(relationOid)); + ScanKeyInit(&key[1], Anum_load_fingerprint_fingerprint, BTEqualStrategyNumber, + F_BYTEAEQ, PointerGetDatum(fp)); + + idxOid = pgcolumnar_index_oid("load_fingerprint_pkey"); + scan = systable_beginscan(rel, idxOid, OidIsValid(idxOid), snapshot, 2, key); + tuple = systable_getnext(scan); + found = HeapTupleIsValid(tuple); + systable_endscan(scan); + table_close(rel, AccessShareLock); + pfree(fp); + + return found; +} + +/* + * PgColumnarRecordLoadFingerprint + * Record that this file has been loaded into this table (#403 item 7). + * + * Called AFTER the data commits, never before. A crash between the two + * leaves data with no fingerprint, so a retry loads again -- the behaviour + * without this feature, and the safe direction. The reverse order would + * leave a fingerprint with no data and refuse rows that were never stored. + */ +void +PgColumnarRecordLoadFingerprint(Oid relationOid, const uint8 *fingerprint, + int fingerprintLen, int64 rows) +{ + Relation rel = open_columnar_table("load_fingerprint", RowExclusiveLock); + TupleDesc tupdesc = RelationGetDescr(rel); + Datum values[Natts_load_fingerprint]; + bool nulls[Natts_load_fingerprint]; + HeapTuple tuple; + bytea *fp; + + memset(nulls, false, sizeof(nulls)); + fp = (bytea *) palloc(VARHDRSZ + fingerprintLen); + SET_VARSIZE(fp, VARHDRSZ + fingerprintLen); + memcpy(VARDATA(fp), fingerprint, fingerprintLen); + + values[Anum_load_fingerprint_relation_oid - 1] = ObjectIdGetDatum(relationOid); + values[Anum_load_fingerprint_fingerprint - 1] = PointerGetDatum(fp); + values[Anum_load_fingerprint_rows - 1] = Int64GetDatum(rows); + values[Anum_load_fingerprint_loaded_at - 1] = + TimestampTzGetDatum(GetCurrentTimestamp()); + + tuple = heap_form_tuple(tupdesc, values, nulls); + metadata_flush_insert(rel, tuple); + heap_freetuple(tuple); + metadata_flush_close(rel, RowExclusiveLock); +} + /* * PgColumnarReadBloomForColumn * One column's bloom filter for one row group, or NULL when it has none diff --git a/src/columnar_parallel_copy.c b/src/columnar_parallel_copy.c index c11c8e2b..1dc7c811 100644 --- a/src/columnar_parallel_copy.c +++ b/src/columnar_parallel_copy.c @@ -41,6 +41,12 @@ #include #include +#include "common/cryptohash.h" +#include "common/sha2.h" + +/* read size for the dedup fingerprint pass */ +#define PCOPY_FINGERPRINT_BUFSZ (256 * 1024) + #include "columnar.h" #include "columnar_write_state.h" @@ -395,6 +401,16 @@ typedef struct PcopyHeader int failed_worker; /* index of the first failed loader, or -1 */ int coord_sqlerrcode; char coord_errmsg[512]; + + /* + * Opt-in load dedup (#403 item 7). dedup is what the caller asked for; + * fingerprint is the SHA-256 of the file, computed by the coordinator while + * the loaders run; skipped_duplicate reports back that the load was refused + * because this table has already taken this file. + */ + bool dedup; + uint8 fingerprint[PG_SHA256_DIGEST_LENGTH]; + bool skipped_duplicate; } PcopyHeader; /* @@ -972,6 +988,88 @@ pgcolumnar_parallel_copy_worker(Datum main_arg) proc_exit(0); } +/* + * pcopy_fingerprint_file + * SHA-256 of a file's bytes, for the opt-in load dedup (#403 item 7). + * + * The whole file rather than its path, size or mtime, because all three of + * those call a changed file the same file, and being wrong in that + * direction discards rows the caller meant to load. + * + * The coordinator runs this while the loaders are already reading the same + * file, so it overlaps their work rather than adding a serial pass in front + * of it. Measured on 264 MiB: the load takes 1.13 s across four workers and + * a single-threaded SHA-256 of the same bytes takes 0.48 s, so doing it + * before dispatch would have cost 42%. + * + * A file that cannot be read is an error, not a reason to load without + * checking: the caller asked for the load to be refused if it was a repeat, + * and silently loading it again is the answer they excluded. + */ +static void +pcopy_fingerprint_file(const char *path, uint8 *out) +{ + pg_cryptohash_ctx *ctx; + char *buf; + int fd; + int nread; + + fd = OpenTransientFile(path, O_RDONLY | PG_BINARY); + if (fd < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not open \"%s\" to fingerprint it for dedup: %m", + path))); + + ctx = pg_cryptohash_create(PG_SHA256); + if (ctx == NULL || pg_cryptohash_init(ctx) < 0) + { + CloseTransientFile(fd); + ereport(ERROR, + (errcode(ERRCODE_INTERNAL_ERROR), + errmsg("could not initialise SHA-256 for the dedup fingerprint"))); + } + + buf = palloc(PCOPY_FINGERPRINT_BUFSZ); + while ((nread = read(fd, buf, PCOPY_FINGERPRINT_BUFSZ)) > 0) + { + if (pg_cryptohash_update(ctx, (const uint8 *) buf, (size_t) nread) < 0) + { + CloseTransientFile(fd); + pg_cryptohash_free(ctx); + ereport(ERROR, + (errcode(ERRCODE_INTERNAL_ERROR), + errmsg("could not compute the dedup fingerprint of \"%s\"", + path))); + } + } + + if (nread < 0) + { + int saved = errno; + + CloseTransientFile(fd); + pg_cryptohash_free(ctx); + errno = saved; + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read \"%s\" to fingerprint it for dedup: %m", + path))); + } + CloseTransientFile(fd); + + if (pg_cryptohash_final(ctx, out, PG_SHA256_DIGEST_LENGTH) < 0) + { + pg_cryptohash_free(ctx); + ereport(ERROR, + (errcode(ERRCODE_INTERNAL_ERROR), + errmsg("could not finalise the dedup fingerprint of \"%s\"", + path))); + } + pg_cryptohash_free(ctx); + pfree(buf); +} + /* * pcopy_finish_prepared * COMMIT PREPARED (isCommit) or ROLLBACK PREPARED a gid from this (coordinator) @@ -1048,6 +1146,7 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) BackgroundWorker bw; BackgroundWorkerHandle **handles; int i; + bool fingerprinted = false; pqsignal(SIGTERM, pcopy_coord_sigterm); BackgroundWorkerUnblockSignals(); @@ -1155,6 +1254,18 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) } if (running == 0 || pcopy_coord_got_sigterm) break; + + /* + * Fingerprint the file once, here rather than before dispatch, so + * the pass overlaps the loaders instead of delaying them (#403 + * item 7). The decision it feeds is taken below, after every loader + * has PREPAREd and before anything commits. + */ + if (hdr->dedup && !fingerprinted) + { + pcopy_fingerprint_file(hdr->filename, hdr->fingerprint); + fingerprinted = true; + } (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, 1000L, PG_WAIT_EXTENSION); ResetLatch(MyLatch); @@ -1194,6 +1305,55 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) if (all_prepared) { + bool duplicate = false; + + /* + * The load is done and prepared; nothing is visible yet. This is the + * only point at which a repeat can be refused for free, because 2PC + * lets the whole thing be thrown away after the work rather than + * before it (#403 item 7). A duplicate therefore costs a full load + * that is discarded, which is the right trade: the duplicate is the + * exceptional case, and checking before the load would mean either a + * serial hash pass in front of every load or a fingerprint that + * depends on the worker count. + * + * NOT serialized against a CONCURRENT identical load. Two loads of + * the same file running at once both check before either records, + * so both commit -- which is the behaviour without dedup, and is the + * safe direction. What this refuses is the case it was built for: a + * committed load whose acknowledgement the client never saw, retried + * afterwards. Serializing the concurrent case needs a lock held + * across the check, the COMMIT PREPAREDs and the record, and + * COMMIT PREPARED cannot run inside a transaction block. + */ + if (hdr->dedup) + { + if (!fingerprinted) + { + pcopy_fingerprint_file(hdr->filename, hdr->fingerprint); + fingerprinted = true; + } + StartTransactionCommand(); + PushActiveSnapshot(GetTransactionSnapshot()); + duplicate = PgColumnarLoadFingerprintSeen(hdr->relid, + hdr->fingerprint, + PG_SHA256_DIGEST_LENGTH, + GetActiveSnapshot()); + PopActiveSnapshot(); + CommitTransactionCommand(); + } + + if (duplicate) + { + /* refuse it: roll every range back, so nothing becomes visible */ + for (i = 0; i < nworkers; i++) + pcopy_rollback_prepared_quietly(slots[i].gid); + hdr->total_rows = 0; + hdr->skipped_duplicate = true; + pg_atomic_write_u32(&hdr->coord_state, PCOPY_COORD_DONE); + } + else + { /* * Decision: commit. Committing N prepared transactions is not itself * one atomic step -- a coordinator crash mid-loop leaves the standard @@ -1207,7 +1367,26 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) total += slots[i].rows; } hdr->total_rows = total; + + /* + * Recorded 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. + */ + if (hdr->dedup) + { + StartTransactionCommand(); + PushActiveSnapshot(GetTransactionSnapshot()); + PgColumnarRecordLoadFingerprint(hdr->relid, hdr->fingerprint, + PG_SHA256_DIGEST_LENGTH, total); + PopActiveSnapshot(); + CommitTransactionCommand(); + } + pg_atomic_write_u32(&hdr->coord_state, PCOPY_COORD_DONE); + } } else { @@ -1287,6 +1466,7 @@ pgcolumnar_parallel_copy(PG_FUNCTION_ARGS) char *path; int workers; int max_prepared; + bool dedup; bool single_table = false; int64 *offs; shm_toc_estimator est; @@ -1309,6 +1489,7 @@ pgcolumnar_parallel_copy(PG_FUNCTION_ARGS) relid = PG_GETARG_OID(0); path = text_to_cstring(PG_GETARG_TEXT_PP(1)); workers = PG_ARGISNULL(2) ? pcopy_auto_workers() : PG_GETARG_INT32(2); + dedup = PG_ARGISNULL(3) ? false : PG_GETARG_BOOL(3); if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_SERVER_FILES)) ereport(ERROR, @@ -1441,6 +1622,9 @@ pgcolumnar_parallel_copy(PG_FUNCTION_ARGS) hdr->failed_worker = -1; hdr->coord_sqlerrcode = 0; hdr->coord_errmsg[0] = '\0'; + hdr->dedup = dedup; + hdr->skipped_duplicate = false; + memset(hdr->fingerprint, 0, sizeof(hdr->fingerprint)); pg_atomic_init_u32(&hdr->coord_state, PCOPY_COORD_PENDING); for (i = 0; i < workers; i++) @@ -1495,8 +1679,22 @@ pgcolumnar_parallel_copy(PG_FUNCTION_ARGS) if (cstate == PCOPY_COORD_DONE) { int64 total = hdr->total_rows; + bool skipped = hdr->skipped_duplicate; dsm_detach(seg); + + /* + * Say what happened rather than returning a silent 0. Discarding rows a + * caller asked to insert is not ordinary INSERT behaviour, so a caller + * that gets 0 back is owed the reason (#403 item 7). + */ + if (skipped) + ereport(NOTICE, + (errmsg("pgcolumnar.parallel_copy skipped a load already applied to \"%s\"", + get_rel_name(relid)), + errdetail("The file has the same contents as a load this table already took."), + errhint("Pass dedup => false to load it again."))); + PG_RETURN_INT64(total); } else diff --git a/test/parallel_copy_dedup.sh b/test/parallel_copy_dedup.sh new file mode 100755 index 00000000..c0771f85 --- /dev/null +++ b/test/parallel_copy_dedup.sh @@ -0,0 +1,110 @@ +#!/usr/bin/env bash +# +# pgColumnar: pgcolumnar.parallel_copy can refuse a load it has already done +# (#403 item 7). +# +# THE DEFECT. A committed load whose acknowledgement the client never sees is +# retried, and the rows go in twice. Measured before this change: the same file +# loaded twice into the same table gives 100,000 rows and then 200,000. +# +# WHY THE UNIT IS THE WHOLE LOAD, not a "part". The paper this comes from keeps +# hashes of the last N inserted PARTS, because there each part commits on its +# own. pgcolumnar.parallel_copy is atomic through 2PC: the loaders PREPARE and a +# coordinator commits all or none, proved by a malformed row at line 50,001 +# leaving 0 rows and 0 prepared transactions. Parts never commit independently +# here, so a part hash would deduplicate nothing that is not already +# all-or-nothing. The load is the unit. +# +# WHY IT IS OPT-IN. Discarding rows a client asked to insert is not SQL INSERT +# behaviour. It is off unless asked for, it is scoped to the bulk-load path, and +# it says what it skipped rather than returning a silent 0. +# +# WHAT MAKES TWO LOADS "THE SAME". The SHA-256 of the file's bytes, so a file +# that changed at the same path is a different load and is loaded. Path, size +# and mtime would all call that file the same one. +# +# THE ORDER OF THE TWO WRITES IS THE SAFETY ARGUMENT. The data commits first and +# the fingerprint is recorded after. A crash between them leaves data with no +# fingerprint, so a retry loads again -- which is exactly today's behaviour and +# is the safe direction. The reverse order would leave a fingerprint with no +# data, and a later load would be refused for rows that were never stored. +# +# Usage: test/parallel_copy_dedup.sh [PG_CONFIG] +# Written fresh for pgColumnar. + +set -uo pipefail +export PGC_EXTRA_CONF=$'max_prepared_transactions=8\nmax_worker_processes=16' +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" + +DATADIR="$PGC_WORKDIR/pcdedup" +mkdir -p "$DATADIR"; chmod 777 "$DATADIR" +if [ "$(id -u)" = "0" ]; then chown postgres "$DATADIR"; fi + +ROWS=20000 +F="$DATADIR/load.txt" +G="$DATADIR/other.txt" + +psql_run "COPY (SELECT g AS id, 'v'||g AS txt FROM generate_series(1,$ROWS) g) + TO '$F' WITH (FORMAT text);" >/dev/null +psql_run "COPY (SELECT g AS id, 'w'||g AS txt FROM generate_series(1,$ROWS) g) + TO '$G' WITH (FORMAT text);" >/dev/null + +q() { + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -Atq \ + -c "$1" 2>&1 | tail -1 +} + +mk() { psql_run "DROP TABLE IF EXISTS $1; CREATE TABLE $1 (id int, txt text) USING pgcolumnar;" >/dev/null; } + +# ---- the control: without dedup the retry doubles the table ---------------- +mk pcd_plain +r1="$(q "SELECT pgcolumnar.parallel_copy('pcd_plain','$F',4)")" +r2="$(q "SELECT pgcolumnar.parallel_copy('pcd_plain','$F',4)")" +check "premise: without dedup a repeated load doubles the table, which is the defect" \ + "$(q "SELECT count(*) FROM pcd_plain")" "$(( ROWS * 2 ))" +check "premise: and each of those loads reported its rows" "$r1/$r2" "$ROWS/$ROWS" + +# ---- the same load twice, with dedup -------------------------------------- +mk pcd +d1="$(q "SELECT pgcolumnar.parallel_copy('pcd','$F',4,true)")" +c1="$(q "SELECT count(*) FROM pcd")" +d2="$(q "SELECT pgcolumnar.parallel_copy('pcd','$F',4,true)")" +c2="$(q "SELECT count(*) FROM pcd")" +echo "-- dedup on: load 1 returned $d1 (table $c1), load 2 returned $d2 (table $c2)" + +check "the first load stores its rows normally (#403 item 7)" "$c1" "$ROWS" +check "and reports them" "$d1" "$ROWS" +check "the second load of the same file stores nothing (#403 item 7)" "$c2" "$ROWS" +check "and reports 0 rather than claiming it loaded them" "$d2" "0" + +# ---- a changed file at the same path is a different load ------------------- +cp "$G" "$F" +d3="$(q "SELECT pgcolumnar.parallel_copy('pcd','$F',4,true)")" +c3="$(q "SELECT count(*) FROM pcd")" +echo "-- after replacing the file at the same path: returned $d3, table $c3" +check "a changed file at the same path is loaded, not skipped (#403 item 7)" "$c3" "$(( ROWS * 2 ))" +check "and reports the rows it loaded" "$d3" "$ROWS" + +# ---- the fingerprint is per table ----------------------------------------- +mk pcd_other +d4="$(q "SELECT pgcolumnar.parallel_copy('pcd_other','$F',4,true)")" +check "the same file loads into a DIFFERENT table (the record is per table)" \ + "$(q "SELECT count(*) FROM pcd_other")" "$ROWS" +check "and reports its rows" "$d4" "$ROWS" + +# ---- dedup off still loads twice, on the same table ------------------------ +d5="$(q "SELECT pgcolumnar.parallel_copy('pcd_other','$F',4,false)")" +check "asking for no dedup loads again even though the file is on record" \ + "$(q "SELECT count(*) FROM pcd_other")" "$(( ROWS * 2 ))" +check "and reports those rows too" "$d5" "$ROWS" + +# ---- the record itself ----------------------------------------------------- +check "the skipped load left exactly one record for that table and file" \ + "$(q "SELECT count(*) FROM pgcolumnar.load_fingerprint + WHERE relation_oid = 'pcd'::regclass")" "2" + +check "no prepared transaction leaked through any of it" \ + "$(q "SELECT count(*) FROM pg_prepared_xacts")" "0" + +pgc_summary diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index 2d44bd77..9c63c00b 100644 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -212,6 +212,7 @@ SUITES=( objstore_userinfo parallel parallel_copy + parallel_copy_dedup parallel_degree parallel_export_parquet parallel_flush_optin From bb97a8696cbe819734ec15ed945b2cdf62e96896 Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Fri, 28 Aug 2026 12:06:09 -0600 Subject: [PATCH 2/4] feat: pgcolumnar.expire drops fully expired row groups (#403 item 5a) The tractable half of the paper's "merge-time data transformation". Our rewrites already retire whole row groups -- pgcolumnar.compact drops every group that is fully deleted, through PgColumnarRetireGroup under ShareUpdateExclusiveLock. Retention is the same operation with a different predicate. The decision is a catalog read. A group's zone map records the maximum value of each column, so a group whose maximum is older than the cutoff holds no row still inside the retention. Nothing is decoded, nothing is rewritten, and readers and writers continue throughout. ## Explicit, not implicit It deletes rows, so it is called by name and never runs on its own. Not wired into VACUUM, compact or autovacuum. ClickHouse folds TTL into its merges; an operation a PostgreSQL user runs for maintenance must not silently drop their data. A table with no declared retention raises an error rather than returning 0, which would read as success. Declared through set_options, beside the other per-table options, as ttl_column and ttl_interval. Either alone means no retention, so a half declaration cannot drop 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. That is the price of deciding by group rather than by row, and it is the safe direction. Measured: 5,000 rows in 5 groups, 1,440 past a three-day retention. expire dropped 1 group, 1,000 rows all 3,560 rows still inside the retention kept including the 440 expired rows sharing the straddling group ## Removal proof The natural wrong implementation is "the group holds an expired row" rather than "every row in the group is expired": decide on the group's MINIMUM instead of its maximum. With that mutation applied and asserted: 2 groups dropped instead of 1, and 3,000 rows left of 3,560 FAIL NO row still inside the retention was dropped (#403 item 5a) Only that check goes red. It is the reason the fixture builds a straddling group deliberately: a suite that only proved expired data disappears would pass just as well on an implementation that dropped everything. ## Gate test/ttl_expire.sh, 11 checks, registered in run_all_versions.sh. native_upgrade_converge passes, which is what proves the two new options columns, set_options' changed signature and the new function reach an upgraded catalog identically to a fresh 1.0-alpha3 install. set_options is called by 88 suites; the change is additive with defaults, and 24 of them were run here, all passing, with the full set left to CI. Built with COPT=-Werror, 0 warnings. One error worth recording: the first run failed every check because the base script's COMMENT ON FUNCTION still named set_options' old signature, so CREATE EXTENSION itself failed. The suite caught it on its first run. Refs #403. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UaQ4vThXcmTCf3KRQpUDrE --- CHANGELOG.md | 27 ++++ docs/sql-reference.md | 37 ++++++ pgcolumnar--1.0-alpha2--1.0-alpha3.sql | 173 ++++++++++++++++++++++++- pgcolumnar--1.0-alpha3.sql | 32 ++++- src/columnar.h | 3 + src/columnar_metadata.c | 59 +++++++++ src/columnar_vacuum.c | 162 +++++++++++++++++++++++ test/run_all_versions.sh | 1 + test/ttl_expire.sh | 130 +++++++++++++++++++ 9 files changed, 616 insertions(+), 8 deletions(-) create mode 100755 test/ttl_expire.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ef21342..b67c2a44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,33 @@ true until the next version shipped. ### 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 diff --git a/docs/sql-reference.md b/docs/sql-reference.md index 458f45c0..37ba96dd 100644 --- a/docs/sql-reference.md +++ b/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 diff --git a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql index dd98fbbb..4eede029 100644 --- a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql +++ b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql @@ -36,10 +36,179 @@ CREATE TABLE pgcolumnar.load_fingerprint ( CREATE INDEX load_fingerprint_pkey ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint); --- Two functions gain a parameter, which changes their signatures, so neither can --- be a CREATE OR REPLACE: sort_status (#761) and parallel_copy (#403 item 7). +-- 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, diff --git a/pgcolumnar--1.0-alpha3.sql b/pgcolumnar--1.0-alpha3.sql index d9759506..91f0ecdc 100644 --- a/pgcolumnar--1.0-alpha3.sql +++ b/pgcolumnar--1.0-alpha3.sql @@ -55,7 +55,11 @@ CREATE TABLE pgcolumnar.options ( compression_level integer, compression name, encode_effort name, - sort_by name[] -- declared physical sort key (#288) + sort_by name[], -- declared physical sort key (#288) + -- Declared retention (#403 item 5a), read only by pgcolumnar.expire. + -- Nothing drops rows on its own: expire is called by name. + ttl_column name, + ttl_interval interval ); /* @@ -393,7 +397,9 @@ CREATE FUNCTION pgcolumnar.set_options( compression name DEFAULT NULL, compression_level int DEFAULT NULL, encode_effort name DEFAULT NULL, - sort_by 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$ @@ -510,9 +516,11 @@ BEGIN INSERT INTO pgcolumnar.options AS o (regclass, chunk_group_row_limit, stripe_row_limit, - compression, compression_level, encode_effort, sort_by) + 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) + 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), @@ -525,11 +533,15 @@ BEGIN encode_effort = COALESCE(EXCLUDED.encode_effort, o.encode_effort), sort_by = - COALESCE(EXCLUDED.sort_by, o.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[]) +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.reset_options( @@ -849,6 +861,14 @@ $sort_status$; COMMENT ON FUNCTION pgcolumnar.sort_status(regclass) IS 'how much of an ordered columnar table is still in its ordered run, and by what kind of ordering (#301, #761)'; +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.vacuum(tablename regclass, stripe_count int DEFAULT 0) RETURNS void LANGUAGE C STRICT diff --git a/src/columnar.h b/src/columnar.h index 4893586e..d864e18c 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -503,6 +503,9 @@ extern NativeZoneMapMetadata *PgColumnarReadZoneMapForColumn(uint64 storageId, extern void PgColumnarCloseZoneMapSession(PgColumnarZoneMapSession *sess); extern void PgColumnarDeleteMetadata(uint64 storageId); +/* declared retention, read by pgcolumnar.expire (#403 item 5a) */ +extern bool PgColumnarReadTtl(Oid relid, char **column, Interval **interval); + /* per-table options catalog (spec 7.4) */ extern bool PgColumnarReadOptions(Oid relid, PgColumnarOptions *opts); extern int pgcolumnar_effective_stripe_row_limit(Oid relid); diff --git a/src/columnar_metadata.c b/src/columnar_metadata.c index 2db59d3a..a8baf9c7 100644 --- a/src/columnar_metadata.c +++ b/src/columnar_metadata.c @@ -44,6 +44,8 @@ #define Anum_options_compression 5 #define Anum_options_encode_effort 6 #define Anum_options_sort_by 7 +#define Anum_options_ttl_column 8 +#define Anum_options_ttl_interval 9 #define Natts_options 7 /* attribute numbers for columnar.projection (gap 26, format 2.2) */ @@ -3108,6 +3110,63 @@ pgcolumnar_effective_stripe_row_limit(Oid relid) * the same command-id-advanced snapshot as PgColumnarReadOptions so a * sort_by set earlier in this transaction is visible. */ +/* + * PgColumnarReadTtl + * The retention declared for this table by set_options (#403 item 5a), or + * false when it has none. Read with the same command-id-advanced snapshot + * as PgColumnarReadOptions, so a retention set earlier in this transaction + * is visible. + * + * Both halves must be present to mean anything: a column with no interval + * names nothing to compare against, and an interval with no column has + * nothing to compare. Either alone reads as "no retention", so a half + * declaration cannot drop rows. + */ +bool +PgColumnarReadTtl(Oid relid, char **column, Interval **interval) +{ + Relation rel = open_columnar_table("options", AccessShareLock); + TupleDesc tupdesc = RelationGetDescr(rel); + ScanKeyData key[1]; + SysScanDesc scan; + HeapTuple tuple; + Snapshot base; + Snapshot snapshot; + bool found = false; + + *column = NULL; + *interval = NULL; + + base = ActiveSnapshotSet() ? GetActiveSnapshot() : GetTransactionSnapshot(); + snapshot = PgColumnarCatalogSnapshot(base); + + ScanKeyInit(&key[0], Anum_options_regclass, BTEqualStrategyNumber, + F_OIDEQ, ObjectIdGetDatum(relid)); + scan = systable_beginscan(rel, InvalidOid, false, snapshot, 1, key); + tuple = systable_getnext(scan); + if (HeapTupleIsValid(tuple)) + { + bool colnull; + bool ivnull; + Datum cold = heap_getattr(tuple, Anum_options_ttl_column, + tupdesc, &colnull); + Datum ivd = heap_getattr(tuple, Anum_options_ttl_interval, + tupdesc, &ivnull); + + if (!colnull && !ivnull) + { + *column = pstrdup(NameStr(*DatumGetName(cold))); + *interval = (Interval *) palloc(sizeof(Interval)); + memcpy(*interval, DatumGetIntervalP(ivd), sizeof(Interval)); + found = true; + } + } + systable_endscan(scan); + table_close(rel, AccessShareLock); + + return found; +} + List * PgColumnarReadSortBy(Oid relid) { diff --git a/src/columnar_vacuum.c b/src/columnar_vacuum.c index c129dd19..cf3819ed 100644 --- a/src/columnar_vacuum.c +++ b/src/columnar_vacuum.c @@ -60,6 +60,7 @@ PG_FUNCTION_INFO_V1(pgcolumnar_vacuum); PG_FUNCTION_INFO_V1(pgcolumnar_vacuum_sorted); PG_FUNCTION_INFO_V1(pgcolumnar_cluster); PG_FUNCTION_INFO_V1(pgcolumnar_compact); +PG_FUNCTION_INFO_V1(pgcolumnar_expire); PG_FUNCTION_INFO_V1(pgcolumnar_compact_rewrite); PG_FUNCTION_INFO_V1(pgcolumnar_recluster); PG_FUNCTION_INFO_V1(pgcolumnar_truncate); @@ -2018,6 +2019,167 @@ pgcolumnar_cluster(PG_FUNCTION_ARGS) * file, or a later F3 pass. Rewriting partially-deleted groups online is * Phase F3b. */ +/* + * pgcolumnar_expire + * Drop every row group whose rows are ALL older than the retention declared + * by set_options(ttl_column, ttl_interval) (#403 item 5a). Returns the + * number of groups dropped. + * + * The decision is a catalog read. A group's zone map records the maximum + * value of each column in it, so a group whose maximum is below the cutoff + * holds no row that is still within the retention. Nothing is decoded and + * nothing is rewritten; the group's catalog rows are retired exactly as + * pgcolumnar.compact retires a fully deleted one, under the same + * ShareUpdateExclusiveLock, so readers and writers continue throughout. + * + * A group that STRADDLES the cutoff is kept whole. Its expired rows survive + * until every row in the group has expired. That is the price of deciding + * by group rather than by row, and it is the safe direction: the + * alternative drops rows that are still inside the retention. + * + * This 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. + */ +Datum +pgcolumnar_expire(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + Relation rel; + TupleDesc tupdesc; + char *ttlColumn; + Interval *ttlInterval; + AttrNumber attno = InvalidAttrNumber; + Form_pg_attribute att; + Datum cutoff; + TypeCacheEntry *tce; + uint64 storageId; + List *groups; + ListCell *lc; + int64 retired = 0; + int i; + + PgColumnarRequireTableOwnerByOid(relid); + + if (!PgColumnarReadTtl(relid, &ttlColumn, &ttlInterval)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("table \"%s\" has no declared retention", + get_rel_name(relid)), + errhint("Declare one with pgcolumnar.set_options(..., ttl_column => ..., ttl_interval => ...)."))); + + /* the lazy lock, as compact takes: readers and writers continue */ + rel = table_open(relid, ShareUpdateExclusiveLock); + + if (!PgColumnarIsColumnarRelation(relid)) + { + table_close(rel, ShareUpdateExclusiveLock); + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("relation \"%s\" is not a columnar table", + RelationGetRelationName(rel)))); + } + + tupdesc = RelationGetDescr(rel); + for (i = 0; i < tupdesc->natts; i++) + { + Form_pg_attribute a = TupleDescAttr(tupdesc, i); + + if (!a->attisdropped && strcmp(NameStr(a->attname), ttlColumn) == 0) + { + attno = a->attnum; + break; + } + } + if (attno == InvalidAttrNumber) + { + table_close(rel, ShareUpdateExclusiveLock); + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_COLUMN), + errmsg("retention column \"%s\" does not exist in table \"%s\"", + ttlColumn, get_rel_name(relid)))); + } + + att = TupleDescAttr(tupdesc, attno - 1); + + /* + * The cutoff is computed in the column's own type, so the comparison below + * is the type's own ordering rather than a coercion invented here. + */ + switch (att->atttypid) + { + case TIMESTAMPTZOID: + cutoff = DirectFunctionCall2(timestamptz_mi_interval, + TimestampTzGetDatum(GetCurrentTimestamp()), + IntervalPGetDatum(ttlInterval)); + break; + case TIMESTAMPOID: + cutoff = DirectFunctionCall2(timestamp_mi_interval, + DirectFunctionCall1(timestamptz_timestamp, + TimestampTzGetDatum(GetCurrentTimestamp())), + IntervalPGetDatum(ttlInterval)); + break; + default: + table_close(rel, ShareUpdateExclusiveLock); + ereport(ERROR, + (errcode(ERRCODE_DATATYPE_MISMATCH), + errmsg("retention column \"%s\" must be timestamp or timestamptz", + ttlColumn))); + cutoff = (Datum) 0; /* keep the compiler quiet */ + break; + } + + tce = lookup_type_cache(att->atttypid, TYPECACHE_CMP_PROC_FINFO); + if (!OidIsValid(tce->cmp_proc_finfo.fn_oid)) + { + table_close(rel, ShareUpdateExclusiveLock); + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_FUNCTION), + errmsg("retention column \"%s\" has no ordering to compare against", + ttlColumn))); + } + + storageId = PgColumnarStorageId(rel); + groups = PgColumnarReadRowGroupList(storageId, GetActiveSnapshot()); + + foreach(lc, groups) + { + NativeRowGroupMetadata *rg = (NativeRowGroupMetadata *) lfirst(lc); + NativeZoneMapMetadata *z; + char *cur; + Datum maxv; + int32 c; + + z = PgColumnarReadZoneMapForColumn(storageId, rg->groupNumber, + attno - 1, GetActiveSnapshot(), NULL); + + /* + * No zone map, or one without min/max, means nothing is known about what + * this group holds. Keeping it is the only safe reading: dropping on an + * absent bound would drop groups whose contents were never examined. + */ + if (z == NULL || !z->hasMinMax) + continue; + + cur = (char *) z->maximum; + maxv = PgColumnarDecodeValue(att, &cur, z->maximum + z->maximumLen, + CurrentMemoryContext); + + c = DatumGetInt32(FunctionCall2Coll(&tce->cmp_proc_finfo, + att->attcollation, maxv, cutoff)); + if (c < 0) + { + PgColumnarRetireGroup(storageId, rg->groupNumber); + retired++; + } + } + + /* keep the lock until end of transaction, as compact does */ + table_close(rel, NoLock); + + PG_RETURN_INT64(retired); +} + Datum pgcolumnar_compact(PG_FUNCTION_ARGS) { diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index 9c63c00b..080cf0ab 100644 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -255,6 +255,7 @@ SUITES=( sorted_projection stats_privilege temporal + ttl_expire ungrouped_vector_agg unique_conc update_conc diff --git a/test/ttl_expire.sh b/test/ttl_expire.sh new file mode 100755 index 00000000..271f81f8 --- /dev/null +++ b/test/ttl_expire.sh @@ -0,0 +1,130 @@ +#!/usr/bin/env bash +# +# pgColumnar: pgcolumnar.expire drops row groups whose rows are all older than a +# declared retention, without reading or rewriting them (#403 item 5a). +# +# WHAT IT IS. The tractable half of #403 item 5, "merge-time data transformation". +# Our rewrites already retire whole row groups: pgcolumnar.compact drops every +# group that is fully deleted, through PgColumnarRetireGroup, under +# ShareUpdateExclusiveLock. Retention is the same operation with a different +# predicate, and the zone map already holds what decides it -- a group whose +# MAXIMUM value in the retention column is older than the cutoff contains no row +# that is still within it. So the decision is a catalog read: no decode, no +# rewrite, no scan of the data. +# +# WHY AN EXPLICIT FUNCTION rather than a hook inside vacuum. This deletes rows. +# ClickHouse folds TTL into its merges; an operation a PostgreSQL user runs for +# maintenance must not silently drop their data. It is asked for by name, it +# returns how many groups it dropped, and a table with no declared retention is +# an error rather than a no-op that reads as success. +# +# THE CHECK THAT MATTERS IS THE STRADDLING GROUP. Dropping a group whose rows are +# all expired is the feature; dropping one that still holds live rows is data +# loss. The fixture builds a group that spans the cutoff deliberately and pins +# that it survives with every one of its rows. A test that only proved expired +# data disappears would pass just as well on an implementation that dropped +# everything. +# +# Usage: test/ttl_expire.sh [PG_CONFIG] +# Written fresh for pgColumnar. + +set -uo pipefail +# Pinned in the cluster config rather than by SET, so the writing session and any +# later session agree about the geometry (#806). +PGC_EXTRA_CONF="${PGC_EXTRA_CONF:-} +pgcolumnar.stripe_row_limit=1000" +export PGC_EXTRA_CONF +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" + +q() { + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -Atq \ + -c "$1" 2>&1 | tail -1 +} + +# 5,000 rows at 1,000 a group = 5 groups, one day apart per 1,000 rows, so each +# group covers a distinct day. Retention of 3 days puts the cutoff INSIDE the +# third group, which is the straddling case. +psql_run "CREATE TABLE ttl_t (id int, ts timestamptz, v text) USING pgcolumnar;" +psql_run "INSERT INTO ttl_t + SELECT g, + now() - make_interval(days => 5) + make_interval(mins => g * 2), + 'v'||g + FROM generate_series(1,5000) g;" +psql_run "ANALYZE ttl_t;" + +GROUPS_BEFORE="$(q "SELECT count(*) FROM pgcolumnar.storage s + JOIN pgcolumnar.row_group rg USING (storage_id) + WHERE s.relation_oid = 'ttl_t'::regclass")" +ROWS_BEFORE="$(q "SELECT count(*) FROM ttl_t")" + +CUTOFF="SELECT now() - make_interval(days => 3)" +EXPIRED_ROWS="$(q "SELECT count(*) FROM ttl_t WHERE ts < ($CUTOFF)")" +LIVE_ROWS="$(q "SELECT count(*) FROM ttl_t WHERE ts >= ($CUTOFF)")" + +# ---- premises ------------------------------------------------------------- +check "premise: the fixture is laid out in several row groups" \ + "$([ "${GROUPS_BEFORE:-0}" -ge 4 ] && echo "many ($GROUPS_BEFORE)" || echo "TOO FEW ($GROUPS_BEFORE)")" \ + "many ($GROUPS_BEFORE)" + +check "premise: some rows are past the retention and some are not" \ + "$([ "${EXPIRED_ROWS:-0}" -gt 0 ] && [ "${LIVE_ROWS:-0}" -gt 0 ] && echo "both" \ + || echo "ONE-SIDED (expired=$EXPIRED_ROWS live=$LIVE_ROWS)")" "both" + +# The whole safety argument rests on this: a group that spans the cutoff exists, +# so "drop every group with an expired row in it" and "drop every group whose +# rows are all expired" give different answers on this fixture. +STRADDLE_ROWS="$(q "SELECT count(*) FROM ttl_t + WHERE ts >= ($CUTOFF) - make_interval(days => 1) + AND ts < ($CUTOFF) + make_interval(days => 1)")" +check "premise: rows exist on both sides of the cutoff within one day of it" \ + "$([ "${STRADDLE_ROWS:-0}" -gt 0 ] && echo "straddled" || echo "NO STRADDLE")" "straddled" + +# ---- declaring the retention ---------------------------------------------- +psql_run "SELECT pgcolumnar.set_options('ttl_t', ttl_column => 'ts', + ttl_interval => '3 days');" +check "the retention is recorded where the other options live" \ + "$(q "SELECT ttl_column || ' / ' || ttl_interval FROM pgcolumnar.options + WHERE regclass = 'ttl_t'::regclass")" "ts / 3 days" + +# ---- expiring -------------------------------------------------------------- +RETIRED="$(q "SELECT pgcolumnar.expire('ttl_t')")" +ROWS_AFTER="$(q "SELECT count(*) FROM ttl_t")" +GROUPS_AFTER="$(q "SELECT count(*) FROM pgcolumnar.storage s + JOIN pgcolumnar.row_group rg USING (storage_id) + WHERE s.relation_oid = 'ttl_t'::regclass")" +LIVE_AFTER="$(q "SELECT count(*) FROM ttl_t WHERE ts >= ($CUTOFF)")" +echo "-- before: $ROWS_BEFORE rows in $GROUPS_BEFORE groups ($EXPIRED_ROWS past retention)" +echo "-- expire retired $RETIRED group(s): $ROWS_AFTER rows in $GROUPS_AFTER groups" + +check "expire retires at least one group (#403 item 5a)" \ + "$([ "${RETIRED:-0}" -gt 0 ] && echo "retired" || echo "RETIRED NOTHING ($RETIRED)")" "retired" + +check "and the table lost exactly the groups it retired" \ + "$(( GROUPS_BEFORE - RETIRED ))" "$GROUPS_AFTER" + +# THE SAFETY CHECK. Every row still inside the retention must still be there. An +# implementation that dropped a straddling group would fail here and nowhere else. +check "NO row still inside the retention was dropped (#403 item 5a)" "$LIVE_AFTER" "$LIVE_ROWS" + +# Something was actually dropped, so the check above is not passing because the +# function did nothing at all. +check "and the table really is smaller than it was" \ + "$([ "${ROWS_AFTER:-0}" -lt "${ROWS_BEFORE:-0}" ] && echo "smaller" \ + || echo "UNCHANGED ($ROWS_AFTER of $ROWS_BEFORE)")" "smaller" + +check "every remaining row reads back its own value" \ + "$(q "SELECT count(*) FROM ttl_t WHERE v <> 'v'||id")" "0" + +# ---- a second run has nothing left to do ---------------------------------- +AGAIN="$(q "SELECT pgcolumnar.expire('ttl_t')")" +check "running it again retires nothing, because nothing new expired" "$AGAIN" "0" + +# ---- a table with no declared retention ------------------------------------ +psql_run "CREATE TABLE ttl_none (id int, ts timestamptz) USING pgcolumnar;" +psql_run "INSERT INTO ttl_none SELECT g, now() FROM generate_series(1,10) g;" +ERR="$(q "SELECT pgcolumnar.expire('ttl_none')")" +check "a table with no declared retention is an error, not a silent success" \ + "$(grep -qiE 'ERROR|no retention|ttl' <<<"$ERR" && echo "refused" || echo "ACCEPTED ($ERR)")" "refused" + +pgc_summary From 642a2311a3a2ba2602900bbb155fd5f2b65da949 Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Fri, 28 Aug 2026 12:41:44 -0600 Subject: [PATCH 3/4] fix: fingerprint the file inside a transaction (review of #814) CI was red on both majors with a coordinator segfault, and the review was right that the body's "suites pass" was describing my box rather than the code. ## Root cause pcopy_fingerprint_file ran in the coordinator's wait loop, which is OUTSIDE a transaction. On a build configured --with-openssl, pg_cryptohash_create registers the hash context with CurrentResourceOwner, and outside a transaction that pointer is NULL. A build without OpenSSL uses the in-core SHA-2, which touches no resource owner and does not care. /usr/lib/postgresql/18 '--with-openssl' crashes /usr/local/pg18_nc no ssl passes That is the whole of why this passed here and failed on CI, and it is the same class as the socket-directory difference: I verified against a build that lacks a feature the CI build has. ## Finding it The suite reproduced the crash immediately once run against the packaged PostgreSQL rather than my source builds. It still crashed at -O0, so it was deterministic rather than an optimization artifact, and the postmaster leaves no core, so the coordinator was instrumented with elog markers to find the site: PCOPYDBG 1 attached nworkers=4 dedup=0 ... 6 committed (both control loads) PCOPYDBG 1 attached nworkers=4 dedup=1 PCOPYDBG 2 storage row done So it died in the wait loop, not in the post-commit record the review suggested as a starting point. Recording that because the hypothesis was reasonable and the evidence pointed elsewhere. ## The fix and its proof Both fingerprint calls now run inside a transaction. The overlap with the loaders is unchanged: it is still one pass, still in the wait loop, still while the loaders read the same file. Removal proof, mutation asserted applied: with the transaction removed the coordinator segfaults again on the packaged build and the cluster reinitialises. Verified on BOTH build kinds now, which is the check that was missing: packaged PG 18.4 (--with-openssl) 14 checks PASSED source pg18_nc (no ssl) 14 checks PASSED parallel_copy, native_upgrade_converge, entry_point_privilege, harness_selftest and docs_style also pass on the packaged build. The suite header now states the requirement, so the next person to run it only against a source build learns why that is not enough. ## Also from the review The index is renamed load_fingerprint_idx. It is neither a primary key nor unique, the non-uniqueness is a load-bearing decision, and a name ending _pkey made the next reader check. Renamed in both catalog scripts and both C lookups. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UaQ4vThXcmTCf3KRQpUDrE --- pgcolumnar--1.0-alpha2--1.0-alpha3.sql | 2 +- pgcolumnar--1.0-alpha3.sql | 2 +- src/columnar_metadata.c | 4 ++-- src/columnar_parallel_copy.c | 14 ++++++++++++-- test/parallel_copy_dedup.sh | 8 ++++++++ 5 files changed, 24 insertions(+), 6 deletions(-) diff --git a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql index dd98fbbb..dbd05e16 100644 --- a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql +++ b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql @@ -33,7 +33,7 @@ CREATE TABLE pgcolumnar.load_fingerprint ( -- 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_pkey +CREATE INDEX load_fingerprint_idx ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint); -- Two functions gain a parameter, which changes their signatures, so neither can diff --git a/pgcolumnar--1.0-alpha3.sql b/pgcolumnar--1.0-alpha3.sql index d9759506..c961309f 100644 --- a/pgcolumnar--1.0-alpha3.sql +++ b/pgcolumnar--1.0-alpha3.sql @@ -281,7 +281,7 @@ CREATE TABLE pgcolumnar.load_fingerprint ( -- 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_pkey +CREATE INDEX load_fingerprint_idx ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint); /* --------------------------------------------------------------------------- diff --git a/src/columnar_metadata.c b/src/columnar_metadata.c index 2db59d3a..1b87bc46 100644 --- a/src/columnar_metadata.c +++ b/src/columnar_metadata.c @@ -2427,7 +2427,7 @@ PgColumnarInsertBloomRow(const NativeBloomMetadata *b) * Has this exact file already been loaded into this table by a committed * pgcolumnar.parallel_copy (#403 item 7)? * - * Keyed by (relation_oid, fingerprint), which is load_fingerprint_pkey, so + * Keyed by (relation_oid, fingerprint), which is load_fingerprint_idx, so * this is an exact index lookup. The fingerprint is the SHA-256 of the * file's bytes: a file that changed at the same path is a different load. */ @@ -2452,7 +2452,7 @@ PgColumnarLoadFingerprintSeen(Oid relationOid, const uint8 *fingerprint, ScanKeyInit(&key[1], Anum_load_fingerprint_fingerprint, BTEqualStrategyNumber, F_BYTEAEQ, PointerGetDatum(fp)); - idxOid = pgcolumnar_index_oid("load_fingerprint_pkey"); + idxOid = pgcolumnar_index_oid("load_fingerprint_idx"); scan = systable_beginscan(rel, idxOid, OidIsValid(idxOid), snapshot, 2, key); tuple = systable_getnext(scan); found = HeapTupleIsValid(tuple); diff --git a/src/columnar_parallel_copy.c b/src/columnar_parallel_copy.c index 1dc7c811..27cbb3b1 100644 --- a/src/columnar_parallel_copy.c +++ b/src/columnar_parallel_copy.c @@ -1260,10 +1260,19 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) * the pass overlaps the loaders instead of delaying them (#403 * item 7). The decision it feeds is taken below, after every loader * has PREPAREd and before anything commits. + * + * INSIDE A TRANSACTION, which is not optional. On a build with + * --with-openssl, pg_cryptohash_create registers the context with + * CurrentResourceOwner, and outside a transaction that pointer is + * NULL: the coordinator segfaults. A build without OpenSSL uses the + * in-core SHA-2, which touches no resource owner and does not care, + * which is why this passed on one build and crashed on another. */ if (hdr->dedup && !fingerprinted) { + StartTransactionCommand(); pcopy_fingerprint_file(hdr->filename, hdr->fingerprint); + CommitTransactionCommand(); fingerprinted = true; } (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, @@ -1328,13 +1337,14 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) */ if (hdr->dedup) { + StartTransactionCommand(); + PushActiveSnapshot(GetTransactionSnapshot()); + /* inside the transaction, for the resource-owner reason above */ if (!fingerprinted) { pcopy_fingerprint_file(hdr->filename, hdr->fingerprint); fingerprinted = true; } - StartTransactionCommand(); - PushActiveSnapshot(GetTransactionSnapshot()); duplicate = PgColumnarLoadFingerprintSeen(hdr->relid, hdr->fingerprint, PG_SHA256_DIGEST_LENGTH, diff --git a/test/parallel_copy_dedup.sh b/test/parallel_copy_dedup.sh index c0771f85..9fdc4439 100755 --- a/test/parallel_copy_dedup.sh +++ b/test/parallel_copy_dedup.sh @@ -29,6 +29,14 @@ # is the safe direction. The reverse order would leave a fingerprint with no # data, and a later load would be refused for rows that were never stored. # +# RUN THIS AGAINST A BUILD WITH OpenSSL. The coordinator fingerprints the file +# from inside a transaction, and that is not tidiness: on a --with-openssl build +# pg_cryptohash_create registers the hash context with CurrentResourceOwner, +# which is NULL outside a transaction, and the coordinator segfaults. A build +# without OpenSSL uses the in-core SHA-2, touches no resource owner, and passes. +# This suite therefore passed on the local source builds and crashed on CI's +# packaged PostgreSQL until the transaction was added. +# # Usage: test/parallel_copy_dedup.sh [PG_CONFIG] # Written fresh for pgColumnar. From 3b31c8c0b01c92d2e05672e0ebe6e04f146af4f0 Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Fri, 28 Aug 2026 12:46:07 -0600 Subject: [PATCH 4/4] fix: include utils/timestamp.h in columnar_vacuum.c (review of #815) The branch did not build on PostgreSQL 19. columnar_vacuum.c uses TimestampTzGetDatum, GetCurrentTimestamp and IntervalPGetDatum and never included utils/timestamp.h; on 15 through 18 it arrived transitively and 19 stopped carrying it. OK 15.18 / 16.14 / 17.10 / 18.4 / 19beta2 0 warnings each built 5 of 5 That last line is #813's, and it is what makes the preflight worth running: it would previously have printed PASSED for a run that built nothing. I had built this branch with COPT=-Werror on 18 only, which is why -Werror did not save me. The cheap half of the matrix is the part that finds this class, and it takes about a minute. Also merges the #814 fix this branch sits on, so it no longer carries the coordinator that segfaulted. ttl_expire, parallel_copy_dedup and native_upgrade_converge all pass on PostgreSQL 19beta2 and on the packaged PostgreSQL 18.4. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UaQ4vThXcmTCf3KRQpUDrE --- src/columnar_vacuum.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/columnar_vacuum.c b/src/columnar_vacuum.c index cf3819ed..35f52751 100644 --- a/src/columnar_vacuum.c +++ b/src/columnar_vacuum.c @@ -48,6 +48,7 @@ #include "utils/builtins.h" #include "utils/lsyscache.h" #include "utils/relcache.h" +#include "utils/timestamp.h" #include "utils/rls.h" #include "utils/rel.h" #include "utils/snapmgr.h"