try maintain error message compat, avoid erroring inside PG destructor

Author: serprex

src/binary.c

@@ -233,13 +233,13 @@ unwrap_for_block_column(const chc_type * t)
 }
 
 pg_noreturn static void
-raise_chc(const chc_err * err, int sqlstate)
+raise_chc(const chc_err * err, int sqlstate, const char *prefix)
 {
 	const char *m = (err && err->msg[0]) ? err->msg : "unknown error";
 
 	ereport(ERROR,
 			(errcode(sqlstate),
-			 errmsg("pg_clickhouse: %s", m)));
+			 errmsg("pg_clickhouse: %s%s", prefix ? prefix : "", m)));
 }
 
 /* ===================================================================== */
@@ -470,7 +470,8 @@ ch_binary_connect(ch_connection_details * details)
 		int			rc = chc_client_init(&s->client, &opts, &pg_chc_alloc, &s->io, &err);
 
 		if (rc != CHC_OK)
-			raise_chc(&err, ERRCODE_FDW_UNABLE_TO_ESTABLISH_CONNECTION);
+			raise_chc(&err, ERRCODE_FDW_UNABLE_TO_ESTABLISH_CONNECTION,
+					  "connection error: ");
 	}
 	PG_CATCH();
 	{
@@ -946,6 +947,7 @@ struct ch_binary_insert_handle
 	bool		array_active;
 	size_t		array_col_idx;
 	bool		started;
+	bool		finalized;		/* finalize_insert has run (success or raised) */
 };
 
 static void
@@ -1050,7 +1052,7 @@ classify_column(ic_col * ic, const chc_type * col_t)
 
 		ereport(ERROR,
 				(errcode(ERRCODE_FDW_INVALID_DATA_TYPE),
-				 errmsg("pg_clickhouse: unsupported column type: %.*s",
+				 errmsg("pg_clickhouse: could not prepare insert - unsupported column type: %.*s",
 						(int) tnlen, tname ? tname : "?")));
 	}
 }
@@ -1067,7 +1069,7 @@ recv_initial_block(struct ch_binary_state *s, ch_binary_insert_handle * h)
 		if (rc != CHC_OK)
 		{
 			s->broken = true;
-			raise_chc(&err, ERRCODE_FDW_ERROR);
+			raise_chc(&err, ERRCODE_FDW_ERROR, "could not prepare insert - ");
 		}
 		if (pkt.kind == CHC_PKT_EXCEPTION)
 		{
@@ -1083,7 +1085,7 @@ recv_initial_block(struct ch_binary_state *s, ch_binary_insert_handle * h)
 			chc_packet_clear(s->client, &pkt);
 			ereport(ERROR,
 					(errcode(ERRCODE_FDW_ERROR),
-					 errmsg("pg_clickhouse: %s", msg_copy)));
+					 errmsg("pg_clickhouse: could not prepare insert - %s", msg_copy)));
 		}
 		if (pkt.kind == CHC_PKT_DATA && pkt.block &&
 			chc_block_n_columns(pkt.block) > 0)
@@ -1189,7 +1191,7 @@ ch_binary_begin_insert(ch_binary_connection_t * conn, const ch_query * query,
 		if (rc != CHC_OK)
 		{
 			s->broken = true;
-			raise_chc(&err, ERRCODE_FDW_ERROR);
+			raise_chc(&err, ERRCODE_FDW_ERROR, "could not prepare insert - ");
 		}
 
 		recv_initial_block(s, h);
@@ -1262,7 +1264,7 @@ resolve_col(ch_binary_insert_handle * h, size_t col_idx, bool isnull)
 
 		ereport(ERROR,
 				(errcode(ERRCODE_NOT_NULL_VIOLATION),
-				 errmsg("pg_clickhouse: cannot append NULL to NOT NULL %.*s column",
+				 errmsg("pg_clickhouse: could not append data to column - cannot append NULL to NOT NULL %.*s column",
 						(int) tnlen, tname ? tname : "?")));
 	}
 	if (!h->array_active && c->is_nullable)
@@ -1978,7 +1980,7 @@ ch_binary_flush_block(ch_binary_insert_handle * h)
 	int			rc = chc_block_builder_init(&bb, &pg_chc_alloc, &err);
 
 	if (rc != CHC_OK)
-		raise_chc(&err, ERRCODE_FDW_ERROR);
+		raise_chc(&err, ERRCODE_FDW_ERROR, "could not append data to column - ");
 
 	for (size_t i = 0; i < h->ncols; i++)
 	{
@@ -2101,7 +2103,8 @@ ch_binary_flush_block(ch_binary_insert_handle * h)
 				}
 		}
 		if (rc != CHC_OK)
