From 44a346563eee0106dc70b35491384109e506e518 Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Tue, 11 Aug 2026 23:54:02 -0600 Subject: [PATCH 1/2] harden: cancel-safe cleanup + skip parallel for new-in-xact relations (#445 slice 4) Two robustness fixes ahead of flipping the default: 1. Cancel-window leak: wrap the parallel flush's wait/collect in PG_ENSURE_ERROR_CLEANUP. On an error (statement cancel inside the wait is the realistic case) it terminates the workers and frees every published-but- uncollected pinned output segment; the collect loop invalidates each handle as it frees it so the cleanup never double-frees. Closes the leak-to-postmaster- restart window flagged on #591. 2. CREATE TABLE ...; INSERT (or CTAS) in one transaction: the table's pg_class row is uncommitted, so a worker's fresh-snapshot table_open fails and the flush did register-fail-warn-redo-serially. rel_new_in_current_xact() detects it via rd_createSubid and keeps the flush serial up front -- silent, no double work. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar_write_state.c | 86 +++++++++++++++++++++++++++++++++++++- 1 file changed, 85 insertions(+), 1 deletion(-) diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 380a06a..11311b6 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -2160,6 +2160,51 @@ pgcolumnar_parallel_flush_worker(Datum main_arg) proc_exit(0); } +/* + * pflush_error_cleanup + * Runs on an error during the parallel flush's wait/collect (#445 slice 4). + * Terminates any still-running workers, then frees every published-but- + * uncollected output segment. The collect loop invalidates each slot's handle + * as it frees it, so this only attaches segments still pinned by a worker -- + * valid handles, safe to map -- and never double-frees one the happy path + * already released. Without this, a statement cancel inside WaitForBackground- + * WorkerShutdown would leak the pinned segments until postmaster restart. + */ +typedef struct PflushCleanup +{ + BackgroundWorkerHandle **handles; + int nstarted; + PflushWorkerSlot *slots; + int nworkers; +} PflushCleanup; + +static void +pflush_error_cleanup(int code, Datum arg) +{ + PflushCleanup *cl = (PflushCleanup *) DatumGetPointer(arg); + int i; + + for (i = 0; i < cl->nstarted; i++) + if (cl->handles[i] != NULL) + { + TerminateBackgroundWorker(cl->handles[i]); + WaitForBackgroundWorkerShutdown(cl->handles[i]); + } + + for (i = 0; i < cl->nworkers; i++) + if (pg_atomic_read_u32(&cl->slots[i].state) == PFLUSH_DONE && + cl->slots[i].outHandle != DSM_HANDLE_INVALID) + { + dsm_segment *s = dsm_attach(cl->slots[i].outHandle); + + if (s != NULL) + { + dsm_unpin_segment(cl->slots[i].outHandle); + dsm_detach(s); + } + } +} + /* * flush_columns_parallel * The #445 slice-3 parallel flush: dispatch flush_one_column across a pool @@ -2195,6 +2240,7 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, uint32 dsmh; BackgroundWorker bw; BackgroundWorkerHandle **handles; + PflushCleanup cleanup; int nstarted = 0; int nWorkerCols = 0; int c; @@ -2317,6 +2363,19 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, } } + /* + * Arm the cancel-safe cleanup over the wait+collect: an error here (a + * statement cancel inside the wait is the realistic one) terminates the + * workers and frees any pinned output segment not yet collected, so nothing + * leaks to postmaster restart. Disarmed at PG_END below on the normal path. + */ + cleanup.handles = handles; + cleanup.nstarted = nstarted; + cleanup.slots = slots; + cleanup.nworkers = nworkers; + + PG_ENSURE_ERROR_CLEANUP(pflush_error_cleanup, PointerGetDatum(&cleanup)); + { /* wait for every started worker to finish (SIGTERM handler is die) */ for (i = 0; i < nstarted; i++) if (handles[i] != NULL) @@ -2379,6 +2438,7 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, nWorkerCols++; } dsm_detach(outseg); + slots[i].outHandle = DSM_HANDLE_INVALID; /* freed; skip in cleanup */ } else if (st == PFLUSH_FAILED) { @@ -2387,6 +2447,8 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, i, slots[i].errmsg[0] ? slots[i].errmsg : "unknown error"))); } } + } + PG_END_ENSURE_ERROR_CLEANUP(pflush_error_cleanup, PointerGetDatum(&cleanup)); dsm_detach(seg); pfree(inputs.data); @@ -2424,6 +2486,27 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, } } +/* + * rel_new_in_current_xact + * True if the relation was created in the current transaction, so its + * pg_class row is not yet committed. A parallel-flush worker opens the + * relation on a fresh snapshot to read its tuple descriptor; for a table + * created and loaded in one transaction (CREATE TABLE ...; INSERT ..., or + * CREATE TABLE AS) that open fails, and the flush would register workers, + * have them all fail, log a WARNING, and redo the whole thing serially. + * Detect it up front and keep the flush serial -- silently, and without the + * wasted registration and double work. (#445 slice 4) + */ +static bool +rel_new_in_current_xact(Oid relid) +{ + Relation rel = table_open(relid, NoLock); /* the INSERT already holds a lock */ + bool isnew = (rel->rd_createSubid != InvalidSubTransactionId); + + table_close(rel, NoLock); + return isnew; +} + /* * pgcolumnar_flush_row_group * Native-format (PGCN v1) flush. Lay out the accumulated rows as one row @@ -2512,7 +2595,8 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) * (synthetic tupdesc a worker could not rebuild from relid) on the serial * path. With the GUC off, keep slice 2's in-backend round-trip loop unchanged. */ - if (pgcolumnar_parallel_flush && natts >= 2 && writeState->tupdescIsRel) + if (pgcolumnar_parallel_flush && natts >= 2 && writeState->tupdescIsRel && + !rel_new_in_current_xact(writeState->relid)) { FlushColumnResult *colResults = palloc0(sizeof(FlushColumnResult) * natts); From b5e9aaec1ca5a3f22c227118074328be76f0ebe9 Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Wed, 12 Aug 2026 09:00:42 -0600 Subject: [PATCH 2/2] docs: parallel_flush stays opt-in; document the measured perf profile (#445 slice 4) Slice 4 measured the default flip and it does not hold up: parallel_flush wins only for a single large flush of many cheap numeric columns (~-14%, a wide bulk load) and REGRESSES the common cases -- small/frequent flushes pay a fixed worker-spawn cost (3.6x), and text-heavy or very large flushes pay an O(bytes) serialization cost (copying buffered data through the dsm) that outweighs the saving. So it stays off by default; the GUC description now says when to use it. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar_tableam.c | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/src/columnar_tableam.c b/src/columnar_tableam.c index bb30d6d..c80b849 100644 --- a/src/columnar_tableam.c +++ b/src/columnar_tableam.c @@ -2687,15 +2687,23 @@ _PG_init(void) DefineCustomBoolVariable("pgcolumnar.parallel_flush", "Dispatch the per-column stripe flush across background " - "workers (#445 slice 3).", - "Off by default so the merged write path is unchanged. " - "When on, a stripe flush of two or more columns fans the " - "per-column encode/compress work out to a pool of " - "background workers and degrades to serial in-backend " - "completion for any column a worker does not finish, so " - "the stored bytes are byte-identical to the serial path " - "either way. Opt-in for testing; slice 4 makes it the " - "measured, eventually-default control.", + "workers (#445).", + "Off by default, and stays that way (#445 slice 4 measured " + "it). When on, a stripe flush of two or more columns fans " + "the per-column encode/compress work out to a pool of " + "background workers, degrading to serial in-backend " + "completion for any column a worker does not finish, so the " + "stored bytes are byte-identical to the serial path either " + "way. It is a targeted opt-in, not a general win: it helps a " + "single large flush of many cheap (numeric) columns -- a " + "wide bulk load -- by up to ~14%, and it REGRESSES the common " + "cases, because it copies the buffered data through shared " + "memory: small or frequent flushes (a low stripe_row_limit, " + "or per-row commits) pay the fixed worker-spawn cost (3.6x " + "slower measured), and text-heavy or very large flushes pay " + "an O(bytes) serialization cost that outweighs the parallel " + "saving. Turn it on only for a wide-numeric bulk load, off " + "otherwise.", &pgcolumnar_parallel_flush, false, PGC_USERSET,