diff --git a/CHANGELOG.md b/CHANGELOG.md index c0dffb95f..8d0ef3c54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -400,6 +400,17 @@ true until the next version shipped. Reported by @OffgridwithJD. ### Fixed +- **A NOLOGIN role can run `parallel_export_parquet`, `parallel_copy` and the + parallel flush on PostgreSQL 15 and 16.** After `SET ROLE` to a role without + LOGIN, the export failed `worker 0 failed: worker exited without reporting a + result`, `parallel_copy` failed `coordinator exited without reporting a result`, + and the parallel flush encoded every column serially. Each background worker + connected as the calling role, which these majors refuse for a role that cannot + log in. A worker now connects as the session's authenticated user and then takes + the caller's user id and security context, as PostgreSQL's own parallel workers + do, so triggers, privilege checks and `current_user` inside it still see the + caller. PostgreSQL 17 and later already worked (#1585). + `test/nologin_workers.sh` and `test_nologin_workers.py`. - **In the upgrade window, a refusal for a missing catalog table names `ALTER EXTENSION UPDATE`.** After the new library is installed and the server restarted, but before `ALTER EXTENSION pgcolumnar UPDATE`, `DROP TABLE`, `TRUNCATE` and diff --git a/src/columnar.h b/src/columnar.h index aea068ac8..7f36a420f 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -1273,4 +1273,58 @@ PgColumnarVarSizeAnyUnalignedBounded(const char *p, const char *end) return len; } +/* + * Connect a background worker that does a caller's work, and act as that caller + * (#1585, #1612). The leader records the caller as userid, with its superuser flag + * and security context (GetUserIdAndSecContext), and GetAuthenticatedUserId() as + * authid. + * + * THE WORKER'S SESSION USER IS THE CALLER on every major, so a RESET ROLE inside the + * worker (a copy trigger can run one) lands on the caller and not on some other user. + * On 17 and later the worker connects as the caller, with the login check bypassed, + * as it always did. Before 17 there is no such bypass and a NOLOGIN caller cannot be + * connected as, so the worker connects as the authenticated user, which logged in, and + * then makes the caller its session user. Connecting as the authenticated user and + * stopping there made it the session user: a trigger's RESET ROLE then became that + * user, the copy PREPAREd as it, and the leader could not finish the transaction + * (#1612 review). + * + * THE SECURITY CONTEXT TRAVELS TOO, as core's ParallelWorkerMain carries it, so a call + * made inside a SECURITY DEFINER function or a security-restricted operation keeps its + * restrictions in the worker: without it, a trigger there could switch to the session + * user (#1612 review). It is taken inside the transaction; see + * columnar_worker_take_context below. + */ +#include "miscadmin.h" +#include "postmaster/bgworker.h" + +static inline void +columnar_worker_connect_as(Oid dbid, Oid authid, Oid userid, bool userIsSuperuser) +{ +#if PG_VERSION_NUM >= 170000 + (void) authid; + (void) userIsSuperuser; + BackgroundWorkerInitializeConnectionByOid(dbid, userid, + BGWORKER_BYPASS_ALLOWCONN | + BGWORKER_BYPASS_ROLELOGINCHECK); +#else + BackgroundWorkerInitializeConnectionByOid(dbid, authid, BGWORKER_BYPASS_ALLOWCONN); + SetSessionAuthorization(userid, userIsSuperuser); /* the current user too */ +#endif +} + +/* + * Take the caller's security context, INSIDE the worker's transaction. A top-level + * transaction must start with none -- StartTransaction asserts it, which is why core's + * ParallelWorkerMain sets it after its transaction begins -- and commit or abort + * restores the transaction's starting state. Set before the first transaction, an + * assert-enabled build aborted the coordinator (#1612 review). Called by the workers + * that run user code: the copy loader (triggers) and the export worker. + */ +static inline void +columnar_worker_take_context(Oid userid, int sec_context) +{ + SetUserIdAndSecContext(userid, sec_context); +} + #endif /* PGCOLUMNAR_H */ diff --git a/src/columnar_parallel_copy.c b/src/columnar_parallel_copy.c index 27cbb3b11..25e559efe 100644 --- a/src/columnar_parallel_copy.c +++ b/src/columnar_parallel_copy.c @@ -388,7 +388,10 @@ typedef struct PcopyWorkerSlot typedef struct PcopyHeader { Oid dbid; - Oid roleid; + Oid roleid; /* the caller's user id, with secContext */ + int secContext; + Oid authid; /* the authenticated user, which workers connect as below 17 (#1585) */ + bool roleIsSuperuser; /* roleid's superuser flag, for the worker's session user */ Oid relid; /* target relation (partitioned parent, or a single columnar table) */ int nworkers; bool single_table; /* target is one columnar table (not partitioned): @@ -816,7 +819,6 @@ pgcolumnar_parallel_copy_worker(Datum main_arg) PcopyWorkerSlot *slots; PcopyWorkerSlot *me; int widx; - uint32 conn_flags = BGWORKER_BYPASS_ALLOWCONN; memcpy(&widx, MyBgworkerEntry->bgw_extra, sizeof(int)); @@ -837,10 +839,8 @@ pgcolumnar_parallel_copy_worker(Datum main_arg) slots = (PcopyWorkerSlot *) shm_toc_lookup(toc, PCOPY_KEY_WORKERS, false); me = &slots[widx]; -#if PG_VERSION_NUM >= 170000 - conn_flags |= BGWORKER_BYPASS_ROLELOGINCHECK; -#endif - BackgroundWorkerInitializeConnectionByOid(hdr->dbid, hdr->roleid, conn_flags); + columnar_worker_connect_as(hdr->dbid, hdr->authid, hdr->roleid, + hdr->roleIsSuperuser); /* * Single-table load: this loader writes the one shared storage concurrently @@ -870,6 +870,8 @@ pgcolumnar_parallel_copy_worker(Datum main_arg) StartTransactionCommand(); BeginTransactionBlock(); CommitTransactionCommand(); /* TBLOCK_BEGIN -> in progress */ + /* the caller's security context, now that a transaction is open (#1612) */ + columnar_worker_take_context(hdr->roleid, hdr->secContext); fd = pcopy_open_regular_file(hdr->filename, NULL); if (lseek(fd, (off_t) me->start_off, SEEK_SET) < 0) @@ -951,6 +953,27 @@ pgcolumnar_parallel_copy_worker(Datum main_arg) * loader holds no transaction, so writing the result slot is a plain * shared-memory store. */ + /* + * PREPARE only as the caller (#1612 review). A trigger can change role + * inside this transaction (SET ROLE to a role the caller belongs to, or a + * transaction-local set_config('role')), and a transaction prepared as + * another user is one the coordinator, running as the caller, cannot + * finish: it would stay prepared, holding its locks, until someone ran + * ROLLBACK PREPARED. Refused here it aborts like any other error, and the + * coordinator rolls the load back. + */ + { + Oid nowUser; + int nowContext; + + GetUserIdAndSecContext(&nowUser, &nowContext); + if (nowUser != hdr->roleid || nowContext != hdr->secContext) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("pgcolumnar parallel_copy worker is no longer running as the role that called parallel_copy"), + errdetail("A trigger changed the role during the load, so its transaction cannot be prepared for the caller to finish."))); + } + if (!PrepareTransactionBlock(me->gid)) ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), @@ -1142,7 +1165,6 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) PcopyHeader *hdr; PcopyWorkerSlot *slots; int nworkers; - uint32 conn_flags = BGWORKER_BYPASS_ALLOWCONN; BackgroundWorker bw; BackgroundWorkerHandle **handles; int i; @@ -1165,10 +1187,8 @@ pgcolumnar_parallel_copy_coordinator(Datum main_arg) slots = (PcopyWorkerSlot *) shm_toc_lookup(toc, PCOPY_KEY_WORKERS, false); nworkers = hdr->nworkers; -#if PG_VERSION_NUM >= 170000 - conn_flags |= BGWORKER_BYPASS_ROLELOGINCHECK; -#endif - BackgroundWorkerInitializeConnectionByOid(hdr->dbid, hdr->roleid, conn_flags); + columnar_worker_connect_as(hdr->dbid, hdr->authid, hdr->roleid, + hdr->roleIsSuperuser); /* * Single-table load: pre-create and COMMIT the storage catalog row before any @@ -1623,7 +1643,9 @@ pgcolumnar_parallel_copy(PG_FUNCTION_ARGS) dsmh = dsm_segment_handle(seg); hdr->dbid = MyDatabaseId; - hdr->roleid = GetUserId(); + GetUserIdAndSecContext(&hdr->roleid, &hdr->secContext); + hdr->authid = GetAuthenticatedUserId(); + hdr->roleIsSuperuser = superuser_arg(hdr->roleid); hdr->relid = relid; hdr->nworkers = workers; hdr->single_table = single_table; diff --git a/src/columnar_parallel_export.c b/src/columnar_parallel_export.c index 73ea1b1fc..b2c637eea 100644 --- a/src/columnar_parallel_export.c +++ b/src/columnar_parallel_export.c @@ -94,7 +94,10 @@ typedef struct PexportWorkerSlot typedef struct PexportHeader { Oid dbid; - Oid roleid; + Oid roleid; /* the caller's user id, with secContext */ + int secContext; + Oid authid; /* the authenticated user, which workers connect as below 17 (#1585) */ + bool roleIsSuperuser; /* roleid's superuser flag, for the worker's session user */ Oid relid; /* single columnar table, or partitioned parent */ uint64 storageId; /* single-table storage id (0 when partitioned) */ int nworkers; @@ -452,7 +455,6 @@ pgcolumnar_parallel_export_worker(Datum main_arg) Oid *leaves; PexportWorkerSlot *me; int widx; - uint32 conn_flags = BGWORKER_BYPASS_ALLOWCONN; Snapshot snap; memcpy(&widx, MyBgworkerEntry->bgw_extra, sizeof(int)); @@ -475,10 +477,8 @@ pgcolumnar_parallel_export_worker(Datum main_arg) leaves = (Oid *) shm_toc_lookup(toc, PEXPORT_KEY_LEAVES, false); me = &slots[widx]; -#if PG_VERSION_NUM >= 170000 - conn_flags |= BGWORKER_BYPASS_ROLELOGINCHECK; -#endif - BackgroundWorkerInitializeConnectionByOid(hdr->dbid, hdr->roleid, conn_flags); + columnar_worker_connect_as(hdr->dbid, hdr->authid, hdr->roleid, + hdr->roleIsSuperuser); /* * Adopt the dispatcher's exact snapshot. ImportSnapshot requires a @@ -486,6 +486,8 @@ pgcolumnar_parallel_export_worker(Datum main_arg) * exactly as SET TRANSACTION SNAPSHOT does. */ StartTransactionCommand(); + /* the caller's security context, now that a transaction is open (#1612) */ + columnar_worker_take_context(hdr->roleid, hdr->secContext); XactIsoLevel = XACT_REPEATABLE_READ; ImportSnapshot(hdr->snapname); snap = GetTransactionSnapshot(); @@ -759,7 +761,9 @@ pgcolumnar_parallel_export_parquet(PG_FUNCTION_ARGS) dsmh = dsm_segment_handle(seg); hdr->dbid = MyDatabaseId; - hdr->roleid = GetUserId(); + GetUserIdAndSecContext(&hdr->roleid, &hdr->secContext); + hdr->authid = GetAuthenticatedUserId(); + hdr->roleIsSuperuser = superuser_arg(hdr->roleid); hdr->relid = relid; hdr->storageId = storageId; hdr->nworkers = workers; diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index d9f72d973..001d7091a 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -101,7 +101,10 @@ typedef struct PflushWorkerSlot typedef struct PflushHeader { Oid dbid; - Oid roleid; + Oid roleid; /* the caller's user id, with secContext */ + int secContext; + Oid authid; /* the authenticated user, which workers connect as below 17 (#1585) */ + bool roleIsSuperuser; /* roleid's superuser flag, for the worker's session user */ Oid relid; uint64 storageId; uint64 groupNumber; @@ -2440,7 +2443,6 @@ pgcolumnar_parallel_flush_worker(Datum main_arg) pg_atomic_uint32 *claim; PflushWorkerSlot *me; int widx; - uint32 conn_flags = BGWORKER_BYPASS_ALLOWCONN; memcpy(&widx, MyBgworkerEntry->bgw_extra, sizeof(int)); @@ -2465,10 +2467,8 @@ pgcolumnar_parallel_flush_worker(Datum main_arg) claim = (pg_atomic_uint32 *) shm_toc_lookup(toc, PFLUSH_KEY_CLAIM, false); me = &slots[widx]; -#if PG_VERSION_NUM >= 170000 - conn_flags |= BGWORKER_BYPASS_ROLELOGINCHECK; -#endif - BackgroundWorkerInitializeConnectionByOid(hdr->dbid, hdr->roleid, conn_flags); + columnar_worker_connect_as(hdr->dbid, hdr->authid, hdr->roleid, + hdr->roleIsSuperuser); /* * A fresh snapshot for catalog/tupdesc reads only. A plain @@ -2770,7 +2770,9 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, pg_atomic_init_u32(claim, 0); hdr->dbid = MyDatabaseId; - hdr->roleid = GetUserId(); + GetUserIdAndSecContext(&hdr->roleid, &hdr->secContext); + hdr->authid = GetAuthenticatedUserId(); + hdr->roleIsSuperuser = superuser_arg(hdr->roleid); hdr->relid = writeState->relid; hdr->storageId = writeState->storageId; hdr->groupNumber = groupNumber; diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index f0e7bbd10..7e688efd0 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -3789,6 +3789,21 @@ native_writer native_writer native row groups (5000 rows / 1000 limit) 15;16;17; native_writer native_writer native row-group row total 15;16;17;18;19 never - native_writer native_writer native storage row written 15;16;17;18;19 never - native_writer native_writer native vector length recorded 15;16;17;18;19 2026-10-16 columnar.h: COLUMNAR_NATIVE_VECTOR_LENGTH 2048 +nologin_workers nologin_workers a NOLOGIN role's parallel flush encodes every column in a worker 15;16;17;18;19 2026-10-21 the workers connect as the calling role, as before #1585, so PG15 and PG16 refuse a NOLOGIN caller +nologin_workers nologin_workers a NOLOGIN role's parallel_copy loads every row 15;16;17;18;19 2026-10-21 the workers connect as the calling role, as before #1585, so PG15 and PG16 refuse a NOLOGIN caller +nologin_workers nologin_workers a NOLOGIN role's parallel_export_parquet exports every row 15;16;17;18;19 2026-10-21 the workers connect as the calling role, as before #1585, so PG15 and PG16 refuse a NOLOGIN caller +nologin_workers nologin_workers a copy run by a SECURITY DEFINER function keeps the definer's restriction in its workers 15;16;17;18;19 2026-10-21 columnar.h: the worker takes the caller's user id without its security context +nologin_workers nologin_workers a copy trigger that resets the role still writes every row as the NOLOGIN role 15;16;17;18;19 2026-10-21 columnar.h: the worker connects as the authenticated user and never makes the caller its session user +nologin_workers nologin_workers a copy trigger that switches to another role is refused before the worker prepares 15;16;17;18;19 2026-10-21 columnar_parallel_copy.c: the loader prepares whatever role a trigger left it as +nologin_workers nologin_workers and that copy leaves no prepared transaction behind 15;16;17;18;19 never - +nologin_workers nologin_workers and the refused copy leaves no prepared transaction behind 15;16;17;18;19 2026-10-21 columnar_parallel_copy.c: the loader prepares whatever role a trigger left it as +nologin_workers nologin_workers and the table holds every row it flushed 15;16;17;18;19 never - +nologin_workers nologin_workers no worker was refused a login as the NOLOGIN role 15;16;17;18;19 2026-10-21 the workers connect as the calling role, as before #1585, so PG15 and PG16 refuse a NOLOGIN caller +nologin_workers nologin_workers parallel_copy's workers wrote every row as the NOLOGIN role 15;16;17;18;19 2026-10-21 columnar.h: the worker connects as the authenticated user and never takes the caller's user id;the workers connect as the calling role, as before #1585, so PG15 and PG16 refuse a NOLOGIN caller +nologin_workers nologin_workers premise: SET ROLE to the NOLOGIN role takes 15;16;17;18;19 never - +nologin_workers nologin_workers premise: the NOLOGIN role cannot open a session 15;16;17;18;19 never - +nologin_workers nologin_workers premise: the server log records the NOLOGIN role's refused login 15;16;17;18;19 never - +nologin_workers nologin_workers the flush is dispatched to workers 15;16;17;18;19 never - objstore_allowlist objstore_allowlist a multi-entry list matches any entry 15;16;17;18;19 never - objstore_allowlist objstore_allowlist a non-superuser cannot widen the list (42501) 15;16;17;18;19 2026-10-14 objstore_allowed_endpoints defined PGC_SIGHUP, so no session may SET it objstore_allowlist objstore_allowlist a superuser session SET works (admin override) 15;16;17;18;19 2026-10-14 objstore_allowed_endpoints defined PGC_SIGHUP, so no session may SET it @@ -4290,6 +4305,63 @@ range_pruning range_pruning premise: the range type collates differently from it range_pruning range_pruning so does a containment probe 15;16;17;18;19 never - range_pruning range_pruning the overlap qual is pushed down, where it used to be dropped 15;16;17;18;19 never - range_pruning range_pruning the same for a containment probe far to its right 15;16;17;18;19 2026-10-17 columnar_write_state.c: the writer ignores an unbounded range, so its unit records only the largest finite upper bound +reader_buffer_reuse reader_buffer_reuse a PROJECTED scan (one column of two) matches too 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse a column ADDED after the groups exist reads back NULL, as on heap 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse a rescan (nested loop) returns the same rows 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse a short group after a full one reads correctly in a whole-row read 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, the whole-group read stops 64 bytes short, so the group's last 64 bytes are the previous group's +reader_buffer_reuse reader_buffer_reuse a short group after a full one reads correctly with column projection off 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, the whole-group read stops 64 bytes short, so the group's last 64 bytes are the previous group's +reader_buffer_reuse reader_buffer_reuse a short group after a full one still reads correctly 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse a whole-row read (a whole-group read) matches a heap mirror 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, the whole-group read stops 64 bytes short, so the group's last 64 bytes are the previous group's +reader_buffer_reuse reader_buffer_reuse an all-NULL column beside a wide one matches 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse premise: that really added a group smaller than the ones before it 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse premise: the all-groups arm really reads every row group 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse premise: the differential fixture spans several row groups 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse premise: the fixture has several row groups, or reuse cannot matter 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse premise: the instrument reports a nonzero fault count at all 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse premise: the one-group arm really reads exactly one row group 15;16;17;18;19 never - +reader_buffer_reuse reader_buffer_reuse projected subset [b] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [i] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [i||':'||t] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [i||b::text||t||u] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [s] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [s||':'||u] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [t] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse projected subset [u] matches the heap twin 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse reading every group does not cost per-group faults (ratio well under the group count) 15;16;17;18;19 2026-10-20 the reader allocates a fresh group buffer for every row group, the pre-reuse behaviour +reader_buffer_reuse reader_buffer_reuse reads after DROP COLUMN match 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse the full scan matches a heap mirror exactly 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes +reader_buffer_reuse reader_buffer_reuse the full scan with column projection off (a whole-group read) matches a heap mirror 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, the whole-group read stops 64 bytes short, so the group's last 64 bytes are the previous group's +recluster_extent recluster_extent a group numbered below the run is not counted as ordered 15;16;17;18;19 2026-10-20 the online recluster records no extent at all;the run claims every group, below and above the rewrite (#342) +recluster_extent recluster_extent an uncontended rewrite claims the whole relation 15;16;17;18;19 2026-10-20 the online recluster records no extent at all +recluster_extent recluster_extent an uncontended rewrite counts every row as sorted 15;16;17;18;19 2026-10-20 the online recluster records no extent at all +recluster_extent recluster_extent every row is present afterwards 15;16;17;18;19 never - +recluster_extent recluster_extent noproj: a full recluster claims every row 15;16;17;18;19 2026-10-20 the online recluster records no extent at all +recluster_extent recluster_extent noproj: a full recluster leaves nothing appended 15;16;17;18;19 2026-10-20 the online recluster records no extent at all +recluster_extent recluster_extent premise: the concurrent insert was not in the rewrite's work set 15;16;17;18;19 never - +recluster_extent recluster_extent the concurrent rows are reported as appended 15;16;17;18;19 2026-10-20 the run claims every group, below and above the rewrite (#342) +recluster_extent recluster_extent the counts still cover every stored row 15;16;17;18;19 never - +recluster_extent recluster_extent the id-drawn-below counts still cover every stored row 15;16;17;18;19 never - +recluster_extent recluster_extent the insert ran while the rewrite was still running 15;16;17;18;19 never - +recluster_extent recluster_extent the rewrite reported groups reclustered 15;16;17;18;19 never - +recluster_extent recluster_extent the rows written below the run are reported as appended 15;16;17;18;19 2026-10-20 the online recluster records no extent at all;the run claims every group, below and above the rewrite (#342) +recluster_extent recluster_extent the run does not claim rows written during the rewrite 15;16;17;18;19 2026-10-20 the online recluster records no extent at all;the run claims every group, below and above the rewrite (#342);the run reaches the rewrite's last id, across ids another writer took (#311) +recluster_extent recluster_extent the table loaded 15;16;17;18;19 never - +recluster_extent recluster_extent withproj: a full recluster claims every row 15;16;17;18;19 2026-10-20 a projection's stripe ids are not merged into the run, so they break it;the online recluster records no extent at all +recluster_extent recluster_extent withproj: a full recluster leaves nothing appended 15;16;17;18;19 2026-10-20 a projection's stripe ids are not merged into the run, so they break it;the online recluster records no extent at all +recluster_extent recluster_extent withproj: add_projection created the projection 15;16;17;18;19 2026-10-21 columnar_projection.c: pgcolumnar_add_projection raises right after reading its arguments, so no projection is created +recluster_gate recluster_gate a repeat recluster by the new key is a no-op 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded;recluster's #415 self-gate removed, so every call rewrites +recluster_gate recluster_gate a third identical recluster is also a no-op 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded;recluster's #415 self-gate removed, so every call rewrites +recluster_gate recluster_gate and it folds the new data back into the run (0 appended after) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing;columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded;the gate ignores groups appended past the recorded run +recluster_gate recluster_gate and it leaves the physical layout byte-identical (group offsets unmoved) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded;recluster's #415 self-gate removed, so every call rewrites +recluster_gate recluster_gate and sort_status now reports the new key 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing;columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded;the gate ignores the key, so a new key reads as already clustered +recluster_gate recluster_gate no-op recluster preserved the exact row set (columnar == heap mirror) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster_online skips the first sorted row it writes back +recluster_gate recluster_gate premise: the insert appended unsorted groups 15;16;17;18;19 never - +recluster_gate recluster_gate recluster after appends does real work again (>0) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing;the gate ignores groups appended past the recorded run +recluster_gate recluster_gate recluster by a NEW key is NOT a no-op -- it re-clusters (>0) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing;the gate ignores the key, so a new key reads as already clustered +recluster_gate recluster_gate recluster by the same key on an unchanged table reclusters 0 groups 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded;recluster's #415 self-gate removed, so every call rewrites +recluster_gate recluster_gate sort_status now reports the clustering key (was NULL before #415) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing;columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded +recluster_gate recluster_gate the first recluster does real work (>0 groups) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing +recluster_gate recluster_gate the run covers the whole relation (0 appended groups) 15;16;17;18;19 2026-10-21 columnar_vacuum.c: recluster's #415 gate over-applied (if (true || sameKey ...)), so every recluster returns 0 and does nothing;columnar_vacuum.c: recluster_online skips record_online_sorted_extent, so no extent or key is recorded recovery recovery premise: s2's INSERT ran in its open transaction before the crash 15;16;17;18;19 never - recovery recovery s1 aggregate 15;16;17;18;19 2026-10-16 columnar_storage.c: a stripe data page written with no WAL record recovery recovery s1 col count 15;16;17;18;19 never - diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index 92c1f333c..08db60974 100644 --- a/test/check_ledger_budget.txt +++ b/test/check_ledger_budget.txt @@ -15,7 +15,7 @@ # # Adding a check to a suite that is already covered does not move it, which is # what makes it safe to bound. -suites_not_covered 125 +suites_not_covered 122 # # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become # one. @@ -4963,4 +4963,87 @@ suites_not_covered 125 # never 3648 + 2 = 3650 # dated 1925 + 5 = 1930 # 3650 + 1930 == 5580 -checks_never_observed_red 3650 +# +# #1341, recluster_gate seeded, with its twin (jdatcmd): 3505 -> 3506, and +# suites_not_covered 149 -> 148. The suite had no rows. It enters with 13 checks from +# five-major runs (13/13 on each of 15-19, shell and twin alike), after two fixes. +# The row-set check compared rg_t with a heap copied from rg_t, so it agreed with +# whatever the reclusters did: with a row corrupted before the copy, it passed. The +# mirror is now built from the generator. And the layout comparison agreed on two +# failed reads; it now needs both. Three checks lost their premise: prefix under +# #1339 (OffgridwithJD's review): the first recluster's return, sort_status's +# recorded key (#415) and the recorded extent (#311) are recluster's own behaviour, +# and a one-line change reddens each. 12 are dated by six mutations, each reddening +# the same names in both twins: the gate removed, appended groups ignored, the key +# ignored, no extent or key recorded, the gate over-applied so every call does +# nothing, and the first sorted row dropped on write-back. 1 arrives never: the +# premise that the insert appended unsorted groups, green under the gate removed, no +# record, the gate over-applied and the dropped row. +# +# ledger rows 5580 + 13 = 5593 +# never 3650 + 1 = 3651 +# dated 1930 + 12 = 1942 +# 3651 + 1942 == 5593 +# +# #1341, recluster_extent seeded, with its twin (jdatcmd): 3503 -> 3510, and +# suites_not_covered 149 -> 148. The suite had no rows. It enters with 18 checks from +# five-major runs (18/18 on each of 15-19, shell and twin alike), after three fixes +# and one added check. +# "the run does not claim rows written during the rewrite" read sorted_rows <= base, +# which a run that recorded nothing meets: with no extent recorded it passed. It now +# needs at least one sorted row. The race check compared two client clocks, but the +# rewrite ends 3-4 ms after the insert commits, so the order flipped between twins; it +# now asks the server whether the recluster is running when the insert begins, and is +# renamed "the insert ran while the rewrite was still running". Both leave +# lossy_arms.tsv, and the projection loop's names are literal. The #345 arm passed +# when its projection was never created, because add_projection's error only reached +# stderr (OffgridwithJD's review): with add_projection failing and #345's merge +# skipped, all 17 passed. "withproj: add_projection created the projection" now reads +# the projection back from the catalog. 11 are dated by five mutations, each reddening +# the same names in both twins: four of the extent recorder, and add_projection +# raising. 7 arrive never. +# +# ledger rows 5593 + 18 = 5611 +# never 3651 + 7 = 3658 +# dated 1942 + 11 = 1953 +# 3658 + 1953 == 5611 +# +# #1341, reader_buffer_reuse seeded, with its twin (jdatcmd): 3503 -> 3511, and +# suites_not_covered 149 -> 148. The suite had no rows. It enters with 26 checks from +# five-major runs (26/26 on each of 15-19, shell and twin alike), after four fixes. +# Both heap mirrors were copied from the tables under test, so every differential check +# agreed with whatever the reader returned: with a row corrupted before the copy, the +# suite passed 22 of 22. They are now built from the generators, and the three +# full-scan comparisons refuse an empty side. Two checks leave lossy_arms.tsv, and the +# projection loop's names are literal. Every read named its columns, so the +# whole-group read into the reused buffer was never compared: with that read stopping +# 64 bytes short on a reused buffer, the suite passed 22 of 22. Four checks are new, a +# read with column projection off and a whole-row read, each before and after the +# short group, against the mirror. 18 are dated on PG17, in both twins with the same +# names: 1 by a fresh buffer per group, the 4 new ones by that short whole-group read, +# and 13 by a projected read that does not refresh a reused buffer, which reddens no +# premise. 8 arrive never, 6 of them premises. +# +# ledger rows 5611 + 26 = 5637 +# never 3658 + 8 = 3666 +# dated 1953 + 18 = 1971 +# 3666 + 1971 == 5637 +# +# #1585, nologin_workers seeded, with its twin (jdatcmd): 3666 -> 3672. A new suite, so +# suites_not_covered does not move: it is registered and covered in the same change. It +# enters with 15 checks from five-major runs (all green on each of 15-19, shell and twin +# alike). 9 are dated by five mutations, each reddening the same names in both twins: +# the workers connecting as the calling role, as before #1585 (on PG16, reddening the +# export, the copy, the copy's writer, the flush's split and the refused-login count); a +# worker that never takes the caller's user id (its copy rows read postgres); one that +# connects as the authenticated user and never makes the caller its session user (a +# copy trigger's RESET ROLE then writes as another user); one that drops the caller's +# security context (a SECURITY DEFINER copy's trigger escapes its restriction); and a +# loader that prepares whatever role a trigger left it as (a role-switching trigger is +# not refused, and its transaction is left prepared). 6 arrive never. +# +# ledger rows 5637 + 15 = 5652 +# never 3666 + 6 = 3672 +# dated 1971 + 9 = 1980 +# 3672 + 1980 == 5652 +checks_never_observed_red 3672 diff --git a/test/lossy_arms.tsv b/test/lossy_arms.tsv index c68384a9f..746e11882 100644 --- a/test/lossy_arms.tsv +++ b/test/lossy_arms.tsv @@ -149,11 +149,7 @@ projection_scan_cost premise: the prunable range is priced above the one-stripe projection_scan_cost while a plain range on the sort key still earns its discount projections fp2 spans multiple projection row groups pushdown_report the executor filters far more rows with pushdown off -reader_buffer_reuse premise: the differential fixture spans several row groups -reader_buffer_reuse reading every group does not cost per-group faults (ratio well under the group count) read_stream multiple stripes present -recluster_extent the insert committed while the rewrite was still running -recluster_extent the run does not claim rows written during the rewrite scan_decode_cost a wide projection costs materially more than a narrow one (decode is priced) scan_decode_cost four wide text columns cost more than eight narrow int ones (#768) truncate_cleanup PREMISE the projection has its own storage, distinct from the base diff --git a/test/nologin_workers.sh b/test/nologin_workers.sh new file mode 100755 index 000000000..9eddc2ce1 --- /dev/null +++ b/test/nologin_workers.sh @@ -0,0 +1,145 @@ +#!/usr/bin/env bash +# +# pgColumnar: a NOLOGIN role's background workers connect and act as that role (#1585). +# +# parallel_export_parquet, parallel_copy and the parallel flush each start background +# workers for the calling role. A worker used to connect AS that role, and before +# PostgreSQL 17 a background worker cannot connect as a role without LOGIN: there is no +# BGWORKER_BYPASS_ROLELOGINCHECK. So after SET ROLE to a NOLOGIN group role, on PG15 and +# PG16, the export failed "worker 0 failed: worker exited without reporting a result", +# parallel_copy failed "coordinator exited without reporting a result", and the parallel +# flush silently encoded every column serially. The server log said why: +# 'role "..." is not permitted to log in'. +# +# A worker now connects as the session's authenticated user, which logged in, and then +# takes the caller's user id, as core's parallel workers do. The second half matters as +# much as the first: a worker that connected as the authenticated user and stopped there +# would run the caller's COPY as that user, typically a superuser. The row trigger on +# nw_dst records current_user inside the copy worker, which is what catches that. +# +# Usage: test/nologin_workers.sh [PG_CONFIG] +# Written fresh for pgColumnar. + +set -uo pipefail +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +# parallel_copy prepares one transaction per worker. +export PGC_EXTRA_CONF=$'max_prepared_transactions=4\nmax_worker_processes=16' +pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" + +ROWS=2000 +outdir="$PGC_WORKDIR/nw_dir" +infile="$PGC_WORKDIR/nw_in.txt" +seq 1 "$ROWS" | sed 's/$/\t/' >"$infile" +chmod 644 "$infile" + +psql_run "DROP ROLE IF EXISTS t_nl;" +psql_run "DROP ROLE IF EXISTS t_nl_other;" +psql_run "CREATE ROLE t_nl NOSUPERUSER NOLOGIN;" +psql_run "GRANT USAGE ON SCHEMA pgcolumnar TO t_nl;" +psql_run "GRANT pg_read_server_files, pg_write_server_files TO t_nl;" +psql_run "CREATE TABLE nw_src (a int, b text) USING pgcolumnar;" +psql_run "INSERT INTO nw_src SELECT g, 'x' || g FROM generate_series(1, $ROWS) g;" +# The trigger runs inside the copy worker, so it records the worker's current_user. +psql_run "CREATE TABLE nw_dst (a int, who text) USING pgcolumnar;" +psql_run "CREATE FUNCTION nw_stamp() RETURNS trigger LANGUAGE plpgsql + AS \$\$BEGIN NEW.who := current_user; RETURN NEW; END\$\$;" +psql_run "CREATE TRIGGER nw_stamp BEFORE INSERT ON nw_dst FOR EACH ROW EXECUTE FUNCTION nw_stamp();" +psql_run "CREATE TABLE nw_wide (a int, b int, c int) USING pgcolumnar;" +# A trigger that resets the role inside the copy worker (#1612 review). The worker must +# not end up as some other user: its rows would be written as that user, and its +# prepared transaction would belong to a user the leader cannot finish it as. +psql_run "CREATE TABLE nw_esc (a int, who text) USING pgcolumnar;" +psql_run "CREATE FUNCTION nw_reset_stamp() RETURNS trigger LANGUAGE plpgsql + AS \$\$BEGIN + BEGIN PERFORM set_config('role', 'none', true); + EXCEPTION WHEN OTHERS THEN NULL; END; + NEW.who := current_user; RETURN NEW; + END\$\$;" +psql_run "CREATE TRIGGER nw_reset_stamp BEFORE INSERT ON nw_esc FOR EACH ROW EXECUTE FUNCTION nw_reset_stamp();" +# A trigger that switches to ANOTHER role the caller belongs to: the worker would then +# PREPARE as that role, which the coordinator, running as the caller, cannot finish. +psql_run "CREATE ROLE t_nl_other NOSUPERUSER NOLOGIN;" +psql_run "GRANT t_nl_other TO t_nl;" +psql_run "CREATE TABLE nw_sw (a int, who text) USING pgcolumnar;" +psql_run "CREATE FUNCTION nw_switch_stamp() RETURNS trigger LANGUAGE plpgsql + AS \$\$BEGIN PERFORM set_config('role', 't_nl_other', true); + NEW.who := current_user; RETURN NEW; END\$\$;" +psql_run "CREATE TRIGGER nw_switch_stamp BEFORE INSERT ON nw_sw FOR EACH ROW EXECUTE FUNCTION nw_switch_stamp();" +psql_run "GRANT SELECT, INSERT ON nw_sw TO t_nl_other;" +psql_run "CREATE TABLE nw_def (a int, who text) USING pgcolumnar;" +psql_run "CREATE TRIGGER nw_def_stamp BEFORE INSERT ON nw_def FOR EACH ROW EXECUTE FUNCTION nw_reset_stamp();" +# The definer: a superuser session calls it, and inside it the caller is t_nl under a +# security-restricted context, which must hold in the workers too. +psql_run "CREATE FUNCTION nw_definer_copy(p text) RETURNS bigint LANGUAGE sql SECURITY DEFINER + AS \$\$SELECT pgcolumnar.parallel_copy('nw_def'::regclass, p, 2)\$\$;" +psql_run "ALTER FUNCTION nw_definer_copy(text) OWNER TO t_nl;" +psql_run "GRANT SELECT ON nw_src TO t_nl;" +psql_run "GRANT SELECT, INSERT ON nw_dst, nw_wide, nw_esc, nw_def, nw_sw TO t_nl;" + +as_nl() { # sql -> its output as t_nl, the debug messages included, after the role took + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -X -qAt -v ON_ERROR_STOP=1 -c "SET ROLE t_nl;" \ + -c "SELECT 'as:' || current_user;" -c "$1" 2>&1 +} +refusals() { # -> lines in the server log refusing t_nl a login + pgc_pg "grep -c 'role \"t_nl\" is not permitted to log in' '$PGC_LOGFILE'" 2>/dev/null || echo 0 +} + +# ---- premises: the role cannot log in, and the log records it when it tries ---- +check "premise: the NOLOGIN role cannot open a session" \ + "$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U t_nl -d "$PGC_DB" -c 'SELECT 1' 2>&1 | + grep -c 'is not permitted to log in')" "1" +check "premise: the server log records the NOLOGIN role's refused login" "$(refusals)" "1" +check "premise: SET ROLE to the NOLOGIN role takes" "$(as_nl 'SELECT 1;' | head -1)" "as:t_nl" + +# ---- parallel_export_parquet ---------------------------------------------------- +reply="$(as_nl "SELECT pgcolumnar.parallel_export_parquet('nw_src'::regclass, '$outdir', 2);")" +check_num "a NOLOGIN role's parallel_export_parquet exports every row" \ + "$(sed -n 2p <<<"$reply")" "$ROWS" + +# ---- parallel_copy ---------------------------------------------------------------- +reply="$(as_nl "SELECT pgcolumnar.parallel_copy('nw_dst'::regclass, '$infile', 2);")" +check_num "a NOLOGIN role's parallel_copy loads every row" "$(sed -n 2p <<<"$reply")" "$ROWS" +check "parallel_copy's workers wrote every row as the NOLOGIN role" \ + "$(q "SELECT string_agg(DISTINCT coalesce(who, ''), ',') FROM nw_dst;")" "t_nl" + +# ---- a trigger that resets the role inside the copy worker (#1612 review) ----------- +# On PG17+ a worker that connected as the authenticated user became that user on RESET +# ROLE, PREPAREd as it, and left the leader "permission denied to finish prepared +# transaction" with two orphaned prepared transactions holding locks. +reply="$(as_nl "SELECT pgcolumnar.parallel_copy('nw_esc'::regclass, '$infile', 2);")" +check "a copy trigger that resets the role still writes every row as the NOLOGIN role" \ + "$(q "SELECT count(*) || ':' || coalesce(string_agg(DISTINCT coalesce(who, ''), ','), '') FROM nw_esc;")" "$ROWS:t_nl" +check_num "and that copy leaves no prepared transaction behind" \ + "$(q "SELECT count(*) FROM pg_prepared_xacts;")" "0" +q "SELECT 'ROLLBACK PREPARED ' || quote_literal(gid) FROM pg_prepared_xacts" | while read -r _rb; do q "$_rb" >/dev/null; done + +# ---- a trigger that switches to another role is refused before it prepares --------- +reply="$(as_nl "SELECT pgcolumnar.parallel_copy('nw_sw'::regclass, '$infile', 2);")" +check "a copy trigger that switches to another role is refused before the worker prepares" \ + "$(grep -q 'no longer running as the role that called parallel_copy' <<<"$reply" && echo refused || echo "not refused: $(sed -n 2p <<<"$reply")")" "refused" +check_num "and the refused copy leaves no prepared transaction behind" \ + "$(q "SELECT count(*) FROM pg_prepared_xacts;")" "0" +q "SELECT 'ROLLBACK PREPARED ' || quote_literal(gid) FROM pg_prepared_xacts" | while read -r _rb; do q "$_rb" >/dev/null; done + +# ---- a copy called from a SECURITY DEFINER function keeps its restriction ----------- +# The worker takes the leader's security context with its user id. Without it, the +# trigger's role reset succeeds inside the worker and runs as the superuser session. +q "SELECT nw_definer_copy('$infile');" >/dev/null +check "a copy run by a SECURITY DEFINER function keeps the definer's restriction in its workers" \ + "$(q "SELECT count(*) || ':' || coalesce(string_agg(DISTINCT coalesce(who, ''), ','), '') FROM nw_def;")" "$ROWS:t_nl" + +# ---- the parallel flush ------------------------------------------------------------ +reply="$(as_nl "SET pgcolumnar.parallel_flush = on; SET client_min_messages = debug1; + INSERT INTO nw_wide SELECT g, g, g FROM generate_series(1, $ROWS) g;")" +check "the flush is dispatched to workers" \ + "$(grep -o 'parallel_flush dispatch: .* -> [a-z]*' <<<"$reply" | sed 's/.* -> //' | head -1)" "parallel" +check "a NOLOGIN role's parallel flush encodes every column in a worker" \ + "$(grep -o 'parallel_flush: [0-9]* of [0-9]* columns' <<<"$reply" | head -1)" "parallel_flush: 3 of 3 columns" +check_num "and the table holds every row it flushed" "$(q 'SELECT count(*) FROM nw_wide;')" "$ROWS" + +# ---- and no worker was refused on the way ------------------------------------------ +check "no worker was refused a login as the NOLOGIN role" "$(refusals)" "1" + +pgc_summary diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index 8c1297e60..4fa71764f 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -288,6 +288,10 @@ behaviour, the source of that number is named. - [240. test_fsst_margin.py: the FSST keep/drop margin decides, and never changes content](#240-test_fsst_marginpy-the-fsst-keepdrop-margin-decides-and-never-changes-content) - [241. test_index_delete_liveness.py: nbtree's deletion passes never destroy an entry for a live row](#241-test_index_delete_livenesspy-nbtrees-deletion-passes-never-destroy-an-entry-for-a-live-row) - [242. test_rewrite_group_scan.py: a rewrite reads each row group once, not once per row](#242-test_rewrite_group_scanpy-a-rewrite-reads-each-row-group-once-not-once-per-row) +- [243. test_recluster_gate.py: recluster skips its rewrite when the table is already this key's run](#243-test_recluster_gatepy-recluster-skips-its-rewrite-when-the-table-is-already-this-keys-run) +- [244. test_recluster_extent.py: an online recluster records only the extent it rewrote](#244-test_recluster_extentpy-an-online-recluster-records-only-the-extent-it-rewrote) +- [245. test_reader_buffer_reuse.py: the reader reuses one group buffer, and the reuse never shows in an answer](#245-test_reader_buffer_reusepy-the-reader-reuses-one-group-buffer-and-the-reuse-never-shows-in-an-answer) +- [246. test_nologin_workers.py: a NOLOGIN role's background workers connect and act as that role](#246-test_nologin_workerspy-a-nologin-roles-background-workers-connect-and-act-as-that-role) ## 1. How to read a test in here @@ -9675,3 +9679,112 @@ A rewrite reads `finished` only when compact_rewrite returned a positive count. | `test_a_group_under_the_cap_still_rewrites_correctly` | a group under the cap still rewrites correctly | | `test_the_index_lookup_reads_the_index` | the index lookup's plan reads rgs_id | | `test_an_index_lookup_finds_a_rewritten_row` | an index lookup finds a rewritten row | + +## 243. test_recluster_gate.py: recluster skips its rewrite when the table is already this key's run + +The pytest twin of `test/recluster_gate.sh` (ported for #1341), every check by the shell's name. recluster returns the number of row groups it rewrote, and its self-gate (#415) makes a repeat on an unchanged table rewrite nothing; an append or a new key makes it work again. On a private database. + +### Every test + +The row-set check reads a heap mirror built from the generator, not copied from the table under test. A layout comparison needs both readings. + +| test | what it holds | +| --- | --- | +| `test_the_first_gate_recluster_does_real_work` | the first recluster does real work (>0 groups) | +| `test_sort_status_reports_the_clustering_key` | sort_status now reports the clustering key (was NULL before #415) | +| `test_the_recluster_run_covers_the_whole_relation` | the run covers the whole relation (0 appended groups) | +| `test_a_repeat_on_an_unchanged_table_reclusters_nothing` | recluster by the same key on an unchanged table reclusters 0 groups | +| `test_the_layout_is_byte_identical` | and it leaves the physical layout byte-identical (group offsets unmoved) | +| `test_a_third_identical_recluster_is_a_no_op` | a third identical recluster is also a no-op | +| `test_the_insert_appended_unsorted_groups` | premise: the insert appended unsorted groups | +| `test_a_recluster_after_appends_does_work` | recluster after appends does real work again (>0) | +| `test_recluster_folds_the_tail_back_into_the_run` | and it folds the new data back into the run (0 appended after) | +| `test_a_recluster_by_a_new_key_does_work` | recluster by a NEW key is NOT a no-op -- it re-clusters (>0) | +| `test_sort_status_reports_the_new_key` | and sort_status now reports the new key | +| `test_a_repeat_by_the_new_key_is_a_no_op` | a repeat recluster by the new key is a no-op | +| `test_the_no_op_reclusters_preserved_the_row_set` | no-op recluster preserved the exact row set (columnar == heap mirror) | + +## 244. test_recluster_extent.py: an online recluster records only the extent it rewrote + +The pytest twin of `test/recluster_extent.sh` (ported for #1341), every check by the shell's name. The online recluster runs while other sessions write, and its run must cover the groups it wrote and nothing else: not rows inserted during it (#311), and not a group whose id another writer drew below the rewrite's (#342). The races run in threads, as the shell runs them in background jobs. On a private database. + +### Every test + +Whether the insert ran during the rewrite is asked of the server before the insert begins. The run must claim at least one row and no more than the base. The projection arm first reads its projection back from the catalog, so a failed `add_projection` cannot turn it into a second arm without one. + +| test | what it holds | +| --- | --- | +| `test_the_table_loaded` | the table loaded | +| `test_the_insert_ran_during_the_rewrite` | the insert ran while the rewrite was still running | +| `test_the_rewrite_reported_groups_reclustered` | the rewrite reported groups reclustered | +| `test_the_concurrent_insert_was_not_in_the_work_set` | premise: the concurrent insert was not in the rewrite's work set | +| `test_every_row_is_present_afterwards` | every row is present afterwards | +| `test_the_run_does_not_claim_rows_written_during_the_rewrite` | the run does not claim rows written during the rewrite | +| `test_the_counts_still_cover_every_stored_row` | the counts still cover every stored row | +| `test_the_concurrent_rows_are_reported_as_appended` | the concurrent rows are reported as appended | +| `test_a_group_numbered_below_the_run_is_not_ordered` | a group numbered below the run is not counted as ordered | +| `test_the_rows_below_the_run_are_appended` | the rows written below the run are reported as appended | +| `test_the_id_drawn_below_counts_cover_every_row` | the id-drawn-below counts still cover every stored row | +| `test_noproj_a_full_recluster_claims_every_row` | noproj: a full recluster claims every row | +| `test_noproj_a_full_recluster_leaves_nothing_appended` | noproj: a full recluster leaves nothing appended | +| `test_withproj_add_projection_created_the_projection` | withproj: add_projection created the projection | +| `test_withproj_a_full_recluster_claims_every_row` | withproj: a full recluster claims every row | +| `test_withproj_a_full_recluster_leaves_nothing_appended` | withproj: a full recluster leaves nothing appended | +| `test_an_uncontended_rewrite_claims_the_whole_relation` | an uncontended rewrite claims the whole relation | +| `test_an_uncontended_rewrite_counts_every_row_as_sorted` | an uncontended rewrite counts every row as sorted | + +## 245. test_reader_buffer_reuse.py: the reader reuses one group buffer, and the reuse never shows in an answer + +The pytest twin of `test/reader_buffer_reuse.sh` (ported for #1341), every check by the shell's name. Reading every group costs well under one group's worth of minor faults per extra group (log_executor_stats), and the full scan, projected scans, whole-group reads, a short group, added and dropped columns and a rescan all match heap mirrors. On a private database, with serial planning as the shell's cluster runs. + +### Every test + +The heap mirrors are built from the generators, not copied from the tables under test. A differential comparison refuses an empty side. + +A read that names its columns fills only their ranges of the reused buffer. A read with no projection reads the whole group into it in one call: a whole-row read, and every read with `pgcolumnar.enable_column_projection = off`. Four tests compare that read with the mirror, before and after the short group. + +| test | what it holds | +| --- | --- | +| `test_the_reuse_fixture_has_several_row_groups` | premise: the fixture has several row groups, or reuse cannot matter | +| `test_the_one_group_arm_reads_one_group` | premise: the one-group arm really reads exactly one row group | +| `test_the_all_groups_arm_reads_every_group` | premise: the all-groups arm really reads every row group | +| `test_the_instrument_reports_faults` | premise: the instrument reports a nonzero fault count at all | +| `test_reading_every_group_reuses_the_buffer` | reading every group does not cost per-group faults (ratio well under the group count) | +| `test_the_full_scan_matches_the_mirror` | the full scan matches a heap mirror exactly | +| `test_a_projected_scan_matches_too` | a PROJECTED scan (one column of two) matches too | +| `test_the_full_scan_with_projection_off_matches_the_mirror` | the full scan with column projection off (a whole-group read) matches a heap mirror | +| `test_a_whole_row_read_matches_the_mirror` | a whole-row read (a whole-group read) matches a heap mirror | +| `test_a_smaller_group_was_added` | premise: that really added a group smaller than the ones before it | +| `test_a_short_group_after_a_full_one_reads_correctly` | a short group after a full one still reads correctly | +| `test_a_short_group_reads_correctly_with_projection_off` | a short group after a full one reads correctly with column projection off | +| `test_a_short_group_reads_correctly_in_a_whole_row_read` | a short group after a full one reads correctly in a whole-row read | +| `test_the_differential_fixture_spans_several_groups` | premise: the differential fixture spans several row groups | +| `test_a_projected_subset_matches_the_heap_twin` | one test per row of `PROJECTIONS` (8 rows), each by the shell's name | +| `test_an_added_column_reads_back_null` | a column ADDED after the groups exist reads back NULL, as on heap | +| `test_an_all_null_column_beside_a_wide_one_matches` | an all-NULL column beside a wide one matches | +| `test_reads_after_drop_column_match` | reads after DROP COLUMN match | +| `test_a_rescan_returns_the_same_rows` | a rescan (nested loop) returns the same rows | + +## 246. test_nologin_workers.py: a NOLOGIN role's background workers connect and act as that role + +The pytest twin of `test/nologin_workers.sh` (#1585), every check by the shell's name. As a NOLOGIN role reached by SET ROLE and holding both server-file roles: `parallel_export_parquet` and `parallel_copy` return every row, a row trigger inside the copy worker records that role and not the authenticated user, the parallel flush encodes every column in a worker, and the server log holds no refused login for the role beyond the one the premise provokes. Before #1585 the workers connected as the calling role, which PostgreSQL 15 and 16 refuse for a role without LOGIN. On a private database. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_the_nologin_role_cannot_open_a_session` | premise: the NOLOGIN role cannot open a session | +| `test_the_server_log_records_the_refused_login` | premise: the server log records the NOLOGIN role's refused login -- the positive control for the log reading the last test relies on | +| `test_set_role_to_the_nologin_role_takes` | premise: SET ROLE to the NOLOGIN role takes | +| `test_a_nologin_roles_parallel_export_exports_every_row` | a NOLOGIN role's parallel_export_parquet exports every row | +| `test_a_nologin_roles_parallel_copy_loads_every_row` | a NOLOGIN role's parallel_copy loads every row | +| `test_parallel_copys_workers_wrote_as_the_nologin_role` | parallel_copy's workers wrote every row as the NOLOGIN role -- a worker that connected as the authenticated user and never took the caller's id writes `postgres` here | +| `test_the_flush_is_dispatched_to_workers` | the flush is dispatched to workers | +| `test_a_nologin_roles_parallel_flush_encodes_every_column_in_a_worker` | a NOLOGIN role's parallel flush encodes every column in a worker -- read from the flush's own DEBUG1 split, because a refused worker falls back to serial and stores the same rows | +| `test_the_table_holds_every_row_it_flushed` | and the table holds every row it flushed | +| `test_no_worker_was_refused_a_login` | no worker was refused a login as the NOLOGIN role | +| `test_a_role_resetting_copy_trigger_writes_as_the_nologin_role` | a copy trigger that resets the role still writes every row as the NOLOGIN role -- the worker's session user is the caller on every major (#1612 review) | +| `test_the_role_resetting_copy_leaves_nothing_prepared` | and that copy leaves no prepared transaction behind | +| `test_a_role_switching_copy_trigger_is_refused_before_prepare` | a copy trigger that switches to another role is refused before the worker prepares: a transaction prepared as another user is one the caller cannot finish | +| `test_the_refused_copy_leaves_nothing_prepared` | and the refused copy leaves no prepared transaction behind | +| `test_a_definer_copy_keeps_its_restriction_in_the_workers` | a copy run by a SECURITY DEFINER function keeps the definer's restriction in its workers: the worker takes the caller's security context, inside its transaction | diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index 71f8bdf80..f9cf02a92 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -983,4 +983,12 @@ guard_tests 467 # 9 tests, one per check, by collection over the cluster files. # cluster_tests 2162 -> 2169 for #1341's test_rewrite_group_scan.py, the twin of rewrite_group_scan.sh: # 7 tests, one per check, by collection over the cluster files. -cluster_tests 2169 +# cluster_tests 2169 -> 2182 for #1341's test_recluster_gate.py, the twin of recluster_gate.sh: +# 13 tests, one per check, by collection over the cluster files. +# cluster_tests 2182 -> 2200 for #1341's test_recluster_extent.py, the twin of recluster_extent.sh: +# 18 tests, one per check, by collection over the cluster files. +# cluster_tests 2200 -> 2226 for #1341's test_reader_buffer_reuse.py, the twin of reader_buffer_reuse.sh: +# 26 tests, one per check, by collection over the cluster files. +# cluster_tests 2226 -> 2241 for #1341's test_nologin_workers.py, the twin of nologin_workers.sh: +# 15 tests, one per check, by collection over the cluster files. +cluster_tests 2241 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index bbf8f3890..09cc39014 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -91,6 +91,10 @@ "smoke", "sort_status", "sort_status_privilege", + "nologin_workers", + "reader_buffer_reuse", + "recluster_extent", + "recluster_gate", "rewrite_group_scan", "index_delete_liveness", "fsst_margin", diff --git a/test/pytest/test_nologin_workers.py b/test/pytest/test_nologin_workers.py new file mode 100644 index 000000000..6203c34d2 --- /dev/null +++ b/test/pytest/test_nologin_workers.py @@ -0,0 +1,256 @@ +"""A NOLOGIN role's background workers connect and act as that role (#1585). The pytest +twin of `test/nologin_workers.sh`, every check by the shell's name. + +parallel_export_parquet, parallel_copy and the parallel flush start background workers +for the calling role. A worker used to connect AS that role, which PostgreSQL 15 and 16 +refuse for a role without LOGIN, so after SET ROLE to a NOLOGIN role the export and the +copy failed with no reason given and the flush fell back to serial. A worker now +connects as the authenticated user and then takes the caller's user id. Asserted, as a +NOLOGIN role holding both server-file roles: the export and the copy return every row; +a row trigger inside the copy worker records that role, not the authenticated user; the +flush encodes every column in a worker; and the server log holds no refused login for +the role beyond the one the premise provoked. + +ON A PRIVATE DATABASE, with the role created for this module and dropped after. +""" +import re + +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_nologin_workers" +ROLE = "t_nl" +OTHER = "t_nl_other" +ROWS = 2000 +REFUSED = rf'role "{ROLE}" is not permitted to log in' + + +def _refusals(cluster): + """-> the lines in the server log refusing the role a login, as the shell's grep -c.""" + try: + text = (cluster.datadir / "server.log").read_text(errors="replace") + except OSError as exc: + return query_error(f"server log unreadable: {exc}") + return sum(1 for line in text.splitlines() if re.search(REFUSED, line)) + + +def _as_nl(dsn, sql, settings=()): + """-> (the first column of the first row as text, or query_error() carrying the failure; + the role current_user reported; the debug messages the statement raised).""" + messages = [] + try: + with psycopg.connect(dsn, autocommit=True) as conn: + conn.add_notice_handler(lambda d: messages.append(d.message_primary or "")) + conn.execute(f"SET ROLE {ROLE}") + who = conn.execute("SELECT current_user").fetchone()[0] + for s in settings: + conn.execute(s) + cur = conn.execute(sql) + row = cur.fetchone() if cur.description else None + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__), None, messages + return ("" if row is None or row[0] is None else str(row[0])), who, messages + + +def _one(dsn, sql): + """-> the first column of the first row as text, or query_error() carrying the failure, + so a failed final reading reds its check by name rather than erroring the fixture.""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + return "" if row is None or row[0] is None else str(row[0]) + + +def _rollback_prepared(dsn): + """Finish whatever a refused copy left prepared, so it cannot hold locks past its check.""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + for (gid,) in conn.execute("SELECT gid FROM pg_prepared_xacts").fetchall(): + conn.execute("ROLLBACK PREPARED %s", (gid,)) + except psycopg.Error: + pass + + +def _int(v): + return int(v) if isinstance(v, str) and v.isdigit() else v + + +@pytest.fixture(scope="module") +def nw(pgc_cluster, server_dir): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"DROP ROLE IF EXISTS {ROLE}") + a.execute(f"DROP ROLE IF EXISTS {OTHER}") + a.execute(f"CREATE DATABASE {DB}") + a.execute(f"CREATE ROLE {ROLE} NOSUPERUSER NOLOGIN") + try: + dsn = pgc_cluster.dsn(dbname=DB) + infile = server_dir / "nw_in.txt" + infile.write_text("".join(f"{g}\t\n" for g in range(1, ROWS + 1))) + infile.chmod(0o644) + outdir = server_dir / "nw_dir" + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar") + conn.execute(f"GRANT USAGE ON SCHEMA pgcolumnar TO {ROLE}") + conn.execute(f"GRANT pg_read_server_files, pg_write_server_files TO {ROLE}") + conn.execute("CREATE TABLE nw_src (a int, b text) USING pgcolumnar") + conn.execute(f"INSERT INTO nw_src SELECT g, 'x' || g FROM generate_series(1, {ROWS}) g") + # The trigger runs inside the copy worker, so it records the worker's current_user. + conn.execute("CREATE TABLE nw_dst (a int, who text) USING pgcolumnar") + conn.execute("CREATE FUNCTION nw_stamp() RETURNS trigger LANGUAGE plpgsql " + "AS $$BEGIN NEW.who := current_user; RETURN NEW; END$$") + conn.execute("CREATE TRIGGER nw_stamp BEFORE INSERT ON nw_dst " + "FOR EACH ROW EXECUTE FUNCTION nw_stamp()") + conn.execute("CREATE TABLE nw_wide (a int, b int, c int) USING pgcolumnar") + conn.execute(f"GRANT SELECT ON nw_src TO {ROLE}") + # A trigger that resets the role inside the copy worker, and one that switches to + # another role the caller belongs to (#1612 review): neither may leave the worker + # writing, or preparing, as anyone but the caller. + conn.execute("CREATE TABLE nw_esc (a int, who text) USING pgcolumnar") + conn.execute("CREATE FUNCTION nw_reset_stamp() RETURNS trigger LANGUAGE plpgsql AS $$BEGIN " + "BEGIN PERFORM set_config('role', 'none', true); " + "EXCEPTION WHEN OTHERS THEN NULL; END; " + "NEW.who := current_user; RETURN NEW; END$$") + conn.execute("CREATE TRIGGER nw_reset_stamp BEFORE INSERT ON nw_esc " + "FOR EACH ROW EXECUTE FUNCTION nw_reset_stamp()") + conn.execute(f"CREATE ROLE {OTHER} NOSUPERUSER NOLOGIN") + conn.execute(f"GRANT {OTHER} TO {ROLE}") + conn.execute("CREATE TABLE nw_sw (a int, who text) USING pgcolumnar") + conn.execute("CREATE FUNCTION nw_switch_stamp() RETURNS trigger LANGUAGE plpgsql AS $$BEGIN " + f"PERFORM set_config('role', '{OTHER}', true); " + "NEW.who := current_user; RETURN NEW; END$$") + conn.execute("CREATE TRIGGER nw_switch_stamp BEFORE INSERT ON nw_sw " + "FOR EACH ROW EXECUTE FUNCTION nw_switch_stamp()") + conn.execute(f"GRANT SELECT, INSERT ON nw_sw TO {OTHER}") + # The definer: a superuser session calls it, and inside it the caller is t_nl + # under a security-restricted context, which must hold in the workers too. + conn.execute("CREATE TABLE nw_def (a int, who text) USING pgcolumnar") + conn.execute("CREATE TRIGGER nw_def_stamp BEFORE INSERT ON nw_def " + "FOR EACH ROW EXECUTE FUNCTION nw_reset_stamp()") + conn.execute("CREATE FUNCTION nw_definer_copy(p text) RETURNS bigint LANGUAGE sql " + "SECURITY DEFINER AS $$SELECT pgcolumnar.parallel_copy('nw_def'::regclass, p, 2)$$") + conn.execute(f"ALTER FUNCTION nw_definer_copy(text) OWNER TO {ROLE}") + conn.execute(f"GRANT SELECT, INSERT ON nw_dst, nw_wide, nw_esc, nw_def, nw_sw TO {ROLE}") + + before = _refusals(pgc_cluster) + try: + with psycopg.connect(dsn.replace("user=postgres", f"user={ROLE}"), connect_timeout=5): + login = "connected" + except psycopg.Error as exc: + login = str(exc) + after_login = _refusals(pgc_cluster) + + _, took, _ = _as_nl(dsn, "SELECT 1") + export, _, _ = _as_nl(dsn, f"SELECT pgcolumnar.parallel_export_parquet('nw_src'::regclass, '{outdir}', 2)") + copy, _, _ = _as_nl(dsn, f"SELECT pgcolumnar.parallel_copy('nw_dst'::regclass, '{infile}', 2)") + _, _, flush = _as_nl(dsn, f"INSERT INTO nw_wide SELECT g, g, g FROM generate_series(1, {ROWS}) g", + settings=("SET pgcolumnar.parallel_flush = on", + "SET client_min_messages = debug1")) + writers = _one(dsn, "SELECT string_agg(DISTINCT coalesce(who, ''), ',') FROM nw_dst") + flushed = _one(dsn, "SELECT count(*) FROM nw_wide") + + stamped = "count(*) || ':' || coalesce(string_agg(DISTINCT coalesce(who, ''), ','), '')" + _as_nl(dsn, f"SELECT pgcolumnar.parallel_copy('nw_esc'::regclass, '{infile}', 2)") + esc = _one(dsn, f"SELECT {stamped} FROM nw_esc") + esc_prepared = _one(dsn, "SELECT count(*) FROM pg_prepared_xacts") + _rollback_prepared(dsn) + switched, _, _ = _as_nl(dsn, f"SELECT pgcolumnar.parallel_copy('nw_sw'::regclass, '{infile}', 2)") + sw_prepared = _one(dsn, "SELECT count(*) FROM pg_prepared_xacts") + _rollback_prepared(dsn) + _one(dsn, f"SELECT nw_definer_copy('{infile}')") + definer = _one(dsn, f"SELECT {stamped} FROM nw_def") + dispatch = [re.sub(r".* -> ", "", m) for m in flush if "parallel_flush dispatch:" in m] + split = [re.search(r"parallel_flush: [0-9]+ of [0-9]+ columns", m) for m in flush] + split = [m.group(0) for m in split if m] + yield { + "login refused": 1 if "is not permitted to log in" in login else 0, + "log records it": (after_login - before) if isinstance(after_login, int) and isinstance(before, int) else after_login, + "took": f"as:{took}", + "export": export, + "copy": copy, + "writers": writers or "", + "dispatch": dispatch[0] if dispatch else "", + "split": split[0] if split else "", + "flushed": _int(flushed), + "esc": esc, + "esc prepared": _int(esc_prepared), + "switch refused": ("refused" if "no longer running as the role that called parallel_copy" in str(switched) + else f"not refused: {switched}"), + "sw prepared": _int(sw_prepared), + "definer": definer, + "refused in all": (_refusals(pgc_cluster) - before) if isinstance(before, int) else before, + } + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"DROP ROLE IF EXISTS {ROLE}") + a.execute(f"DROP ROLE IF EXISTS {OTHER}") + + +def test_the_nologin_role_cannot_open_a_session(nw, expect): + expect.num(nw["login refused"], 1, "premise: the NOLOGIN role cannot open a session") + + +def test_the_server_log_records_the_refused_login(nw, expect): + expect.num(nw["log records it"], 1, "premise: the server log records the NOLOGIN role's refused login") + + +def test_set_role_to_the_nologin_role_takes(nw, expect): + expect.text(nw["took"], f"as:{ROLE}", "premise: SET ROLE to the NOLOGIN role takes") + + +def test_a_nologin_roles_parallel_export_exports_every_row(nw, expect): + expect.num(_int(nw["export"]), ROWS, "a NOLOGIN role's parallel_export_parquet exports every row") + + +def test_a_nologin_roles_parallel_copy_loads_every_row(nw, expect): + expect.num(_int(nw["copy"]), ROWS, "a NOLOGIN role's parallel_copy loads every row") + + +def test_parallel_copys_workers_wrote_as_the_nologin_role(nw, expect): + expect.text(nw["writers"], ROLE, "parallel_copy's workers wrote every row as the NOLOGIN role") + + +def test_the_flush_is_dispatched_to_workers(nw, expect): + expect.text(nw["dispatch"], "parallel", "the flush is dispatched to workers") + + +def test_a_nologin_roles_parallel_flush_encodes_every_column_in_a_worker(nw, expect): + expect.text(nw["split"], "parallel_flush: 3 of 3 columns", + "a NOLOGIN role's parallel flush encodes every column in a worker") + + +def test_the_table_holds_every_row_it_flushed(nw, expect): + expect.num(nw["flushed"], ROWS, "and the table holds every row it flushed") + + +def test_no_worker_was_refused_a_login(nw, expect): + # The premise's own refused login is the one line expected, as in the shell. + expect.num(nw["refused in all"], 1, "no worker was refused a login as the NOLOGIN role") + + +def test_a_role_resetting_copy_trigger_writes_as_the_nologin_role(nw, expect): + expect.text(nw["esc"], f"{ROWS}:{ROLE}", + "a copy trigger that resets the role still writes every row as the NOLOGIN role") + + +def test_the_role_resetting_copy_leaves_nothing_prepared(nw, expect): + expect.num(nw["esc prepared"], 0, "and that copy leaves no prepared transaction behind") + + +def test_a_role_switching_copy_trigger_is_refused_before_prepare(nw, expect): + expect.text(nw["switch refused"], "refused", + "a copy trigger that switches to another role is refused before the worker prepares") + + +def test_the_refused_copy_leaves_nothing_prepared(nw, expect): + expect.num(nw["sw prepared"], 0, "and the refused copy leaves no prepared transaction behind") + + +def test_a_definer_copy_keeps_its_restriction_in_the_workers(nw, expect): + expect.text(nw["definer"], f"{ROWS}:{ROLE}", + "a copy run by a SECURITY DEFINER function keeps the definer's restriction in its workers") diff --git a/test/pytest/test_reader_buffer_reuse.py b/test/pytest/test_reader_buffer_reuse.py new file mode 100644 index 000000000..9184d0523 --- /dev/null +++ b/test/pytest/test_reader_buffer_reuse.py @@ -0,0 +1,272 @@ +"""The columnar reader reuses one group buffer across row groups instead of allocating a +fresh one per group, and the reuse never shows in an answer. The pytest twin of +`test/reader_buffer_reuse.sh`, every check by the shell's name (#1341). + +Asserted: reading every group of a 600,000-row table costs well under one group's worth +of minor page faults per extra group (log_executor_stats, as the shell reads it); the +full scan, a projected scan and a short group after full ones match a heap mirror, as +does the whole-group read (column projection off, and a whole-row read) before and after +the short group; and +on a five-column table, eight projected subsets, a column added afterwards, an all-NULL +column, and reads after DROP COLUMN all match their heap twin, as does a rescan. + +THE HEAP MIRRORS ARE BUILT FROM THE GENERATORS, not copied from the tables under test: +a copy agrees with whatever the reader returned, and with a row corrupted before the +copy the shell's old form passed. A differential comparison refuses an empty side. + +ON A PRIVATE DATABASE, with max_parallel_workers_per_gather = 0 as the shell's cluster +runs, built in the shell's order by a module fixture. +""" +import re + +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_reader_buffer_reuse" +STRIPE = 150000 +ROWS = 600000 +OFF = ["SET pgcolumnar.enable_ungrouped_vector_agg = off", "SET pgcolumnar.enable_vectorization = off"] +VAL = "md5(g::text)||md5((g*7)::text)||md5((g*13)::text)||md5((g*17)::text)" +DIGEST = "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM {t}" +SUMS = "SELECT count(*)||'/'||sum(id) FROM {t}" +WHOLEROW = "SELECT count(*)||'/'||md5(string_agg(r::text, ',' ORDER BY (r).id)) FROM {t} r" +NOPROJ = "-c pgcolumnar.enable_column_projection=off" +ONE = f"SELECT count(v) FROM rb WHERE id <= {STRIPE}" +ALL = "SELECT count(v) FROM rb" +PROJECTIONS = [ + ("projected subset [i] matches the heap twin", "i"), + ("projected subset [s] matches the heap twin", "s"), + ("projected subset [b] matches the heap twin", "b"), + ("projected subset [t] matches the heap twin", "t"), + ("projected subset [u] matches the heap twin", "u"), + ("projected subset [i||':'||t] matches the heap twin", "i||':'||t"), + ("projected subset [s||':'||u] matches the heap twin", "s||':'||u"), + ("projected subset [i||b::text||t||u] matches the heap twin", "i||b::text||t||u"), +] + + +class _Db: + def __init__(self, dsn): + self.dsn = dsn + + def run(self, sql): + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + conn.execute(sql) + except psycopg.Error as exc: + print(f"-- step failed: {str(exc).splitlines()[0]}") + + def q(self, sql, options=None): + """-> the first column of the first row as text; a connection or query error as a value. + options: libpq startup options, as the shell passes PGOPTIONS.""" + try: + with psycopg.connect(self.dsn, autocommit=True, **({"options": options} if options else {})) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + if row is None or row[0] is None: + return "" + return ("t" if row[0] else "f") if isinstance(row[0], bool) else str(row[0]) + + def n(self, sql): + v = self.q(sql) + return int(v) if v.isdigit() else v + + def groups_read(self, sql): + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + for s in OFF: + conn.execute(s) + plan = "\n".join(r[0] for r in conn.execute( + f"EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) {sql}")) + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0]) + m = re.search(r"Columnar Chunk Groups Read: (\d+)", plan) + return int(m.group(1)) if m else query_error("no 'Chunk Groups Read' line") + + def faults(self, sql): + """-> the minor page faults of the third of three runs, from log_executor_stats.""" + notes = [] + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + conn.add_notice_handler(lambda d: notes.append(d.message_detail or "")) + conn.execute("SET client_min_messages = log") + conn.execute("SET log_executor_stats = on") + for s in OFF: + conn.execute(s) + for _ in range(3): + conn.execute(sql).fetchall() + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0]) + found = re.findall(r"0/(\d+) \[0/\d+\] page faults", "\n".join(notes)) + return int(found[-1]) if found else query_error("no page-fault line in the executor stats") + + def dq_pair(self, proj): + a = self.q(f"SELECT md5(string_agg(x, ',' ORDER BY x)) FROM (SELECT ({proj})::text x FROM rbd) z") + b = self.q(f"SELECT md5(string_agg(x, ',' ORDER BY x)) FROM (SELECT ({proj})::text x FROM rbd_h) z") + return "match" if a and not a.startswith("QUERY_ERROR") and a == b else f"differ({a} vs {b})" + + +@pytest.fixture(scope="module") +def rb(pgc_cluster): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + # serial, as the shell's cluster runs: a Gather splits the groups between + # workers, and the leader's count then read 2 of 4 + a.execute(f"ALTER DATABASE {DB} SET max_parallel_workers_per_gather = 0") + try: + d = _Db(pgc_cluster.dsn(dbname=DB)) + d.run("CREATE EXTENSION IF NOT EXISTS pgcolumnar") + r = {} + d.run("CREATE TABLE rb (id int, v text) USING pgcolumnar") + d.run(f"SELECT pgcolumnar.set_options('rb', stripe_row_limit => {STRIPE}, " + "compression => 'none', encode_effort => 'fast')") + d.run(f"INSERT INTO rb SELECT g, {VAL} FROM generate_series(1, {ROWS}) g") + r["ngroups"] = d.n("SELECT count(*) FROM pgcolumnar.stats('rb')") + r["g one"], r["g all"] = d.groups_read(ONE), d.groups_read(ALL) + r["f one"], r["f all"] = d.faults(ONE), d.faults(ALL) + + # the mirror comes from the generator, not from rb + d.run("CREATE TABLE rb_h (id int, v text)") + d.run(f"INSERT INTO rb_h SELECT g, {VAL} FROM generate_series(1, {ROWS}) g") + r["full"] = (d.q(DIGEST.format(t="rb")), d.q(DIGEST.format(t="rb_h"))) + r["projected"] = (d.q(SUMS.format(t="rb")), d.q(SUMS.format(t="rb_h"))) + # the whole-group read: no projection, so the group is read into the reused + # buffer in one call + r["noproj"] = (d.q(DIGEST.format(t="rb"), options=NOPROJ), d.q(DIGEST.format(t="rb_h"))) + r["wholerow"] = (d.q(WHOLEROW.format(t="rb")), d.q(WHOLEROW.format(t="rb_h"))) + for t in ("rb", "rb_h"): + d.run(f"INSERT INTO {t} SELECT g, {VAL} FROM generate_series({ROWS + 1}, {ROWS + 2000}) g") + r["short premise"] = d.q("SELECT (SELECT min(rowcount) FROM pgcolumnar.stats('rb')) < " + "(SELECT max(rowcount) FROM pgcolumnar.stats('rb'))") + r["short"] = (d.q(DIGEST.format(t="rb")), d.q(DIGEST.format(t="rb_h"))) + r["short noproj"] = (d.q(DIGEST.format(t="rb"), options=NOPROJ), d.q(DIGEST.format(t="rb_h"))) + r["short wholerow"] = (d.q(WHOLEROW.format(t="rb")), d.q(WHOLEROW.format(t="rb_h"))) + + d.run("CREATE TABLE rbd (i int, s smallint, b bigint, t text, u text) USING pgcolumnar") + d.run("SELECT pgcolumnar.set_options('rbd', stripe_row_limit => 20000, encode_effort => 'fast')") + d.run("INSERT INTO rbd SELECT g, (g%100)::smallint, g::bigint*1000, md5(g::text), " + "repeat(md5((g*7)::text), 3) FROM generate_series(1, 120000) g") + d.run("CREATE TABLE rbd_h AS SELECT g AS i, (g%100)::smallint AS s, g::bigint*1000 AS b, " + "md5(g::text) AS t, repeat(md5((g*7)::text), 3) AS u FROM generate_series(1, 120000) g") + r["rbd groups"] = d.n("SELECT count(*) FROM pgcolumnar.stats('rbd')") + for name, proj in PROJECTIONS: + r[name] = d.dq_pair(proj) + for t in ("rbd", "rbd_h"): + d.run(f"ALTER TABLE {t} ADD COLUMN added int") + r["added"] = d.dq_pair("coalesce(added::text,'N')||'|'||t") + for t in ("rbd", "rbd_h"): + d.run(f"ALTER TABLE {t} ADD COLUMN allnull text") + d.run(f"UPDATE {t} SET allnull = NULL WHERE i <= 10") + r["allnull"] = d.dq_pair("coalesce(allnull,'N')||'|'||u") + for t in ("rbd", "rbd_h"): + d.run(f"ALTER TABLE {t} DROP COLUMN b") + r["dropped"] = d.dq_pair("i||':'||t||':'||u") + r["rescan"] = d.q("SELECT count(*) FROM (SELECT 1 FROM generate_series(1,3) s, rb WHERE rb.id <= 100) x") + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def _at_least(n, k, what): + return ("yes" if n >= k else f"no ({n} {what})") if isinstance(n, int) else f"no ({n})" + + +def test_the_reuse_fixture_has_several_row_groups(rb, expect): + expect.text(_at_least(rb["ngroups"], 3, "groups"), "yes", + "premise: the fixture has several row groups, or reuse cannot matter") + + +def test_the_one_group_arm_reads_one_group(rb, expect): + expect.num(rb["g one"], 1, "premise: the one-group arm really reads exactly one row group") + + +def test_the_all_groups_arm_reads_every_group(rb, expect): + expect.num(rb["g all"], rb["ngroups"], "premise: the all-groups arm really reads every row group") + + +def test_the_instrument_reports_faults(rb, expect): + f1, fa = rb["f one"], rb["f all"] + ok = isinstance(f1, int) and isinstance(fa, int) and f1 > 0 and fa > 0 + expect.text("yes" if ok else f"no ({f1}, {fa})", "yes", + "premise: the instrument reports a nonzero fault count at all") + + +def test_reading_every_group_reuses_the_buffer(rb, expect): + f1, fa, n = rb["f one"], rb["f all"], rb["ngroups"] + if not all(isinstance(v, int) for v in (f1, fa, n)) or f1 <= 0: + got = f"no reading ({f1}, {fa}, {n})" + else: + ratio = fa / f1 + got = "reused" if ratio < 1 + (n - 1) * 0.5 else f"per-group (ratio {ratio:.2f} over {n} groups)" + expect.text(got, "reused", + "reading every group does not cost per-group faults (ratio well under the group count)") + + +def test_the_full_scan_matches_the_mirror(rb, expect): + got, want = rb["full"] + expect.text(got, want, "the full scan matches a heap mirror exactly") + + +def test_a_projected_scan_matches_too(rb, expect): + got, want = rb["projected"] + expect.text(got, want, "a PROJECTED scan (one column of two) matches too") + + +def test_the_full_scan_with_projection_off_matches_the_mirror(rb, expect): + got, want = rb["noproj"] + expect.text(got, want, "the full scan with column projection off (a whole-group read) matches a heap mirror") + + +def test_a_whole_row_read_matches_the_mirror(rb, expect): + got, want = rb["wholerow"] + expect.text(got, want, "a whole-row read (a whole-group read) matches a heap mirror") + + +def test_a_smaller_group_was_added(rb, expect): + expect.text(rb["short premise"], "t", "premise: that really added a group smaller than the ones before it") + + +def test_a_short_group_after_a_full_one_reads_correctly(rb, expect): + got, want = rb["short"] + expect.text(got, want, "a short group after a full one still reads correctly") + + +def test_a_short_group_reads_correctly_with_projection_off(rb, expect): + got, want = rb["short noproj"] + expect.text(got, want, "a short group after a full one reads correctly with column projection off") + + +def test_a_short_group_reads_correctly_in_a_whole_row_read(rb, expect): + got, want = rb["short wholerow"] + expect.text(got, want, "a short group after a full one reads correctly in a whole-row read") + + +def test_the_differential_fixture_spans_several_groups(rb, expect): + expect.text(_at_least(rb["rbd groups"], 3, "groups"), "yes", + "premise: the differential fixture spans several row groups") + + +@pytest.mark.parametrize("name,proj", PROJECTIONS) +def test_a_projected_subset_matches_the_heap_twin(rb, name, proj, expect): + expect.text(rb[name], "match", name) + + +def test_an_added_column_reads_back_null(rb, expect): + expect.text(rb["added"], "match", "a column ADDED after the groups exist reads back NULL, as on heap") + + +def test_an_all_null_column_beside_a_wide_one_matches(rb, expect): + expect.text(rb["allnull"], "match", "an all-NULL column beside a wide one matches") + + +def test_reads_after_drop_column_match(rb, expect): + expect.text(rb["dropped"], "match", "reads after DROP COLUMN match") + + +def test_a_rescan_returns_the_same_rows(rb, expect): + expect.text(rb["rescan"], "300", "a rescan (nested loop) returns the same rows") diff --git a/test/pytest/test_recluster_extent.py b/test/pytest/test_recluster_extent.py new file mode 100644 index 000000000..b0c3894b7 --- /dev/null +++ b/test/pytest/test_recluster_extent.py @@ -0,0 +1,223 @@ +"""An online recluster records only the extent it actually rewrote (#311, #342). The pytest +twin of `test/recluster_extent.sh`, every check by the shell's name (#1341). + +The online recluster runs while other sessions write. Its run must cover the groups it +wrote and nothing else: not rows inserted while it ran, and not a group whose id another +writer drew below the rewrite's. Asserted, with the races played out in threads as the +shell plays them in background jobs: the concurrent insert commits during the rewrite, +every row survives, the run claims some of the base and no more, the concurrent rows are +reported as appended, a group numbered below the run is not counted as ordered, a full +recluster with and without a projection claims every row (the projection read back from +the catalog, so a failed add_projection cannot turn the projection arm into a second +arm without one), and an uncontended rewrite claims the whole relation. + +A run that recorded nothing claims nothing, so "does not claim rows written during the +rewrite" needs at least one sorted row; 0 used to pass it. + +ON A PRIVATE DATABASE, built in the shell's order by a module fixture. +""" +import threading +import time + +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_recluster_extent" +BASE = 1500000 + + +class _Db: + def __init__(self, dsn): + self.dsn = dsn + + def run(self, sql): + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + conn.execute(sql.encode()) # simple protocol, as psql -c: several statements + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + return "" + + def q(self, sql): + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + if row is None or row[0] is None: + return "" + return ("true" if row[0] else "false") if isinstance(row[0], bool) else str(row[0]) + + def n(self, sql): + v = self.q(sql) + return int(v) if v.lstrip("-").isdigit() else v + + def ss(self, col, table="t"): + return self.n(f"SELECT {col} FROM pgcolumnar.sort_status('{table}')") + + +def _prj(d, arm, projection=False): + """-> (sorted, appended, projections): projections is how many projections named + p_ARM the catalog holds after add_projection, as a value (a failed read included), + or None when none was asked for.""" + d.run(f"DROP TABLE IF EXISTS pr_{arm}") + d.run(f"CREATE TABLE pr_{arm} (id int, k int, v text) USING pgcolumnar") + d.run(f"SELECT pgcolumnar.set_options('pr_{arm}', stripe_row_limit => 20000)") + d.run(f"INSERT INTO pr_{arm} SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g FROM generate_series(1, 200000) g") + made = None + if projection: + d.run(f"SELECT pgcolumnar.add_projection('pr_{arm}', 'p_{arm}', ARRAY['k','id'], ARRAY['k'])") + # read back from the catalog: add_projection's own result was discarded + made = d.n(f"SELECT count(*) FROM pgcolumnar.projection WHERE name = 'p_{arm}' AND projection_id > 0") + d.run(f"SELECT pgcolumnar.recluster('pr_{arm}', 'id', 'k')") + return d.ss("sorted_rows", f"pr_{arm}"), d.ss("appended_rows", f"pr_{arm}"), made + + +@pytest.fixture(scope="module") +def re_(pgc_cluster): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + try: + d = _Db(pgc_cluster.dsn(dbname=DB)) + d.run("CREATE EXTENSION IF NOT EXISTS pgcolumnar") + r = {} + d.run("CREATE TABLE t (id int, k int, v text) USING pgcolumnar") + d.run("SELECT pgcolumnar.set_options('t', stripe_row_limit => 20000, chunk_group_row_limit => 2048)") + d.run(f"INSERT INTO t SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g FROM generate_series(1, {BASE}) g") + r["loaded"] = d.n("SELECT count(*) FROM t") + + # the rewrite in one session, an insert in another two seconds in + out = {} + + def recluster(): + out["retired"] = d.n("SELECT pgcolumnar.recluster('t', 'id', 'k')") + th = threading.Thread(target=recluster) + th.start() + time.sleep(2) + # asked of the server, before the insert: the rewrite's end comes moments after + # the insert commits, so nothing read afterwards can order the two + active = d.n("SELECT count(*) FROM pg_stat_activity WHERE state = 'active' " + "AND query LIKE 'SELECT pgcolumnar.recluster(''t''%'") + d.run(f"INSERT INTO t SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g " + f"FROM generate_series({BASE + 1}, {BASE + 50000}) g") + th.join() + r["during"] = {1: "yes", 0: "no (the rewrite was not running when the insert began)"}.get( + active, f"no reading ({active})") + r["retired"] = out["retired"] + r["rows"] = d.n("SELECT count(*) FROM t") + r["sorted"], r["appended"] = d.ss("sorted_rows"), d.ss("appended_rows") + r["cover"] = d.q("SELECT (sorted_rows + appended_rows = (SELECT sum(rowcount) FROM pgcolumnar.stats('t')))::text " + "FROM pgcolumnar.sort_status('t')") + + # a writer draws its group id before the rewrite and commits after it + d.run("DROP TABLE IF EXISTS d") + d.run("CREATE TABLE d (id int, k int, v text) USING pgcolumnar") + d.run("SELECT pgcolumnar.set_options('d', stripe_row_limit => 20000, chunk_group_row_limit => 2048)") + d.run("INSERT INTO d SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g FROM generate_series(1, 100000) g") + hold = threading.Thread(target=d.run, args=( + "BEGIN; INSERT INTO d SELECT g, 1, 'late' || g FROM generate_series(900001, 905000) g; " + "SELECT pg_sleep(12); COMMIT;",)) + hold.start() + time.sleep(3) + d.run("SELECT pgcolumnar.recluster('d', 'id', 'k')") + hold.join() + r["d sorted"], r["d appended"] = d.ss("sorted_rows", "d"), d.ss("appended_rows", "d") + r["d cover"] = d.q("SELECT (sorted_rows + appended_rows = (SELECT sum(rowcount) FROM pgcolumnar.stats('d')))::text " + "FROM pgcolumnar.sort_status('d')") + + r["noproj"] = _prj(d, "noproj") + r["withproj"] = _prj(d, "withproj", projection=True) + + d.run("SELECT pgcolumnar.recluster('t', 'id', 'k')") + r["final appended"], r["final sorted"] = d.ss("appended_rows"), d.ss("sorted_rows") + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def test_the_table_loaded(re_, expect): + expect.num(re_["loaded"], BASE, "the table loaded") + + +def test_the_insert_ran_during_the_rewrite(re_, expect): + expect.text(re_["during"], "yes", "the insert ran while the rewrite was still running") + + +def test_the_rewrite_reported_groups_reclustered(re_, expect): + n = re_["retired"] + expect.text(("yes" if n > 0 else "no") if isinstance(n, int) else f"no ({n})", "yes", + "the rewrite reported groups reclustered") + + +def test_the_concurrent_insert_was_not_in_the_work_set(re_, expect): + expect.num(re_["retired"], BASE // 20000, "premise: the concurrent insert was not in the rewrite's work set") + + +def test_every_row_is_present_afterwards(re_, expect): + expect.num(re_["rows"], BASE + 50000, "every row is present afterwards") + + +def test_the_run_does_not_claim_rows_written_during_the_rewrite(re_, expect): + s = re_["sorted"] + if not isinstance(s, int): + got = f"no reading ({s})" + elif s < 1: + got = "no (the run claims no rows at all)" + else: + got = "yes" if s <= BASE else f"no ({s} sorted, more than the {BASE} the rewrite read)" + expect.text(got, "yes", "the run does not claim rows written during the rewrite") + + +def test_the_counts_still_cover_every_stored_row(re_, expect): + expect.text(re_["cover"], "true", "the counts still cover every stored row") + + +def test_the_concurrent_rows_are_reported_as_appended(re_, expect): + a = re_["appended"] + expect.text(("yes" if a > 0 else "no") if isinstance(a, int) else f"no ({a})", "yes", + "the concurrent rows are reported as appended") + + +def test_a_group_numbered_below_the_run_is_not_ordered(re_, expect): + expect.num(re_["d sorted"], 100000, "a group numbered below the run is not counted as ordered") + + +def test_the_rows_below_the_run_are_appended(re_, expect): + expect.num(re_["d appended"], 5000, "the rows written below the run are reported as appended") + + +def test_the_id_drawn_below_counts_cover_every_row(re_, expect): + expect.text(re_["d cover"], "true", "the id-drawn-below counts still cover every stored row") + + +def test_noproj_a_full_recluster_claims_every_row(re_, expect): + expect.num(re_["noproj"][0], 200000, "noproj: a full recluster claims every row") + + +def test_noproj_a_full_recluster_leaves_nothing_appended(re_, expect): + expect.num(re_["noproj"][1], 0, "noproj: a full recluster leaves nothing appended") + + +def test_withproj_add_projection_created_the_projection(re_, expect): + # without it, withproj is a second noproj arm and passes whatever #345's merge does + expect.num(re_["withproj"][2], 1, "withproj: add_projection created the projection") + + +def test_withproj_a_full_recluster_claims_every_row(re_, expect): + expect.num(re_["withproj"][0], 200000, "withproj: a full recluster claims every row") + + +def test_withproj_a_full_recluster_leaves_nothing_appended(re_, expect): + expect.num(re_["withproj"][1], 0, "withproj: a full recluster leaves nothing appended") + + +def test_an_uncontended_rewrite_claims_the_whole_relation(re_, expect): + expect.num(re_["final appended"], 0, "an uncontended rewrite claims the whole relation") + + +def test_an_uncontended_rewrite_counts_every_row_as_sorted(re_, expect): + expect.num(re_["final sorted"], BASE + 50000, "an uncontended rewrite counts every row as sorted") diff --git a/test/pytest/test_recluster_gate.py b/test/pytest/test_recluster_gate.py new file mode 100644 index 000000000..6323e7b8f --- /dev/null +++ b/test/pytest/test_recluster_gate.py @@ -0,0 +1,153 @@ +"""pgcolumnar.recluster() skips its rewrite when the table is already this key's run with +nothing appended (#415). The pytest twin of `test/recluster_gate.sh`, every check by the +shell's name (#1341). + +recluster returns the number of row groups it rewrote. Asserted: the first recluster +does work and records the key; repeating it on an unchanged table rewrites nothing and +leaves the layout byte-identical; an append makes it work again and fold the tail into +the run; a new key re-clusters and is recorded; and the no-op calls left the row set +exactly as the generator wrote it. + +THE ROW-SET CHECK reads a heap mirror built from the generator, not one copied from the +table under test: a copy agrees with whatever the reclusters did, and with a row +corrupted before the copy the shell's old form passed. A layout comparison needs both +readings, since two failed reads agreed. + +ON A PRIVATE DATABASE, built in the shell's order by a module fixture. +""" +import psycopg +import pytest + +from pgc_vacuity import QUERY_ERROR, query_error + +DB = "pgc_recluster_gate" +CG = 1000 +ROWSET = "SELECT count(*)||'/'||sum(a)||'/'||sum(b)||'/'||md5(string_agg(pad,',' ORDER BY b)) FROM {t}" +LAYOUT = ("SELECT md5(string_agg(stripeid::text||':'||fileoffset::text||':'||rowcount::text, ',' " + "ORDER BY stripeid)) FROM pgcolumnar.stats('rg_t')") + + +class _Db: + def __init__(self, dsn): + self.dsn = dsn + + def run(self, sql): + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + conn.execute(sql) + except psycopg.Error as exc: + print(f"-- step failed: {str(exc).splitlines()[0]}") + + def q(self, sql): + """-> the first column as psql -At prints it, or a query_error().""" + try: + with psycopg.connect(self.dsn, autocommit=True) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + if row is None or row[0] is None: + return "" + return ("t" if row[0] else "f") if isinstance(row[0], bool) else str(row[0]) + + def recluster(self, key): + v = self.q(f"SELECT pgcolumnar.recluster('rg_t','{key}')") + return int(v) if v.isdigit() else v + + +def _work(n): + return ("yes" if n > 0 else "no") if isinstance(n, int) else f"no ({n})" + + +@pytest.fixture(scope="module") +def rc(pgc_cluster): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + try: + d = _Db(pgc_cluster.dsn(dbname=DB)) + d.run("CREATE EXTENSION IF NOT EXISTS pgcolumnar") + r = {} + d.run("CREATE TABLE rg_t (a int, b int, pad text) USING pgcolumnar") + d.run(f"SELECT pgcolumnar.set_options('rg_t', chunk_group_row_limit => {CG})") + d.run("INSERT INTO rg_t SELECT (g*2654435761)::bigint % 1000, g, md5(g::text) FROM generate_series(1,20000) g") + r["first"] = d.recluster("a") + r["key"] = d.q("SELECT sort_key::text FROM pgcolumnar.sort_status('rg_t')") + r["appended"] = d.q("SELECT appended_groups FROM pgcolumnar.sort_status('rg_t')") + before = d.q(LAYOUT) + r["second"] = d.recluster("a") + after = d.q(LAYOUT) + ok = all(v and not v.startswith(QUERY_ERROR) for v in (before, after)) + r["layout"] = ("same" if before == after else "rewritten") if ok \ + else f"no reading: [{before}] -> [{after}]" + r["third"] = d.recluster("a") + d.run("INSERT INTO rg_t SELECT (g*2654435761)::bigint % 1000, g, md5(g::text) FROM generate_series(1,5000) g") + r["tail"] = d.q("SELECT appended_groups > 0 FROM pgcolumnar.sort_status('rg_t')") + r["after append"] = d.recluster("a") + r["folded"] = d.q("SELECT appended_groups FROM pgcolumnar.sort_status('rg_t')") + r["new key"] = d.recluster("b") + r["key b"] = d.q("SELECT sort_key::text FROM pgcolumnar.sort_status('rg_t')") + r["repeat b"] = d.recluster("b") + # the mirror comes from the generator: g = 1..20000, then 1..5000 again + d.run("CREATE TABLE rg_h AS " + "SELECT ((g*2654435761)::bigint % 1000)::int AS a, g AS b, md5(g::text) AS pad " + "FROM generate_series(1,20000) g UNION ALL " + "SELECT ((g*2654435761)::bigint % 1000)::int, g, md5(g::text) FROM generate_series(1,5000) g") + r["rowset"], r["mirror"] = d.q(ROWSET.format(t="rg_t")), d.q(ROWSET.format(t="rg_h")) + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +# Not premises (#1339): the next three read recluster's own work, its recorded key (#415) +# and its recorded extent (#311), and a one-line change to recluster reddens each. +def test_the_first_gate_recluster_does_real_work(rc, expect): + expect.text(_work(rc["first"]), "yes", "the first recluster does real work (>0 groups)") + + +def test_sort_status_reports_the_clustering_key(rc, expect): + expect.text(rc["key"], "{a}", "sort_status now reports the clustering key (was NULL before #415)") + + +def test_the_recluster_run_covers_the_whole_relation(rc, expect): + expect.text(rc["appended"], "0", "the run covers the whole relation (0 appended groups)") + + +def test_a_repeat_on_an_unchanged_table_reclusters_nothing(rc, expect): + expect.num(rc["second"], 0, "recluster by the same key on an unchanged table reclusters 0 groups") + + +def test_the_layout_is_byte_identical(rc, expect): + expect.text(rc["layout"], "same", "and it leaves the physical layout byte-identical (group offsets unmoved)") + + +def test_a_third_identical_recluster_is_a_no_op(rc, expect): + expect.num(rc["third"], 0, "a third identical recluster is also a no-op") + + +def test_the_insert_appended_unsorted_groups(rc, expect): + expect.text(rc["tail"], "t", "premise: the insert appended unsorted groups") + + +def test_a_recluster_after_appends_does_work(rc, expect): + expect.text(_work(rc["after append"]), "yes", "recluster after appends does real work again (>0)") + + +def test_recluster_folds_the_tail_back_into_the_run(rc, expect): + expect.text(rc["folded"], "0", "and it folds the new data back into the run (0 appended after)") + + +def test_a_recluster_by_a_new_key_does_work(rc, expect): + expect.text(_work(rc["new key"]), "yes", "recluster by a NEW key is NOT a no-op -- it re-clusters (>0)") + + +def test_sort_status_reports_the_new_key(rc, expect): + expect.text(rc["key b"], "{b}", "and sort_status now reports the new key") + + +def test_a_repeat_by_the_new_key_is_a_no_op(rc, expect): + expect.num(rc["repeat b"], 0, "a repeat recluster by the new key is a no-op") + + +def test_the_no_op_reclusters_preserved_the_row_set(rc, expect): + expect.text(rc["rowset"], rc["mirror"], "no-op recluster preserved the exact row set (columnar == heap mirror)") diff --git a/test/pytest/unported_suites.txt b/test/pytest/unported_suites.txt index 4f2a0c826..f0f1e0c3e 100644 --- a/test/pytest/unported_suites.txt +++ b/test/pytest/unported_suites.txt @@ -101,8 +101,5 @@ projection_rename_restore pushdown_report qual_order_selectivity read_stream -reader_buffer_reuse -recluster_extent -recluster_gate row_triggers scan_direction diff --git a/test/reader_buffer_reuse.sh b/test/reader_buffer_reuse.sh index 99e02be6a..d54755cee 100755 --- a/test/reader_buffer_reuse.sh +++ b/test/reader_buffer_reuse.sh @@ -78,7 +78,7 @@ ALL="SELECT count(v) FROM rb" G_ONE="$(groups_read "$ONE")" G_ALL="$(groups_read "$ALL")" check_num "premise: the one-group arm really reads exactly one row group" "$G_ONE" "1" -check "premise: the all-groups arm really reads every row group" \ +check_num "premise: the all-groups arm really reads every row group" \ "$G_ALL" "$NGROUPS" F_ONE="$(faults "$ONE")" @@ -94,7 +94,7 @@ check "premise: the instrument reports a nonzero fault count at all" \ RATIO="$(awk -v a="$F_ALL" -v b="$F_ONE" 'BEGIN{ printf "%.2f", (b>0)? a/b : 999 }')" echo "-- faults(all)/faults(one) = $RATIO with $NGROUPS groups" check "reading every group does not cost per-group faults (ratio well under the group count)" \ - "$(awk -v r="$RATIO" -v n="$NGROUPS" 'BEGIN{ print (r < 1 + (n-1)*0.5) ? "reused" : "per-group" }')" \ + "$(awk -v r="$RATIO" -v n="$NGROUPS" 'BEGIN{ print (r < 1 + (n-1)*0.5) ? "reused" : "per-group (ratio " r " over " n " groups)" }')" \ "reused" # ---- correctness: the reused buffer must not leak one group into the next ---- @@ -103,14 +103,36 @@ check "reading every group does not cost per-group faults (ratio well under the # projected read -- which deliberately does not read every column's range -- is # where a leak would show. psql_run "CREATE TABLE rb_h (id int, v text);" -psql_run "INSERT INTO rb_h SELECT id, v FROM rb;" -check "the full scan matches a heap mirror exactly" \ +# From the generator, not copied from rb: a copy agrees with whatever the reader +# returned, so a full-scan error showed on both sides and passed. +psql_run "INSERT INTO rb_h SELECT g, + md5(g::text)||md5((g*7)::text)||md5((g*13)::text)||md5((g*17)::text) + FROM generate_series(1, $ROWS) g;" +check_text "the full scan matches a heap mirror exactly" \ "$(q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb;")" \ "$(q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb_h;")" -check "a PROJECTED scan (one column of two) matches too" \ +check_text "a PROJECTED scan (one column of two) matches too" \ "$(q "SELECT count(*)||'/'||sum(id) FROM rb;")" \ "$(q "SELECT count(*)||'/'||sum(id) FROM rb_h;")" +# ---- the whole-group read ---------------------------------------------------- +# Every read above names its columns, so it takes the projected read, which fills +# only the wanted columns' ranges of the reused buffer. The reader has a second +# path: with no projection it reads the whole group into the reused buffer in one +# call. It takes that path for a whole-row Var, and for every read when +# pgcolumnar.enable_column_projection = off. A whole-group read that stops short +# leaves the previous group's bytes in the tail, so compare both forms with the +# generator-built mirror. A setting that fails to apply errors the session, and +# check_text refuses the empty side. +NOPROJ="-c pgcolumnar.enable_column_projection=off" +WHOLEROW="SELECT count(*)||'/'||md5(string_agg(r::text, ',' ORDER BY (r).id)) FROM" +check_text "the full scan with column projection off (a whole-group read) matches a heap mirror" \ + "$(PGOPTIONS="$NOPROJ" q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb;")" \ + "$(q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb_h;")" +check_text "a whole-row read (a whole-group read) matches a heap mirror" \ + "$(q "$WHOLEROW rb r;")" \ + "$(q "$WHOLEROW rb_h r;")" + # A group SMALLER than the one before it is the case that exposes a stale tail: # the buffer is not shrunk, so the bytes past this group's end are the previous # group's. Append a short group and read the whole table again. @@ -123,9 +145,15 @@ psql_run "INSERT INTO rb_h SELECT g, check "premise: that really added a group smaller than the ones before it" \ "$(q "SELECT (SELECT min(rowcount) FROM pgcolumnar.stats('rb')) < (SELECT max(rowcount) FROM pgcolumnar.stats('rb'));")" \ "t" -check "a short group after a full one still reads correctly" \ +check_text "a short group after a full one still reads correctly" \ "$(q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb;")" \ "$(q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb_h;")" +check_text "a short group after a full one reads correctly with column projection off" \ + "$(PGOPTIONS="$NOPROJ" q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb;")" \ + "$(q "SELECT count(*)||'/'||md5(string_agg(v, ',' ORDER BY id)) FROM rb_h;")" +check_text "a short group after a full one reads correctly in a whole-row read" \ + "$(q "$WHOLEROW rb r;")" \ + "$(q "$WHOLEROW rb_h r;")" # ---- differential arms: the shapes where a stale read would actually show ---- # These exist because of a finding on the #776 review: the arms above pass @@ -152,9 +180,13 @@ psql_run "SELECT pgcolumnar.set_options('rbd', stripe_row_limit => 20000, encode psql_run "INSERT INTO rbd SELECT g, (g%100)::smallint, g::bigint*1000, md5(g::text), repeat(md5((g*7)::text), 3) FROM generate_series(1, 120000) g;" -psql_run "CREATE TABLE rbd_h AS SELECT * FROM rbd;" +# From the generator too, for the same reason. +psql_run "CREATE TABLE rbd_h AS SELECT g AS i, (g%100)::smallint AS s, g::bigint*1000 AS b, + md5(g::text) AS t, repeat(md5((g*7)::text), 3) AS u + FROM generate_series(1, 120000) g;" +RBD_GROUPS="$(q "SELECT count(*) FROM pgcolumnar.stats('rbd');")" check "premise: the differential fixture spans several row groups" \ - "$([ "$(q "SELECT count(*) FROM pgcolumnar.stats('rbd');")" -ge 3 ] && echo yes || echo no)" "yes" + "$([ "${RBD_GROUPS:-0}" -ge 3 ] 2>/dev/null && echo yes || echo "no ($RBD_GROUPS groups)")" "yes" dq_pair() { # same projection against columnar and its heap twin local proj="$1" @@ -163,9 +195,14 @@ dq_pair() { # same projection against columnar and its heap twin b="$(q "SELECT md5(string_agg(x, ',' ORDER BY x)) FROM (SELECT ($proj)::text x FROM rbd_h) z;")" [ -n "$a" ] && [ "$a" = "$b" ] && echo match || echo "differ($a vs $b)" } -for proj in "i" "s" "b" "t" "u" "i||':'||t" "s||':'||u" "i||b::text||t||u"; do - check "projected subset [$proj] matches the heap twin" "$(dq_pair "$proj")" "match" -done +check "projected subset [i] matches the heap twin" "$(dq_pair "i")" "match" +check "projected subset [s] matches the heap twin" "$(dq_pair "s")" "match" +check "projected subset [b] matches the heap twin" "$(dq_pair "b")" "match" +check "projected subset [t] matches the heap twin" "$(dq_pair "t")" "match" +check "projected subset [u] matches the heap twin" "$(dq_pair "u")" "match" +check "projected subset [i||':'||t] matches the heap twin" "$(dq_pair "i||':'||t")" "match" +check "projected subset [s||':'||u] matches the heap twin" "$(dq_pair "s||':'||u")" "match" +check "projected subset [i||b::text||t||u] matches the heap twin" "$(dq_pair "i||b::text||t||u")" "match" psql_run "ALTER TABLE rbd ADD COLUMN added int;" psql_run "ALTER TABLE rbd_h ADD COLUMN added int;" diff --git a/test/recluster_extent.sh b/test/recluster_extent.sh index c78c13b3d..cee0cadc3 100755 --- a/test/recluster_extent.sh +++ b/test/recluster_extent.sh @@ -52,17 +52,19 @@ RECL_PID=$! # Give the rewrite time to take its group locks and begin, then insert. The # insert is small so it commits quickly and lands inside the rewrite's window. sleep 2 -INS_START=$(date +%s%N) +# Asked of the server, and before the insert: is the rewrite still running? Its +# end comes moments after the insert commits (measured: 3 and 4 ms), so two client +# clocks, or a status read after the insert, cannot order them. +RECL_ACTIVE="$(raw "SELECT count(*) FROM pg_stat_activity + WHERE state = 'active' AND query LIKE 'SELECT pgcolumnar.recluster(''t''%';")" raw "INSERT INTO t SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g FROM generate_series($((BASE + 1)), $((BASE + 50000))) g;" >/dev/null -INS_DONE=$(date +%s%N) wait $RECL_PID -RECL_DONE=$(date +%s%N) # The premise. If the insert committed after the rewrite had already finished, # nothing concurrent happened and the rest of this suite proves nothing. -check "the insert committed while the rewrite was still running" \ - "$( [ "$INS_DONE" -lt "$RECL_DONE" ] && echo yes || echo no )" "yes" +check "the insert ran while the rewrite was still running" \ + "$(case "$RECL_ACTIVE" in 1) echo yes ;; 0) echo "no (the rewrite was not running when the insert began)" ;; *) echo "no reading ($RECL_ACTIVE)" ;; esac)" "yes" # The real premise, and the one the wall-clock comparison above cannot establish # (#342). recluster returns the number of groups it retired, which is exactly the # set it read. The base load is BASE rows at stripe_row_limit, so a rewrite that @@ -87,8 +89,13 @@ echo " base $BASE rows, 50000 inserted during the rewrite; sorted $SORTED, # cannot have ordered the ones inserted while it ran, so the run must not claim # more rows than it began with. A mark taken from the highest live group number # would claim them. +# A run that recorded nothing claims nothing too: 0 sorted rows used to pass here. +# The rewrite reordered the base, so the run must claim some of it and no more. check "the run does not claim rows written during the rewrite" \ - "$( [ "$SORTED" -le "$BASE" ] && echo yes || echo no )" "yes" + "$(if ! pgc_is_number "$SORTED"; then echo "no reading ($SORTED)" + elif [ "$SORTED" -lt 1 ]; then echo "no (the run claims no rows at all)" + elif [ "$SORTED" -le "$BASE" ]; then echo yes + else echo "no ($SORTED sorted, more than the $BASE the rewrite read)"; fi)" "yes" # And the concurrent rows must be visible as decay rather than silently dropped # from both counts. @@ -156,23 +163,43 @@ check "the id-drawn-below counts still cover every stored row" \ # # Both arms are identical except for the projection, so the projection is # established as the cause rather than assumed. -for arm in noproj withproj; do - raw "DROP TABLE IF EXISTS pr_$arm;" >/dev/null - raw "CREATE TABLE pr_$arm (id int, k int, v text) USING pgcolumnar;" >/dev/null - raw "SELECT pgcolumnar.set_options('pr_$arm', stripe_row_limit => 20000);" >/dev/null - raw "INSERT INTO pr_$arm SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g +# prj ARM [PROJECTION] -> "SORTED APPENDED PROJECTIONS" after a full recluster of a +# 200000-row table, with a projection added first when asked. PROJECTIONS is how many +# projections named p_ARM the catalog holds after add_projection, or "-" when none was +# asked for. One literal set of names per arm. +prj() { + local p + raw "DROP TABLE IF EXISTS pr_$1;" >/dev/null + raw "CREATE TABLE pr_$1 (id int, k int, v text) USING pgcolumnar;" >/dev/null + raw "SELECT pgcolumnar.set_options('pr_$1', stripe_row_limit => 20000);" >/dev/null + raw "INSERT INTO pr_$1 SELECT g, ((g::bigint * 7919) % 100000)::int, 'v' || g FROM generate_series(1, 200000) g;" >/dev/null - if [ "$arm" = withproj ]; then - E="$(raw "SELECT pgcolumnar.add_projection('pr_$arm', 'p_$arm', ARRAY['k','id'], ARRAY['k']);")" - case "$E" in *ERROR*) echo " add_projection failed: $E";; esac + if [ -n "${2:-}" ]; then + local e + e="$(raw "SELECT pgcolumnar.add_projection('pr_$1', 'p_$1', ARRAY['k','id'], ARRAY['k']);")" + case "$e" in *ERROR*) echo " add_projection failed: $e" >&2;; esac + # Read back from the catalog, whatever add_projection printed. One word, so a + # failed read cannot shift the fields read after it. + p="$(raw "SELECT count(*) FROM pgcolumnar.projection + WHERE name = 'p_$1' AND projection_id > 0;")" + p="${p//[[:space:]]/_}" + else + p=- fi - raw "SELECT pgcolumnar.recluster('pr_$arm', 'id', 'k');" >/dev/null - S="$(raw "SELECT sorted_rows FROM pgcolumnar.sort_status('pr_$arm');")" - A="$(raw "SELECT appended_rows FROM pgcolumnar.sort_status('pr_$arm');")" - echo " $arm: sorted $S, appended $A" - check "$arm: a full recluster claims every row" "$S" "200000" - check "$arm: a full recluster leaves nothing appended" "$A" "0" -done + raw "SELECT pgcolumnar.recluster('pr_$1', 'id', 'k');" >/dev/null + echo "$(raw "SELECT sorted_rows FROM pgcolumnar.sort_status('pr_$1');") $(raw "SELECT appended_rows FROM pgcolumnar.sort_status('pr_$1');") ${p:-no_reading}" +} +read -r S A _ <<<"$(prj noproj)" +echo " noproj: sorted $S, appended $A" +check "noproj: a full recluster claims every row" "$S" "200000" +check "noproj: a full recluster leaves nothing appended" "$A" "0" +read -r S A P <<<"$(prj withproj yes)" +echo " withproj: sorted $S, appended $A" +# Without the projection, withproj is a second noproj arm, and its two checks pass +# whatever #345's merge does. add_projection's error only reached stderr. +check "withproj: add_projection created the projection" "$P" "1" +check "withproj: a full recluster claims every row" "$S" "200000" +check "withproj: a full recluster leaves nothing appended" "$A" "0" # ---------------------------------------------------- the uncontended case diff --git a/test/recluster_gate.sh b/test/recluster_gate.sh index 89555ed8e..379014924 100755 --- a/test/recluster_gate.sh +++ b/test/recluster_gate.sh @@ -34,11 +34,13 @@ physlayout() { q "SELECT md5(string_agg(stripeid::text||':'||fileoffset::text||' # ---- fixture: cluster once by (a) -------------------------------------------- mk rg_t FIRST="$(q "SELECT pgcolumnar.recluster('rg_t','a');")" -check "premise: the first recluster does real work (>0 groups)" \ +# Not premises (#1339): these three read recluster's own work, its recorded key (#415) +# and its recorded extent (#311), and a one-line change to recluster reddens each. +check "the first recluster does real work (>0 groups)" \ "$([ "${FIRST:-0}" -gt 0 ] && echo yes || echo no)" "yes" -check "premise: sort_status now reports the clustering key (was NULL before #415)" \ +check "sort_status now reports the clustering key (was NULL before #415)" \ "$(q "SELECT sort_key::text FROM pgcolumnar.sort_status('rg_t');")" "{a}" -check "premise: the run covers the whole relation (0 appended groups)" \ +check "the run covers the whole relation (0 appended groups)" \ "$(q "SELECT appended_groups FROM pgcolumnar.sort_status('rg_t');")" "0" # ---- the gate: a second recluster by the same key is a no-op ----------------- @@ -46,8 +48,10 @@ PHYS_BEFORE="$(physlayout rg_t)" SECOND="$(q "SELECT pgcolumnar.recluster('rg_t','a');")" PHYS_AFTER="$(physlayout rg_t)" check_num "recluster by the same key on an unchanged table reclusters 0 groups" "$SECOND" "0" +# Two failed layout reads agreed, so a missing reading is "no reading", not "same". check "and it leaves the physical layout byte-identical (group offsets unmoved)" \ - "$([ "$PHYS_BEFORE" = "$PHYS_AFTER" ] && echo same || echo rewritten)" "same" + "$(if [ -z "$PHYS_BEFORE" ] || [ -z "$PHYS_AFTER" ]; then echo "no reading: [$PHYS_BEFORE] -> [$PHYS_AFTER]" + elif [ "$PHYS_BEFORE" = "$PHYS_AFTER" ]; then echo same; else echo rewritten; fi)" "same" THIRD="$(q "SELECT pgcolumnar.recluster('rg_t','a');")" check_num "a third identical recluster is also a no-op" "$THIRD" "0" @@ -72,9 +76,14 @@ check_num "a repeat recluster by the new key is a no-op" \ "$(q "SELECT pgcolumnar.recluster('rg_t','b');")" "0" # ---- correctness: the data is unchanged by the no-op (oracle vs heap) -------- -psql_run "CREATE TABLE rg_h (a int, b int, pad text);" -psql_run "INSERT INTO rg_h SELECT a,b,pad FROM rg_t;" -check "no-op recluster preserved the exact row set (columnar == heap mirror)" \ +# The mirror is built from the generator, NOT copied from rg_t: a copy of rg_t agrees +# with rg_t whatever the reclusters did to it (with a row corrupted before the copy, +# this passed). rg_t holds g = 1..20000, then g = 1..5000 again from the append. +psql_run "CREATE TABLE rg_h AS + SELECT ((g*2654435761)::bigint % 1000)::int AS a, g AS b, md5(g::text) AS pad FROM generate_series(1,20000) g + UNION ALL + SELECT ((g*2654435761)::bigint % 1000)::int, g, md5(g::text) FROM generate_series(1,5000) g;" +check_text "no-op recluster preserved the exact row set (columnar == heap mirror)" \ "$(q "SELECT count(*)||'/'||sum(a)||'/'||sum(b)||'/'||md5(string_agg(pad,',' ORDER BY b)) FROM rg_t;")" \ "$(q "SELECT count(*)||'/'||sum(a)||'/'||sum(b)||'/'||md5(string_agg(pad,',' ORDER BY b)) FROM rg_h;")" diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index d19b5f380..f3b986ad1 100755 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -230,6 +230,7 @@ SUITES=( native_zonemap native_zonemap_narrow native_zonemap_session + nologin_workers objstore_addressing objstore_allowlist objstore_credentials