Improved remote error and notice handling.

- Register libpq notice handler with PQsetNoticeReceiver
- Extract details from err/notice with PQresultErrorField
  and pass to ereport().
- Improve state-machine in another_result().

expected/plproxy_encoding.out

@@ -108,6 +108,7 @@ select * from test_encoding2('クライアント側のデータ');
 (1 row)
 
 select * from test_encoding3('クライアント側のデータ');
+NOTICE:  public.test_encoding3(1): REMOTE NOTICE: got: クライアント側のデータ
  id |         コラム         
 ----+------------------------
   3 | クライアント側のデータ
@@ -214,6 +215,7 @@ select * from test_encoding2('クライアント側のデータ');
 (1 row)
 
 select * from test_encoding3('クライアント側のデータ');
+NOTICE:  public.test_encoding3(1): REMOTE NOTICE: got: クライアント側のデータ
  id |         コラム         
 ----+------------------------
   3 | クライアント側のデータ

expected/plproxy_test.out

@@ -363,3 +363,35 @@ select * from test_simple(0);
            0
 (1 row)
 
+-- test error passing
+\c test_part
+create function test_error1() returns int4
+as $$
+begin
+    select line2err;
+    return 0;
+end;
+$$ language plpgsql;
+\c regression
+create function test_error1() returns int4
+as $$
+    cluster 'testcluster';
+    run on 0;
+$$ language plproxy;
+select * from test_error1();
+ERROR:  public.test_error1(0): REMOTE ERROR: column "line2err" does not exist
+LINE 1: select line2err
+               ^
+QUERY:  select line2err
+CONTEXT:  Remote context: PL/pgSQL function "test_error1" line 2 at SQL statement
+create function test_error2() returns int4
+as $$
+    cluster 'testcluster';
+    run on 0;
+    select err;
+$$ language plproxy;
+select * from test_error2();
+NOTICE:  PL/Proxy: dropping stale conn
+ERROR:  public.test_error2(0): REMOTE ERROR: column "err" does not exist
+LINE 1: select * from test_error2();
+               ^

sql/plproxy_test.sql

@@ -231,3 +231,29 @@ as $$
 $$ language plproxy;
 select * from test_simple(0);
 
+-- test error passing
+\c test_part
+create function test_error1() returns int4
+as $$
+begin
+    select line2err;
+    return 0;
+end;
+$$ language plpgsql;
+\c regression
+create function test_error1() returns int4
+as $$
+    cluster 'testcluster';
+    run on 0;
+$$ language plproxy;
+select * from test_error1();
+
+create function test_error2() returns int4
+as $$
+    cluster 'testcluster';
+    run on 0;
+    select err;
+$$ language plproxy;
+select * from test_error2();
+
+

src/execute.c

@@ -258,6 +258,13 @@ check_old_conn(ProxyFunction *func, ProxyConnection *conn, struct timeval * now)
 	return true;
 }
 
+static void
+handle_notice(void *arg, const PGresult *res)
+{
+	ProxyCluster *cluster = arg;
+	plproxy_remote_error(cluster->cur_func, res, false);
+}
+
 /* check existing conn status or launch new conn */
 static void
 prepare_conn(ProxyFunction *func, ProxyConnection *conn)
@@ -301,6 +308,9 @@ prepare_conn(ProxyFunction *func, ProxyConnection *conn)
 
 	if (PQstatus(conn->db) == CONNECTION_BAD)
 		conn_error(func, conn, "PQconnectStart");
+
+	/* override default notice handler */
+	PQsetNoticeReceiver(conn->db, handle_notice, func->cur_cluster);
 }
 
 /*
@@ -329,15 +339,29 @@ another_result(ProxyFunction *func, ProxyConnection *conn)
 	{
 		case PGRES_TUPLES_OK:
 			if (conn->res)
+			{
+				PQclear(res);
 				conn_error(func, conn, "double result?");
+			}
 			conn->res = res;
 			break;
 		case PGRES_COMMAND_OK:
 			PQclear(res);
 			break;
+		case PGRES_FATAL_ERROR:
+			if (conn->res)
+				PQclear(conn->res);
+			conn->res = res;
+
+			plproxy_remote_error(func, res, true);
+			break;
 		default:
-			PQclear(res);
-			conn_error(func, conn, "remote error");
+			if (conn->res)
+				PQclear(conn->res);
+			conn->res = res;
+
+			plproxy_error(func, "Unexpected result type: %s", PQresStatus(PQresultStatus(res)));
+			break;
 	}
 	return true;
 }
@@ -993,6 +1017,7 @@ plproxy_exec(ProxyFunction *func, FunctionCallInfo fcinfo)
 	PG_TRY();
 	{
 		func->cur_cluster->busy = true;
+		func->cur_cluster->cur_func = func;
 
 		/* clean old results */
 		plproxy_clean_results(func->cur_cluster);
