diff --git a/CHANGELOG.md b/CHANGELOG.md index c5020ab..461f23a 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/docs/sql-reference.md b/docs/sql-reference.md index a54ef82..37ba96d 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 @@ -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; diff --git a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql b/pgcolumnar--1.0-alpha2--1.0-alpha3.sql index 7827734..4f80f4b 100644 --- a/pgcolumnar--1.0-alpha2--1.0-alpha3.sql +++ b/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, diff --git a/pgcolumnar--1.0-alpha3.sql b/pgcolumnar--1.0-alpha3.sql index 758de65..70a5220 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 ); /* @@ -258,6 +262,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_idx + ON pgcolumnar.load_fingerprint USING btree (relation_oid, fingerprint); + /* --------------------------------------------------------------------------- * pgcolumnar.free_space (Phase F physical reclaim) * @@ -367,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$ @@ -484,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), @@ -499,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( @@ -823,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 @@ -1205,13 +1251,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 23d0379..0736c2d 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); @@ -645,6 +648,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 383cf30..8ef2d6c 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) */ @@ -109,6 +111,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 +2424,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_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. + */ +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_idx"); + 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 @@ -3106,6 +3193,63 @@ pgcolumnar_written_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_parallel_copy.c b/src/columnar_parallel_copy.c index c11c8e2..27cbb3b 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,27 @@ 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. + * + * 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, 1000L, PG_WAIT_EXTENSION); ResetLatch(MyLatch); @@ -1194,6 +1314,56 @@ 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) + { + StartTransactionCommand(); + PushActiveSnapshot(GetTransactionSnapshot()); + /* inside the transaction, for the resource-owner reason above */ + if (!fingerprinted) + { + pcopy_fingerprint_file(hdr->filename, hdr->fingerprint); + fingerprinted = true; + } + 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 +1377,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 +1476,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 +1499,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 +1632,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 +1689,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/src/columnar_vacuum.c b/src/columnar_vacuum.c index c129dd1..35f5275 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" @@ -60,6 +61,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 +2020,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/parallel_copy_dedup.sh b/test/parallel_copy_dedup.sh new file mode 100755 index 0000000..9fdc443 --- /dev/null +++ b/test/parallel_copy_dedup.sh @@ -0,0 +1,118 @@ +#!/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. +# +# 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. + +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 1fe6004..f21305d 100644 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -213,6 +213,7 @@ SUITES=( objstore_userinfo parallel parallel_copy + parallel_copy_dedup parallel_degree parallel_export_parquet parallel_flush_optin @@ -255,6 +256,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 0000000..271f81f --- /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