Support remote functions

Author: knizhnik

multimaster.c

@@ -68,6 +68,7 @@
 #include "parser/analyze.h"
 #include "parser/parse_relation.h"
 #include "parser/parse_type.h"
+#include "parser/parse_func.h"
 #include "catalog/pg_class.h"
 #include "catalog/pg_type.h"
 #include "tcop/pquery.h"
@@ -158,6 +159,7 @@ static void	  MtmInitializeSequence(int64* start, int64* step);
 static void*  MtmCreateSavepointContext(void);
 static void	  MtmRestoreSavepointContext(void* ctx);
 static void	  MtmReleaseSavepointContext(void* ctx);
+static void   MtmSetRemoteFunction(char const* list, void* extra);
 
 static void MtmCheckClusterLock(void);
 static void MtmCheckSlots(void);
@@ -184,6 +186,7 @@ MtmConnectionInfo* MtmConnections;
 
 HTAB* MtmXid2State;
 HTAB* MtmGid2State;
+static HTAB* MtmRemoteFunctions;
 static HTAB* MtmLocalTables;
 
 static bool MtmIsRecoverySession;
@@ -258,6 +261,7 @@ bool  MtmMajorNode;
 TransactionId  MtmUtilityProcessedInXid;
 
 static char* MtmConnStrs;
+static char* MtmRemoteFunctionsList;
 static char* MtmClusterName;
 static int	 MtmQueueSize;
 static int	 MtmWorkers;
@@ -2229,7 +2233,7 @@ MtmCreateLocalTableMap(void)
 		"MtmLocalTables",
 		MULTIMASTER_MAX_LOCAL_TABLES, MULTIMASTER_MAX_LOCAL_TABLES,
 		&info,
-		HASH_ELEM
+		HASH_ELEM | HASH_BLOBS
 	);
 	return htab;
 }
@@ -2423,6 +2427,48 @@ MtmShmemStartup(void)
 	MtmInitialize();
 }
 
+static void MtmSetRemoteFunction(char const* list, void* extra)
+{
+	if (MtmRemoteFunctions) {
+		hash_destroy(MtmRemoteFunctions);
+		MtmRemoteFunctions = NULL;
+	}
+}
+
+static void MtmInitializeRemoteFunctionsMap()
+{
+	HASHCTL info;
+	char* p, *q;
+	int n_funcs = 1;
+	FuncCandidateList clist;
+
+	for (p = MtmRemoteFunctionsList; (q = strchr(p, ',')) != NULL; p = q + 1, n_funcs++);
+
+	Assert(MtmRemoteFunctions == NULL);
+
+	memset(&info, 0, sizeof(info));
+	info.entrysize = info.keysize = sizeof(Oid);
+	info.hcxt = TopMemoryContext;
+	MtmRemoteFunctions = hash_create("MtmRemoteFunctions", n_funcs, &info,
+									 HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+
+	p = pstrdup(MtmRemoteFunctionsList);
+	do {
+		q = strchr(p, ',');
+		if (q != NULL) {
+			*q++ = '\0';
+		}
+		clist = FuncnameGetCandidates(stringToQualifiedNameList(p), -1, NIL, false, false, true);
+		if (clist == NULL) {
+			MTM_ELOG(ERROR, "Failed to lookup function %s", p);
+		} else if (clist->next != NULL) {
+			MTM_ELOG(ERROR, "Ambigious function %s", p);
+		}
+		hash_search(MtmRemoteFunctions, &clist->oid, HASH_ENTER, NULL);
+		p = q;
+	} while (p != NULL);
+}
+
 /*
  * Parse node connection string.
  * This function is called at cluster startup and while adding new cluster node
@@ -3052,6 +3098,19 @@ _PG_init(void)
 		NULL		 /* GucShowHook show_hook */
 	);
 