@@ -1013,6 +1038,10 @@ plproxy_exec(ProxyFunction *func, FunctionCallInfo fcinfo)
 
 		if (geterrcode() == ERRCODE_QUERY_CANCELED)
 			remote_cancel(func);
+
+		/* plproxy_remote_error() cannot clean itself, do it here */
+		plproxy_clean_results(func->cur_cluster);
+
 		PG_RE_THROW();
 	}
 	PG_END_TRY();

src/main.c

@@ -79,6 +79,41 @@ plproxy_error(ProxyFunction *func, const char *fmt,...)
 		 func->name, func->arg_count, msg);
 }
 
+/*
+ * Pass remote error/notice/warning through.
+ */
+void
+plproxy_remote_error(ProxyFunction *func, const PGresult *res, bool iserr)
+{
+	const char *ss = PQresultErrorField(res, PG_DIAG_SQLSTATE);
+	const char *sev = PQresultErrorField(res, PG_DIAG_SEVERITY);
+	const char *msg = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
+	const char *det = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
+	const char *hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
+	const char *spos = PQresultErrorField(res, PG_DIAG_STATEMENT_POSITION);
+	const char *ipos = PQresultErrorField(res, PG_DIAG_INTERNAL_POSITION);
+	const char *iquery = PQresultErrorField(res, PG_DIAG_INTERNAL_QUERY);
+	const char *ctx = PQresultErrorField(res, PG_DIAG_CONTEXT);
+	int elevel;
+
+	if (iserr)
+		/* must ignore remote level, as it may be FATAL/PANIC */
+		elevel = ERROR;
+	else
+		/* cannot look at sev here, as it may be localized */
+		elevel = !strncmp(ss, "00", 2) ? NOTICE : WARNING;
+
+	ereport(elevel, (
+		errcode(MAKE_SQLSTATE(ss[0], ss[1], ss[2], ss[3], ss[4])),
+		errmsg("%s(%d): REMOTE %s: %s", func->name, func->arg_count, sev, msg),
+		det ? errdetail("Remote detail: %s", det) : 0,
+		hint ? errhint("Remote hint: %s", det) : 0,
+		spos ? errposition(atoi(spos)) : 0,
+		ipos ? internalerrposition(atoi(ipos)) : 0,
+		iquery ? internalerrquery(iquery) : 0,
+		ctx ? errcontext("Remote context: %s", ctx) : 0));
+}
+
 /*
  * Library load-time initialization.
  * Do the initialization when SPI is active to simplify the code.

src/plproxy.h

@@ -75,7 +75,6 @@
 #define PG_DETOAST_DATUM_PACKED(x) PG_DETOAST_DATUM(x)
 #endif
 
-
 /*
  * Determine if this argument is to SPLIT
  */
@@ -186,6 +185,9 @@ typedef struct ProxyCluster
 	 */
 	ItemPointerData		clusterTupleId;
 	ItemPointerData		umTupleId;
+
+	/* notice processing: provide info about currently executing function */
+	struct ProxyFunction	*cur_func;
 } ProxyCluster;
 
 /*
@@ -328,6 +330,7 @@ typedef struct ProxyFunction
 /* main.c */
 Datum		plproxy_call_handler(PG_FUNCTION_ARGS);
 void		plproxy_error(ProxyFunction *func, const char *fmt,...);
+void		plproxy_remote_error(ProxyFunction *func, const PGresult *res, bool iserr);
 
 /* function.c */
 void		plproxy_function_cache_init(void);