Improve DDL handling

Author: knizhnik

contrib/mmts/multimaster.c

@@ -64,6 +64,7 @@
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_constraint_fn.h"
+#include "catalog/pg_proc.h"
 #include "pglogical_output/hooks.h"
 #include "parser/analyze.h"
 #include "parser/parse_relation.h"
@@ -255,8 +256,6 @@ bool  MtmUseDtm;
 bool  MtmPreserveCommitOrder;
 bool  MtmVolksWagenMode; /* Pretend to be normal postgres. This means skip some NOTICE's and use local sequences */
 
-TransactionId  MtmUtilityProcessedInXid;
-
 static char* MtmConnStrs;
 static char* MtmRemoteFunctionsList;
 static char* MtmClusterName;
@@ -275,6 +274,7 @@ static bool	 MtmClusterLocked;
 static bool	 MtmInsideTransaction;
 static bool  MtmReferee;
 static bool  MtmMonotonicSequences;
+static void const* MtmDDLStatement;
 
 static ExecutorStart_hook_type PreviousExecutorStartHook;
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
@@ -923,6 +923,7 @@ MtmResetTransaction()
 	x->csn = INVALID_CSN;
 	x->status = TRANSACTION_STATUS_UNKNOWN;
 	x->gid[0] = '\0';
+	MtmDDLStatement = NULL;
 }
 
 #if 0
@@ -986,6 +987,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 			MtmCheckClusterLock();
 		}
 		MtmInsideTransaction = true;
+		MtmDDLStatement = NULL;
 		Mtm->nRunningTransactions += 1;
 
 		x->snapshot = MtmAssignCSN();
@@ -3447,7 +3449,7 @@ _PG_init(void)
 		&MtmRemoteFunctionsList,
 		"lo_create,lo_unlink",
 		PGC_USERSET, /* context */
-		0,			 /* flags */
+		GUC_LIST_INPUT | GUC_LIST_QUOTE, /* flags */
 		NULL,		 /* GucStringCheckHook check_hook */
 		MtmSetRemoteFunction,		 /* GucStringAssignHook assign_hook */
 		NULL		 /* GucShowHook show_hook */
@@ -4961,14 +4963,17 @@ static void MtmGucDiscard()
 	dlist_init(&MtmGucList);
 
 	hash_destroy(MtmGucHash);
-	MtmGucInit();
+	MtmGucHash = NULL;
 }
 
 static inline void MtmGucUpdate(const char *key, char *value)
 {
 	MtmGucEntry *hentry;
 	bool found;
 
+	if (!MtmGucHash)
+		MtmGucInit();
+
 	hentry = (MtmGucEntry*)hash_search(MtmGucHash, key, HASH_ENTER, &found);
 	if (found)
 	{
@@ -4984,6 +4989,9 @@ static inline void MtmGucRemove(const char *key)
 	MtmGucEntry *hentry;
 	bool found;
 
+	if (!MtmGucHash)
+		MtmGucInit();
+
 	hentry = (MtmGucEntry*)hash_search(MtmGucHash, key, HASH_FIND, &found);
 	if (found)
 	{
@@ -5042,23 +5050,19 @@ char* MtmGucSerialize(void)
 
 	serialized_gucs = makeStringInfo();
 
-	/*
-	 * Crutch for scheduler. It sets search_path through SetConfigOption()
-	 * so our callback do not react on that.
-	 */
-	search_path = GetConfigOption("search_path", false, true);
-	appendStringInfo(serialized_gucs, "SET search_path TO %s; ", search_path);
-
 	dlist_foreach(iter, &MtmGucList)
 	{
 		MtmGucEntry *cur_entry = dlist_container(MtmGucEntry, list_node, iter.cur);
 
+		if (strcmp(cur_entry->key, "search_path") == 0)
+			continue;
+
 		appendStringInfoString(serialized_gucs, "SET ");
 		appendStringInfoString(serialized_gucs, cur_entry->key);
 		appendStringInfoString(serialized_gucs, " TO ");
 
 		/* quite a crutch */
-		if (strstr(cur_entry->key, "_mem") != NULL || *(cur_entry->value) == '\0' || strchr(cur_entry->value, ',') != NULL)
+		if (strstr(cur_entry->key, "_mem") != NULL || *(cur_entry->value) == '\0')
 		{
 			appendStringInfoString(serialized_gucs, "'");
 			appendStringInfoString(serialized_gucs, cur_entry->value);
@@ -5071,6 +5075,13 @@ char* MtmGucSerialize(void)
 		appendStringInfoString(serialized_gucs, "; ");
 	}
 
+	/*
+	 * Crutch for scheduler. It sets search_path through SetConfigOption()
+	 * so our callback do not react on that.
+	 */
+	search_path = GetConfigOption("search_path", false, true);
+	appendStringInfo(serialized_gucs, "SET search_path TO %s; ", search_path);
+
 	return serialized_gucs->data;
 }
 
@@ -5363,6 +5374,11 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 						 return;
 					}
 				}
+				else if (stmt->removeType == OBJECT_FUNCTION && MtmTx.isReplicated)
+				{
+					/* Make it possible to drop functions which were not replicated */
+					stmt->missing_ok = true;
+				}
 			}
 			break;
 
