fix: a parallel flush worker commits before it pins its output, and the collect loop frees what it did not collect (#1747)

CHANGELOG.md

@@ -418,6 +418,12 @@ true until the next version shipped.
   `_SUCCESS` writers, now copies the error out first and re-throws it afterwards
   (#1741). `test/objstore_export_double_cancel.sh` and
   `test_objstore_export_double_cancel.py`.
+- **A parallel flush worker that fails after publishing its output no longer leaks
+  it.** A worker pinned its shared memory output and then committed; an error in
+  between reported the worker as failed, and the dispatcher, which collects only
+  finished workers, never freed the segment, so it stayed until the server
+  restarted. The worker now commits before it publishes, and the dispatcher frees
+  any output it did not collect (#1747).
 - **A second cancel during a parallel flush's error cleanup no longer leaks the
   workers' output segments.** With `pgcolumnar.parallel_flush` on, a cancel during
   the flush terminates the workers and frees the shared memory segments they

src/columnar_write_state.c

@@ -2496,7 +2496,16 @@ pgcolumnar_parallel_flush_worker(Datum main_arg)
 		pgcolumnar_fsst_min_gain_percent = hdr->fsstMinGainPercent;
 		pgcolumnar_encoding_sample_rows = hdr->encodingSampleRows;
 
-		initStringInfo(&recbuf);
+		/*
+		 * The record outlives the transaction: it is copied into the OUTPUT
+		 * segment after the commit below (#1747).
+		 */
+		{
+			MemoryContext oldcxt = MemoryContextSwitchTo(TopMemoryContext);
+
+			initStringInfo(&recbuf);
+			MemoryContextSwitchTo(oldcxt);
+		}
 
 		for (;;)
 		{
@@ -2551,7 +2560,18 @@ pgcolumnar_parallel_flush_worker(Datum main_arg)
 		 * 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.
+		 *
+		 * COMMIT FIRST, PIN LAST (#1747). A worker that raised after the pin
+		 * reported FAILED, and the dispatcher's collect loop unpins only a DONE
+		 * slot, so the pinned segment stayed until postmaster restart. Everything
+		 * that can raise -- the commit, the create -- now runs before the pin;
+		 * after it there is only the detach and the DONE write. A segment created
+		 * and not yet pinned is freed when this worker exits.
 		 */
+		table_close(rel, AccessShareLock);
+		PopActiveSnapshot();
+		CommitTransactionCommand();
+
 		outseg = dsm_create((Size) sizeof(uint32) + recbuf.len, 0);
 		obase = (char *) dsm_segment_address(outseg);
 		memcpy(obase, &count, sizeof(uint32));
@@ -2563,10 +2583,6 @@ pgcolumnar_parallel_flush_worker(Datum main_arg)
 		pg_write_barrier();
 		dsm_pin_segment(outseg);
 		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();
@@ -2942,6 +2958,24 @@ flush_columns_parallel(PgColumnarWriteState *writeState, uint64 groupNumber,
 					(errmsg("pgcolumnar parallel_flush worker %d failed, completing its columns serially: %s",
 							i, slots[i].errmsg[0] ? slots[i].errmsg : "unknown error")));
 		}
+
+		/*
+		 * A segment this loop did not collect is freed here (#1747): the worker
+		 * pins last, so nothing should raise after it, but one it named and did
+		 * not report DONE for would otherwise stay pinned to postmaster restart.
+		 * A named segment that was never pinned is already gone (attach NULL).
+		 */
+		if (slots[i].outHandle != DSM_HANDLE_INVALID)
+		{
+			dsm_segment *s = dsm_attach(slots[i].outHandle);
+
+			if (s != NULL)
+			{
+				dsm_unpin_segment(slots[i].outHandle);
+				dsm_detach(s);
+			}
+			slots[i].outHandle = DSM_HANDLE_INVALID;
+		}
 	}
 	}
 	PG_END_ENSURE_ERROR_CLEANUP(pflush_error_cleanup, PointerGetDatum(&cleanup));

test/check_ledger.tsv

@@ -5553,6 +5553,7 @@ parallel_export_parquet	parallel_export_parquet	single(4): wrote exactly 4 files
 parallel_export_parquet	parallel_export_parquet	the export was cancelled mid-flight, not completed first (fixture premise)	15;16;17;18;19	never	-
 parallel_export_parquet	parallel_export_parquet	the retry writes a fresh _SUCCESS marker	15;16;17;18;19	2026-10-22	columnar_parallel_export.c: the _SUCCESS marker is never written;columnar_parallel_export.c: the error cleanup of a cancelled export does not remove its part files
 parallel_flush_optin	parallel_flush_optin	a default flush dispatches serial	15;16;17;18;19	2026-10-21	columnar_tableam.c: pgcolumnar.parallel_flush boot default set to true
+parallel_flush_optin	parallel_flush_optin	a flush worker runs nothing that can raise after pinning its output, to the end of its PG_TRY block	15;16;17;18;19	2026-10-22	columnar_write_state.c: the flush worker commits after pinning its output (table_close, PopActiveSnapshot, CommitTransactionCommand moved after dsm_pin_segment)
 parallel_flush_optin	parallel_flush_optin	a twice-cancelled parallel flush leaves no worker segment pinned	15;16;17;18;19	2026-10-22	columnar_write_state.c: pflush_error_cleanup waits for its workers without HOLD_INTERRUPTS, so a second cancel leaves before the unpin loop
 parallel_flush_optin	parallel_flush_optin	a twice-cancelled parallel flush reports the first cancel	15;16;17;18;19	2026-10-22	columnar_write_state.c: pflush_error_cleanup waits for its workers without HOLD_INTERRUPTS, so a second cancel leaves before the unpin loop
 parallel_flush_optin	parallel_flush_optin	opting in stores byte-identical chunks to the serial path	15;16;17;18;19	2026-10-21	columnar_write_state.c: flush_columns_parallel stores each worker column under the next index (colResults[(col + 1) % natts])

test/check_ledger_budget.txt

@@ -6996,18 +6996,29 @@ suites_not_covered 28
 #     dated         3414 + 2 = 3416
 #     4545 + 3416 == 7961
 #
-# #1753, folded into #1746 at jdatcmd's suggestion, in parallel_flush_optin
-# (OffgridwithJD): 4545 -> 4545. The collect loop unpinned a DONE worker's output
-# segment and kept its handle until after the detach, so an error in between sent
-# pflush_error_cleanup to attach it again: "can't attach the same segment more than
-# once" replaced the original error and cut the cleanup short (an ERROR injected after
-# the first unpin, measured on PG17: 3 segments left; with the handle forgotten at the
-# unpin, the injected error is reported and 0 are left). Arm 5 reads the statement
-# after the unpin, comments blanked; it is dated by the handle invalidated after the
-# detach again, which reddens it alone in both twins (main's code reddens it too).
-#
-#     ledger rows   7961 + 1 = 7962
+# #1747, a parallel flush worker that fails after pinning its output, in
+# parallel_flush_optin (OffgridwithJD): 4545 -> 4545. On the normal path the collect
+# loop unpinned only a DONE slot, so a worker that raised after dsm_pin_segment (its
+# commit ran after the pin) reported FAILED and left the segment pinned to postmaster
+# restart: 8 segments from a two-stripe flush, measured on main with an injected ERROR
+# that is not committed. The worker now commits first and pins last, and arm 6 reads
+# the source: between the pin and the DONE write only the detach may run. Dated by the
+# worker committing after the pin again, which reddens the same name in both twins.
+#
+#     ledger rows   7962 + 1 = 7963
 #     never         4545 + 0 = 4545
-#     dated         3416 + 1 = 3417
-#     4545 + 3417 == 7962
+#     dated         3417 + 1 = 3418
+#     4545 + 3418 == 7963
+#
+# #1748's review (jdatcmd), in parallel_flush_optin (OffgridwithJD): 4545 -> 4545. Arm 6's
+# structure check read from dsm_pin_segment to the DONE write, so a commit moved AFTER
+# the DONE write passed it (an ERROR there lands in PG_CATCH, which overwrites DONE with
+# FAILED, and the segment leaks). The stretch now runs to the end of the PG_TRY block and
+# must be the detach and the DONE write alone; the row is renamed for it and keeps its
+# date (the commit moved back before the DONE write still reddens it).
+#
+#     ledger rows   7963 + 0 = 7963
+#     never         4545 + 0 = 4545
+#     dated         3418 + 0 = 3418
+#     4545 + 3418 == 7963
 checks_never_observed_red 4545

test/parallel_flush_optin.sh

@@ -222,4 +222,26 @@ after_unpin="$(perl -0777 -ne '
 check_text "the collect loop forgets a worker's handle in the statement after it unpins the segment" \
 	"$after_unpin" "slots[i].outHandle = DSM_HANDLE_INVALID;"
 
+# ---- 6. nothing raises between a worker's pin and its DONE (#1747) ----------
+#
+# A worker pins its OUTPUT segment so it outlives the worker. On the NORMAL path the
+# dispatcher's collect loop unpins only a DONE slot; a worker that raised after the
+# pin reported FAILED, and its pinned segment stayed until postmaster restart
+# (measured on main with an injected ERROR after the pin: 8 segments left by a
+# two-stripe flush). The worker now commits first and pins last, so nothing after
+# dsm_pin_segment can raise. The stretch read runs to the end of the PG_TRY block, not
+# just to the DONE write: an ERROR after DONE also lands in PG_CATCH, which overwrites
+# DONE with FAILED, so a commit moved after the DONE write leaks the segment too
+# (#1748 review). Comments blanked, it must be the detach and the DONE write alone.
+after_pin="$(perl -0777 -ne '
+	if (/\npgcolumnar_parallel_flush_worker\(Datum main_arg\)\n\{(.*?)\n\}\n/s) {
+		my $f = $1;
+		if ($f =~ /dsm_pin_segment\(outseg\);(.*?)\}\s*PG_CATCH\(\);/s) {
+			my $b = $1; $b =~ s{/\*.*?\*/}{}gs; $b =~ s{//[^\n]*}{}g; $b =~ s/\s+/ /g; $b =~ s/^ | $//g;
+			print $b eq "" ? "(nothing)" : $b;
+		} else { print "the pin and the PG_CATCH are not both in the worker" }
+	} else { print "pgcolumnar_parallel_flush_worker not found" }' "$SRC/columnar_write_state.c")"
+check_text "a flush worker runs nothing that can raise after pinning its output, to the end of its PG_TRY block" \
+	"$after_pin" "dsm_detach(outseg); pg_atomic_write_u32(&me->state, PFLUSH_DONE);"
+
 pgc_summary

test/pytest/TESTS.md

@@ -10302,9 +10302,12 @@ The ON check reads the split line ("N of M columns by K worker(s)") as well as t
 | `test_double_cancel_reports_the_first_cancel` | a twice-cancelled parallel flush reports the first cancel |
 | `test_double_cancel_leaves_no_segment_pinned` | a twice-cancelled parallel flush leaves no worker segment pinned |
 | `test_the_collect_loop_forgets_a_handle_after_its_unpin` | the collect loop forgets a worker's handle in the statement after it unpins the segment |
+| `test_a_flush_worker_raises_nothing_after_its_pin` | a flush worker runs nothing that can raise after pinning its output, to the end of its PG_TRY block |
 
 Arm 4 (#1745) runs on a private cluster with `dynamic_shared_memory_type = mmap`, so the workers' output segments are files in that cluster's own `pg_dynshmem` and can be counted from SQL. A 2s `statement_timeout` cancels a parallel flush while one worker still encodes a zstd-19 text column, and `pg_cancel_backend` cancels it again 1.5s later, inside the error cleanup's wait. No new segment may survive, and the error reported must be the first cancel's.
 
+Arm 6 (#1747) reads `pgcolumnar_parallel_flush_worker` with comments blanked: from `dsm_pin_segment` to the end of its PG_TRY block, only `dsm_detach` and the DONE write may run, because the collect loop unpins only a DONE slot and a worker that raised after its pin left the segment pinned to postmaster restart. The stretch runs past the DONE write because an ERROR there lands in PG_CATCH, which overwrites DONE with FAILED (#1748 review).
+
 ## 263. test_parquet_export_stats.py: the Parquet files we export carry correct per-row-group statistics
 
 The pytest twin of `test/parquet_export_stats.sh` (ported for #1341), every check by the shell's name: 162 tests, three of the functions parametrized over a literal table of the shell's names. A 196609-row table is exported in four row groups and `test/parquet_stats.py` reads the footer with no Parquet library. The bounds, null counts, float rules (#850), column orders and the FDW's row-group skipping on the exported file are asserted. On a private database; the exported files live in `server_dir`.
@@ -12837,3 +12840,4 @@ Every test reads one value from a module fixture that runs the whole scenario on
 | `test_the_server_log_has_no_errstart_was_not_called` | double cancel: the server log has no 'errstart was not called' |
 | `test_no_completed_object_remains_at_the_prefix` | double cancel: no completed object remains at the prefix |
 | `test_the_server_survived` | double cancel: the server survived |
+

test/pytest/expected_tests.txt

@@ -1191,6 +1191,8 @@ guard_tests 839
 # cluster_tests 2748 -> 2753 for #1706's five timestamp-flag tests in test_native_parquet_schema.py, by collection.
 # cluster_tests 4221 -> 4225 for #1745's arm 4 in test_parallel_flush_optin.py: four tests,
 # one per check, by collection over the cluster files.
+# cluster_tests 4226 -> 4227 for #1747's arm 6 in test_parallel_flush_optin.py: one test, by
+# collection over the cluster files.
 # cluster_tests 4214 -> 4221 for #1341's test_objstore_export_double_cancel.py, the twin of objstore_export_double_cancel.sh:
 # 7 checks in 7 tests, by collection over the cluster files.
 # cluster_tests 4200 -> 4214 for #1738's fourteen timestamp/timestamptz tests in test_native_skip.py, by collection.
@@ -1206,4 +1208,4 @@ guard_tests 839
 # 54 tests, one per check, by collection over the cluster files.
 # cluster_tests 4225 -> 4226 for #1753's arm 5 in test_parallel_flush_optin.py (folded into #1746):
 # one structural test, by collection over the cluster files.
-cluster_tests 4226
+cluster_tests 4227

test/pytest/test_parallel_flush_optin.py

@@ -286,3 +286,29 @@ def _after_unpin():
 def test_the_collect_loop_forgets_a_handle_after_its_unpin(expect):
     expect.text(_after_unpin(), "slots[i].outHandle = DSM_HANDLE_INVALID;",
                 "the collect loop forgets a worker's handle in the statement after it unpins the segment")
+
+
+# ---- 6. nothing raises between a worker's pin and its DONE (#1747) ----------
+#
+# The shell's arm 6. On the NORMAL path the collect loop unpins only a DONE slot, so a
+# worker that raised after pinning its OUTPUT segment left it pinned until postmaster
+# restart. The worker now commits first and pins last: from dsm_pin_segment to the
+# end of the PG_TRY block, with comments blanked, only the detach and the DONE write
+# may run (an ERROR after DONE lands in PG_CATCH too, which overwrites it with FAILED).
+def _after_pin():
+    text = (SRC / "columnar_write_state.c").read_text()
+    m = re.search(r"\npgcolumnar_parallel_flush_worker\(Datum main_arg\)\n\{(.*?)\n\}\n", text, re.S)
+    if not m:
+        return "pgcolumnar_parallel_flush_worker not found"
+    b = re.search(r"dsm_pin_segment\(outseg\);(.*?)\}\s*PG_CATCH\(\);", m.group(1), re.S)
+    if not b:
+        return "the pin and the PG_CATCH are not both in the worker"
+    body = re.sub(r"/\*.*?\*/", "", b.group(1), flags=re.S)
+    body = re.sub(r"//[^\n]*", "", body)
+    body = re.sub(r"\s+", " ", body).strip()
+    return body or "(nothing)"
+
+
+def test_a_flush_worker_raises_nothing_after_its_pin(expect):
+    expect.text(_after_pin(), "dsm_detach(outseg); pg_atomic_write_u32(&me->state, PFLUSH_DONE);",
+                "a flush worker runs nothing that can raise after pinning its output, to the end of its PG_TRY block")