-			raise_chc(&err, ERRCODE_FDW_ERROR);
+			raise_chc(&err, ERRCODE_FDW_ERROR,
+					  "could not append data to column - ");
 	}
 
 	rc = chc_client_send_data(h->client, bb, &err);
@@ -2110,7 +2113,7 @@ ch_binary_flush_block(ch_binary_insert_handle * h)
 	{
 		if (h->state)
 			h->state->broken = true;
-		raise_chc(&err, ERRCODE_FDW_ERROR);
+		raise_chc(&err, ERRCODE_FDW_ERROR, "could not insert columns - ");
 	}
 
 	/* Reset per-column buffers for next batch. */
@@ -2132,62 +2135,72 @@ ch_binary_flush_block(ch_binary_insert_handle * h)
 	MemoryContextSwitchTo(old);
 }
 
+/*
+ * Send final empty Data + drain. May ereport on server exception or
+ * transport failure. Idempotent via h->finalized. Leaves h->cxt alive;
+ * ch_binary_release_insert deletes it.
+ */
 void
-ch_binary_end_insert(ch_binary_insert_handle * h)
+ch_binary_finalize_insert(ch_binary_insert_handle * h)
 {
-	if (!h)
+	if (!h || h->finalized)
+		return;
+
+	/*
+	 * Set early so an ereport(ERROR) below still leaves h in the "do not
+	 * touch the wire" state for the release callback.
+	 */
+	h->finalized = true;
+
+	if (!(h->started && h->client))
 		return;
 
 	MemoryContext old = MemoryContextSwitchTo(h->cxt);
 	char	   *exc_msg = NULL;
 	bool		broke = false;
+	chc_err		err = {0};
+	int			rc = chc_client_send_data(h->client, NULL, &err);
 
-	if (h->started && h->client)
+	if (rc != CHC_OK)
 	{
-		chc_err		err = {0};
-		int			rc = chc_client_send_data(h->client, NULL, &err);
-
-		if (rc != CHC_OK)
-		{
-			broke = true;
-			exc_msg = pstrdup(err.msg[0] ? err.msg : "send_data failed");
-		}
-		else
+		broke = true;
+		exc_msg = pstrdup(err.msg[0] ? err.msg : "send_data failed");
+	}
+	else
+	{
+		/* Drain until EOS or exception. */
+		for (;;)
 		{
-			/* Drain until EOS or exception. */
-			for (;;)
-			{
-				chc_packet	pkt = {0};
+			chc_packet	pkt = {0};
 
-				err = (chc_err)
-				{
-					0
-				};
-				rc = chc_client_recv_packet(h->client, &pkt, &err);
-				if (rc != CHC_OK)
-				{
-					broke = true;
-					exc_msg = pstrdup(err.msg[0] ? err.msg : "recv_packet failed");
-					chc_packet_clear(h->client, &pkt);
-					break;
-				}
-				if (pkt.kind == CHC_PKT_EXCEPTION)
-				{
-					const char *msg = "server exception";
-
-					if (pkt.exception && pkt.exception->display_text)
-						msg = pkt.exception->display_text;
-					else if (pkt.exception && pkt.exception->name)
-						msg = pkt.exception->name;
-					exc_msg = pstrdup(msg);
-					broke = true;
-					chc_packet_clear(h->client, &pkt);
-					break;
-				}
+			err = (chc_err)
+			{
+				0
+			};
+			rc = chc_client_recv_packet(h->client, &pkt, &err);
+			if (rc != CHC_OK)
+			{
+				broke = true;
+				exc_msg = pstrdup(err.msg[0] ? err.msg : "recv_packet failed");
 				chc_packet_clear(h->client, &pkt);
-				if (pkt.kind == CHC_PKT_END_OF_STREAM)
-					break;
+				break;
+			}
+			if (pkt.kind == CHC_PKT_EXCEPTION)
+			{
+				const char *msg = "server exception";
+
+				if (pkt.exception && pkt.exception->display_text)
+					msg = pkt.exception->display_text;
+				else if (pkt.exception && pkt.exception->name)
+					msg = pkt.exception->name;
+				exc_msg = pstrdup(msg);
+				broke = true;
+				chc_packet_clear(h->client, &pkt);
+				break;
 			}
+			chc_packet_clear(h->client, &pkt);
+			if (pkt.kind == CHC_PKT_END_OF_STREAM)
+				break;
 		}
 	}
 
@@ -2198,20 +2211,33 @@ ch_binary_end_insert(ch_binary_insert_handle * h)
 	if (broke && h->state)
 		h->state->broken = true;
 
-	/*
-	 * Copy exc_msg into the parent context before deleting h->cxt so the
-	 * ereport message survives.
-	 */
+	MemoryContextSwitchTo(old);
+
 	if (exc_msg)
 	{
-		MemoryContextSwitchTo(old);
+		/* exc_msg lives in h->cxt; copy into parent before raising. */
 		char	   *parent_msg = pstrdup(exc_msg);
 
-		MemoryContextDelete(h->cxt);
 		ereport(ERROR,
 				(errcode(ERRCODE_FDW_ERROR),
-				 errmsg("pg_clickhouse: %s", parent_msg)));
+				 errmsg("pg_clickhouse: could not finish INSERT - %s", parent_msg)));
 	}