@@ -5395,16 +5411,14 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			break;
 	}
 
-	if (!skipCommand && !MtmTx.isReplicated && (context == PROCESS_UTILITY_TOPLEVEL || MtmUtilityProcessedInXid != GetCurrentTransactionId()))
+	if (!skipCommand && !MtmTx.isReplicated && !MtmDDLStatement)
 	{
-		MtmUtilityProcessedInXid = GetCurrentTransactionId();
-		if (context == PROCESS_UTILITY_TOPLEVEL || !ActivePortal) {
-			MtmProcessDDLCommand(queryString, true);
-		} else {
-			MtmProcessDDLCommand(ActivePortal->sourceText, true);
-		}
+		MTM_LOG3("Process DDL statement '%s', MtmTx.isReplicated=%d, MtmIsLogicalReceiver=%d", queryString, MtmTx.isReplicated, MtmIsLogicalReceiver);
+		MtmProcessDDLCommand(queryString, true);
 		executed = true;
+		MtmDDLStatement = queryString;
 	}
+	else MTM_LOG3("Skip utility statement '%s': skip=%d, insideDDL=%d", queryString, skipCommand, MtmDDLStatement != NULL);
 
 	if (PreviousProcessUtilityHook != NULL)
 	{
@@ -5423,16 +5437,17 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 #endif
 	if (MyXactAccessedTempRel)
 	{
-		MTM_LOG1("Xact accessed temp table, stopping replication");
+		MTM_LOG1("Xact accessed temp table, stopping replication of statement '%s'", queryString);
 		MtmTx.isDistributed = false; /* Skip */
 		MtmTx.snapshot = INVALID_CSN;
 	}
 
 	if (executed)
 	{
 		MtmFinishDDLCommand();
+		MtmDDLStatement = NULL;
 	}
-	if (nodeTag(parsetree) == T_CreateStmt)
+	if (IsA(parsetree, CreateStmt))
 	{
 		CreateStmt* create = (CreateStmt*)parsetree;
 		Oid relid = RangeVarGetRelid(create->relation, NoLock, true);
@@ -5449,15 +5464,12 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			}
 		}
 	}
