From fa9dc5b49715d0b85de171363f0879fd273d8dfd Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Wed, 12 Aug 2026 13:38:50 -0600 Subject: [PATCH] feat: pin parallel_flush as opt-in; a default-on gate is refuted (#445) #445's serial-load lever, parallel_flush, is off by default because it helps one shape (a large flush of many cheap numeric columns, ~14%) and regresses the rest. This asked whether a gate could default it on while declining the losing shapes. Answer: no. No metric computable at dispatch -- natts, buffered bytes, bytes per value -- separates the win from the losses, because the deciding variable is per-column encode CPU and its balance across columns, which the buffered .len fields do not carry. Verified by hand: 20 int2 and 5 int8 columns buffer the same bytes yet do 6.4x different work (3680ms vs 578ms/500k rows), and a random vs a constant column are byte-identical to the metric and opposite in cost. A threshold over such a metric would mispredict. Shipped, no gate: - A dispatch observability line at DEBUG1 (rows/natts/bufbytes/valbytes/valcount -> parallel|serial) via pflush_metrics(), a read-only sum over the .len fields. Byte-neutral; differential/native_writer/native_zonemap/native_bloom unchanged. - A refutation comment at the dispatch so no one re-adds a (natts, bufbytes, width) gate without a new per-column-cost signal. - test/parallel_flush_optin.sh pins default-off (removal-proof: flip the boot default -> the arm goes RED, verified), opt-in byte-identity (on == off), and the metric collision (a random and a constant column log an identical dispatch metric). Registered; harness_selftest + docs_style green. - design/ISSUE_445_PARALLEL_FLUSH_GATE.md records the verdict and the re-shape (per-column encode-cost estimate + per-column partitioning) as the only path to default-on, a separate design. Green on PG17; default stays off, opt-in path byte-identical and still ~14% on the wide-numeric shape. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01WmQJqcXdwyuoAiHHt2znBr --- design/ISSUE_445_PARALLEL_FLUSH_GATE.md | 88 ++++++++++++++++++ src/columnar_write_state.c | 70 ++++++++++++++- test/parallel_flush_optin.sh | 115 ++++++++++++++++++++++++ test/run_all_versions.sh | 1 + 4 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 design/ISSUE_445_PARALLEL_FLUSH_GATE.md create mode 100755 test/parallel_flush_optin.sh diff --git a/design/ISSUE_445_PARALLEL_FLUSH_GATE.md b/design/ISSUE_445_PARALLEL_FLUSH_GATE.md new file mode 100644 index 0000000..5940b21 --- /dev/null +++ b/design/ISSUE_445_PARALLEL_FLUSH_GATE.md @@ -0,0 +1,88 @@ +# Issue #445: a default-on gate for parallel_flush + +Plan, written before code, per the house rule. This finishes #445's serial-load +story: `pgcolumnar.parallel_flush` landed as a proven, opt-in control (#589/#591/ +#592). It is off by default because it only pays on one shape and regresses +several others. This work asked whether a gate could let it default ON while +auto-declining the shapes it hurts. + +## VERDICT (2026-08-12): NO. Keep it opt-in. + +Measured, designed, and adversarially refuted (a workflow of 7 agents plus a +hand-check). **No metric computable at dispatch time -- natts, buffered bytes, or +bytes per value -- separates the win from the losses**, because the deciding +variable is per-column encode CPU and its BALANCE across columns, which the +buffered `.len` fields do not carry. Verified by hand: 20 int2 columns and 5 int8 +columns buffer the same bytes but do **6.4x different work** (3680 ms vs 578 ms to +load 500k rows), so buffered bytes is not a work proxy; a random and a constant +column are byte-identical to the metric and opposite in cost. Moving a threshold +cannot fix a metric that lacks the deciding term. + +(The specific win/regress ratios the sweep reported are wall-clock noisy -- the +peer's ClickBench ran concurrently on the box -- and should not be quoted. The +structural refutation above is noise-independent and is what the verdict rests +on.) + +**Disposition shipped**: default stays `false`; a dispatch observability line +records the metric (helps opt-in users, and makes the refutation checkable); +`test/parallel_flush_optin.sh` pins the default-off + byte-identity + the metric +collision, so the refuted gate cannot be re-added silently. The only path to +default-on is a re-SHAPE of the dispatch around a per-column encode-cost estimate +with per-column partitioning -- a separate design, not a re-calibration of a +bytes/natts threshold. The original plan follows, kept as the record of what was +tried. + +## What is already measured (do not re-derive; verify) + +From the #592 review and the crossover data on that PR: + +- **Wins**: one large flush of many CHEAP (numeric, fixed-width) columns. 41-col + int / 500k rows: ON 916 ms vs OFF 1058 ms, ~14% faster. Byte-identical. +- **Regresses, badly**: frequent small flushes. 50 flushes of a 5-col 50k load: + OFF 47 ms vs ON 170 ms, 3.6x slower. Worker spawn + shmem round-trip per flush + dominates when the flush is small. +- **Regresses**: a wide TEXT-heavy flush, ~+16% (peer). The parallel path copies + the buffered column bytes through shared memory; varlena buffers are large, so + the copy overhead exceeds the parallel encode/compress benefit. + +So the decision boundary is not just "big vs small". It is whether the +parallelisable per-column encode/compress work exceeds the fixed cost (worker +spawn) plus the shmem-copy cost (proportional to buffered bytes). Numeric-wide is +the only measured win; text and small are losses. + +## The question this work must answer FIRST (prove, not assume) + +**Can a metric computable at dispatch time reliably separate the win from the +losses?** The dispatch point already gates on `pgcolumnar_parallel_flush && natts +>= 2 && tupdescIsRel && !rel_new_in_current_xact`. The gate adds a size/shape +term. Candidate metrics, to be chosen by measurement not taste: + +- total buffered bytes across the stripe's columns (the shmem-copy cost proxy); +- buffered bytes per column, or the varlena/fixed-width split; +- natts, rows-in-stripe, or their product. + +If no metric cleanly separates the cases on real shapes, the honest outcome is +**keep it opt-in** and close #445's default-on question as not worth the +misprediction risk. That verdict is a valid result of this work. + +## Order of work (TDD, one slice) + +1. **Measure the crossover** on a private lane (NOT the peer's pg18 bench lane): + sweep column count, stripe size, and column type, `parallel_flush` on vs off, + to find where parallel starts winning and which metric predicts it. +2. **Design the gate** from the measured boundary: the metric, the threshold, and + how to compute it at the dispatch point without a new pass over the data. +3. **RED test** (`test/parallel_flush_gate.sh`): a wide-numeric stripe must + dispatch to workers, a small stripe and a text-heavy stripe must stay serial, + all byte-identical, asserted on an observable (an EXPLAIN/log signal that the + flush went parallel), with a removal proof. +4. **Implement** the gate, flip the default to on-behind-the-gate, and verify: the + full suite matrix does not regress (the #592 timeout must not return), the + wide-numeric win survives, small/text stay serial. + +## Adversarial bar + +Before implementing, a shape that the proposed gate MISPREDICTS -- goes parallel +and regresses, or stays serial and misses a clear win -- must be searched for and +not found (or the threshold moved until it is not). The gate is only as good as +the worst shape it misjudges. diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 11311b6..57659c9 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -2205,6 +2205,43 @@ pflush_error_cleanup(int code, Datum arg) } } +/* + * pflush_metrics + * Sum a stripe's buffered column sizes for the parallel_flush dispatch log + * (#445). One pass over the .len fields already in hand; reads no data and + * allocates nothing, so it does not perturb the flush. bufBytes is the total + * the parallel path would copy through shared memory; valBytes and valCount + * describe the value payload alone. + */ +static void +pflush_metrics(PgColumnarWriteState *writeState, uint64 *bufBytes, + uint64 *valBytes, uint64 *valCount) +{ + ListCell *lc; + uint64 bb = 0; + uint64 vb = 0; + uint64 vc = 0; + + foreach(lc, writeState->chunkGroups) + { + ChunkGroupBuffer *group = (ChunkGroupBuffer *) lfirst(lc); + int c; + + for (c = 0; c < writeState->natts; c++) + { + ColumnChunkBuffer *col = &group->columns[c]; + + bb += (uint64) col->existsStream.len + (uint64) col->valueStream.len + + (uint64) col->hashBuf.len; + vb += (uint64) col->valueStream.len; + vc += col->valueCount; + } + } + *bufBytes = bb; + *valBytes = vb; + *valCount = vc; +} + /* * flush_columns_parallel * The #445 slice-3 parallel flush: dispatch flush_one_column across a pool @@ -2546,6 +2583,7 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) ListCell *lc; int c; bool pushedSnapshot = false; + bool goParallel; List *zoneRows = NIL; /* NativeZoneMapMetadata * to insert (D5) */ List *bloomRows = NIL; /* NativeBloomMetadata * to insert (D5b) */ @@ -2594,9 +2632,37 @@ pgcolumnar_flush_row_group(PgColumnarWriteState *writeState) * 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. + * + * The GUC stays OFF by default. A gate to turn it on by default was measured + * and REFUTED (#445): no metric computable here -- natts, buffered bytes, or + * bytes per value -- separates the win from the losses, because the deciding + * variable is per-column encode CPU and its balance across columns, which the + * buffered .len fields do not carry. Two shapes with byte-identical + * (natts, bufbytes) had opposite best: 20 int2 and 5 int8 both buffer the same + * bytes yet one wins and one regresses, and a random int column ties a + * constant one on the metric while differing three-fold in fact. So do NOT add + * a (natts, bufbytes, width) gate here; it would mispredict. The dispatch log + * below records the metrics, so the refutation stays checkable and an opt-in + * user can see how a flush was split. test/parallel_flush_optin.sh pins this. */ - if (pgcolumnar_parallel_flush && natts >= 2 && writeState->tupdescIsRel && - !rel_new_in_current_xact(writeState->relid)) + goParallel = pgcolumnar_parallel_flush && natts >= 2 && + writeState->tupdescIsRel && !rel_new_in_current_xact(writeState->relid); + + if (message_level_is_interesting(DEBUG1)) + { + uint64 bufBytes; + uint64 valBytes; + uint64 valCount; + + pflush_metrics(writeState, &bufBytes, &valBytes, &valCount); + elog(DEBUG1, "pgcolumnar parallel_flush dispatch: rows=%lu natts=%d " + "bufbytes=%lu valbytes=%lu valcount=%lu -> %s", + (unsigned long) rowCount, natts, (unsigned long) bufBytes, + (unsigned long) valBytes, (unsigned long) valCount, + goParallel ? "parallel" : "serial"); + } + + if (goParallel) { FlushColumnResult *colResults = palloc0(sizeof(FlushColumnResult) * natts); diff --git a/test/parallel_flush_optin.sh b/test/parallel_flush_optin.sh new file mode 100755 index 0000000..4a94b68 --- /dev/null +++ b/test/parallel_flush_optin.sh @@ -0,0 +1,115 @@ +#!/usr/bin/env bash +# +# pgColumnar: parallel_flush stays OPT-IN, and the metric a default-on gate would +# need cannot separate the shapes (#445). +# +# #445 measured that pgcolumnar.parallel_flush helps one narrow shape -- a large +# flush of many cheap numeric columns -- and regresses the common cases (frequent +# small flushes, wide text, cheap fixed-width). A gate to turn it on by default +# was designed and REFUTED: the metric computable at dispatch time, over the +# buffered .len fields (natts, buffered bytes, bytes per value), does not carry +# per-column encode CPU, which is the variable that decides the outcome. Two +# shapes with byte-identical metrics have opposite best, so no threshold over the +# metric can gate them. +# +# This suite pins the decision, not a gate. It asserts three things: +# 1. the default is OFF, and a default flush dispatches serial; +# 2. opting in is byte-identical to the serial path (and actually dispatches +# parallel, so the identity is not a silent fall-back); +# 3. the refuted metric COLLIDES: a random and a constant column, opposite in +# encode cost, produce a byte-identical dispatch metric -- so the metric +# cannot tell apart shapes whose real best differs. +# +# It is GREEN on current source. Its value is the removal proofs, marked below: +# flip the boot default, or mutate the parallel assembly, and the named arm goes +# RED. The collision arm can only go RED if someone adds a distinguishing term to +# the metric -- which is exactly the signal that a real gate has become possible. +# +# Written fresh for pgColumnar. + +set -uo pipefail +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" + +# The ON arms need worker slots; a bulk flush registers min(natts, <=8) workers. +psql_run "ALTER SYSTEM SET max_worker_processes = 16;" +env PATH="$PGC_BINDIR:$PATH" pg_ctl -D "$PGC_PGDATA" restart -w -o "-p $PGC_PORT" >/dev/null 2>&1 || true +sleep 1 + +q() { + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -Atqc "$1" 2>/dev/null +} + +# The dispatch line the flush emits once per stripe at DEBUG1. gucSql runs in the +# same session before the insert, so a SET reaches the flush. Returns the tail of +# the line: "rows=.. natts=.. bufbytes=.. valbytes=.. valcount=.. -> serial". +dispatch() { # dispatch + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -c "SET client_min_messages=debug1;" -c "$1" -c "$2" 2>&1 | + grep -oE 'parallel_flush dispatch: .*-> (parallel|serial)' | tail -1 | + sed 's/^.*dispatch: //' +} + +# LSN-independent storage fingerprint over the stored chunk metadata. +fp() { # fp + q "SELECT md5(string_agg(group_number || ':' || column_index || ':' || value_count || ':' || encode(encoding_descriptor, 'hex') || ':' || block_codec || ':' || page_length, '|' ORDER BY group_number, column_index)) FROM pgcolumnar.column_chunk WHERE storage_id = pgcolumnar.get_storage_id('$1')" +} + +# ---- 1. the default is OFF, and stays that way ----------------------------- + +check_text "premise: parallel_flush is off by default" \ + "$(q "SHOW pgcolumnar.parallel_flush")" "off" + +q "CREATE TABLE wnum (a int, b int, c int, d int, e int, f int, g int, h int) USING pgcolumnar;" >/dev/null + +# DEFAULT-OFF PIN. A fresh flush with no SET must dispatch SERIAL. +# REMOVAL PROOF: set the boot default to true in columnar_tableam.c (the +# DefineCustomBoolVariable for pgcolumnar.parallel_flush) -> this line logs +# "-> parallel" and the check goes RED. That red proves the default, not +# incidental structure, keeps the machine off. +d_default="$(dispatch "SELECT 1;" "INSERT INTO wnum SELECT g,g,g,g,g,g,g,g FROM generate_series(1,300000) g;")" +check_text "a default flush dispatches serial" \ + "$([ "${d_default##*-> }" = "serial" ] && echo serial || echo "parallel:[$d_default]")" \ + "serial" + +# ---- 2. opting in is byte-identical to serial ------------------------------ + +q "TRUNCATE wnum;" >/dev/null +d_on="$(dispatch "SET pgcolumnar.parallel_flush=on;" "INSERT INTO wnum SELECT g,g,g,g,g,g,g,g FROM generate_series(1,300000) g;")" +# The identity below is vacuous if ON silently fell back to serial, so assert the +# wide-numeric flush really took the worker path. +check_text "with the GUC on, a wide-numeric flush dispatches parallel (not a silent serial fall-back)" \ + "$([ "${d_on##*-> }" = "parallel" ] && echo parallel || echo "serial:[$d_on]")" \ + "parallel" +fp_on="$(fp wnum)" + +q "TRUNCATE wnum;" >/dev/null +q "SET pgcolumnar.parallel_flush=off; INSERT INTO wnum SELECT g,g,g,g,g,g,g,g FROM generate_series(1,300000) g;" >/dev/null +fp_off="$(fp wnum)" + +# REMOVAL PROOF: mutate flush_columns_parallel's assembly to reorder or alter a +# column's bytes -> fp_on diverges from fp_off and this goes RED. +check_text "opting in stores byte-identical chunks to the serial path" "$fp_on" "$fp_off" + +# ---- 3. the refuted metric collides ---------------------------------------- +# +# i5_rand and i5_const have the identical schema and row count, so the dispatch +# metric (rows, natts, bufbytes, valbytes, valcount) is identical -- yet their +# encode cost is opposite: random bigints are incompressible and carry the heavy +# per-column work parallelism spreads, constants are trivial. #445 measured +# opposite best for exactly this kind of pair. A gate over this metric cannot +# tell them apart, which is why default-on is refused. +q "CREATE TABLE i5_rand (a bigint,b bigint,c bigint,d bigint,e bigint) USING pgcolumnar;" >/dev/null +q "CREATE TABLE i5_const (a bigint,b bigint,c bigint,d bigint,e bigint) USING pgcolumnar;" >/dev/null +m_rand="$(dispatch "SET pgcolumnar.parallel_flush=on;" "INSERT INTO i5_rand SELECT (random()*9e18)::bigint,(random()*9e18)::bigint,(random()*9e18)::bigint,(random()*9e18)::bigint,(random()*9e18)::bigint FROM generate_series(1,400000) g;")" +m_const="$(dispatch "SET pgcolumnar.parallel_flush=on;" "INSERT INTO i5_const SELECT 7::bigint,7::bigint,7::bigint,7::bigint,7::bigint FROM generate_series(1,400000) g;")" +# strip the -> decision; compare only the metric the gate would read. +check_text "the dispatch metric is byte-identical for a random and a constant column (the metric cannot separate opposite-cost shapes)" \ + "${m_rand% -> *}" "${m_const% -> *}" + +# REMOVAL PROOF / future signal: this can only go RED if a distinguishing term +# (e.g. per-column encode cost) is added to the metric -- which is the signal a +# real gate has become possible. A comment at the dispatch says so. + +pgc_summary diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index 8b0151d..1902e59 100755 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -149,6 +149,7 @@ SUITES=( parallel_copy parallel_degree parallel_export_parquet + parallel_flush_optin parallel_vector_agg parquet_count_bounds parquet_export