-	MemoryContextSwitchTo(old);
+}
+
+/*
+ * Teardown counterpart to finalize. Safe from a MemoryContext reset
+ * callback: never raises, never talks to the server. If finalize did not
+ * run (mid-query abort), flags the connection broken so it rebuilds on
+ * next use.
+ */
+void
+ch_binary_release_insert(ch_binary_insert_handle * h)
+{
+	if (!h)
+		return;
+
+	if (!h->finalized && h->started && h->state)
+		h->state->broken = true;
+
 	MemoryContextDelete(h->cxt);
 }

src/binary_encode.c

@@ -381,6 +381,11 @@ ch_binary_insert_state_free(void *c)
 	if (state->insert_block == NULL)
 		return;
 
-	ch_binary_end_insert(state->insert_block);
+	/*
+	 * Reset-callback context: cannot ereport. Finalize runs from the FDW
+	 * happy path before we get here; if it did not (mid-query abort), the
+	 * release call flags the connection broken so the next query rebuilds.
+	 */
+	ch_binary_release_insert(state->insert_block);
 	state->insert_block = NULL;
 }

src/fdw.c

@@ -1420,6 +1420,15 @@ clickhouseEndForeignInsert(EState * estate,
 		/* flush */
 		oldcontext = MemoryContextSwitchTo(fmstate->temp_cxt);
 		fmstate->conn.methods->insert_tuple(fmstate->state, NULL);
+
+		/*
+		 * Finalize on the happy path so the binary driver can ereport on a
+		 * server-side INSERT exception without raising from inside the
+		 * MemoryContext reset callback that fires during abort.
+		 */
+		if (fmstate->conn.methods->finalize_insert)
+			fmstate->conn.methods->finalize_insert(fmstate->state);
+
 		MemoryContextSwitchTo(oldcontext);
 		MemoryContextReset(fmstate->temp_cxt);
 

src/include/binary.h

@@ -130,10 +130,18 @@ extern uint32_t ch_binary_column_datetime64_precision(const ch_binary_insert_han
 extern void ch_binary_flush_block(ch_binary_insert_handle * h);
 
 /*
- * End insert and free handle. Idempotent: passing NULL is a no-op.
- * ereports if server raised after final empty Data.
+ * Finish the insert: send final empty Data, drain the response, ereport
+ * if the server raised. Idempotent; call exactly once from the FDW happy
+ * path before tearing down the handle.
  */
-extern void ch_binary_end_insert(ch_binary_insert_handle * h);
+extern void ch_binary_finalize_insert(ch_binary_insert_handle * h);
+
+/*
+ * Tear down the handle. Never raises and never talks to the server, so it
+ * is safe to call from a MemoryContext reset callback during transaction
+ * abort. Flags the connection broken if finalize did not run.
+ */
+extern void ch_binary_release_insert(ch_binary_insert_handle * h);
 
 /* PG-typed surface follows. */
 

src/include/fdw.h

@@ -75,6 +75,7 @@ typedef Datum * (*cursor_fetch_row_method) (ChFdwScanRowContext * ctx);
 typedef void *(*prepare_insert_method) (void *conn, ResultRelInfo *, List *,
 										const ch_query *, char *);
 typedef void (*insert_tuple_method) (void *state, TupleTableSlot * slot);
+typedef void (*finalize_insert_method) (void *state);
 typedef ch_cursor * (*streaming_query_method) (void *conn,
 											   const ch_query * query,
 											   int32 fetch_size);
@@ -87,6 +88,8 @@ typedef struct
 	cursor_fetch_row_method fetch_row;
 	prepare_insert_method prepare_insert;
 	insert_tuple_method insert_tuple;
+	finalize_insert_method finalize_insert; /* NULL if no explicit finalize
+											 * step needed */
 	streaming_query_method streaming_query; /* NULL if not supported */
 	cursor_fetch_row_method streaming_fetch_row;	/* NULL if not supported */
 	is_broken_method is_broken; /* NULL means connection is never broken */

src/pglink.c

@@ -63,6 +63,7 @@ static bool binary_is_broken(const void *conn);
 /* static void binary_simple_insert(void *conn, const char *query); */
 static Datum * binary_fetch_row(ChFdwScanRowContext * ctx);
 static void binary_insert_tuple(void *, TupleTableSlot * slot);
+static void binary_finalize_insert(void *istate);
 static void *binary_prepare_insert(void *, ResultRelInfo *, List *,
 								   const ch_query * query, char *table_name);
 static char *ch_escape_string(const char *s, size_t len);
@@ -77,6 +78,7 @@ static libclickhouse_methods binary_methods =
 		.fetch_row = binary_fetch_row,
 		.prepare_insert = binary_prepare_insert,
 		.insert_tuple = binary_insert_tuple,
+		.finalize_insert = binary_finalize_insert,
 		.streaming_query = NULL,
 		.streaming_fetch_row = NULL,
 		.is_broken = binary_is_broken,
@@ -1157,6 +1159,15 @@ binary_insert_tuple(void *istate, TupleTableSlot * slot)
 	}
 }
 
