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, 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);