From c421c5317fb771ba3063daac734fa9c9b5f8c371 Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 17:22:34 -0600 Subject: [PATCH 1/2] refactor: extract pure flush_one_column() from the flush loop (#445 slice 1) Slice 1 of the #445 in-COPY parallelism design (#588): move the per-column flush body of pgcolumnar_flush_row_group into a standalone flush_one_column(inputs) -> {chunk, descriptor, codec, zonemap, bloom} that reads only its arguments, no writeState reach-through. The backend assembles the column chunks into the stripe in column order and keeps all I/O and catalog writes. No workers yet; output is byte-identical to the serial path. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_017N82wDmsawqSWoWkmxtHmW --- src/columnar_write_state.c | 843 ++++++++++++++++++++----------------- 1 file changed, 447 insertions(+), 396 deletions(-) diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index cc773d20..71233110 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -906,6 +906,438 @@ PgColumnarBufferedRowByNumber(Relation rel, uint64 rowNumber, return false; } +typedef struct FlushColumnResult +{ + StringInfo chunk; /* [validity][finalData] column-chunk bytes; NULL if skipped */ + char *descriptor; /* encoding descriptor bytes; NULL if the column was skipped */ + uint32 descriptorLen; + int blockCodec; + List *zoneRows; /* NativeZoneMapMetadata * for this column (per-vector + whole-chunk) */ + NativeBloomMetadata *bloomRow; /* per-chunk bloom for this column, or NULL */ +} FlushColumnResult; + +/* + * flush_one_column + * Produce one column chunk's bytes, descriptor, block codec, zone maps and + * bloom from that column's buffered input, reading only its arguments (no + * writeState reach-through). The caller assembles the column chunks into the + * stripe in column order and performs all I/O and catalog writes. Output is + * byte-identical to the in-loop code this was extracted from (#445 slice 1). + */ +static FlushColumnResult +flush_one_column(Form_pg_attribute att, List *chunkGroups, + PgColumnarColumnDef *def, uint64 rowCount, int validityBytes, + int encodeEffort, int compressionType, int compressionLevel, + uint64 storageId, uint64 groupNumber, int columnIndex) +{ + FlushColumnResult result; + List *zoneRows = NIL; + NativeBloomMetadata *bloomRow = NULL; + StringInfo chunk = makeStringInfo(); + ListCell *lc; + uint8 *validity = (uint8 *) palloc0(validityBytes); + uint64 rowIdx = 0; + StringInfo encoded = makeStringInfo(); + StringInfo desc = makeStringInfo(); + uint8 descVersion = COLUMNAR_NATIVE_ENCDESC_VERSION; + uint8 descReserved = 0; + uint32 vectorCount = (uint32) list_length(chunkGroups); + char *fsstTable = NULL; /* chunk-shared FSST table (E3b), or NULL */ + uint32 fsstTableLen = 0; + char *finalData; + uint32 finalLen; + int blockCodec = COLUMNAR_COMPRESSION_NONE; + int vec = 0; + bool chunkHasMinMax = false; + Datum chunkMin = (Datum) 0; + Datum chunkMax = (Datum) 0; + uint64 chunkValueCount = 0; + int64 chunkSum = 0; + + /* + * Virtual generated columns (attgenerated 'v', PostgreSQL 18+) are computed + * on read from their base columns and never stored, so writing an all-null + * chunk for them wastes space. Skip the chunk entirely: the reader finds no + * column_chunk for this column, treats it as absent, and returns its missing + * value (NULL via getmissingattr), while the executor expands the generated + * expression regardless. A NULL descriptor marks the column skipped for the + * column_chunk insertion pass below. ('v' is never set on PG15-17.) + */ + if (att->attgenerated == 'v') + { + result.chunk = NULL; + result.descriptor = NULL; + result.descriptorLen = 0; + result.blockCodec = COLUMNAR_COMPRESSION_NONE; + result.zoneRows = NIL; + result.bloomRow = NULL; + pfree(validity); + return result; + } + + foreach(lc, chunkGroups) + { + ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); + ColumnChunkBuffer *col = &group->columns[columnIndex]; + char *existsBytes = col->existsStream.data; + uint64 i; + + for (i = 0; i < group->rowCount; i++, rowIdx++) + if (existsBytes[i]) + validity[rowIdx >> 3] |= (uint8) (1 << (rowIdx & 7)); + } + appendBinaryStringInfo(chunk, (char *) validity, validityBytes); + + /* descriptor header */ + appendBinaryStringInfo(desc, (char *) &descVersion, 1); + appendBinaryStringInfo(desc, (char *) &descReserved, 1); + appendBinaryStringInfo(desc, (char *) &vectorCount, sizeof(uint32)); + + /* + * E3b: build one FSST symbol table for the whole column chunk from a + * bounded sample of its value streams, so the costly table build is paid + * once here rather than once per vector. It is stored once as a trailing + * descriptor region and reused by every FSST vector below. Non-varlena + * columns and columns FSST cannot help leave it NULL. + */ + if (att->attlen == -1) + { + StringInfoData corpus; + uint32 sampleLen = 0; + /* `def` is the enclosing block's, at the top of this per-column + * loop. Re-declaring it here shadowed that one, which this project + * builds with -Wshadow=compatible-local and treats as an error. */ + bool reuseVerdict; + + initStringInfo(&corpus); + foreach(lc, chunkGroups) + { + ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); + ColumnChunkBuffer *col = &group->columns[columnIndex]; + + if (col->valueStream.len > 0) + appendBinaryStringInfo(&corpus, col->valueStream.data, + col->valueStream.len); + if (sampleLen == 0 && corpus.len >= 262144) + sampleLen = (uint32) corpus.len; /* matches FSST_SAMPLE_CAP: + * train the one per-chunk + * table on a broad sample */ + if (corpus.len >= COLUMNAR_FSST_DECIDE_CAP) + break; + } + if (sampleLen == 0) + sampleLen = (uint32) corpus.len; + + /* + * encode_effort = fast skips the FSST substring search entirely: + * no symbol table, so no whole-corpus decision below and no + * per-vector encode either, since all three are reached only + * through a non-NULL fsstTable. + * + * This is where a text column's write cost lives (issue #155). + * Measured on 1,000,000 rows, one text column, the load runs 1.2x + * to 5.7x faster without it -- and on five of the seven shapes + * measured it produced byte-for-byte identical storage, so that + * time bought nothing at all. On the two where FSST does win it + * costs 2.7% and 12.2% more space, which is why this is a choice + * offered rather than a default changed. + */ + /* + * Skip the FSST symbol-table build when a cheap distinct probe shows + * the dictionary wins outright (#155): the build is the single largest + * cost of a text load, and for a low-cardinality column the table is + * built and then never used per vector. The probe reads the same corpus + * the keep/drop decision uses, and only skips when the dictionary is + * viable and wins for every vector, so the stored bytes are identical. + */ + /* + * Reuse this column's previous verdict when it is young enough + * (#472). A HURTS verdict skips the build as well as the question, + * since the vectors then take their ordinary encoding, which is + * what a freshly taken HURTS would have produced. + */ + reuseVerdict = (pgcolumnar_fsst_verdict_reuse > 0 && + def->fsstVerdict != COLUMNAR_FSST_UNKNOWN && + def->fsstVerdictAge < pgcolumnar_fsst_verdict_reuse); + + if (corpus.len > 0 && + encodeEffort != COLUMNAR_ENCODE_EFFORT_FAST && + !(reuseVerdict && def->fsstVerdict == COLUMNAR_FSST_HURTS) && + !PgColumnarFsstDictWins(corpus.data, (uint32) corpus.len)) + PgColumnarFsstBuildChunkTable(corpus.data, sampleLen, att, + &fsstTable, &fsstTableLen); + + /* + * A table that shrinks every vector can still enlarge the chunk, + * because what lands on disk is this stream after the codec below + * has run, and FSST codes compress far worse than the text they + * replace. Ask before committing to it, and drop the table when the + * answer is no: the vectors below then take their ordinary encoding + * and skip the FSST attempt altogether, so the check pays for itself + * in write time exactly when it saves space. + * + * This is asked over a much longer run of bytes than the table is + * trained on, because the answer moves with volume and the sample + * size is not neutral: zstd needs a good deal of FSST output before + * it finds the structure in it. Measured on 300,000 e-mail-shaped + * rows, the 256 kB training sample says FSST is 24% worse while over + * the whole column it is 23% better -- a verdict that is not merely + * imprecise but inverted, so no margin on the sample would be safe. + */ + if (fsstTable != NULL) + { + bool helps; + + /* + * A reused HELPS verdict still builds the table above, because + * the table is trained on THIS row group's corpus and stored + * with the chunk: reusing the table itself would change the + * stored bytes. Only the whole-corpus question is skipped, and + * that is the expensive half. + */ + if (reuseVerdict) + { + helps = (def->fsstVerdict == COLUMNAR_FSST_HELPS); + def->fsstVerdictAge++; + } + else + { + helps = PgColumnarFsstHelpsCompressed(corpus.data, + (uint32) corpus.len, + fsstTable, fsstTableLen, + compressionType, + compressionLevel); + def->fsstVerdict = helps ? COLUMNAR_FSST_HELPS + : COLUMNAR_FSST_HURTS; + def->fsstVerdictAge = 0; + } + + if (!helps) + { + pfree(fsstTable); + fsstTable = NULL; + fsstTableLen = 0; + } + } + else if (reuseVerdict && def->fsstVerdict == COLUMNAR_FSST_HURTS) + { + /* + * The build was skipped on the strength of the cached verdict, + * so this row group counts as a reuse too. Without this the age + * would never advance on the common path and the bound would + * never re-take the verdict. + */ + def->fsstVerdictAge++; + } + + pfree(corpus.data); + } + + /* encode each vector (chunk group) and record its descriptor entry */ + foreach(lc, chunkGroups) + { + ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); + ColumnChunkBuffer *col = &group->columns[columnIndex]; + char *encData; + uint32 encLen; + int encType; + uint8 entryType; + uint32 entryValueCount; + uint32 entryRawLen; + + encType = PgColumnarEncodeChunk(col->valueStream.data, + col->valueStream.len, att, + col->valueCount, fsstTable, fsstTableLen, + &encData, &encLen); + + if (encLen > 0) + appendBinaryStringInfo(encoded, encData, encLen); + + entryType = (uint8) encType; + entryValueCount = (uint32) col->valueCount; + entryRawLen = (uint32) col->valueStream.len; + appendBinaryStringInfo(desc, (char *) &entryType, 1); + appendBinaryStringInfo(desc, (char *) &entryValueCount, sizeof(uint32)); + appendBinaryStringInfo(desc, (char *) &entryRawLen, sizeof(uint32)); + appendBinaryStringInfo(desc, (char *) &encLen, sizeof(uint32)); + + /* per-vector zone map (native spec 7.1, D5) */ + { + NativeZoneMapMetadata *z = palloc0(sizeof(NativeZoneMapMetadata)); + + z->storageId = storageId; + z->groupNumber = groupNumber; + z->columnIndex = columnIndex; + z->vectorIndex = vec; + z->valueCount = col->valueCount; + z->nullCount = group->rowCount - col->valueCount; + + if (def->summableInt && col->valueCount > 0) + { + z->hasSum = true; + z->sum = DirectFunctionCall1(int8_numeric, + Int64GetDatum(col->sum)); + } + + if (col->hasMinMax) + { + StringInfoData mn; + StringInfoData mx; + + initStringInfo(&mn); + initStringInfo(&mx); + PgColumnarEncodeValue(&mn, att, col->minValue); + PgColumnarEncodeValue(&mx, att, col->maxValue); + z->hasMinMax = true; + z->minimum = mn.data; + z->minimumLen = (uint32) mn.len; + z->maximum = mx.data; + z->maximumLen = (uint32) mx.len; + + /* fold into the whole-chunk min/max via the btree cmp proc */ + if (!chunkHasMinMax) + { + chunkMin = col->minValue; + chunkMax = col->maxValue; + chunkHasMinMax = true; + } + else + { + if (DatumGetInt32(FunctionCall2Coll(&def->cmpFn, + def->collation, + col->minValue, + chunkMin)) < 0) + chunkMin = col->minValue; + if (DatumGetInt32(FunctionCall2Coll(&def->cmpFn, + def->collation, + col->maxValue, + chunkMax)) > 0) + chunkMax = col->maxValue; + } + } + + chunkValueCount += col->valueCount; + chunkSum += col->sum; + zoneRows = lappend(zoneRows, z); + } + vec++; + } + + /* + * E3b: trailing chunk-shared FSST table region (descriptor version 2). + * sharedTableLen is 0 when the chunk has no shared table; FSST vectors + * above reference this one table instead of embedding their own. + */ + appendBinaryStringInfo(desc, (char *) &fsstTableLen, sizeof(uint32)); + if (fsstTableLen > 0) + appendBinaryStringInfo(desc, fsstTable, fsstTableLen); + + /* whole-chunk zone map (vector_index -1) */ + { + NativeZoneMapMetadata *z = palloc0(sizeof(NativeZoneMapMetadata)); + + z->storageId = storageId; + z->groupNumber = groupNumber; + z->columnIndex = columnIndex; + z->vectorIndex = -1; + z->valueCount = chunkValueCount; + z->nullCount = rowCount - chunkValueCount; + + if (def->summableInt && chunkValueCount > 0) + { + z->hasSum = true; + z->sum = DirectFunctionCall1(int8_numeric, + Int64GetDatum(chunkSum)); + } + + if (chunkHasMinMax) + { + StringInfoData mn; + StringInfoData mx; + + initStringInfo(&mn); + initStringInfo(&mx); + PgColumnarEncodeValue(&mn, att, chunkMin); + PgColumnarEncodeValue(&mx, att, chunkMax); + z->hasMinMax = true; + z->minimum = mn.data; + z->minimumLen = (uint32) mn.len; + z->maximum = mx.data; + z->maximumLen = (uint32) mx.len; + } + + zoneRows = lappend(zoneRows, z); + } + + /* per-column-chunk bloom over hashable values (native spec 7.2, D5b) */ + if (def->bloomable) + { + StringInfoData hashes; + char *bloom; + uint32 bloomLen; + + initStringInfo(&hashes); + foreach(lc, chunkGroups) + { + ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); + ColumnChunkBuffer *col = &group->columns[columnIndex]; + + if (col->hashBuf.len > 0) + appendBinaryStringInfo(&hashes, col->hashBuf.data, + col->hashBuf.len); + } + if (hashes.len > 0 && + PgColumnarBloomBuild((const uint32 *) hashes.data, + hashes.len / sizeof(uint32), + &bloom, &bloomLen)) + { + NativeBloomMetadata *b = palloc0(sizeof(NativeBloomMetadata)); + + b->storageId = storageId; + b->groupNumber = groupNumber; + b->columnIndex = columnIndex; + b->filter = bloom; + b->filterLen = bloomLen; + bloomRow = b; + } + } + + /* optional block codec over the whole encoded region (spec 6) */ + finalData = encoded->data; + finalLen = encoded->len; + if (compressionType != COLUMNAR_COMPRESSION_NONE && + encoded->len > 0) + { + char *compData; + uint32 compLen; + int usedType; + int usedLevel; + + PgColumnarCompressValueStream(encoded->data, encoded->len, + compressionType, + compressionLevel, + &compData, &compLen, + &usedType, &usedLevel); + if (usedType != COLUMNAR_COMPRESSION_NONE) + { + finalData = compData; + finalLen = compLen; + blockCodec = usedType; + } + } + + if (finalLen > 0) + appendBinaryStringInfo(chunk, finalData, finalLen); + + result.chunk = chunk; + result.descriptor = desc->data; + result.descriptorLen = (uint32) desc->len; + result.blockCodec = blockCodec; + result.zoneRows = zoneRows; + result.bloomRow = bloomRow; + return result; +} + /* * pgcolumnar_flush_row_group * Native-format (PGCN v1) flush. Lay out the accumulated rows as one row @@ -989,405 +1421,24 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) for (c = 0; c < natts; c++) { Form_pg_attribute att = TupleDescAttr(writeState->tupdesc, c); - uint8 *validity = (uint8 *) palloc0(validityBytes); - uint64 rowIdx = 0; - StringInfo encoded = makeStringInfo(); - StringInfo desc = makeStringInfo(); - uint8 descVersion = COLUMNAR_NATIVE_ENCDESC_VERSION; - uint8 descReserved = 0; - uint32 vectorCount = (uint32) list_length(writeState->chunkGroups); - char *fsstTable = NULL; /* chunk-shared FSST table (E3b), or NULL */ - uint32 fsstTableLen = 0; - char *finalData; - uint32 finalLen; - int blockCodec = COLUMNAR_COMPRESSION_NONE; - PgColumnarColumnDef *def = &writeState->colDefs[c]; - int vec = 0; - bool chunkHasMinMax = false; - Datum chunkMin = (Datum) 0; - Datum chunkMax = (Datum) 0; - uint64 chunkValueCount = 0; - int64 chunkSum = 0; + 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; - - /* - * Virtual generated columns (attgenerated 'v', PostgreSQL 18+) are computed - * on read from their base columns and never stored, so writing an all-null - * chunk for them wastes space. Skip the chunk entirely: the reader finds no - * column_chunk for this column, treats it as absent, and returns its missing - * value (NULL via getmissingattr), while the executor expands the generated - * expression regardless. A NULL descriptor marks the column skipped for the - * column_chunk insertion pass below. ('v' is never set on PG15-17.) - */ - if (att->attgenerated == 'v') - { - chunkLength[c] = 0; - chunkDescriptor[c] = NULL; - chunkDescriptorLen[c] = 0; - chunkBlockCodec[c] = COLUMNAR_COMPRESSION_NONE; - pfree(validity); - continue; - } - - foreach(lc, writeState->chunkGroups) - { - ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); - ColumnChunkBuffer *col = &group->columns[c]; - char *existsBytes = col->existsStream.data; - uint64 i; - - for (i = 0; i < group->rowCount; i++, rowIdx++) - if (existsBytes[i]) - validity[rowIdx >> 3] |= (uint8) (1 << (rowIdx & 7)); - } - appendBinaryStringInfo(data, (char *) validity, validityBytes); - - /* descriptor header */ - appendBinaryStringInfo(desc, (char *) &descVersion, 1); - appendBinaryStringInfo(desc, (char *) &descReserved, 1); - appendBinaryStringInfo(desc, (char *) &vectorCount, sizeof(uint32)); - - /* - * E3b: build one FSST symbol table for the whole column chunk from a - * bounded sample of its value streams, so the costly table build is paid - * once here rather than once per vector. It is stored once as a trailing - * descriptor region and reused by every FSST vector below. Non-varlena - * columns and columns FSST cannot help leave it NULL. - */ - if (att->attlen == -1) - { - StringInfoData corpus; - uint32 sampleLen = 0; - /* `def` is the enclosing block's, at the top of this per-column - * loop. Re-declaring it here shadowed that one, which this project - * builds with -Wshadow=compatible-local and treats as an error. */ - bool reuseVerdict; - - initStringInfo(&corpus); - foreach(lc, writeState->chunkGroups) - { - ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); - ColumnChunkBuffer *col = &group->columns[c]; - - if (col->valueStream.len > 0) - appendBinaryStringInfo(&corpus, col->valueStream.data, - col->valueStream.len); - if (sampleLen == 0 && corpus.len >= 262144) - sampleLen = (uint32) corpus.len; /* matches FSST_SAMPLE_CAP: - * train the one per-chunk - * table on a broad sample */ - if (corpus.len >= COLUMNAR_FSST_DECIDE_CAP) - break; - } - if (sampleLen == 0) - sampleLen = (uint32) corpus.len; - - /* - * encode_effort = fast skips the FSST substring search entirely: - * no symbol table, so no whole-corpus decision below and no - * per-vector encode either, since all three are reached only - * through a non-NULL fsstTable. - * - * This is where a text column's write cost lives (issue #155). - * Measured on 1,000,000 rows, one text column, the load runs 1.2x - * to 5.7x faster without it -- and on five of the seven shapes - * measured it produced byte-for-byte identical storage, so that - * time bought nothing at all. On the two where FSST does win it - * costs 2.7% and 12.2% more space, which is why this is a choice - * offered rather than a default changed. - */ - /* - * Skip the FSST symbol-table build when a cheap distinct probe shows - * the dictionary wins outright (#155): the build is the single largest - * cost of a text load, and for a low-cardinality column the table is - * built and then never used per vector. The probe reads the same corpus - * the keep/drop decision uses, and only skips when the dictionary is - * viable and wins for every vector, so the stored bytes are identical. - */ - /* - * Reuse this column's previous verdict when it is young enough - * (#472). A HURTS verdict skips the build as well as the question, - * since the vectors then take their ordinary encoding, which is - * what a freshly taken HURTS would have produced. - */ - reuseVerdict = (pgcolumnar_fsst_verdict_reuse > 0 && - def->fsstVerdict != COLUMNAR_FSST_UNKNOWN && - def->fsstVerdictAge < pgcolumnar_fsst_verdict_reuse); - - if (corpus.len > 0 && - writeState->encodeEffort != COLUMNAR_ENCODE_EFFORT_FAST && - !(reuseVerdict && def->fsstVerdict == COLUMNAR_FSST_HURTS) && - !PgColumnarFsstDictWins(corpus.data, (uint32) corpus.len)) - PgColumnarFsstBuildChunkTable(corpus.data, sampleLen, att, - &fsstTable, &fsstTableLen); - - /* - * A table that shrinks every vector can still enlarge the chunk, - * because what lands on disk is this stream after the codec below - * has run, and FSST codes compress far worse than the text they - * replace. Ask before committing to it, and drop the table when the - * answer is no: the vectors below then take their ordinary encoding - * and skip the FSST attempt altogether, so the check pays for itself - * in write time exactly when it saves space. - * - * This is asked over a much longer run of bytes than the table is - * trained on, because the answer moves with volume and the sample - * size is not neutral: zstd needs a good deal of FSST output before - * it finds the structure in it. Measured on 300,000 e-mail-shaped - * rows, the 256 kB training sample says FSST is 24% worse while over - * the whole column it is 23% better -- a verdict that is not merely - * imprecise but inverted, so no margin on the sample would be safe. - */ - if (fsstTable != NULL) - { - bool helps; - - /* - * A reused HELPS verdict still builds the table above, because - * the table is trained on THIS row group's corpus and stored - * with the chunk: reusing the table itself would change the - * stored bytes. Only the whole-corpus question is skipped, and - * that is the expensive half. - */ - if (reuseVerdict) - { - helps = (def->fsstVerdict == COLUMNAR_FSST_HELPS); - def->fsstVerdictAge++; - } - else - { - helps = PgColumnarFsstHelpsCompressed(corpus.data, - (uint32) corpus.len, - fsstTable, fsstTableLen, - writeState->compressionType, - writeState->compressionLevel); - def->fsstVerdict = helps ? COLUMNAR_FSST_HELPS - : COLUMNAR_FSST_HURTS; - def->fsstVerdictAge = 0; - } - - if (!helps) - { - pfree(fsstTable); - fsstTable = NULL; - fsstTableLen = 0; - } - } - else if (reuseVerdict && def->fsstVerdict == COLUMNAR_FSST_HURTS) - { - /* - * The build was skipped on the strength of the cached verdict, - * so this row group counts as a reuse too. Without this the age - * would never advance on the common path and the bound would - * never re-take the verdict. - */ - def->fsstVerdictAge++; - } - - pfree(corpus.data); - } - - /* encode each vector (chunk group) and record its descriptor entry */ - foreach(lc, writeState->chunkGroups) - { - ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); - ColumnChunkBuffer *col = &group->columns[c]; - char *encData; - uint32 encLen; - int encType; - uint8 entryType; - uint32 entryValueCount; - uint32 entryRawLen; - - encType = PgColumnarEncodeChunk(col->valueStream.data, - col->valueStream.len, att, - col->valueCount, fsstTable, fsstTableLen, - &encData, &encLen); - - if (encLen > 0) - appendBinaryStringInfo(encoded, encData, encLen); - - entryType = (uint8) encType; - entryValueCount = (uint32) col->valueCount; - entryRawLen = (uint32) col->valueStream.len; - appendBinaryStringInfo(desc, (char *) &entryType, 1); - appendBinaryStringInfo(desc, (char *) &entryValueCount, sizeof(uint32)); - appendBinaryStringInfo(desc, (char *) &entryRawLen, sizeof(uint32)); - appendBinaryStringInfo(desc, (char *) &encLen, sizeof(uint32)); - - /* per-vector zone map (native spec 7.1, D5) */ - { - NativeZoneMapMetadata *z = palloc0(sizeof(NativeZoneMapMetadata)); - - z->storageId = writeState->storageId; - z->groupNumber = groupNumber; - z->columnIndex = c; - z->vectorIndex = vec; - z->valueCount = col->valueCount; - z->nullCount = group->rowCount - col->valueCount; - - if (def->summableInt && col->valueCount > 0) - { - z->hasSum = true; - z->sum = DirectFunctionCall1(int8_numeric, - Int64GetDatum(col->sum)); - } - - if (col->hasMinMax) - { - StringInfoData mn; - StringInfoData mx; - - initStringInfo(&mn); - initStringInfo(&mx); - PgColumnarEncodeValue(&mn, att, col->minValue); - PgColumnarEncodeValue(&mx, att, col->maxValue); - z->hasMinMax = true; - z->minimum = mn.data; - z->minimumLen = (uint32) mn.len; - z->maximum = mx.data; - z->maximumLen = (uint32) mx.len; - - /* fold into the whole-chunk min/max via the btree cmp proc */ - if (!chunkHasMinMax) - { - chunkMin = col->minValue; - chunkMax = col->maxValue; - chunkHasMinMax = true; - } - else - { - if (DatumGetInt32(FunctionCall2Coll(&def->cmpFn, - def->collation, - col->minValue, - chunkMin)) < 0) - chunkMin = col->minValue; - if (DatumGetInt32(FunctionCall2Coll(&def->cmpFn, - def->collation, - col->maxValue, - chunkMax)) > 0) - chunkMax = col->maxValue; - } - } - - chunkValueCount += col->valueCount; - chunkSum += col->sum; - zoneRows = lappend(zoneRows, z); - } - vec++; - } - - /* - * E3b: trailing chunk-shared FSST table region (descriptor version 2). - * sharedTableLen is 0 when the chunk has no shared table; FSST vectors - * above reference this one table instead of embedding their own. - */ - appendBinaryStringInfo(desc, (char *) &fsstTableLen, sizeof(uint32)); - if (fsstTableLen > 0) - appendBinaryStringInfo(desc, fsstTable, fsstTableLen); - - /* whole-chunk zone map (vector_index -1) */ - { - NativeZoneMapMetadata *z = palloc0(sizeof(NativeZoneMapMetadata)); - - z->storageId = writeState->storageId; - z->groupNumber = groupNumber; - z->columnIndex = c; - z->vectorIndex = -1; - z->valueCount = chunkValueCount; - z->nullCount = rowCount - chunkValueCount; - - if (def->summableInt && chunkValueCount > 0) - { - z->hasSum = true; - z->sum = DirectFunctionCall1(int8_numeric, - Int64GetDatum(chunkSum)); - } - - if (chunkHasMinMax) - { - StringInfoData mn; - StringInfoData mx; - - initStringInfo(&mn); - initStringInfo(&mx); - PgColumnarEncodeValue(&mn, att, chunkMin); - PgColumnarEncodeValue(&mx, att, chunkMax); - z->hasMinMax = true; - z->minimum = mn.data; - z->minimumLen = (uint32) mn.len; - z->maximum = mx.data; - z->maximumLen = (uint32) mx.len; - } - - zoneRows = lappend(zoneRows, z); - } - - /* per-column-chunk bloom over hashable values (native spec 7.2, D5b) */ - if (def->bloomable) - { - StringInfoData hashes; - char *bloom; - uint32 bloomLen; - - initStringInfo(&hashes); - foreach(lc, writeState->chunkGroups) - { - ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); - ColumnChunkBuffer *col = &group->columns[c]; - - if (col->hashBuf.len > 0) - appendBinaryStringInfo(&hashes, col->hashBuf.data, - col->hashBuf.len); - } - if (hashes.len > 0 && - PgColumnarBloomBuild((const uint32 *) hashes.data, - hashes.len / sizeof(uint32), - &bloom, &bloomLen)) - { - NativeBloomMetadata *b = palloc0(sizeof(NativeBloomMetadata)); - - b->storageId = writeState->storageId; - b->groupNumber = groupNumber; - b->columnIndex = c; - b->filter = bloom; - b->filterLen = bloomLen; - bloomRows = lappend(bloomRows, b); - } - } - - /* optional block codec over the whole encoded region (spec 6) */ - finalData = encoded->data; - finalLen = encoded->len; - if (writeState->compressionType != COLUMNAR_COMPRESSION_NONE && - encoded->len > 0) - { - char *compData; - uint32 compLen; - int usedType; - int usedLevel; - - PgColumnarCompressValueStream(encoded->data, encoded->len, - writeState->compressionType, - writeState->compressionLevel, - &compData, &compLen, - &usedType, &usedLevel); - if (usedType != COLUMNAR_COMPRESSION_NONE) - { - finalData = compData; - finalLen = compLen; - blockCodec = usedType; - } - } - - if (finalLen > 0) - appendBinaryStringInfo(data, finalData, finalLen); - + if (res.chunk != NULL && res.chunk->len > 0) + appendBinaryStringInfo(data, res.chunk->data, res.chunk->len); chunkLength[c] = data->len - chunkOffset[c]; - chunkDescriptor[c] = desc->data; - chunkDescriptorLen[c] = (uint32) desc->len; - chunkBlockCodec[c] = blockCodec; + 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; From dd7bccf815d1de4d8b86f65cec7629459ad7abfa Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Tue, 11 Aug 2026 19:30:04 -0600 Subject: [PATCH 2/2] 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 71233110..b2f6f6f5 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;