+static void
+binary_finalize_insert(void *istate)
+{
+	ch_binary_insert_state *state = istate;
+
+	if (state && state->insert_block)
+		ch_binary_finalize_insert(state->insert_block);
+}
+
 /*
  * Query to generate table for doc/pg_clickhouse.md. Keep in sync with
  * str_types_map below. On change, re-run and paste the output into

test/expected/binary_inserts.out

@@ -490,15 +490,15 @@ INSERT INTO default_vals VALUES(
 	   NULL, NULL, NULL, NULL, ARRAY[NULL]::int[],
 	   NULL, NULL, NULL
 );
-ERROR:  pg_clickhouse: cannot append NULL to NOT NULL UInt8 column
+ERROR:  pg_clickhouse: could not append data to column - cannot append NULL to NOT NULL UInt8 column
 SELECT * FROM default_vals;
  c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 | c9 | c10 | c11 | c12 | c13 | c14 | c15 | c16 | c17 | c18 | c19 | c20 | c21 | c22 | c23 | c24 | c25 | c26 | c27 | c28 
 ----+----+----+----+----+----+----+----+----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----
 (0 rows)
 
 -- Test unsupported Nullables.
 INSERT INTO not_nullable VALUES (1, 'x', 'x', NULL);
-ERROR:  pg_clickhouse: cannot append NULL to NOT NULL LowCardinality(Nullable(String)) column
+ERROR:  pg_clickhouse: could not append data to column - cannot append NULL to NOT NULL LowCardinality(Nullable(String)) column
 DROP USER MAPPING FOR CURRENT_USER SERVER binary_inserts_loopback;
 SELECT clickhouse_raw_query('DROP DATABASE binary_inserts_test');
  clickhouse_raw_query