diff --git a/CHANGELOG.md b/CHANGELOG.md index 13ed22283..3d1baa30c 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index 811d2902b..0760c6c62 100644 --- a/src/columnar_write_state.c +++ b/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)); diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index 417adc493..053c08f87 100644 --- a/test/check_ledger.tsv +++ b/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]) diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index e7e423856..2cea03ef9 100644 --- a/test/check_ledger_budget.txt +++ b/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 diff --git a/test/parallel_flush_optin.sh b/test/parallel_flush_optin.sh index 1f05454fe..9590d2deb 100755 --- a/test/parallel_flush_optin.sh +++ b/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 diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index e92c48b11..cedfcb647 100644 --- a/test/pytest/TESTS.md +++ b/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 | + diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index c49fd3bce..38492c1df 100644 --- a/test/pytest/expected_tests.txt +++ b/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 diff --git a/test/pytest/test_parallel_flush_optin.py b/test/pytest/test_parallel_flush_optin.py index bd36c1b3b..4655c0fc8 100644 --- a/test/pytest/test_parallel_flush_optin.py +++ b/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")