-	if (context == PROCESS_UTILITY_TOPLEVEL) {
-		MtmUtilityProcessedInXid = InvalidTransactionId;
-	}
 }
 
 static void
 MtmExecutorStart(QueryDesc *queryDesc, int eflags)
 {
-	if (!MtmTx.isReplicated && ActivePortal)
+	if (!MtmTx.isReplicated && !MtmDDLStatement)
 	{
 		ListCell   *tlist;
 
@@ -5471,11 +5483,32 @@ MtmExecutorStart(QueryDesc *queryDesc, int eflags)
 			TargetEntry *tle = (TargetEntry *) lfirst(tlist);
 			if (tle->expr && IsA(tle->expr, FuncExpr))
 			{
-				if (hash_search(MtmRemoteFunctions, &((FuncExpr*)tle->expr)->funcid, HASH_FIND, NULL))
+				Oid func_oid = ((FuncExpr*)tle->expr)->funcid;
+				if (!hash_search(MtmRemoteFunctions, &func_oid, HASH_FIND, NULL))
 				{
-					MtmProcessDDLCommand(ActivePortal->sourceText, true);
-					break;
+					Form_pg_proc funcform;
+					bool is_sec_def;
+					HeapTuple func_tuple = SearchSysCache1(PROCOID, ObjectIdGetDatum(func_oid));
+					if (!HeapTupleIsValid(func_tuple))
+						elog(ERROR, "cache lookup failed for function %u", func_oid);
+					funcform = (Form_pg_proc) GETSTRUCT(func_tuple);
+					is_sec_def = funcform->prosecdef;
+					ReleaseSysCache(func_tuple);
+					elog(LOG, "Function %s security defined=%d", tle->resname, is_sec_def);
+					if (!is_sec_def)
+					{
+						continue;
+					}
 				}
+				/*
+				 * Execute security defined functions or functions marked as remote at replicated nodes.
+				 * Them are executed as DDL statements.
+				 * All data modifications done inside this function are not replicated.
+				 * As a result generated content can vary at different nodes.
+				 */
+				MtmProcessDDLCommand(queryDesc->sourceText, true);
+				MtmDDLStatement = queryDesc;
+				break;
 			}
 		}
 	}
@@ -5524,6 +5557,12 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 	{
 		standard_ExecutorFinish(queryDesc);
 	}
+
+	if (MtmDDLStatement == queryDesc)
+	{
+		MtmFinishDDLCommand();
+		MtmDDLStatement = NULL;
+	}
 }
 
 static void MtmSeqNextvalHook(Oid seqid, int64 next)

src/backend/commands/functioncmds.c

@@ -1114,17 +1114,13 @@ CreateFunction(CreateFunctionStmt *stmt, const char *queryString)
  * Note: this is also used for aggregate deletion, since the OIDs of
  * both functions and aggregates point to pg_proc.
  */
+extern bool MyXactAccessedTempRel;
 void
 RemoveFunctionById(Oid funcOid)
 {
 	Relation	relation;
 	HeapTuple	tup;
 	bool		isagg;
-	Oid language_oid;
-	HeapTuple languageTuple;
-	Form_pg_language languageStruct;
-	Oid	languageValidator;
-	bool save_check_function_bodies;
 
 	/*
 	 * Delete the pg_proc tuple.
@@ -1137,24 +1133,6 @@ RemoveFunctionById(Oid funcOid)
 
 	isagg = ((Form_pg_proc) GETSTRUCT(tup))->proisagg;
 
-	/*
-	 * MTM-CRUTCH: We need to know wheteher our function had
-	 * accessed temp relation or not. So validate function body
-	 * again -- that will set MyXactAccessedTempRel.
-	 */
-	save_check_function_bodies = check_function_bodies;
-	check_function_bodies = false;
-	language_oid = ((Form_pg_proc) GETSTRUCT(tup))->prolang;
-	languageTuple = SearchSysCache1(LANGOID, language_oid);
-	languageStruct = (Form_pg_language) GETSTRUCT(languageTuple);
-	languageValidator = languageStruct->lanvalidator;
-	if OidIsValid(languageValidator) {
-		/* Language validator is optional feature of language */
-		OidFunctionCall1(languageValidator, ObjectIdGetDatum(funcOid));
-	}
-	ReleaseSysCache(languageTuple);
-	check_function_bodies = save_check_function_bodies;
-
 	simple_heap_delete(relation, &tup->t_self);
 
 	ReleaseSysCache(tup);