+	DefineCustomStringVariable(
+		"multimaster.remote_functions",
+		"List of fnuction names which should be executed remotely at all multimaster nodes instead of executing them at master and replicating result of their work",
+		NULL,
+		&MtmRemoteFunctionsList,
+		"lo_create,lo_unlink",
+		PGC_USERSET, /* context */
+		0,			 /* flags */
+		NULL,		 /* GucStringCheckHook check_hook */
+		MtmSetRemoteFunction,		 /* GucStringAssignHook assign_hook */
+		NULL		 /* GucShowHook show_hook */
+	);
+
 	DefineCustomStringVariable(
 		"multimaster.cluster_name",
 		"Name of the cluster",
@@ -3541,7 +3600,7 @@ lsn_t MtmGetFlushPosition(int nodeId)
  * Keep track of progress of WAL writer.
  * We need to notify WAL senders at other nodes which logical records
  * are flushed to the disk and so can survive failure. In asynchronous commit mode
- * WAL is flushed by WAL writer. Current flish position can be obtained by GetFlushRecPtr().
+ * WAL is flushed by WAL writer. Current flush position can be obtained by GetFlushRecPtr().
  * So on applying new logical record we insert it in the MtmLsnMapping and compare
  * their poistions in local WAL log with current flush position.
  * The records which are flushed to the disk by WAL writer are removed from the list
@@ -4656,7 +4715,7 @@ char* MtmGucSerialize(void)
 		appendStringInfoString(serialized_gucs, " TO ");
 
 		/* quite a crutch */
-		if (strstr(cur_entry->key, "_mem") != NULL || *(cur_entry->value) == '\0')
+		if (strstr(cur_entry->key, "_mem") != NULL || *(cur_entry->value) == '\0' || strchr(cur_entry->value, ',') != NULL)
 		{
 			appendStringInfoString(serialized_gucs, "'");
 			appendStringInfoString(serialized_gucs, cur_entry->value);
@@ -4686,10 +4745,7 @@ static void MtmProcessDDLCommand(char const* queryString, bool transactional)
 	if (transactional)
 	{
 		char *gucCtx = MtmGucSerialize();
-		if (*gucCtx)
-			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s %s", gucCtx, queryString);
-		else
-			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
+		queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s %s", gucCtx, queryString);
 
 		/* Transactional DDL */
 		MTM_LOG3("Sending DDL: %s", queryString);
@@ -5058,29 +5114,28 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 static void
 MtmExecutorStart(QueryDesc *queryDesc, int eflags)
 {
-	bool ddl_generating_call = false;
-	ListCell   *tlist;
-
-	foreach(tlist, queryDesc->plannedstmt->planTree->targetlist)
+	if (!MtmTx.isReplicated && ActivePortal)
 	{
-		TargetEntry *tle = (TargetEntry *) lfirst(tlist);
+		ListCell   *tlist;
 
-		if (tle->resname && strcmp(tle->resname, "lo_create") == 0)
+		if (!MtmRemoteFunctions)
 		{
-			ddl_generating_call = true;
-			break;
+			MtmInitializeRemoteFunctionsMap();
 		}
 
-		if (tle->resname && strcmp(tle->resname, "lo_unlink") == 0)
+		foreach(tlist, queryDesc->plannedstmt->planTree->targetlist)
 		{
-			ddl_generating_call = true;
-			break;
+			TargetEntry *tle = (TargetEntry *) lfirst(tlist);
+			if (tle->expr && IsA(tle->expr, FuncExpr))
+			{
+				if (hash_search(MtmRemoteFunctions, &((FuncExpr*)tle->expr)->funcid, HASH_FIND, NULL))
+				{
+					MtmProcessDDLCommand(ActivePortal->sourceText, true);
+					break;
+				}
+			}
 		}
 	}
-
-	if (ddl_generating_call && !MtmTx.isReplicated)
-		MtmProcessDDLCommand(ActivePortal->sourceText, true);
-
 	if (PreviousExecutorStartHook != NULL)
 		PreviousExecutorStartHook(queryDesc, eflags);
 	else

multimaster.h

@@ -51,22 +51,22 @@
 		fprintf(stderr, MTM_TAG "%s, %lld, %s, %d\n", tx->gid, (long long)MtmGetSystemTime(), event, MyProcPid)
 // #endif
 
-#define MULTIMASTER_NAME                "multimaster"
-#define MULTIMASTER_SCHEMA_NAME         "mtm"
-#define MULTIMASTER_LOCAL_TABLES_TABLE  "local_tables"
-#define MULTIMASTER_SLOT_PATTERN        "mtm_slot_%d"
-#define MULTIMASTER_MIN_PROTO_VERSION   1
-#define MULTIMASTER_MAX_PROTO_VERSION   1
-#define MULTIMASTER_MAX_GID_SIZE        32
-#define MULTIMASTER_MAX_SLOT_NAME_SIZE  16
-#define MULTIMASTER_MAX_CONN_STR_SIZE   128
-#define MULTIMASTER_MAX_HOST_NAME_SIZE  64
-#define MULTIMASTER_MAX_LOCAL_TABLES    256
-#define MULTIMASTER_MAX_CTL_STR_SIZE    256
-#define MULTIMASTER_LOCK_BUF_INIT_SIZE  4096
-#define MULTIMASTER_BROADCAST_SERVICE   "mtm_broadcast"
-#define MULTIMASTER_ADMIN               "mtm_admin"
-#define MULTIMASTER_PRECOMMITTED        "precommitted"
+#define MULTIMASTER_NAME                 "multimaster"
+#define MULTIMASTER_SCHEMA_NAME          "mtm"
+#define MULTIMASTER_LOCAL_TABLES_TABLE   "local_tables"
+#define MULTIMASTER_SLOT_PATTERN         "mtm_slot_%d"
+#define MULTIMASTER_MIN_PROTO_VERSION    1
+#define MULTIMASTER_MAX_PROTO_VERSION    1
+#define MULTIMASTER_MAX_GID_SIZE         32
+#define MULTIMASTER_MAX_SLOT_NAME_SIZE   16
+#define MULTIMASTER_MAX_CONN_STR_SIZE    128
+#define MULTIMASTER_MAX_HOST_NAME_SIZE   64
+#define MULTIMASTER_MAX_LOCAL_TABLES     256
+#define MULTIMASTER_MAX_CTL_STR_SIZE     256
+#define MULTIMASTER_LOCK_BUF_INIT_SIZE   4096
+#define MULTIMASTER_BROADCAST_SERVICE    "mtm_broadcast"
+#define MULTIMASTER_ADMIN                "mtm_admin"
+#define MULTIMASTER_PRECOMMITTED         "precommitted"
 
 #define MULTIMASTER_DEFAULT_ARBITER_PORT 5433
 

pglogical_receiver.c

@@ -485,8 +485,8 @@ pglogical_receiver_main(Datum main_arg)
 					{
 						int64 now = feGetCurrentTimestamp();
 
-						/* Leave is feedback is not sent properly */
 						MtmUpdateLsnMapping(nodeId, walEnd);
+						/* Leave if feedback is not sent properly */
 						if (!sendFeedback(conn, now, nodeId)) {
 							goto OnError;
 						}
@@ -635,7 +635,6 @@ pglogical_receiver_main(Datum main_arg)
 				{
 					int64 now = feGetCurrentTimestamp();
 
-					/* Leave is feedback is not sent properly */
 					MtmUpdateLsnMapping(nodeId, INVALID_LSN);
 					sendFeedback(conn, now, nodeId);
 				}
@@ -731,4 +730,3 @@ void MtmStartReceivers(void)
 		}
 	}
 }
-

pglogical_relid_map.c

@@ -20,29 +20,19 @@ static void
 pglogical_relid_map_init(void)
 {
 	HASHCTL	ctl;
-	int hash_flags = HASH_ELEM;
-
 	Assert(relid_map == NULL);
 
 	MemSet(&ctl, 0, sizeof(ctl));
 	ctl.keysize = sizeof(Oid);
 	ctl.entrysize = sizeof(PGLRelidMapEntry);
-
-#if PG_VERSION_NUM >= 90500
-	hash_flags |= HASH_BLOBS;
-#else
-	ctl.hash = tag_hash;
-	hash_flags |= HASH_FUNCTION;
-#endif
-
-	relid_map = hash_create("pglogical_relid_map", PGL_INIT_RELID_MAP_SIZE, &ctl, hash_flags);
+	relid_map = hash_create("pglogical_relid_map", PGL_INIT_RELID_MAP_SIZE, &ctl, HASH_ELEM | HASH_BLOBS);
 
 	Assert(relid_map != NULL);
 }
 
 Oid pglogical_relid_map_get(Oid relid)
 {
-	if (relid_map != NULL) { 
+	if (relid_map != NULL) {
 		PGLRelidMapEntry* entry = (PGLRelidMapEntry*)hash_search(relid_map, &relid, HASH_FIND, NULL);
 		return entry ? entry->local_relid : InvalidOid;
 	}
@@ -51,23 +41,23 @@ Oid pglogical_relid_map_get(Oid relid)
 
 bool pglogical_relid_map_put(Oid remote_relid, Oid local_relid)
 {
-	bool found;	
+	bool found;
     PGLRelidMapEntry* entry;
-    if (relid_map == NULL) { 
+    if (relid_map == NULL) {
         pglogical_relid_map_init();
     }
     entry = hash_search(relid_map, &remote_relid, HASH_ENTER, &found);
   	if (found) {
 	    entry->local_relid = local_relid;
-		return false;	    
+		return false;
     }
     entry->local_relid = local_relid;
 	return true;
 }
 
 void pglogical_relid_map_reset(void)
 {
-	if (relid_map != NULL) { 
+	if (relid_map != NULL) {
 		hash_destroy(relid_map);
 		relid_map = NULL;
 	}