fix: a second cancel during a remote export's cleanup keeps the export's own error (#1741)

.github/scripts/unrunnable-arm-names.py

@@ -52,8 +52,8 @@
 mismatch green: silent. So refusal wrappers are followed transitively, while the
 twin set is restricted to `lib.sh` and to suite-local definitions spelled
 `check` or `check_<word>`. `check[a-z_]*` is NOT that spelling: it also matches
-`checks_in` in `decode_interrupts.sh`, a counting utility that records nothing
-(@jdatcmd). A population named after a prefix is not a population named after a
+`checks_in`, the counting utility `decode_interrupts.sh` had until #1341, which records
+nothing (@jdatcmd). A population named after a prefix is not a population named after a
 behaviour.
 
 AND THE SUBJECT MUST NOT BE IN THE REFERENCE SET. The first sweep of this

.github/workflows/ci.yml

@@ -185,6 +185,13 @@ jobs:
       - name: fetch the release tags, for the fixture-blob comparisons
         run: git fetch --depth=1 origin '+refs/tags/v1.0-alpha*:refs/tags/v1.0-alpha*'
 
+      - name: Resolve the object-storage suite host names
+        # test_objstore_addressing.py, the twin of objstore_addressing.sh, reads
+        # through virtual-host-style S3 addressing (pgc-bucket.s3.local) against
+        # its own loopback fixture server, as the suites job's shells do. Without
+        # the names its tests decline UNAVAILABLE_ENDPOINT and the run ends 67.
+        run: echo "127.0.0.1 s3.local pgc-bucket.s3.local" | sudo tee -a /etc/hosts
+
       - name: Add the PGDG repository and refresh the package lists
         run: |
           set -euo pipefail

.github/workflows/nightly.yml

@@ -745,6 +745,13 @@ jobs:
       - name: fetch the release tags, for the fixture-blob comparisons
         run: git fetch --depth=1 origin '+refs/tags/v1.0-alpha*:refs/tags/v1.0-alpha*'
 
+      - name: Resolve the object-storage suite host names
+        # test_objstore_addressing.py, the twin of objstore_addressing.sh, reads
+        # through virtual-host-style S3 addressing (pgc-bucket.s3.local) against
+        # its own loopback fixture server, as the suites job's shells do. Without
+        # the names its tests decline UNAVAILABLE_ENDPOINT and the run ends 67.
+        run: echo "127.0.0.1 s3.local pgc-bucket.s3.local" | sudo tee -a /etc/hosts
+
       - name: Add the PGDG repository and install PostgreSQL ${{ matrix.pg }} with its headers
         run: |
           set -euo pipefail

CHANGELOG.md

@@ -400,6 +400,24 @@ true until the next version shipped.
   Reported by @OffgridwithJD.
 
 ### Fixed
+- **A `timestamp` column compared with a `timestamptz` constant no longer loses
+  rows across a spring-forward gap.** The zone maps compared each chunk group's
+  min/max through the cross-type ordering proc, which reads a wall time as an
+  instant in the session zone, and that is not order-preserving inside the gap:
+  in New York on 2024-03-10, `02:30` is 07:30Z but `03:00` is 07:00Z, so a group
+  whose max was `03:00` was skipped for `ts > '07:15+00'` although its `02:30` row
+  matched. That pair is no longer used to skip; the reverse pair and same-type
+  comparisons still are (#1738). `test/native_skip.sh` and `test_native_skip.py`.
+- **A second cancel during a remote export's cleanup no longer reports "errstart
+  was not called".** When a `parallel_export_parquet` to `s3://` was cancelled
+  again while the first cancel's cleanup was still deleting its parts, the export
+  failed with `XX000: errstart was not called` instead of its own 57014. The
+  cleanup ran with the first cancel's error still on the error stack, and the
+  object-store module's best-effort catch, swallowing the second cancel, emptied
+  the whole stack. That cleanup, and the sink abort in the Parquet, Arrow and
+  `_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`.
 - **`export_parquet` no longer marks a `timestamp` column as a UTC instant.** Both
   `timestamp` and `timestamptz` were written as the converted type `TIMESTAMP_MICROS`,
   which the Parquet spec defines as adjusted to UTC. Every other reader (pyarrow,
@@ -417,6 +435,15 @@ true until the next version shipped.
   each, `stats()` read 104 blocks of the table and `sort_status()` 170. Each now reads
   the storage id once: 1 block (#1637). `test/catalog_filter_reads.sh` and
   `test_catalog_filter_reads.py`.
+- **`pgcolumnar.recluster` no longer duplicates rows inserted while it waits for
+  its groups.** The online recluster listed the groups it would retire, then
+  locked them and read the table under a later snapshot, and retired only the
+  groups it had listed. A row committed in between, including while the recluster
+  waited for another session's group lock, sat in a group it never listed: it was
+  read and rewritten, and its own group was kept, so the row appeared twice. The
+  rewrite now reads only the groups it listed. A row written in that window stays
+  where it is and is counted as appended, and a row deleted in that window stays
+  deleted (#1633). `test/recluster_extent.sh` and `test_recluster_extent.py`.
 - **A NOLOGIN role can run `parallel_export_parquet`, `parallel_copy` and the
   parallel flush on PostgreSQL 15 and 16.** After `SET ROLE` to a role without
   LOGIN, the export failed `worker 0 failed: worker exited without reporting a

docs/configuration.md

@@ -123,7 +123,7 @@ answer.
 | Setting | Type | Default | Description |
 | --- | --- | --- | --- |
 | `pgcolumnar.bulk_parallel_writer` | boolean | `off` | Internal. Set by `pgcolumnar.parallel_copy` loader workers so they skip the storage-row creation lock when the row already exists committed, which is what lets several atomic writers load one table at once. Marked `GUC_NOT_IN_SAMPLE`; leave it alone. Setting it by hand is safe but pointless: the skip only fires when the storage row is already committed, which is exactly when the lock guards nothing. |
-| `pgcolumnar.maintenance_hold_ms` | integer | `0` | Internal, for tests. A maintenance verb holds `ShareUpdateExclusiveLock` this many milliseconds, interruptibly, so a test can observe the daemon yield to a stronger lock. `recluster` also waits, after it has fixed the rows it will rewrite, so a test can write while it runs. `0` disables it. Range 0 to 600000. Leave it at `0`. |
+| `pgcolumnar.maintenance_hold_ms` | integer | `0` | Internal, for tests. A maintenance verb holds `ShareUpdateExclusiveLock` this many milliseconds, interruptibly, so a test can observe the daemon yield to a stronger lock. `recluster` also waits, after it has listed the groups it will rewrite and before it locks them, so a test can write while it runs. `0` disables it. Range 0 to 600000. Leave it at `0`. |
 | `pgcolumnar.sink_fail_after` | integer | `-1` | Internal, for tests. A fault-injection point that fails an export write after this many bytes, by the path a full disk takes. `-1` disables it. Range -1 to INT_MAX. Leave it at `-1`. |
 
 ## Per-table storage options

objstore/columnar_objstore_module.c

@@ -245,8 +245,15 @@ os_wait(PgColumnarObjHandle *h, uint32 sockEvents, int *quiet)
 {
 	int			rc;
 
+	/*
+	 * WL_TIMEOUT, or the slice is not a slice: without it WaitLatchOrSocket
+	 * ignores the timeout and waits for the socket or the latch alone, so a
+	 * silent peer never counts a quiet slice. Under proc_exit (a terminated
+	 * worker aborting its upload, #1727) interrupts are held and the latch
+	 * alone never ends the wait either.
+	 */
 	rc = WaitLatchOrSocket(MyLatch,
-						   WL_LATCH_SET | WL_EXIT_ON_PM_DEATH | sockEvents,
+						   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH | sockEvents,
 						   h->fd, OS_WAIT_SLICE_MS, PG_WAIT_EXTENSION);
 	if (rc & WL_LATCH_SET)
 		ResetLatch(MyLatch);
@@ -2002,8 +2009,8 @@ os_flush_part(PgColumnarObjSink *s)
 						s->h->url, resp.status)));
 	/* The part's ETag is the server's word for its bytes; CompleteMultipart
 	 * validates the list against it on a real S3, so echo it verbatim (already
-	 * quoted by the server). The fixture ignores it and concatenates by part
-	 * number, so both are satisfied. */
+	 * quoted by the server). The fixture checks it the same way and assembles
+	 * the listed parts, so both are satisfied. */
 	etag = resp.etag[0] ? resp.etag : "\"\"";
 	appendStringInfo(&s->completeXml,
 					 "<Part><PartNumber>%d</PartNumber><ETag>%s</ETag></Part>",
@@ -2170,6 +2177,14 @@ objstore_sink_abort(PgColumnarObjSink *s)
 			char	   *encId = os_uriencode_query(s->uploadId);
 			char	   *query = psprintf("uploadId=%s", encId);
 
+			/*
+			 * The abort can land mid-request (a cancel or a SIGTERM inside
+			 * os_wait), with a part's body still on this connection: the
+			 * DELETE would be read as more of that body, or its answer
+			 * confused with the part's. A fresh connection makes it a request
+			 * of its own (#1722).
+			 */
+			os_disconnect(s->h);
 			os_write_request(s->h, "DELETE", query, NULL, 0, &resp, NULL);
 			pfree(query);
 			pfree(encId);

src/columnar_arrow.c

@@ -1171,8 +1171,7 @@ pgcolumnar_export_arrow(PG_FUNCTION_ARGS)
 	}
 	PG_CATCH();
 	{
-		PgColumnarSinkAbort(snk);
-		PG_RE_THROW();
+		PgColumnarSinkAbortAndRethrow(snk);	/* never returns (#1741) */
 	}
 	PG_END_TRY();
 

src/columnar_parallel_export.c

@@ -419,8 +419,7 @@ pexport_write_success_marker(const char *dir)
 	}
 	PG_CATCH();
 	{
-		PgColumnarSinkAbort(snk);
-		PG_RE_THROW();
+		PgColumnarSinkAbortAndRethrow(snk);	/* never returns (#1741) */
 	}
 	PG_END_TRY();
 }
@@ -439,6 +438,13 @@ pexport_cleanup(int code, Datum arg)
 	pexport_remove_outputs(s->dirpath, s->nkeys);
 }
 
+/* the ERROR path's form of pexport_cleanup, for PgColumnarCleanupAndRethrow */
+static void
+pexport_cleanup_in_catch(void *arg)
+{
+	pexport_cleanup(0, PointerGetDatum(arg));
+}
+
 /*
  * pgcolumnar_parallel_export_worker
  *		Background-worker entry: attach the DSM, connect, import the dispatcher's
@@ -815,11 +821,17 @@ pgcolumnar_parallel_export_parquet(PG_FUNCTION_ARGS)
 	bw.bgw_notify_pid = MyProcPid;
 
 	/*
-	 * PG_ENSURE_ERROR_CLEANUP so a FATAL in this backend still terminates the
-	 * workers before it exits, rather than orphaning them to read on after the
-	 * exported snapshot is gone.
+	 * PG_ENSURE_ERROR_CLEANUP's shape, spelled out: before_shmem_exit so a FATAL
+	 * in this backend still terminates the workers before it exits, rather than
+	 * orphaning them to read on after the exported snapshot is gone. The ERROR
+	 * path runs the same cleanup through PgColumnarCleanupAndRethrow, not the
+	 * macro's PG_CATCH (#1741): the cleanup's remote DELETEs wait on the store,
+	 * a second cancel lands there, and swallowing it with the cancel being
+	 * handled still on the error stack lost that error ("errstart was not
+	 * called").
 	 */
-	PG_ENSURE_ERROR_CLEANUP(pexport_cleanup, PointerGetDatum(&spawn));
+	before_shmem_exit(pexport_cleanup, PointerGetDatum(&spawn));
+	PG_TRY();
 	{
 		for (i = 0; i < workers; i++)
 		{
@@ -854,7 +866,13 @@ pgcolumnar_parallel_export_parquet(PG_FUNCTION_ARGS)
 			ResetLatch(MyLatch);
 		}
 	}
-	PG_END_ENSURE_ERROR_CLEANUP(pexport_cleanup, PointerGetDatum(&spawn));
+	PG_CATCH();
+	{
+		cancel_before_shmem_exit(pexport_cleanup, PointerGetDatum(&spawn));
+		PgColumnarCleanupAndRethrow(pexport_cleanup_in_catch, &spawn);
+	}
+	PG_END_TRY();
+	cancel_before_shmem_exit(pexport_cleanup, PointerGetDatum(&spawn));
 
 	for (i = 0; i < workers; i++)
 	{

src/columnar_parquet.c

@@ -1384,8 +1384,7 @@ PgColumnarWriteParquetFile(Relation rel, Snapshot snapshot, const char *filepath
 	}
 	PG_CATCH();
 	{
-		PgColumnarSinkAbort(snk);
-		PG_RE_THROW();
+		PgColumnarSinkAbortAndRethrow(snk);	/* never returns (#1741) */
 	}
 	PG_END_TRY();
 

src/columnar_reader.c

@@ -951,6 +951,19 @@ pgcolumnar_make_predicates(SkipPredicate *out, int nkeys, ScanKey keys,
 
 			if (!OidIsValid(opclass))
 				continue;
+
+			/*
+			 * Not a timestamp column against a timestamptz constant (#1738).
+			 * timestamp_cmp_timestamptz turns the column's value into an instant
+			 * in the session zone, and across a spring-forward gap that is not
+			 * order-preserving: in New York on 2024-03-10 the nonexistent 02:30
+			 * is read as standard time, 07:30Z, while 03:00 is 07:00Z. A group
+			 * whose max is 03:00 then looked wholly below 07:15Z although its
+			 * 02:30 row is above it, and was skipped. The reverse pair converts
+			 * the constant once, and the date pairs carry no zone, so both stay.
+			 */
+			if (colType == TIMESTAMPOID && argType == TIMESTAMPTZOID)
+				continue;
 			opfamily = get_opclass_family(opclass);
 			cmpProc = get_opfamily_proc(opfamily, colType, argType,
 										BTORDER_PROC);

src/columnar_sink.c

@@ -16,8 +16,12 @@
 
 #include <unistd.h>
 
+#include "access/xact.h"
+#include "lib/ilist.h"
 #include "miscadmin.h"
 #include "storage/fd.h"
+#include "utils/memutils.h"
+#include "storage/ipc.h"
 
 #include "columnar_objstore.h"
 #include "columnar_sink.h"
@@ -34,8 +38,62 @@ struct PqSink
 	/* remote (#394): NULL api => local sink */
 	const PgColumnarObjStoreApi *api;
 	PgColumnarObjSink *rsink;
+	dlist_node	openNode;		/* on sink_open_remote while rsink is live */
 };
 
+/*
+ * Remote sinks with an upload in progress (#1722). A FATAL exit (SIGTERM, as
+ * pexport_cleanup sends each parallel_export worker on cancel, or
+ * pg_terminate_backend) calls proc_exit without unwinding, so the writers'
+ * PG_CATCH never reaches PgColumnarSinkAbort and the multipart upload stays
+ * open on the store, billed until a lifecycle rule reaps it. A
+ * before_shmem_exit callback aborts whatever is still on this list. It is
+ * registered after InitPostgres's ShutdownPostgres, so it runs first, while
+ * the transaction and the module's handles are still intact. Transaction end
+ * empties the list: every writer finishes or aborts its sink inside its own
+ * statement, so an entry still listed then belongs to memory being released.
+ */
+static dlist_head sink_open_remote = DLIST_STATIC_INIT(sink_open_remote);
+static bool sink_exit_registered = false;
+
+static void
+sink_abort_open_at_exit(int code, Datum arg)
+{
+	dlist_mutable_iter it;
+
+	/*
+	 * The module's abort never raises through PG_TRY, but under proc_exit an
+	 * ERROR is promoted to FATAL, which no CATCH sees. A failed DELETE then
+	 * ends this callback with one more FATAL line; shmem_exit has already
+	 * popped it, so the remaining callbacks still run.
+	 */
+	dlist_foreach_modify(it, &sink_open_remote)
+	{
+		PqSink	   *snk = dlist_container(PqSink, openNode, it.cur);
+
+		dlist_delete(&snk->openNode);
+		snk->api->sink_abort(snk->rsink);
+		snk->rsink = NULL;
+	}
+}
+
+static void
+sink_xact_callback(XactEvent event, void *arg)
+{
+	switch (event)
+	{
+		case XACT_EVENT_COMMIT:
+		case XACT_EVENT_PARALLEL_COMMIT:
+		case XACT_EVENT_ABORT:
+		case XACT_EVENT_PARALLEL_ABORT:
+		case XACT_EVENT_PREPARE:
+			dlist_init(&sink_open_remote);
+			break;
+		default:
+			break;
+	}
+}
+
 PqSink *
 PgColumnarSinkOpenLocal(const char *path)
 {
@@ -83,6 +141,13 @@ PgColumnarSinkOpen(const char *path)
 		snk->finalPath = pstrdup(path);
 		snk->api = api;
 		snk->rsink = api->sink_create(path, NULL);
+		if (!sink_exit_registered)
+		{
+			before_shmem_exit(sink_abort_open_at_exit, (Datum) 0);
+			RegisterXactCallback(sink_xact_callback, NULL);
+			sink_exit_registered = true;
+		}
+		dlist_push_head(&sink_open_remote, &snk->openNode);
 		return snk;
 	}
 	return PgColumnarSinkOpenLocal(path);
@@ -140,6 +205,8 @@ PgColumnarSinkFinish(PqSink *snk)
 	if (snk->api != NULL)
 	{
 		snk->api->sink_finish(snk->rsink);
+		/* only now: a complete that raised leaves it listed for the abort */
+		dlist_delete(&snk->openNode);
 		snk->rsink = NULL;
 		return;
 	}
@@ -173,6 +240,7 @@ PgColumnarSinkAbort(PqSink *snk)
 	{
 		if (snk->rsink != NULL)
 		{
+			dlist_delete(&snk->openNode);
 			snk->api->sink_abort(snk->rsink);
 			snk->rsink = NULL;
 		}
@@ -185,3 +253,48 @@ PgColumnarSinkAbort(PqSink *snk)
 	}
 	(void) unlink(snk->tmpPath);
 }
+
+/*
+ * PgColumnarCleanupAndRethrow
+ *		Called from a PG_CATCH. The error being handled is copied out and the
+ *		error stack emptied BEFORE the cleanup runs, so whatever the cleanup
+ *		raises and swallows (the module's best-effort DELETEs flush the error
+ *		stack, and a second cancel lands exactly there, #1741) cannot take the
+ *		handled error with it. An error the cleanup does not swallow ends the
+ *		cleanup early and is dropped: the error to report is the one being
+ *		handled. That one is then re-thrown, as PG_RE_THROW would have.
+ */
+void
+PgColumnarCleanupAndRethrow(void (*cleanup) (void *arg), void *arg)
+{
+	/* in a transaction for every caller; the copy only needs to outlive this */
+	MemoryContext keep = TopTransactionContext ? TopTransactionContext : TopMemoryContext;
+	ErrorData  *edata;
+
+	MemoryContextSwitchTo(keep);
+	edata = CopyErrorData();
+	FlushErrorState();
+	PG_TRY();
+	{
+		cleanup(arg);
+	}
+	PG_CATCH();
+	{
+		MemoryContextSwitchTo(keep);
+		FlushErrorState();
+	}
+	PG_END_TRY();
+	ReThrowError(edata);
+}
+
+static void
+sink_abort_cleanup(void *arg)
+{
+	PgColumnarSinkAbort((PqSink *) arg);
+}
+
+void
+PgColumnarSinkAbortAndRethrow(PqSink *snk)
+{
+	PgColumnarCleanupAndRethrow(sink_abort_cleanup, snk);
+}