From dd7bccf815d1de4d8b86f65cec7629459ad7abfa Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 19:30:04 -0600 Subject: [PATCH 1/5] feat: round-trip flush_one_column through a dsm segment, serial (#445 slice 2) Slice 2 of the #445 in-COPY parallelism design (#588): serialise each column's flush_one_column input (its per-chunk-group buffers + min/max Datums + counts) into a dsm segment and its result (chunk bytes, descriptor, codec, zone rows, bloom) back, run serially in the backend with no workers. Proves the input/output serialisation is byte-identical before slice 3 adds the worker pool. Reconstructs a minimal per-column chunkGroups on the read side so slice 1's flush_one_column signature is untouched; every buffer is copied out of the dsm before detach. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar_write_state.c | 539 +++++++++++++++++++++++++++++++++++-- 1 file changed, 523 insertions(+), 16 deletions(-) diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 7123311..b2f6f6f 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -24,6 +24,7 @@ #include "catalog/pg_type.h" #include "executor/tuptable.h" #include "miscadmin.h" +#include "storage/dsm.h" #include "storage/lmgr.h" #include "utils/builtins.h" #include "utils/datum.h" @@ -1338,6 +1339,479 @@ 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; +} + /* * pgcolumnar_flush_row_group * Native-format (PGCN v1) flush. Lay out the accumulated rows as one row @@ -1418,27 +1892,60 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) * so an incompressible column stays byte-for-byte the D2b baseline plus the * descriptor. */ + /* + * slice 2: round-trip each column's input and result through a dsm segment, + * still serially in this backend, to prove the serialisation before slice 3's + * workers. + */ 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); - + PgColumnarColumnDef *def = &writeState->colDefs[c]; + StringInfoData inbuf; + dsm_segment *inseg; + List *rtGroups; + FlushColumnResult res; + StringInfoData outbuf; + dsm_segment *outseg; + FlushColumnResult rtRes; + + /* serialise this column's input, ship it through a dsm segment, read it back */ + initStringInfo(&inbuf); + serialize_column_input(&inbuf, att, writeState->chunkGroups, c); + inseg = dsm_create(inbuf.len + sizeof(uint32), 0); + memcpy(dsm_segment_address(inseg), &inbuf.len, sizeof(uint32)); + memcpy((char *) dsm_segment_address(inseg) + sizeof(uint32), inbuf.data, inbuf.len); + rtGroups = deserialize_column_input(dsm_segment_address(inseg), att, c); + + /* run the pure function on the round-tripped input */ + res = flush_one_column(att, rtGroups, def, rowCount, validityBytes, + writeState->encodeEffort, writeState->compressionType, + writeState->compressionLevel, writeState->storageId, + groupNumber, c); + + /* serialise the result, ship it through a dsm segment, read it back */ + initStringInfo(&outbuf); + serialize_column_result(&outbuf, &res); + outseg = dsm_create(outbuf.len + sizeof(uint32), 0); + memcpy(dsm_segment_address(outseg), &outbuf.len, sizeof(uint32)); + memcpy((char *) dsm_segment_address(outseg) + sizeof(uint32), outbuf.data, outbuf.len); + rtRes = deserialize_column_result(dsm_segment_address(outseg)); + + /* assemble from the round-tripped result (identical to slice 1's assembly) */ chunkOffset[c] = data->len; - if (res.chunk != NULL && res.chunk->len > 0) - appendBinaryStringInfo(data, res.chunk->data, res.chunk->len); + if (rtRes.chunk != NULL && rtRes.chunk->len > 0) + appendBinaryStringInfo(data, rtRes.chunk->data, rtRes.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); + chunkDescriptor[c] = rtRes.descriptor; + chunkDescriptorLen[c] = rtRes.descriptorLen; + chunkBlockCodec[c] = rtRes.blockCodec; + if (rtRes.zoneRows != NIL) + zoneRows = list_concat(zoneRows, rtRes.zoneRows); + if (rtRes.bloomRow != NULL) + bloomRows = lappend(bloomRows, rtRes.bloomRow); + + dsm_detach(inseg); + dsm_detach(outseg); } dataLength = data->len; From 014b70dabcc06439be440977a8e84878de44fd5c Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 20:27:03 -0600 Subject: [PATCH 2/5] docs: design for slice 3 (worker pool for the flush) (#445) Design-before-code for the concurrency slice: dispatch flush_one_column across background workers reading the slice-2 dsm, backend collects in column order and keeps all I/O, degrading to serial when worker slots are unavailable. Captures the FSST-verdict-cache threading needed for byte-identity, the slot-starvation degradation, and two open questions for the design owner. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- design/ISSUE_445_SLICE3_WORKERS.md | 102 +++++++++++++++++++++++++++++ 1 file changed, 102 insertions(+) create mode 100644 design/ISSUE_445_SLICE3_WORKERS.md 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). From 827fa4701ec020da587270d69b0fe9c4953ff326 Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 21:14:14 -0600 Subject: [PATCH 3/5] feat: dispatch the stripe flush across a worker pool, off by default (#445 slice 3) Slice 3 of the #445 in-COPY parallelism design. A new pgcolumnar.parallel_flush GUC (default off) dispatches flush_one_column across background workers reading the slice-2 dsm; the backend collects results in column order, applies the FSST verdict updates, and keeps all I/O and catalog writes. Degrades to serial completion in the backend for any column no worker produced (slot starvation or worker failure), so the row count is always correct. Off by default => no behaviour change; on => byte-identical to serial. build_column_def is extracted so the worker and the backend build an identical PgColumnarColumnDef. Encoding GUCs the encoder reads live are threaded through the dsm header so a worker's bytes match the launching session. Projection inner writers flush serially (their tupdesc is synthetic, not the relation's). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar.h | 1 + src/columnar_tableam.c | 17 + src/columnar_write_state.c | 827 ++++++++++++++++++++++++++++++++----- 3 files changed, 735 insertions(+), 110 deletions(-) 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 b2f6f6f..4d456f8 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -24,18 +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. @@ -166,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 */ @@ -225,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 @@ -271,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 @@ -288,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]); } } @@ -456,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; @@ -1812,6 +1943,446 @@ deserialize_column_result(void *dsmaddr) 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 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 */ + + 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; + } + dsm_unpin_segment(slots[i].outHandle); + 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); + + /* + * 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 @@ -1893,59 +2464,95 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) * descriptor. */ /* - * slice 2: round-trip each column's input and result through a dsm segment, - * still serially in this backend, to prove the serialisation before slice 3's - * workers. + * #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. */ - for (c = 0; c < natts; c++) + if (pgcolumnar_parallel_flush && natts >= 2 && writeState->tupdescIsRel) { - Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); - PgColumnarColumnDef *def = &writeState->colDefs[c]; - StringInfoData inbuf; - dsm_segment *inseg; - List *rtGroups; - FlushColumnResult res; - StringInfoData outbuf; - dsm_segment *outseg; - FlushColumnResult rtRes; - - /* serialise this column's input, ship it through a dsm segment, read it back */ - initStringInfo(&inbuf); - serialize_column_input(&inbuf, att, writeState->chunkGroups, c); - inseg = dsm_create(inbuf.len + sizeof(uint32), 0); - memcpy(dsm_segment_address(inseg), &inbuf.len, sizeof(uint32)); - memcpy((char *) dsm_segment_address(inseg) + sizeof(uint32), inbuf.data, inbuf.len); - rtGroups = deserialize_column_input(dsm_segment_address(inseg), att, c); - - /* run the pure function on the round-tripped input */ - res = flush_one_column(att, rtGroups, def, rowCount, validityBytes, - writeState->encodeEffort, writeState->compressionType, - writeState->compressionLevel, writeState->storageId, - groupNumber, c); - - /* serialise the result, ship it through a dsm segment, read it back */ - initStringInfo(&outbuf); - serialize_column_result(&outbuf, &res); - outseg = dsm_create(outbuf.len + sizeof(uint32), 0); - memcpy(dsm_segment_address(outseg), &outbuf.len, sizeof(uint32)); - memcpy((char *) dsm_segment_address(outseg) + sizeof(uint32), outbuf.data, outbuf.len); - rtRes = deserialize_column_result(dsm_segment_address(outseg)); - - /* assemble from the round-tripped result (identical to slice 1's assembly) */ - chunkOffset[c] = data->len; - if (rtRes.chunk != NULL && rtRes.chunk->len > 0) - appendBinaryStringInfo(data, rtRes.chunk->data, rtRes.chunk->len); - chunkLength[c] = data->len - chunkOffset[c]; - chunkDescriptor[c] = rtRes.descriptor; - chunkDescriptorLen[c] = rtRes.descriptorLen; - chunkBlockCodec[c] = rtRes.blockCodec; - if (rtRes.zoneRows != NIL) - zoneRows = list_concat(zoneRows, rtRes.zoneRows); - if (rtRes.bloomRow != NULL) - bloomRows = lappend(bloomRows, rtRes.bloomRow); - - dsm_detach(inseg); - dsm_detach(outseg); + 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 + { + /* + * slice 2: round-trip each column's input and result through a dsm segment, + * still serially in this backend, to prove the serialisation before slice 3's + * workers. + */ + for (c = 0; c < natts; c++) + { + Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); + PgColumnarColumnDef *def = &writeState->colDefs[c]; + StringInfoData inbuf; + dsm_segment *inseg; + List *rtGroups; + FlushColumnResult res; + StringInfoData outbuf; + dsm_segment *outseg; + FlushColumnResult rtRes; + + /* serialise this column's input, ship it through a dsm segment, read it back */ + initStringInfo(&inbuf); + serialize_column_input(&inbuf, att, writeState->chunkGroups, c); + inseg = dsm_create(inbuf.len + sizeof(uint32), 0); + memcpy(dsm_segment_address(inseg), &inbuf.len, sizeof(uint32)); + memcpy((char *) dsm_segment_address(inseg) + sizeof(uint32), inbuf.data, inbuf.len); + rtGroups = deserialize_column_input(dsm_segment_address(inseg), att, c); + + /* run the pure function on the round-tripped input */ + res = flush_one_column(att, rtGroups, def, rowCount, validityBytes, + writeState->encodeEffort, writeState->compressionType, + writeState->compressionLevel, writeState->storageId, + groupNumber, c); + + /* serialise the result, ship it through a dsm segment, read it back */ + initStringInfo(&outbuf); + serialize_column_result(&outbuf, &res); + outseg = dsm_create(outbuf.len + sizeof(uint32), 0); + memcpy(dsm_segment_address(outseg), &outbuf.len, sizeof(uint32)); + memcpy((char *) dsm_segment_address(outseg) + sizeof(uint32), outbuf.data, outbuf.len); + rtRes = deserialize_column_result(dsm_segment_address(outseg)); + + /* assemble from the round-tripped result (identical to slice 1's assembly) */ + chunkOffset[c] = data->len; + if (rtRes.chunk != NULL && rtRes.chunk->len > 0) + appendBinaryStringInfo(data, rtRes.chunk->data, rtRes.chunk->len); + chunkLength[c] = data->len - chunkOffset[c]; + chunkDescriptor[c] = rtRes.descriptor; + chunkDescriptorLen[c] = rtRes.descriptorLen; + chunkBlockCodec[c] = rtRes.blockCodec; + if (rtRes.zoneRows != NIL) + zoneRows = list_concat(zoneRows, rtRes.zoneRows); + if (rtRes.bloomRow != NULL) + bloomRows = lappend(bloomRows, rtRes.bloomRow); + + dsm_detach(inseg); + dsm_detach(outseg); + } } dataLength = data->len; From d90d3cbde1800f5f23ee414d6c8943d4567a92fe Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 21:20:51 -0600 Subject: [PATCH 4/5] harden: unpin worker output early + parallelism observability (#445 slice 3) Move dsm_unpin_segment to right after attach so a throw during collection frees the segment via the resource owner instead of leaking it to postmaster restart (shrinks the leak to the narrow wait-window; the full cleanup callback for that window is the top slice-4 hardening item). Add a DEBUG1 line reporting how the flush split across workers vs serial, which is also the test premise that a starved parallel run did not silently fall back unnoticed. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar_write_state.c | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 4d456f8..2099889 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -2173,6 +2173,7 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, BackgroundWorker bw; BackgroundWorkerHandle **handles; int nstarted = 0; + int nWorkerCols = 0; int c; int i; @@ -2319,6 +2320,14 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, 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); @@ -2344,8 +2353,8 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, writeState->colDefs[col].fsstVerdict = verdict; writeState->colDefs[col].fsstVerdictAge = age; done[col] = true; + nWorkerCols++; } - dsm_unpin_segment(slots[i].outHandle); dsm_detach(outseg); } else if (st == PFLUSH_FAILED) @@ -2359,6 +2368,15 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber, 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 From fa22d63a7cb9d4b6813a2ff30c566db8ac74dbbd Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 23:06:33 -0600 Subject: [PATCH 5/5] perf: route the serial flush path around the dsm (#445 slice 3) Per jdatcmd's #591/#590 review: the dsm serialize round-trip only earns its cost crossing into a worker. When parallel_flush is off (or natts<2, or a column falls back), call flush_one_column directly instead of serialize->dsm_create-> deserialize. OFF is now byte-identical to and as fast as main (was +22% from the per-column dsm_create); ON keeps its ~14% win, dsm crossing intact on the worker path. This also retires #590's standalone regression -- the serial dsm round-trip is no longer on any default path; the serialize/deserialize helpers remain, used solely by the worker dispatch. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar_write_state.c | 68 ++++++++++++-------------------------- 1 file changed, 22 insertions(+), 46 deletions(-) diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 2099889..e1b7395 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -2517,59 +2517,35 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) else { /* - * slice 2: round-trip each column's input and result through a dsm segment, - * still serially in this backend, to prove the serialisation before slice 3's - * workers. + * 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); - PgColumnarColumnDef *def = &writeState->colDefs[c]; - StringInfoData inbuf; - dsm_segment *inseg; - List *rtGroups; - FlushColumnResult res; - StringInfoData outbuf; - dsm_segment *outseg; - FlushColumnResult rtRes; - - /* serialise this column's input, ship it through a dsm segment, read it back */ - initStringInfo(&inbuf); - serialize_column_input(&inbuf, att, writeState->chunkGroups, c); - inseg = dsm_create(inbuf.len + sizeof(uint32), 0); - memcpy(dsm_segment_address(inseg), &inbuf.len, sizeof(uint32)); - memcpy((char *) dsm_segment_address(inseg) + sizeof(uint32), inbuf.data, inbuf.len); - rtGroups = deserialize_column_input(dsm_segment_address(inseg), att, c); - - /* run the pure function on the round-tripped input */ - res = flush_one_column(att, rtGroups, def, rowCount, validityBytes, - writeState->encodeEffort, writeState->compressionType, - writeState->compressionLevel, writeState->storageId, - groupNumber, c); - - /* serialise the result, ship it through a dsm segment, read it back */ - initStringInfo(&outbuf); - serialize_column_result(&outbuf, &res); - outseg = dsm_create(outbuf.len + sizeof(uint32), 0); - memcpy(dsm_segment_address(outseg), &outbuf.len, sizeof(uint32)); - memcpy((char *) dsm_segment_address(outseg) + sizeof(uint32), outbuf.data, outbuf.len); - rtRes = deserialize_column_result(dsm_segment_address(outseg)); + FlushColumnResult res = flush_one_column(att, writeState->chunkGroups, + &writeState->colDefs[c], rowCount, + validityBytes, writeState->encodeEffort, + writeState->compressionType, + writeState->compressionLevel, + writeState->storageId, groupNumber, c); - /* assemble from the round-tripped result (identical to slice 1's assembly) */ chunkOffset[c] = data->len; - if (rtRes.chunk != NULL && rtRes.chunk->len > 0) - appendBinaryStringInfo(data, rtRes.chunk->data, rtRes.chunk->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] = rtRes.descriptor; - chunkDescriptorLen[c] = rtRes.descriptorLen; - chunkBlockCodec[c] = rtRes.blockCodec; - if (rtRes.zoneRows != NIL) - zoneRows = list_concat(zoneRows, rtRes.zoneRows); - if (rtRes.bloomRow != NULL) - bloomRows = lappend(bloomRows, rtRes.bloomRow); - - dsm_detach(inseg); - dsm_detach(outseg); + 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); } }