Fix definitions of SQL functions

Author: knizhnik

contrib/mmts/bgwpool.c

@@ -109,7 +109,14 @@ size_t BgwPoolGetQueueSize(BgwPool* pool)
 
 void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
 {
-    Assert(size+4 <= pool->size);
+    if (size+4 > pool->size) {
+		/* 
+		 * Size of work is larger than size of shared buffer: 
+		 * run it immediately
+		 */
+		pool->executor(0, work, size);
+		return;
+	}
 
     SpinLockAcquire(&pool->lock);
     while (true) { 

contrib/mmts/multimaster--1.0.sql

@@ -24,13 +24,13 @@ AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;
 
 
-CREATE TYPE mtm.node_state(id integer, disabled bool, disconnected bool, catchUp boolean, slotLag bigint, connStr text);
+CREATE TYPE mtm.node_state AS (id integer, disabled bool, disconnected bool, catchUp bool, slotLag bigint, connStr text);
 
 CREATE FUNCTION mtm.get_nodes_state() RETURNS SETOF mtm.node_state
 AS 'MODULE_PATHNAME','mtm_get_nodes_state'
 LANGUAGE C;
 
-CREATE TYPE mtm.cluster_state(status text, disabledNodeMask bigint, disconnectedNodeMask bigint, catchUpNodeMask bigint, nNodes integer, nActiveQueries integer, queueSize bigint);
+CREATE TYPE mtm.cluster_state AS (status text, disabledNodeMask bigint, disconnectedNodeMask bigint, catchUpNodeMask bigint, nNodes integer, nActiveQueries integer, queueSize bigint);
 
 CREATE FUNCTION mtm.get_cluster_state() RETURNS mtm.cluster_state 
 AS 'MODULE_PATHNAME','mtm_get_cluster_state'

contrib/mmts/multimaster.c

@@ -523,7 +523,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 		if (x->isDistributed && dtm->status != MTM_ONLINE) { 
 			/* reject all user's transactions at offline cluster */
 			MtmUnlock();			
-			elog(ERROR, "Multimaster node is not online");
+			elog(ERROR, "Multimaster node is not online: current status %s", MtmNodeStatusMnem[dtm->status]);
 		}
 		x->containsDML = false;
 		x->isPrepared = false;
@@ -1159,7 +1159,7 @@ void MtmReceiverStarted(int nodeId)
 		BIT_SET(dtm->pglogicalNodeMask, nodeId-1);
 		if (++dtm->nReceivers == dtm->nNodes-1) {
 			Assert(dtm->status == MTM_CONNECTED);
-			MtmSwitchClusterMode(MTM_OFFLINE);
+			MtmSwitchClusterMode(MTM_ONLINE);
 		}
      }
 	SpinLockRelease(&dtm->spinlock);	
@@ -1293,6 +1293,7 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	MtmGetNodeStateCtx* usrfctx;
 	MemoryContext oldcontext;
 	char* p;
+	int64 lag;
     bool is_first_call = SRF_IS_FIRSTCALL();
 
     if (is_first_call) { 
@@ -1315,7 +1316,9 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	usrfctx->values[1] = BoolGetDatum(BIT_CHECK(dtm->disabledNodeMask, usrfctx->nodeId-1));
 	usrfctx->values[2] = BoolGetDatum(BIT_CHECK(dtm->connectivityMask, usrfctx->nodeId-1));
 	usrfctx->values[3] = BoolGetDatum(BIT_CHECK(dtm->nodeLockerMask, usrfctx->nodeId-1));
-	usrfctx->values[4] = Int64GetDatum(MtmGetSlotLag(usrfctx->nodeId));
+	lag = MtmGetSlotLag(usrfctx->nodeId);
+	usrfctx->values[4] = Int64GetDatum(lag);
+	usrfctx->nulls[4] = lag < 0;
 	p = strchr(usrfctx->connStrPtr, ',');
 	if (p != NULL) { 
 		*p++ = '\0';
@@ -1333,7 +1336,6 @@ mtm_get_cluster_state(PG_FUNCTION_ARGS)
 	TupleDesc desc;
     Datum     values[7];
     bool      nulls[7] = {false};
-
 	get_call_result_type(fcinfo, NULL, &desc);
 
 	values[0] = CStringGetTextDatum(MtmNodeStatusMnem[dtm->status]);

contrib/mmts/pglogical_receiver.c

@@ -331,8 +331,8 @@ pglogical_receiver_main(Datum main_arg)
 		if (rc & WL_POSTMASTER_DEATH)
 			proc_exit(1);
 
-		if (ds->status != MTM_ONLINE && (ds->status != MTM_RECOVERY || ds->recoverySlot != args->remote_node)) {
-			ereport(LOG, (errmsg("%s: terminating WAL receiver because node is switched to %s mode", worker_proc, MtmNodeStatusMnem[ds->status])));
+		if (ds->status == MTM_OFFLINE || (ds->status == MTM_RECOVERY && ds->recoverySlot != args->remote_node)) {
+			ereport(LOG, (errmsg("%s: terminating WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[ds->status])));
 			proc_exit(0);
 		}