feat: pin parallel_flush as opt-in; a default-on gate is refuted (#445)

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.

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

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 <gucSql> <insertSql>
+	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 <table>
+	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

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