diff --git a/design/ISSUE_445_SLICE3_WORKERS.md b/design/ISSUE_445_SLICE3_WORKERS.md new file mode 100644 index 0000000..979f0cb --- /dev/null +++ b/design/ISSUE_445_SLICE3_WORKERS.md @@ -0,0 +1,102 @@ +# Issue #445 slice 3 — dispatch flush_one_column across a worker pool + +Design before code (house rule). Builds on slice 1 (`flush_one_column`, #589) and +slice 2 (the dsm serialize/deserialize helpers, #590). Slice 3 runs the per-column +flush on background workers instead of serially in the backend, and **degrades to +the serial path whenever it cannot** — never producing a wrong row count. + +## Invariants (the bar this slice is held to) + +1. **Byte-identical** stripe `data` bytes and catalog rows vs the serial path, ON and OFF. +2. **OFF by default** — a new boolean GUC `pgcolumnar.parallel_flush` (default off) gates it, + so the merged default behaviour is unchanged and the parallel path is opt-in for testing. + (Slice 4 turns this into the measured, eventually-default control.) +3. **Slot starvation degrades to serial, never to a wrong row count.** If fewer than the + needed workers register, or any worker fails, the backend completes every unfinished + column itself, serially, in-process. The set of columns written is always exactly [0,natts). + +## Why the FSST verdict cache must be threaded (byte-identity, subtle) + +The #472 cache lives in `writeState->colDefs[c].fsstVerdict/fsstVerdictAge` and persists across +row-group flushes within one statement. The serial path seeds each `flush_one_column` from it +and writes the updated verdict back. A worker with a fresh (`UNKNOWN`) verdict would **re-decide** +every group; for a group whose corpus would flip the cached verdict, re-deciding yields DIFFERENT +stored bytes than the serial reuse. So the cache is not a pure perf detail here — to stay +byte-identical, slice 3 threads it: +- The input dsm carries a `{fsstVerdict, fsstVerdictAge}` array, one per column, seeded from + `writeState->colDefs`. +- Each worker seeds its reconstructed `def` from that array before calling `flush_one_column`, + and writes the updated `{verdict, age}` into its result. +- The backend applies the returned verdicts back into `writeState->colDefs` in column order, + so the next flush sees exactly the state the serial path would have. Each column is owned by + exactly one worker per flush, so there is no cross-worker cache race. + +## dsm layout (one input segment, shm_toc keyed like the export path) + +- KEY_HEADER: `{ dbid, roleid, relid, storageId, groupNumber, rowCount, validityBytes, + encodeEffort, compressionType, compressionLevel, natts, nworkers, snapname[64] }`. +- KEY_INPUTS: all columns' `serialize_column_input` blobs concatenated. +- KEY_INOFFS: `uint64 offset[natts+1]` into KEY_INPUTS (offset[c]..offset[c+1] is column c). +- KEY_VERDICTS: `struct { int8 verdict; int32 age; } [natts]` seed values. +- KEY_SLOTS: `WorkerSlot[nworkers]` — `{ pg_atomic_uint32 state; dsm_handle outHandle; + uint32 outLen; int sqlerrcode; char errmsg[512]; }`. Result bytes do NOT live here (variable + size); each worker publishes its own OUTPUT dsm segment and stores the handle here. +- KEY_CLAIM: one `pg_atomic_uint32` next-column counter; workers claim columns with + `pg_atomic_fetch_add_u32` until it reaches natts (self-balancing; no static partition). + +## Worker (`pgcolumnar_parallel_flush_worker`, modelled on the export worker) + +attach seg + shm_toc; read widx from `bgw_extra`; `BackgroundWorkerInitializeConnectionByOid`; +Start txn + `ImportSnapshot` + push snapshot (for catalog reads only — the value data is in the +dsm, not the table). `table_open(relid, AccessShareLock)` once for the tupdesc. Then loop: +`c = pg_atomic_fetch_add_u32(claim, 1); if (c >= natts) break;` deserialize column c's input +(`deserialize_column_input` on KEY_INPUTS+offset[c]); reconstruct `att` from the tupdesc and +`def` from the column type via the SAME logic the write-state setup uses (extract that into a +shared `build_column_def(att, &def)` helper so worker and backend agree exactly); seed +`def.fsstVerdict/Age` from KEY_VERDICTS[c]; `flush_one_column(...)`; `serialize_column_result` +into a per-worker StringInfo AND record `(c, updated verdict/age)`. After the loop, `dsm_create` +one OUTPUT segment holding `{ uint32 count; per column: uint32 c, int8 verdict, int32 age, +uint32 resultLen, result bytes }`, store its handle+len in the slot, mark DONE. PG_CATCH → +record error + mark FAILED. detach; `proc_exit(0)`. + +## Backend (`pgcolumnar_flush_row_group`, parallel path) + +If GUC off OR natts < 2 → slice-2/serial path unchanged. Else: +1. Build input dsm (serialize every column via `serialize_column_input`, offsets, verdict seeds). + `ExportSnapshot(GetActiveSnapshot())`. +2. `nworkers = min(natts, auto_workers())`. Register workers; **if any registration fails, do + NOT error** — remember how many started (`nstarted`); if `nstarted == 0`, run the whole flush + serially (slice-2 path) and skip to catalog. Workers self-balance via the claim counter, so + `nstarted < nworkers` still covers all columns. +3. Wait for all started workers (`WaitForBackgroundWorkerShutdown`). +4. Collect: a `done[natts]` bitmap. For each DONE worker, attach its output dsm, and for each + column it produced: `deserialize_column_result`, stash by column index, mark done, apply the + returned verdict to `writeState->colDefs[c]`. For any worker FAILED, leave its columns undone + (its error is recorded; a failed worker is not fatal by itself as long as the column gets done). +5. **Serial completion of the remainder:** for every `c` still not done, run + `flush_one_column` in the backend now (seeded from `colDefs[c]`), and apply its verdict. This + is the degradation path and it also covers a worker that failed or a slot that never started. + If a worker FAILED with a hard error (not just "didn't run"), re-raise it after completion — + TBD in review whether a worker error should abort or just fall back; default to fall back + + log, since the serial redo produces the correct bytes. +6. Assemble in column order + write + catalog, exactly as slices 1–2 (backend-only I/O). + +## Verification plan + +- **OFF path byte-identical:** all write suites + differential with GUC off (default). +- **ON path byte-identical:** same suites with `pgcolumnar.parallel_flush=on` — differential, + native_zonemap, native_bloom, write_fsst_compressed (FSST verdict threading), native_dml. +- **Slot starvation:** set `max_worker_processes` below need; assert the load still completes + with the exact row count and byte-identical output (degradation proven, not assumed — + removal-proof style: with the fallback line deleted, a starved run must red/short-count). +- **Real parallelism:** confirm workers actually ran (a counter / log), so an ON run that + silently fell back serially isn't mistaken for a passing parallel test. +- Two toolchains: forced ASAN+UBSAN and pg18a assert; -Wshadow/-Werror clean. + +## Open questions for jdatcmd (his design) + +- Worker-error policy (step 5): fall back + log, or abort the statement? Fallback keeps the + write correct; aborting surfaces a real bug faster. +- Whether the worker should reuse `columnar_parallel_export.c`'s connection/snapshot scaffolding + by extracting it, or carry its own copy (the two differ: export imports the snapshot for data + reads; flush needs it only for catalog/tupdesc, so a plain fresh snapshot may suffice). diff --git a/src/columnar.h b/src/columnar.h index 536dbae..2e21b52 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -199,6 +199,7 @@ extern int pgcolumnar_groupagg_max_groups; /* plan-time group-count cap (#289) * extern bool pgcolumnar_enable_read_stream; /* stream/prefetch block reads (PG17+) */ extern bool pgcolumnar_enable_index_only_scan; /* allow index-only scans (gap 28) */ extern bool pgcolumnar_bulk_parallel_writer; /* internal: parallel_copy loader skips the storage-row creation lock (#300) */ +extern bool pgcolumnar_parallel_flush; /* dispatch the per-column stripe flush across bgworkers (#445 slice 3) */ extern bool pgcolumnar_enable_projection_scan; /* scan a covering projection (gap 26) */ extern bool pgcolumnar_enable_index_fetch_penalty; /* price a columnar index scan's per-row fetch (#355) */ diff --git a/src/columnar_tableam.c b/src/columnar_tableam.c index 7001ab6..bb30d6d 100644 --- a/src/columnar_tableam.c +++ b/src/columnar_tableam.c @@ -2685,6 +2685,23 @@ _PG_init(void) GUC_NOT_IN_SAMPLE, NULL, NULL, NULL); + 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.", + &pgcolumnar_parallel_flush, + false, + PGC_USERSET, + 0, + NULL, NULL, NULL); + DefineCustomBoolVariable("pgcolumnar.enable_unique_insert_lock", "Serialize concurrent inserts of the same unique key.", "Takes a transaction-scoped advisory lock per unique " diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 7123311..e1b7395 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -24,17 +24,101 @@ #include "catalog/pg_type.h" #include "executor/tuptable.h" #include "miscadmin.h" +#include "libpq/pqsignal.h" +#include "port/atomics.h" +#include "postmaster/bgworker.h" +#include "storage/dsm.h" +#include "storage/ipc.h" #include "storage/lmgr.h" +#include "storage/shm_toc.h" +#include "tcop/tcopprot.h" #include "utils/builtins.h" #include "utils/datum.h" #include "utils/fmgroids.h" #include "utils/fmgrprotos.h" +#include "utils/guc.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/rel.h" #include "utils/snapmgr.h" #include "utils/typcache.h" +#ifndef DSM_HANDLE_INVALID +#define DSM_HANDLE_INVALID 0 +#endif + +/* ------------------------------------------------------------------------- + * #445 slice 3: dispatch the per-column flush across background workers. + * + * One input DSM segment carries every column's flush_one_column input (the + * slice-2 serialize_column_input blobs) plus the shared FSST verdict cache; + * N workers claim columns off an atomic counter, run flush_one_column, and + * publish their results in a per-worker OUTPUT segment whose handle they store + * in their slot. The backend waits, collects every DONE worker's columns, and + * completes any column a worker did not finish serially in-process, so the set + * of columns written is always exactly [0, natts) -- slot starvation or a + * worker error degrades to the serial path, never to a wrong row count. + * ------------------------------------------------------------------------- */ +#define PFLUSH_MAGIC 0x50464c53 /* 'PFLS' */ +#define PFLUSH_KEY_HEADER 0 +#define PFLUSH_KEY_INPUTS 1 +#define PFLUSH_KEY_INOFFS 2 +#define PFLUSH_KEY_VERDICTS 3 +#define PFLUSH_KEY_SLOTS 4 +#define PFLUSH_KEY_CLAIM 5 +#define PFLUSH_MAX_WORKERS 8 + +typedef enum PflushState +{ + PFLUSH_PENDING = 0, /* not finished (also: worker never started) */ + PFLUSH_DONE, /* columns written; outHandle is valid */ + PFLUSH_FAILED /* worker caught an error; see errmsg */ +} PflushState; + +typedef struct PflushWorkerSlot +{ + pg_atomic_uint32 state; /* PflushState */ + dsm_handle outHandle; /* this worker's OUTPUT segment (valid on DONE) */ + uint32 outLen; /* its byte length */ + int sqlerrcode; + char errmsg[512]; +} PflushWorkerSlot; + +typedef struct PflushHeader +{ + Oid dbid; + Oid roleid; + Oid relid; + uint64 storageId; + uint64 groupNumber; + uint64 rowCount; + int validityBytes; + int encodeEffort; + int compressionType; + int compressionLevel; + int natts; + int nworkers; + bool bloomEnabled; + + /* + * Byte-identity: these encoding GUCs are read live inside flush_one_column + * (and its encoder), not captured into the write state, so a worker -- a + * fresh backend that never saw the launching session's SET -- would encode + * different bytes than the serial path under a non-default session value. + * Thread the launcher's live values so the worker reproduces them exactly. + */ + int fsstVerdictReuse; + int fsstMinGainPercent; + int encodingSampleRows; +} PflushHeader; + +/* KEY_VERDICTS: the FSST verdict cache, one seed per column (#472). */ +typedef struct PflushVerdict +{ + int8 verdict; + int32 age; +} PflushVerdict; + /* * How many bytes of a column chunk to run through both candidates when deciding * whether FSST still pays once the block compressor has had its turn. @@ -165,6 +249,16 @@ struct PgColumnarWriteState SubTransactionId subid; /* subtransaction that owns the buffer */ TupleDesc tupdesc; /* copy owned by writeContext */ int natts; + + /* + * True when `tupdesc` is relid's on-disk descriptor, so a #445-slice-3 + * parallel-flush worker may rebuild each column's Form_pg_attribute from + * table_open(relid). Set only by the base writer; false (fail-closed) for a + * projection's inner writer, whose tupdesc is a synthetic {rownumber, proj + * cols...} descriptor that does NOT match its base-table relid -- those flush + * on the serial path, which is byte-identical. + */ + bool tupdescIsRel; int stripeRowLimit; int chunkGroupRowLimit; int compressionType; /* columnar.compression at open time */ @@ -224,11 +318,25 @@ struct PgColumnarWriteState static MemoryContext PgColumnarWriteContext = NULL; static List *PgColumnarWriteStates = NIL; +/* + * #445 slice 3 GUC: dispatch the per-column flush across background workers. + * Off by default so the merged behaviour is unchanged; the parallel path is + * opt-in for testing (slice 4 makes it the measured, eventually-default + * control). Externed in columnar.h; the DefineCustomBoolVariable lives with the + * other GUCs in columnar_tableam.c. + */ +bool pgcolumnar_parallel_flush = false; + static void pgcolumnar_flush_row_group(PgColumnarWriteState *writeState); static void flush_ws_projections(PgColumnarWriteState *writeState); static ChunkGroupBuffer *pgcolumnar_start_chunk_group(PgColumnarWriteState *writeState); static uint64 *grow_uint64_array(uint64 *arr, int oldSize, int newSize); static void pgcolumnar_init_col_defs(PgColumnarWriteState *writeState); +static void build_column_def(Form_pg_attribute att, bool bloomEnabled, + MemoryContext cxt, PgColumnarColumnDef *def); + +/* bgworker entry: found by name via RegisterDynamicBackgroundWorker (#445 slice 3) */ +PGDLLEXPORT void pgcolumnar_parallel_flush_worker(Datum main_arg); /* * pgcolumnar_cmp_value @@ -270,6 +378,86 @@ pgcolumnar_cmp_value(PgColumnarColumnDef *def, Datum a, Datum b) return DatumGetInt32(FunctionCall2Coll(&def->cmpFn, def->collation, a, b)); } +/* + * build_column_def + * Resolve one column's skip metadata into *def: the btree comparison proc + * (for the per-chunk min/max skip list, spec 7.2) and the hash proc (for + * the per-chunk bloom filter, I7), plus the direct-comparison kind and the + * int2/int4 summable flag. fmgr procs are copied into `cxt` (the caller's + * long-lived context). A dropped column is left fully zero. + * + * Extracted from pgcolumnar_init_col_defs (#445 slice 3) so the flush + * workers reconstruct a column's def with byte-for-byte the same logic the + * backend's write-state setup uses -- worker and backend must agree exactly + * or the stored bytes diverge. *def is zeroed first, so a caller may pass a + * stack def; the FSST verdict cache (fsstVerdict/Age) is seeded separately. + */ +static void +build_column_def(Form_pg_attribute att, bool bloomEnabled, MemoryContext cxt, + PgColumnarColumnDef *def) +{ + TypeCacheEntry *tce; + + memset(def, 0, sizeof(PgColumnarColumnDef)); + + if (att->attisdropped) + return; + + tce = lookup_type_cache(att->atttypid, + TYPECACHE_CMP_PROC_FINFO | + TYPECACHE_HASH_PROC_FINFO); + if (OidIsValid(tce->cmp_proc_finfo.fn_oid)) + { + def->orderable = true; + fmgr_info_copy(&def->cmpFn, &tce->cmp_proc_finfo, cxt); + def->collation = att->attcollation; + + /* + * Resolve a direct comparison where the type permits one. These + * compare byte-for-byte under any collation, so the fast path + * cannot disagree with the operator it replaces; the zone map it + * feeds is read back through the same ordering. + */ + switch (att->atttypid) + { + case INT2OID: + def->fastCmp = COLUMNAR_FASTCMP_I16; + break; + case INT4OID: + case DATEOID: + def->fastCmp = COLUMNAR_FASTCMP_I32; + break; + case INT8OID: + case TIMESTAMPOID: + case TIMESTAMPTZOID: + def->fastCmp = COLUMNAR_FASTCMP_I64; + break; + default: + def->fastCmp = COLUMNAR_FASTCMP_NONE; + break; + } + } + + /* int2/int4: exact sum fits int64, carried in the zone map (D5) */ + def->summableInt = + (att->atttypid == INT2OID || att->atttypid == INT4OID); + + /* + * Bloom filter for hashable columns whose collation is safe (I7, gap + * 25): non-collatable types and deterministic collations, so a value + * hashes consistently between this build and an equality probe. A + * nondeterministic collation is left unbloomed. + */ + if (bloomEnabled && + OidIsValid(tce->hash_proc_finfo.fn_oid) && + PgColumnarCollationIsDeterministic(att->attcollation)) + { + def->bloomable = true; + fmgr_info_copy(&def->hashFn, &tce->hash_proc_finfo, cxt); + def->hashCollation = att->attcollation; + } +} + /* * pgcolumnar_init_col_defs * Allocate and fill writeState->colDefs: for each column, resolve the btree @@ -287,66 +475,9 @@ pgcolumnar_init_col_defs(PgColumnarWriteState *writeState) for (c = 0; c < writeState->natts; c++) { Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); - TypeCacheEntry *tce; - - if (att->attisdropped) - continue; - - tce = lookup_type_cache(att->atttypid, - TYPECACHE_CMP_PROC_FINFO | - TYPECACHE_HASH_PROC_FINFO); - if (OidIsValid(tce->cmp_proc_finfo.fn_oid)) - { - writeState->colDefs[c].orderable = true; - fmgr_info_copy(&writeState->colDefs[c].cmpFn, - &tce->cmp_proc_finfo, PgColumnarWriteContext); - writeState->colDefs[c].collation = att->attcollation; - - /* - * Resolve a direct comparison where the type permits one. These - * compare byte-for-byte under any collation, so the fast path - * cannot disagree with the operator it replaces; the zone map it - * feeds is read back through the same ordering. - */ - switch (att->atttypid) - { - case INT2OID: - writeState->colDefs[c].fastCmp = COLUMNAR_FASTCMP_I16; - break; - case INT4OID: - case DATEOID: - writeState->colDefs[c].fastCmp = COLUMNAR_FASTCMP_I32; - break; - case INT8OID: - case TIMESTAMPOID: - case TIMESTAMPTZOID: - writeState->colDefs[c].fastCmp = COLUMNAR_FASTCMP_I64; - break; - default: - writeState->colDefs[c].fastCmp = COLUMNAR_FASTCMP_NONE; - break; - } - } - - /* int2/int4: exact sum fits int64, carried in the zone map (D5) */ - writeState->colDefs[c].summableInt = - (att->atttypid == INT2OID || att->atttypid == INT4OID); - /* - * Bloom filter for hashable columns whose collation is safe (I7, gap - * 25): non-collatable types and deterministic collations, so a value - * hashes consistently between this build and an equality probe. A - * nondeterministic collation is left unbloomed. - */ - if (writeState->bloomEnabled && - OidIsValid(tce->hash_proc_finfo.fn_oid) && - PgColumnarCollationIsDeterministic(att->attcollation)) - { - writeState->colDefs[c].bloomable = true; - fmgr_info_copy(&writeState->colDefs[c].hashFn, - &tce->hash_proc_finfo, PgColumnarWriteContext); - writeState->colDefs[c].hashCollation = att->attcollation; - } + build_column_def(att, writeState->bloomEnabled, PgColumnarWriteContext, + &writeState->colDefs[c]); } } @@ -455,6 +586,7 @@ PgColumnarGetWriteState(Relation rel) * CreateTupleDescCopy would clear. */ writeState->tupdesc = CreateTupleDescCopyConstr(RelationGetDescr(rel)); writeState->natts = writeState->tupdesc->natts; + writeState->tupdescIsRel = true; /* rebuildable by a parallel-flush worker */ writeState->stripeRowLimit = pgcolumnar_stripe_row_limit; writeState->chunkGroupRowLimit = pgcolumnar_chunk_group_row_limit; writeState->compressionType = pgcolumnar_compression; @@ -1338,6 +1470,937 @@ flush_one_column(Form_pg_attribute att, List *chunkGroups, return result; } +/* + * serialize_column_input + * Serialise one column's flush_one_column input (its per-chunk-group + * buffers) into `out`, in the INPUT wire format (#445 slice 2). The bytes + * are later copied into a dsm segment behind a uint32 length prefix. Only + * the fields flush_one_column reads are written; the min/max Datums use + * datumSerialize with the column's byval/len. + */ +static void +serialize_column_input(StringInfo out, Form_pg_attribute att, List *chunkGroups, + int columnIndex) +{ + uint32 vectorCount = (uint32) list_length(chunkGroups); + ListCell *lc; + + appendBinaryStringInfo(out, (char *) &vectorCount, sizeof(uint32)); + + foreach(lc, chunkGroups) + { + ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); + ColumnChunkBuffer *col = &group->columns[columnIndex]; + uint64 groupRowCount = group->rowCount; + uint64 valueCount = col->valueCount; + int64 sum = col->sum; + uint8 hasMinMax = (uint8) (col->hasMinMax ? 1 : 0); + uint32 existsLen = (uint32) col->existsStream.len; + uint32 valueLen = (uint32) col->valueStream.len; + uint32 hashLen = (uint32) col->hashBuf.len; + + appendBinaryStringInfo(out, (char *) &groupRowCount, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &valueCount, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &sum, sizeof(int64)); + appendBinaryStringInfo(out, (char *) &hasMinMax, sizeof(uint8)); + + appendBinaryStringInfo(out, (char *) &existsLen, sizeof(uint32)); + if (existsLen > 0) + appendBinaryStringInfo(out, col->existsStream.data, existsLen); + appendBinaryStringInfo(out, (char *) &valueLen, sizeof(uint32)); + if (valueLen > 0) + appendBinaryStringInfo(out, col->valueStream.data, valueLen); + appendBinaryStringInfo(out, (char *) &hashLen, sizeof(uint32)); + if (hashLen > 0) + appendBinaryStringInfo(out, col->hashBuf.data, hashLen); + + if (col->hasMinMax) + { + Size minSpace = datumEstimateSpace(col->minValue, false, + att->attbyval, att->attlen); + Size maxSpace = datumEstimateSpace(col->maxValue, false, + att->attbyval, att->attlen); + char *ptr; + + enlargeStringInfo(out, (int) minSpace); + ptr = out->data + out->len; + datumSerialize(col->minValue, false, att->attbyval, att->attlen, + &ptr); + out->len += (int) minSpace; + + enlargeStringInfo(out, (int) maxSpace); + ptr = out->data + out->len; + datumSerialize(col->maxValue, false, att->attbyval, att->attlen, + &ptr); + out->len += (int) maxSpace; + + out->data[out->len] = '\0'; + } + } +} + +/* + * deserialize_column_input + * Rebuild the List *chunkGroups flush_one_column expects from a dsm segment + * written by serialize_column_input (#445 slice 2). dsmaddr points at a + * uint32 payload length followed by the payload. Every buffer is copied out + * of the dsm into freshly palloc'd memory so nothing points into the segment + * after it is detached. Each ChunkGroupBuffer's columns array is sized + * (columnIndex + 1) and only [columnIndex] is populated. + */ +static List * +deserialize_column_input(void *dsmaddr, Form_pg_attribute att, int columnIndex) +{ + char *base = (char *) dsmaddr; + uint32 payloadLen PG_USED_FOR_ASSERTS_ONLY; + char *cursor; + uint32 vectorCount; + uint32 v; + List *chunkGroups = NIL; + + memcpy(&payloadLen, base, sizeof(uint32)); + cursor = base + sizeof(uint32); + + memcpy(&vectorCount, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + + for (v = 0; v < vectorCount; v++) + { + ChunkGroupBuffer *group = palloc0(sizeof(ChunkGroupBuffer)); + ColumnChunkBuffer *col; + uint64 groupRowCount; + uint64 valueCount; + int64 sum; + uint8 hasMinMax; + uint32 existsLen; + uint32 valueLen; + uint32 hashLen; + + group->columns = palloc0(sizeof(ColumnChunkBuffer) * (columnIndex + 1)); + col = &group->columns[columnIndex]; + + memcpy(&groupRowCount, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&valueCount, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&sum, cursor, sizeof(int64)); + cursor += sizeof(int64); + memcpy(&hasMinMax, cursor, sizeof(uint8)); + cursor += sizeof(uint8); + + group->rowCount = groupRowCount; + col->valueCount = valueCount; + col->sum = sum; + col->hasMinMax = (hasMinMax != 0); + + memcpy(&existsLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + initStringInfo(&col->existsStream); + if (existsLen > 0) + { + appendBinaryStringInfo(&col->existsStream, cursor, existsLen); + cursor += existsLen; + } + + memcpy(&valueLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + initStringInfo(&col->valueStream); + if (valueLen > 0) + { + appendBinaryStringInfo(&col->valueStream, cursor, valueLen); + cursor += valueLen; + } + + memcpy(&hashLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + initStringInfo(&col->hashBuf); + if (hashLen > 0) + { + appendBinaryStringInfo(&col->hashBuf, cursor, hashLen); + cursor += hashLen; + } + + if (col->hasMinMax) + { + bool isnull; + + col->minValue = datumRestore(&cursor, &isnull); + col->maxValue = datumRestore(&cursor, &isnull); + } + + chunkGroups = lappend(chunkGroups, group); + } + + Assert(cursor == base + sizeof(uint32) + payloadLen); + return chunkGroups; +} + +/* + * serialize_column_result + * Serialise one column's flush_one_column result into `out`, in the RESULT + * wire format (#445 slice 2). The numeric zone sum is a varlena Datum + * (byval=false, len=-1); the encoded min/max and bloom filter are opaque + * byte buffers copied verbatim. + */ +static void +serialize_column_result(StringInfo out, FlushColumnResult *res) +{ + uint8 hasChunk = (uint8) (res->chunk != NULL ? 1 : 0); + uint8 hasDescriptor = (uint8) (res->descriptor != NULL ? 1 : 0); + int32 blockCodec = (int32) res->blockCodec; + uint32 zoneCount = (uint32) list_length(res->zoneRows); + uint8 hasBloom = (uint8) (res->bloomRow != NULL ? 1 : 0); + ListCell *lc; + + appendBinaryStringInfo(out, (char *) &hasChunk, sizeof(uint8)); + if (res->chunk != NULL) + { + uint32 chunkLen = (uint32) res->chunk->len; + + appendBinaryStringInfo(out, (char *) &chunkLen, sizeof(uint32)); + if (chunkLen > 0) + appendBinaryStringInfo(out, res->chunk->data, chunkLen); + } + + appendBinaryStringInfo(out, (char *) &hasDescriptor, sizeof(uint8)); + if (res->descriptor != NULL) + { + uint32 descLen = res->descriptorLen; + + appendBinaryStringInfo(out, (char *) &descLen, sizeof(uint32)); + if (descLen > 0) + appendBinaryStringInfo(out, res->descriptor, descLen); + } + + appendBinaryStringInfo(out, (char *) &blockCodec, sizeof(int32)); + appendBinaryStringInfo(out, (char *) &zoneCount, sizeof(uint32)); + + foreach(lc, res->zoneRows) + { + NativeZoneMapMetadata *z = (NativeZoneMapMetadata *) lfirst(lc); + uint64 storageId = z->storageId; + uint64 groupNumber = z->groupNumber; + int32 columnIndex = (int32) z->columnIndex; + int32 vectorIndex = (int32) z->vectorIndex; + uint64 valueCount = z->valueCount; + uint64 nullCount = z->nullCount; + uint8 hasSum = (uint8) (z->hasSum ? 1 : 0); + uint8 zHasMinMax = (uint8) (z->hasMinMax ? 1 : 0); + + appendBinaryStringInfo(out, (char *) &storageId, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &groupNumber, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &columnIndex, sizeof(int32)); + appendBinaryStringInfo(out, (char *) &vectorIndex, sizeof(int32)); + appendBinaryStringInfo(out, (char *) &valueCount, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &nullCount, sizeof(uint64)); + + appendBinaryStringInfo(out, (char *) &hasSum, sizeof(uint8)); + if (z->hasSum) + { + Size sumSpace = datumEstimateSpace(z->sum, false, false, -1); + char *ptr; + + enlargeStringInfo(out, (int) sumSpace); + ptr = out->data + out->len; + datumSerialize(z->sum, false, false, -1, &ptr); + out->len += (int) sumSpace; + out->data[out->len] = '\0'; + } + + appendBinaryStringInfo(out, (char *) &zHasMinMax, sizeof(uint8)); + if (z->hasMinMax) + { + uint32 minLen = z->minimumLen; + uint32 maxLen = z->maximumLen; + + appendBinaryStringInfo(out, (char *) &minLen, sizeof(uint32)); + if (minLen > 0) + appendBinaryStringInfo(out, z->minimum, minLen); + appendBinaryStringInfo(out, (char *) &maxLen, sizeof(uint32)); + if (maxLen > 0) + appendBinaryStringInfo(out, z->maximum, maxLen); + } + } + + appendBinaryStringInfo(out, (char *) &hasBloom, sizeof(uint8)); + if (res->bloomRow != NULL) + { + NativeBloomMetadata *b = res->bloomRow; + uint64 storageId = b->storageId; + uint64 groupNumber = b->groupNumber; + int32 columnIndex = (int32) b->columnIndex; + uint32 filterLen = b->filterLen; + + appendBinaryStringInfo(out, (char *) &storageId, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &groupNumber, sizeof(uint64)); + appendBinaryStringInfo(out, (char *) &columnIndex, sizeof(int32)); + appendBinaryStringInfo(out, (char *) &filterLen, sizeof(uint32)); + if (filterLen > 0) + appendBinaryStringInfo(out, b->filter, filterLen); + } +} + +/* + * deserialize_column_result + * Rebuild a FlushColumnResult from a dsm segment written by + * serialize_column_result (#445 slice 2). dsmaddr points at a uint32 payload + * length followed by the payload. Every buffer (chunk, descriptor, zone + * min/max, bloom filter) and the numeric sum Datum are copied out of the dsm + * into palloc'd memory so nothing points into the segment after detach. + */ +static FlushColumnResult +deserialize_column_result(void *dsmaddr) +{ + char *base = (char *) dsmaddr; + uint32 payloadLen PG_USED_FOR_ASSERTS_ONLY; + char *cursor; + FlushColumnResult result; + uint8 hasChunk; + uint8 hasDescriptor; + int32 blockCodec; + uint32 zoneCount; + uint32 z; + uint8 hasBloom; + List *zoneRows = NIL; + + memcpy(&payloadLen, base, sizeof(uint32)); + cursor = base + sizeof(uint32); + + result.chunk = NULL; + result.descriptor = NULL; + result.descriptorLen = 0; + result.blockCodec = COLUMNAR_COMPRESSION_NONE; + result.zoneRows = NIL; + result.bloomRow = NULL; + + memcpy(&hasChunk, cursor, sizeof(uint8)); + cursor += sizeof(uint8); + if (hasChunk) + { + uint32 chunkLen; + StringInfo chunk = makeStringInfo(); + + memcpy(&chunkLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + if (chunkLen > 0) + { + appendBinaryStringInfo(chunk, cursor, chunkLen); + cursor += chunkLen; + } + result.chunk = chunk; + } + + memcpy(&hasDescriptor, cursor, sizeof(uint8)); + cursor += sizeof(uint8); + if (hasDescriptor) + { + uint32 descLen; + + memcpy(&descLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + if (descLen > 0) + { + char *desc = palloc(descLen); + + memcpy(desc, cursor, descLen); + cursor += descLen; + result.descriptor = desc; + } + else + result.descriptor = palloc(0); + result.descriptorLen = descLen; + } + + memcpy(&blockCodec, cursor, sizeof(int32)); + cursor += sizeof(int32); + result.blockCodec = blockCodec; + + memcpy(&zoneCount, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + + for (z = 0; z < zoneCount; z++) + { + NativeZoneMapMetadata *zm = palloc0(sizeof(NativeZoneMapMetadata)); + uint64 storageId; + uint64 groupNumber; + int32 columnIndex; + int32 vectorIndex; + uint64 valueCount; + uint64 nullCount; + uint8 hasSum; + uint8 zHasMinMax; + + memcpy(&storageId, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&groupNumber, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&columnIndex, cursor, sizeof(int32)); + cursor += sizeof(int32); + memcpy(&vectorIndex, cursor, sizeof(int32)); + cursor += sizeof(int32); + memcpy(&valueCount, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&nullCount, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + + zm->storageId = storageId; + zm->groupNumber = groupNumber; + zm->columnIndex = columnIndex; + zm->vectorIndex = vectorIndex; + zm->valueCount = valueCount; + zm->nullCount = nullCount; + + memcpy(&hasSum, cursor, sizeof(uint8)); + cursor += sizeof(uint8); + if (hasSum) + { + bool isnull; + + zm->hasSum = true; + zm->sum = datumRestore(&cursor, &isnull); + } + + memcpy(&zHasMinMax, cursor, sizeof(uint8)); + cursor += sizeof(uint8); + if (zHasMinMax) + { + uint32 minLen; + uint32 maxLen; + + zm->hasMinMax = true; + + memcpy(&minLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + if (minLen > 0) + { + char *mn = palloc(minLen); + + memcpy(mn, cursor, minLen); + cursor += minLen; + zm->minimum = mn; + } + else + zm->minimum = palloc(0); + zm->minimumLen = minLen; + + memcpy(&maxLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + if (maxLen > 0) + { + char *mx = palloc(maxLen); + + memcpy(mx, cursor, maxLen); + cursor += maxLen; + zm->maximum = mx; + } + else + zm->maximum = palloc(0); + zm->maximumLen = maxLen; + } + + zoneRows = lappend(zoneRows, zm); + } + result.zoneRows = zoneRows; + + memcpy(&hasBloom, cursor, sizeof(uint8)); + cursor += sizeof(uint8); + if (hasBloom) + { + NativeBloomMetadata *b = palloc0(sizeof(NativeBloomMetadata)); + uint64 storageId; + uint64 groupNumber; + int32 columnIndex; + uint32 filterLen; + + memcpy(&storageId, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&groupNumber, cursor, sizeof(uint64)); + cursor += sizeof(uint64); + memcpy(&columnIndex, cursor, sizeof(int32)); + cursor += sizeof(int32); + memcpy(&filterLen, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + + b->storageId = storageId; + b->groupNumber = groupNumber; + b->columnIndex = columnIndex; + if (filterLen > 0) + { + char *f = palloc(filterLen); + + memcpy(f, cursor, filterLen); + cursor += filterLen; + b->filter = f; + } + else + b->filter = palloc(0); + b->filterLen = filterLen; + + result.bloomRow = b; + } + + Assert(cursor == base + sizeof(uint32) + payloadLen); + return result; +} + +/* + * pflush_auto_workers + * Worker count for the parallel flush: half the admin's max_parallel_workers + * budget, at least one, capped at PFLUSH_MAX_WORKERS. Mirrors the export + * path's pexport_auto_workers (that copy is static in another file). + */ +static int +pflush_auto_workers(void) +{ + const char *s = GetConfigOption("max_parallel_workers", true, false); + int budget = (s != NULL) ? atoi(s) : 8; + int n = budget / 2; + + if (n < 1) + n = 1; + if (n > PFLUSH_MAX_WORKERS) + n = PFLUSH_MAX_WORKERS; + return n; +} + +/* + * pgcolumnar_parallel_flush_worker + * Background-worker entry (#445 slice 3): attach the input DSM, connect, + * and run flush_one_column for every column it can claim off the shared + * atomic counter, publishing the results in its own OUTPUT segment. + * + * Modelled on pgcolumnar_parallel_export_worker. Unlike the export path + * this does NOT import the launcher's snapshot: the value data lives in the + * DSM, not the table, so the worker reads only committed catalog (the + * relation's tupdesc). StartTransactionCommand plus a pushed + * GetTransactionSnapshot is enough to open the relation; no + * ExportSnapshot/ImportSnapshot is needed (owner's default decision). + */ +PGDLLEXPORT void +pgcolumnar_parallel_flush_worker(Datum main_arg) +{ + dsm_segment *seg; + shm_toc *toc; + PflushHeader *hdr; + PflushWorkerSlot *slots; + char *inputs; + uint64 *inoffs; + PflushVerdict *verds; + pg_atomic_uint32 *claim; + PflushWorkerSlot *me; + int widx; + uint32 conn_flags = BGWORKER_BYPASS_ALLOWCONN; + + memcpy(&widx, MyBgworkerEntry->bgw_extra, sizeof(int)); + + pqsignal(SIGTERM, die); + BackgroundWorkerUnblockSignals(); + + seg = dsm_attach(DatumGetUInt32(main_arg)); + if (seg == NULL) + ereport(ERROR, + (errcode(ERRCODE_INTERNAL_ERROR), + errmsg("pgcolumnar parallel_flush worker could not attach to the shared segment"))); + toc = shm_toc_attach(PFLUSH_MAGIC, dsm_segment_address(seg)); + if (toc == NULL) + ereport(ERROR, + (errcode(ERRCODE_INTERNAL_ERROR), + errmsg("pgcolumnar parallel_flush worker found a bad shared segment"))); + hdr = (PflushHeader *) shm_toc_lookup(toc, PFLUSH_KEY_HEADER, false); + inputs = (char *) shm_toc_lookup(toc, PFLUSH_KEY_INPUTS, false); + inoffs = (uint64 *) shm_toc_lookup(toc, PFLUSH_KEY_INOFFS, false); + verds = (PflushVerdict *) shm_toc_lookup(toc, PFLUSH_KEY_VERDICTS, false); + slots = (PflushWorkerSlot *) shm_toc_lookup(toc, PFLUSH_KEY_SLOTS, false); + 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); + + /* + * A fresh snapshot for catalog/tupdesc reads only. A plain + * GetTransactionSnapshot pushed onto the active stack lets table_open resolve + * the relation; no ImportSnapshot, because the worker never reads the + * backend's uncommitted rows -- the value bytes are all in the DSM. + */ + StartTransactionCommand(); + PushActiveSnapshot(GetTransactionSnapshot()); + + PG_TRY(); + { + Relation rel = table_open(hdr->relid, AccessShareLock); + TupleDesc tupdesc = RelationGetDescr(rel); + StringInfoData recbuf; + uint32 count = 0; + dsm_segment *outseg; + char *obase; + + /* + * Adopt the launcher's live encoding GUCs so the bytes this worker + * produces match the serial path exactly (see PflushHeader). + */ + pgcolumnar_fsst_verdict_reuse = hdr->fsstVerdictReuse; + pgcolumnar_fsst_min_gain_percent = hdr->fsstMinGainPercent; + pgcolumnar_encoding_sample_rows = hdr->encodingSampleRows; + + initStringInfo(&recbuf); + + for (;;) + { + uint32 c = pg_atomic_fetch_add_u32(claim, 1); + Form_pg_attribute att; + PgColumnarColumnDef def; + List *groups; + FlushColumnResult res; + StringInfoData outbuf; + uint32 resultLen; + int8 verdict; + int32 age; + + if (c >= (uint32) hdr->natts) + break; + + att = TupleDescAttr(tupdesc, c); + + /* reconstruct the column def with the SAME logic the backend uses */ + build_column_def(att, hdr->bloomEnabled, CurrentMemoryContext, &def); + def.fsstVerdict = verds[c].verdict; + def.fsstVerdictAge = verds[c].age; + + /* rebuild this column's input from the DSM and flush it */ + groups = deserialize_column_input(inputs + inoffs[c], att, (int) c); + res = flush_one_column(att, groups, &def, hdr->rowCount, + hdr->validityBytes, hdr->encodeEffort, + hdr->compressionType, hdr->compressionLevel, + hdr->storageId, hdr->groupNumber, (int) c); + + /* record (c, updated verdict/age, serialized result) */ + initStringInfo(&outbuf); + serialize_column_result(&outbuf, &res); + resultLen = (uint32) outbuf.len; + verdict = def.fsstVerdict; + age = (int32) def.fsstVerdictAge; + + appendBinaryStringInfo(&recbuf, (char *) &c, sizeof(uint32)); + appendBinaryStringInfo(&recbuf, (char *) &verdict, sizeof(int8)); + appendBinaryStringInfo(&recbuf, (char *) &age, sizeof(int32)); + appendBinaryStringInfo(&recbuf, (char *) &resultLen, sizeof(uint32)); + if (resultLen > 0) + appendBinaryStringInfo(&recbuf, outbuf.data, outbuf.len); + count++; + } + + /* + * Publish one OUTPUT segment: {uint32 count; per column: uint32 c, + * int8 verdict, int32 age, uint32 resultLen, result bytes}. The + * (resultLen, result bytes) pair is exactly the length-prefixed payload + * deserialize_column_result reads, so the backend deserialises straight + * from the record. Pin the segment so it survives this worker's exit -- + * the backend attaches only after WaitForBackgroundWorkerShutdown, by + * which point an unpinned segment (zero mappings) would be gone. + */ + outseg = dsm_create((Size) sizeof(uint32) + recbuf.len, 0); + obase = (char *) dsm_segment_address(outseg); + memcpy(obase, &count, sizeof(uint32)); + if (recbuf.len > 0) + memcpy(obase + sizeof(uint32), recbuf.data, recbuf.len); + dsm_pin_segment(outseg); + me->outHandle = dsm_segment_handle(outseg); + me->outLen = (uint32) (sizeof(uint32) + recbuf.len); + dsm_detach(outseg); /* pinned: persists for the backend to attach */ + + table_close(rel, AccessShareLock); + PopActiveSnapshot(); + CommitTransactionCommand(); + pg_atomic_write_u32(&me->state, PFLUSH_DONE); + } + PG_CATCH(); + { + ErrorData *edata; + + MemoryContextSwitchTo(TopMemoryContext); + edata = CopyErrorData(); + me->sqlerrcode = edata->sqlerrcode; + strlcpy(me->errmsg, + edata->message ? edata->message : "parallel_flush worker failed", + sizeof(me->errmsg)); + FlushErrorState(); + FreeErrorData(edata); + AbortOutOfAnyTransaction(); + pg_atomic_write_u32(&me->state, PFLUSH_FAILED); + } + PG_END_TRY(); + + dsm_detach(seg); + proc_exit(0); +} + +/* + * flush_columns_parallel + * The #445 slice-3 parallel flush: dispatch flush_one_column across a pool + * of background workers and fill colResults[natts] with every column's + * result. Registration failure or a worker error is not fatal -- the + * backend completes any column a worker did not produce serially, in this + * process, so colResults ends up populated for all of [0, natts). The FSST + * verdict cache in writeState->colDefs is updated in column order (each + * column is owned by exactly one flush, so there is no cross-worker race). + * The caller assembles colResults and does all I/O and catalog writes. + */ +static void +flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, + uint64 rowCount, int validityBytes, + FlushColumnResult *colResults) +{ + int natts = writeState->natts; + int nworkers = Min(natts, pflush_auto_workers()); + bool *done = palloc0(sizeof(bool) * natts); + StringInfoData inputs; + uint64 *inoffs = palloc(sizeof(uint64) * (natts + 1)); + shm_toc_estimator est; + Size segsize; + Size inputsChunk; + dsm_segment *seg; + shm_toc *toc; + PflushHeader *hdr; + char *inbuf; + uint64 *offs; + PflushVerdict *verds; + PflushWorkerSlot *slots; + pg_atomic_uint32 *claim; + uint32 dsmh; + BackgroundWorker bw; + BackgroundWorkerHandle **handles; + int nstarted = 0; + int nWorkerCols = 0; + int c; + int i; + + /* concatenate every column's length-prefixed serialize_column_input blob */ + initStringInfo(&inputs); + for (c = 0; c < natts; c++) + { + Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); + StringInfoData one; + uint32 len; + + inoffs[c] = (uint64) inputs.len; + initStringInfo(&one); + serialize_column_input(&one, att, writeState->chunkGroups, c); + len = (uint32) one.len; + appendBinaryStringInfo(&inputs, (char *) &len, sizeof(uint32)); + if (len > 0) + appendBinaryStringInfo(&inputs, one.data, one.len); + pfree(one.data); + } + inoffs[natts] = (uint64) inputs.len; + + /* one input DSM segment: header, inputs, offsets, verdict seeds, slots, claim */ + inputsChunk = (inputs.len > 0) ? (Size) inputs.len : 1; + shm_toc_initialize_estimator(&est); + shm_toc_estimate_chunk(&est, sizeof(PflushHeader)); + shm_toc_estimate_chunk(&est, inputsChunk); + shm_toc_estimate_chunk(&est, mul_size(sizeof(uint64), natts + 1)); + shm_toc_estimate_chunk(&est, mul_size(sizeof(PflushVerdict), natts)); + shm_toc_estimate_chunk(&est, mul_size(sizeof(PflushWorkerSlot), nworkers)); + shm_toc_estimate_chunk(&est, sizeof(pg_atomic_uint32)); + shm_toc_estimate_keys(&est, 6); + segsize = shm_toc_estimate(&est); + + seg = dsm_create(segsize, 0); + toc = shm_toc_create(PFLUSH_MAGIC, dsm_segment_address(seg), segsize); + + hdr = (PflushHeader *) shm_toc_allocate(toc, sizeof(PflushHeader)); + shm_toc_insert(toc, PFLUSH_KEY_HEADER, hdr); + inbuf = (char *) shm_toc_allocate(toc, inputsChunk); + shm_toc_insert(toc, PFLUSH_KEY_INPUTS, inbuf); + offs = (uint64 *) shm_toc_allocate(toc, mul_size(sizeof(uint64), natts + 1)); + shm_toc_insert(toc, PFLUSH_KEY_INOFFS, offs); + verds = (PflushVerdict *) shm_toc_allocate(toc, mul_size(sizeof(PflushVerdict), natts)); + shm_toc_insert(toc, PFLUSH_KEY_VERDICTS, verds); + slots = (PflushWorkerSlot *) shm_toc_allocate(toc, + mul_size(sizeof(PflushWorkerSlot), nworkers)); + shm_toc_insert(toc, PFLUSH_KEY_SLOTS, slots); + claim = (pg_atomic_uint32 *) shm_toc_allocate(toc, sizeof(pg_atomic_uint32)); + shm_toc_insert(toc, PFLUSH_KEY_CLAIM, claim); + + if (inputs.len > 0) + memcpy(inbuf, inputs.data, inputs.len); + memcpy(offs, inoffs, sizeof(uint64) * (natts + 1)); + for (c = 0; c < natts; c++) + { + verds[c].verdict = writeState->colDefs[c].fsstVerdict; + verds[c].age = (int32) writeState->colDefs[c].fsstVerdictAge; + } + pg_atomic_init_u32(claim, 0); + + hdr->dbid = MyDatabaseId; + hdr->roleid = GetUserId(); + hdr->relid = writeState->relid; + hdr->storageId = writeState->storageId; + hdr->groupNumber = groupNumber; + hdr->rowCount = rowCount; + hdr->validityBytes = validityBytes; + hdr->encodeEffort = writeState->encodeEffort; + hdr->compressionType = writeState->compressionType; + hdr->compressionLevel = writeState->compressionLevel; + hdr->natts = natts; + hdr->nworkers = nworkers; + hdr->bloomEnabled = writeState->bloomEnabled; + hdr->fsstVerdictReuse = pgcolumnar_fsst_verdict_reuse; + hdr->fsstMinGainPercent = pgcolumnar_fsst_min_gain_percent; + hdr->encodingSampleRows = pgcolumnar_encoding_sample_rows; + + for (i = 0; i < nworkers; i++) + { + pg_atomic_init_u32(&slots[i].state, PFLUSH_PENDING); + slots[i].outHandle = DSM_HANDLE_INVALID; + slots[i].outLen = 0; + slots[i].sqlerrcode = 0; + slots[i].errmsg[0] = '\0'; + } + + dsmh = dsm_segment_handle(seg); + + /* + * Register the workers. A registration failure is NOT an error: the started + * workers self-balance via the claim counter (a single worker will claim + * every column), and any column left unclaimed is completed serially below. + */ + handles = (BackgroundWorkerHandle **) + palloc0(sizeof(BackgroundWorkerHandle *) * nworkers); + + memset(&bw, 0, sizeof(bw)); + bw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION; + bw.bgw_start_time = BgWorkerStart_RecoveryFinished; + bw.bgw_restart_time = BGW_NEVER_RESTART; + strlcpy(bw.bgw_library_name, "pgcolumnar", BGW_MAXLEN); + strlcpy(bw.bgw_function_name, "pgcolumnar_parallel_flush_worker", BGW_MAXLEN); + snprintf(bw.bgw_name, BGW_MAXLEN, "pgcolumnar parallel_flush worker"); + snprintf(bw.bgw_type, BGW_MAXLEN, "pgcolumnar parallel_flush worker"); + bw.bgw_main_arg = UInt32GetDatum(dsmh); + bw.bgw_notify_pid = MyProcPid; + + for (i = 0; i < nworkers; i++) + { + memcpy(bw.bgw_extra, &i, sizeof(int)); + if (RegisterDynamicBackgroundWorker(&bw, &handles[i])) + nstarted++; + else + { + handles[i] = NULL; + break; /* slot exhausted: the rest is done serially */ + } + } + + /* wait for every started worker to finish (SIGTERM handler is die) */ + for (i = 0; i < nstarted; i++) + if (handles[i] != NULL) + WaitForBackgroundWorkerShutdown(handles[i]); + + /* + * Collect. For each DONE worker, attach its OUTPUT segment and stash every + * column it produced, applying the returned FSST verdict. A FAILED worker's + * columns are left undone (a WARNING is logged); a worker that never started + * leaves state PENDING. Everything not done is completed serially afterwards. + */ + for (i = 0; i < nstarted; i++) + { + uint32 st = pg_atomic_read_u32(&slots[i].state); + + if (st == PFLUSH_DONE && slots[i].outHandle != DSM_HANDLE_INVALID) + { + dsm_segment *outseg = dsm_attach(slots[i].outHandle); + char *obase; + char *cursor; + uint32 count; + uint32 k; + + if (outseg == NULL) + continue; /* lost the segment: fall to serial completion */ + + /* + * Release the worker's pin now that we hold a mapping: the segment + * lives until our dsm_detach below, and if anything between here and + * there throws, the resource owner detaches our (now unpinned) mapping + * and the segment is freed rather than leaked to postmaster restart. + */ + dsm_unpin_segment(slots[i].outHandle); + + obase = (char *) dsm_segment_address(outseg); + memcpy(&count, obase, sizeof(uint32)); + cursor = obase + sizeof(uint32); + for (k = 0; k < count; k++) + { + uint32 col; + int8 verdict; + int32 age; + uint32 resultLen; + + memcpy(&col, cursor, sizeof(uint32)); + cursor += sizeof(uint32); + memcpy(&verdict, cursor, sizeof(int8)); + cursor += sizeof(int8); + memcpy(&age, cursor, sizeof(int32)); + cursor += sizeof(int32); + memcpy(&resultLen, cursor, sizeof(uint32)); + + /* cursor points at [uint32 resultLen][payload] -> deserialize */ + colResults[col] = deserialize_column_result(cursor); + cursor += sizeof(uint32) + resultLen; + + writeState->colDefs[col].fsstVerdict = verdict; + writeState->colDefs[col].fsstVerdictAge = age; + done[col] = true; + nWorkerCols++; + } + dsm_detach(outseg); + } + else if (st == PFLUSH_FAILED) + { + ereport(WARNING, + (errmsg("pgcolumnar parallel_flush worker %d failed, completing its columns serially: %s", + i, slots[i].errmsg[0] ? slots[i].errmsg : "unknown error"))); + } + } + + dsm_detach(seg); + pfree(inputs.data); + + /* + * Observability + the test premise: how the flush was actually split. A run + * with the GUC on but every column done serially (slot starvation) reports 0 + * worker columns, so a silent fall-back to serial cannot be mistaken for a + * working parallel flush. + */ + elog(DEBUG1, "pgcolumnar parallel_flush: %d of %d columns by %d worker(s), %d serial", + nWorkerCols, natts, nstarted, natts - nWorkerCols); + + /* + * Serial completion of the remainder: every column no worker produced + * (unstarted slot, failed worker, or lost segment). flush_one_column mutates + * writeState->colDefs[c] in place, so its verdict is applied by the call. + */ + for (c = 0; c < natts; c++) + { + Form_pg_attribute att; + PgColumnarColumnDef *def; + + if (done[c]) + continue; + + att = TupleDescAttr(writeState->tupdesc, c); + def = &writeState->colDefs[c]; + colResults[c] = flush_one_column(att, writeState->chunkGroups, def, + rowCount, validityBytes, + writeState->encodeEffort, + writeState->compressionType, + writeState->compressionLevel, + writeState->storageId, groupNumber, c); + } +} + /* * pgcolumnar_flush_row_group * Native-format (PGCN v1) flush. Lay out the accumulated rows as one row @@ -1418,27 +2481,72 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) * so an incompressible column stays byte-for-byte the D2b baseline plus the * descriptor. */ - for (c = 0; c < natts; c++) + /* + * #445 slice 3: with pgcolumnar.parallel_flush on and at least two columns + * to spread, dispatch flush_one_column across a worker pool, degrading to + * serial completion of any column a worker did not produce (byte-identical to + * the serial path either way). tupdescIsRel keeps a projection's inner writer + * (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) { - Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); - FlushColumnResult res = flush_one_column(att, writeState->chunkGroups, - &writeState->colDefs[c], rowCount, - validityBytes, writeState->encodeEffort, - writeState->compressionType, - writeState->compressionLevel, - writeState->storageId, groupNumber, c); - - chunkOffset[c] = data->len; - if (res.chunk != NULL && res.chunk->len > 0) - appendBinaryStringInfo(data, res.chunk->data, res.chunk->len); - chunkLength[c] = data->len - chunkOffset[c]; - chunkDescriptor[c] = res.descriptor; - chunkDescriptorLen[c] = res.descriptorLen; - chunkBlockCodec[c] = res.blockCodec; - if (res.zoneRows != NIL) - zoneRows = list_concat(zoneRows, res.zoneRows); - if (res.bloomRow != NULL) - bloomRows = lappend(bloomRows, res.bloomRow); + FlushColumnResult *colResults = palloc0(sizeof(FlushColumnResult) * natts); + + flush_columns_parallel(writeState, groupNumber, rowCount, validityBytes, + colResults); + + /* assemble in column order (identical to the serial assembly) */ + for (c = 0; c < natts; c++) + { + FlushColumnResult *r = &colResults[c]; + + chunkOffset[c] = data->len; + if (r->chunk != NULL && r->chunk->len > 0) + appendBinaryStringInfo(data, r->chunk->data, r->chunk->len); + chunkLength[c] = data->len - chunkOffset[c]; + chunkDescriptor[c] = r->descriptor; + chunkDescriptorLen[c] = r->descriptorLen; + chunkBlockCodec[c] = r->blockCodec; + if (r->zoneRows != NIL) + zoneRows = list_concat(zoneRows, r->zoneRows); + if (r->bloomRow != NULL) + bloomRows = lappend(bloomRows, r->bloomRow); + } + } + else + { + /* + * Serial / non-dispatched path: call flush_one_column directly, no dsm. + * The serialize/dsm round-trip only earns its cost when a worker will read + * the bytes across a process boundary; in-backend-serial it is pure + * overhead (a per-column dsm_create pair), so the non-dispatched path is + * the same direct call as slice 1 (#589) and OFF stays byte-identical to + * and as fast as main. The dsm crossing lives solely in + * flush_columns_parallel, on the worker path where it pays for itself. + */ + for (c = 0; c < natts; c++) + { + Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); + FlushColumnResult res = flush_one_column(att, writeState->chunkGroups, + &writeState->colDefs[c], rowCount, + validityBytes, writeState->encodeEffort, + writeState->compressionType, + writeState->compressionLevel, + writeState->storageId, groupNumber, c); + + chunkOffset[c] = data->len; + if (res.chunk != NULL && res.chunk->len > 0) + appendBinaryStringInfo(data, res.chunk->data, res.chunk->len); + chunkLength[c] = data->len - chunkOffset[c]; + chunkDescriptor[c] = res.descriptor; + chunkDescriptorLen[c] = res.descriptorLen; + chunkBlockCodec[c] = res.blockCodec; + if (res.zoneRows != NIL) + zoneRows = list_concat(zoneRows, res.zoneRows); + if (res.bloomRow != NULL) + bloomRows = lappend(bloomRows, res.bloomRow); + } } dataLength = data->len;