Porting multimaster to postgres 14

Mikhail Rutman · Mikhail Rutman · commit 1b59ad0ace80 · 2022-04-26T15:14:33.000+07:00
This commit ports multimaster to Postgres 14:
- removed tablespace-setup from mmts/Makefile
  (according to commit 6c788d9f6aadb41d76a72d56149268371a7895ee);
- add readOnlyTree flag to MtmProcessUtility()
  (according to commit 7c337b6b527b7052e6a751f966d5734c56f668b5);
- explicitly specify the hash function HASH_STRING
  (according to commit b3817f5f774663d55931dd4fab9c5a94a15ae7ab);
- refactor the use of EState
  (according to commit 1375422c7826a2bf387be29895e961614f69de4b);
- add 'include_out_arguments' to the function FuncnameGetCandidates()
  (caccording to commit e56bce5d43789cce95d099554ae9593ada92b3b7);
- use MyProc-&gt;xmin instead of MyPgXact-&gt;xmin in dmq.c
  (according to commit 73487a60fc1063ba4b5178b69aee4ee210c182c4);
- remove EE pooler support;
- use SearchNamedReplicationSlot() instead of ReplicationSlotAcquire()
diff --git a/Cluster.pm b/Cluster.pm
@@ -141,6 +141,7 @@ sub init
 	{
 		$node->init(allows_streaming => 'logical');
 		$node->append_conf('postgresql.conf', qq{
+			enable_self_join_removal = off
 			max_connections = 50
 			log_line_prefix = '%m [%p] [xid%x] %i '
 			log_statement = all
diff --git a/Makefile b/Makefile
@@ -69,7 +69,6 @@ all: multimaster.so
 
 submake-regress:
 	$(MAKE) -C $(top_builddir)/src/test/regress all
-	$(MAKE) -C $(top_builddir)/src/test/regress tablespace-setup
 
 # all .pl tests should pass now, but let's see what the buildfarm says
 # ifndef MTM_ALL
diff --git a/expected/regression_ee.diff b/expected/regression_ee.diff
diff --git a/src/ddl.c b/src/ddl.c
@@ -119,19 +119,21 @@ static void MtmSeqNextvalHook(Oid seqid, int64 next);
 static void MtmExecutorStart(QueryDesc *queryDesc, int eflags);
 static void MtmExecutorFinish(QueryDesc *queryDesc);
 
-static void MtmProcessUtility(PlannedStmt *pstmt, const char *queryString,
+static void MtmProcessUtility(PlannedStmt *pstmt, const char *queryString, bool readOnlyTree,
 							  ProcessUtilityContext context, ParamListInfo params,
 							  QueryEnvironment *queryEnv, DestReceiver *dest,
 							  QueryCompletion *qc);
 
 static void MtmProcessUtilityReceiver(PlannedStmt *pstmt,
 						  const char *queryString,
+						  bool readOnlyTree,
 						  ProcessUtilityContext context, ParamListInfo params,
 						  QueryEnvironment *queryEnv, DestReceiver *dest,
 						  QueryCompletion *qc);
 
 static void MtmProcessUtilitySender(PlannedStmt *pstmt,
 						const char *queryString,
+						bool readOnlyTree,
 						ProcessUtilityContext context, ParamListInfo params,
 						QueryEnvironment *queryEnv, DestReceiver *dest,
 						QueryCompletion *qc);
@@ -359,7 +361,7 @@ MtmGucInit(void)
 	MtmGucHash = hash_create("MtmGucHash",
 							 MTM_GUC_HASHSIZE,
 							 &hash_ctl,
-							 HASH_ELEM | HASH_CONTEXT);
+							 HASH_ELEM | HASH_CONTEXT | HASH_STRINGS);
 
 	/*
 	 * If current role is not equal to MtmDatabaseUser, than set it before any
@@ -661,7 +663,7 @@ MtmFinishDDLCommand()
 
 
 static void
-MtmProcessUtility(PlannedStmt *pstmt, const char *queryString,
+MtmProcessUtility(PlannedStmt *pstmt, const char *queryString, bool readOnlyTree,
 				  ProcessUtilityContext context, ParamListInfo params,
 				  QueryEnvironment *queryEnv, DestReceiver *dest,
 				  QueryCompletion *qc)
@@ -676,13 +678,13 @@ MtmProcessUtility(PlannedStmt *pstmt, const char *queryString,
 	{
 		if (PreviousProcessUtilityHook != NULL)
 		{
-			PreviousProcessUtilityHook(pstmt, queryString,
+			PreviousProcessUtilityHook(pstmt, queryString, readOnlyTree,
 									   context, params, queryEnv,
 									   dest, qc);
 		}
 		else
 		{
-			standard_ProcessUtility(pstmt, queryString,
+			standard_ProcessUtility(pstmt, queryString, readOnlyTree,
 									context, params, queryEnv,
 									dest, qc);
 		}
@@ -691,12 +693,12 @@ MtmProcessUtility(PlannedStmt *pstmt, const char *queryString,
 
 	if (MtmIsLogicalReceiver)
 	{
-		MtmProcessUtilityReceiver(pstmt, queryString, context, params,
+		MtmProcessUtilityReceiver(pstmt, queryString, context, readOnlyTree, params,
 								  queryEnv, dest, qc);
 	}
 	else
 	{
-		MtmProcessUtilitySender(pstmt, queryString, context, params,
+		MtmProcessUtilitySender(pstmt, queryString, readOnlyTree, context, params,
 								queryEnv, dest, qc);
 	}
 
@@ -718,7 +720,7 @@ MtmProcessUtility(PlannedStmt *pstmt, const char *queryString,
  * receiver (e.g calling DDL from trigger) this hook does nothing.
  */
 static void
-MtmProcessUtilityReceiver(PlannedStmt *pstmt, const char *queryString,
+MtmProcessUtilityReceiver(PlannedStmt *pstmt, const char *queryString, bool readOnlyTree,
 						  ProcessUtilityContext context, ParamListInfo params,
 						  QueryEnvironment *queryEnv, DestReceiver *dest,
 						  QueryCompletion *qc)
@@ -839,22 +841,18 @@ MtmProcessUtilityReceiver(PlannedStmt *pstmt, const char *queryString,
 	}
 
 	if (PreviousProcessUtilityHook != NULL)
-	{
-		PreviousProcessUtilityHook(pstmt, queryString,
+		PreviousProcessUtilityHook(pstmt, queryString, readOnlyTree,
 								   context, params, queryEnv,
 								   dest, qc);
-	}
 	else
-	{
-		standard_ProcessUtility(pstmt, queryString,
+		standard_ProcessUtility(pstmt, queryString, readOnlyTree,
 								context, params, queryEnv,
 								dest, qc);
-	}
 }
 
 
 static void
-MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString,
+MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString, bool readOnlyTree,
 						ProcessUtilityContext context, ParamListInfo params,
 						QueryEnvironment *queryEnv, DestReceiver *dest,
 						QueryCompletion *qc)
@@ -1186,17 +1184,13 @@ MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString,
 				stmt_string, skipCommand, MtmDDLStatement != NULL);
 
 	if (PreviousProcessUtilityHook != NULL)
-	{
-		PreviousProcessUtilityHook(pstmt, queryString,
+		PreviousProcessUtilityHook(pstmt, queryString, readOnlyTree,
 								   context, params, queryEnv,
 								   dest, qc);
-	}
 	else
-	{
-		standard_ProcessUtility(pstmt, queryString,
+		standard_ProcessUtility(pstmt, queryString, readOnlyTree,
 								context, params, queryEnv,
 								dest, qc);
-	}
 
 	/* Catch GUC assignment */
 	if (nodeTag(parsetree) == T_VariableSetStmt)
@@ -1311,11 +1305,12 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 		if (operation == CMD_INSERT || operation == CMD_UPDATE ||
 			operation == CMD_DELETE || pstmt->hasModifyingCTE)
 		{
-			int			i;
+			ListCell   *l;
 
-			for (i = 0; i < estate->es_num_result_relations; i++)
+			foreach(l, estate->es_opened_result_relations)
 			{
-				Relation	rel = estate->es_result_relations[i].ri_RelationDesc;
+				ResultRelInfo *resultRelInfo = lfirst(l);
+				Relation	rel = resultRelInfo->ri_RelationDesc;
 
 				/*
 				 * Don't run 3pc unless we modified at least one non-local table.
@@ -1709,7 +1704,7 @@ MtmInitializeRemoteFunctionsMap()
 		if (q != NULL)
 			*q++ = '\0';
 
-		clist = FuncnameGetCandidates(stringToQualifiedNameList(p), -1, NIL, false, false, true);
+		clist = FuncnameGetCandidates(stringToQualifiedNameList(p), -1, NIL, false, false, true, true);
 		if (clist == NULL)
 			mtm_log(DEBUG1, "Can't resolve function '%s', postponing that", p);
 		else
@@ -1724,7 +1719,7 @@ MtmInitializeRemoteFunctionsMap()
 		p = q;
 	} while (p != NULL);
 
-	clist = FuncnameGetCandidates(stringToQualifiedNameList("mtm.alter_sequences"), -1, NIL, false, false, true);
+	clist = FuncnameGetCandidates(stringToQualifiedNameList("mtm.alter_sequences"), -1, NIL, false, false, true, true);
 	if (clist != NULL)
 		hash_search(MtmRemoteFunctions, &clist->oid, HASH_ENTER, NULL);
 
diff --git a/src/dmq.c b/src/dmq.c
@@ -353,7 +353,7 @@ dmq_shmem_startup_hook(void)
 									  DMQ_MAX_SUBS_PER_BACKEND * MaxBackends,
 									  DMQ_MAX_SUBS_PER_BACKEND * MaxBackends,
 									  &hash_info,
-									  HASH_ELEM);
+									  HASH_ELEM | HASH_STRINGS);
 
 	LWLockRelease(AddinShmemInitLock);
 }
@@ -1443,7 +1443,7 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 		extra = dmq_receiver_start_hook(sender_name);
 
 	/* do not hold globalxmin. XXX: try to carefully release snaps */
-	MyPgXact->xmin = InvalidTransactionId;
+	MyProc->xmin = InvalidTransactionId;
 
 	for (;;)
 	{
diff --git a/src/global_tx.c b/src/global_tx.c
@@ -185,7 +185,7 @@ MtmGlobalTxShmemStartup(void)
 		gtx_shared->lock = &(GetNamedLWLockTranche("mtm-gtx-lock"))->lock;
 
 	gtx_shared->gid2gtx = ShmemInitHash("gid2gtx", 2*MaxConnections, 2*MaxConnections,
-							&info, HASH_ELEM);
+							&info, HASH_ELEM | HASH_STRINGS);
 
 	LWLockRelease(AddinShmemInitLock);
 }
diff --git a/src/include/compat.h b/src/include/compat.h
@@ -1,12 +1,7 @@
 #ifndef MTMCOMPAT_H
 #define MTMCOMPAT_H
 
-/* EE pooler gets rid of static variable */
-#ifdef PGPRO_EE
-#define FeBeWaitSetCompat() (MyProcPort->pqcomm_waitset)
-#else
 #define FeBeWaitSetCompat() (FeBeWaitSet)
-#endif
 
 #ifdef PGPRO_EE /* atx */
 #define BeginTransactionBlockCompat() (BeginTransactionBlock(false, NIL))
@@ -16,11 +11,6 @@
 #define UserAbortTransactionBlockCompat(chain) (UserAbortTransactionBlock(chain))
 #endif
 
-/* atx renames this for some reason */
-#ifdef PGPRO_EE
-#define on_commits_compat() (pg_on_commit_actions)
-#else
 #define on_commits_compat() (on_commits)
-#endif
 
 #endif							/* MTMCOMPAT_H */
diff --git a/src/pglogical_apply.c b/src/pglogical_apply.c
@@ -166,24 +166,27 @@ create_rel_estate(Relation rel)
 	EState	   *estate;
 	ResultRelInfo *resultRelInfo;
 	RangeTblEntry *rte;
+	List *rangeTable;
 
 	estate = CreateExecutorState();
 
 	resultRelInfo = makeNode(ResultRelInfo);
 	InitResultRelInfo(resultRelInfo, rel, 1, NULL, 0);
 
-	estate->es_result_relations = resultRelInfo;
-	estate->es_num_result_relations = 1;
-	estate->es_result_relation_info = resultRelInfo;
-	estate->es_output_cid = GetCurrentCommandId(true);
-
 	rte = makeNode(RangeTblEntry);
 	rte->rtekind = RTE_RELATION;
 	rte->relid = RelationGetRelid(rel);
 	rte->relkind = rel->rd_rel->relkind;
 	rte->rellockmode = AccessShareLock;
+	rangeTable = list_make1(rte);
 	ExecInitRangeTable(estate, list_make1(rte));
 
+	ExecInitRangeTable(estate, rangeTable);
+	ExecInitResultRelation(estate, resultRelInfo, 1);
+
+	estate->es_result_relation_info = resultRelInfo;
+	estate->es_output_cid = GetCurrentCommandId(true);
+
 	/* Prepare to catch AFTER triggers. */
 	AfterTriggerBeginQuery();
 
@@ -1238,9 +1241,10 @@ process_remote_insert(StringInfo s, Relation rel)
 			if (relinfo->ri_NumIndices > 0)
 			{
 				List	   *recheckIndexes;
+				ResultRelInfo *resultRelInfo = estate->es_result_relation_info;
 
-				recheckIndexes = ExecInsertIndexTuples(bufferedSlots[i],
-													   estate, false, NULL, NIL);
+				recheckIndexes = ExecInsertIndexTuples(resultRelInfo, bufferedSlots[i],
+													   estate, false, false, NULL, NIL);
 
 				/* AFTER ROW INSERT Triggers */
 				ExecARInsertTriggers(estate, relinfo, bufferedSlots[i],
@@ -1267,11 +1271,12 @@ process_remote_insert(StringInfo s, Relation rel)
 	else
 	{
 		TupleTableSlot *newslot;
+		ResultRelInfo *resultRelInfo = estate->es_result_relation_info;
 
 		newslot = ExecInitExtraTupleSlot(estate, tupDesc, &TTSOpsHeapTuple);
 		tuple_to_slot(estate, rel, &new_tuple, newslot);
 
-		ExecSimpleRelationInsert(estate, newslot);
+		ExecSimpleRelationInsert(resultRelInfo, estate, newslot);
 	}
 	ExecCloseIndices(estate->es_result_relation_info);
 	if (ActiveSnapshotSet())
@@ -1367,6 +1372,7 @@ process_remote_update(StringInfo s, Relation rel)
 	if (found)
 	{
 		HeapTuple	remote_tuple = NULL;
+		ResultRelInfo *resultRelInfo = estate->es_result_relation_info;
 
 		remote_tuple = heap_modify_tuple(ExecFetchSlotHeapTuple(localslot, true, NULL),
 										 tupDesc,
@@ -1376,7 +1382,7 @@ process_remote_update(StringInfo s, Relation rel)
 		ExecStoreHeapTuple(remote_tuple, remoteslot, false);
 
 		EvalPlanQualSetSlot(&epqstate, remoteslot);
-		ExecSimpleRelationUpdate(estate, &epqstate, localslot, remoteslot);
+		ExecSimpleRelationUpdate(resultRelInfo, estate, &epqstate, localslot, remoteslot);
 	}
 	else
 	{
@@ -1444,8 +1450,10 @@ process_remote_delete(StringInfo s, Relation rel)
 
 	if (found)
 	{
+		ResultRelInfo *resultRelInfo = estate->es_result_relation_info;
+
 		EvalPlanQualSetSlot(&epqstate, localslot);
-		ExecSimpleRelationDelete(estate, &epqstate, localslot);
+		ExecSimpleRelationDelete(resultRelInfo, estate, &epqstate, localslot);
 	}
 	else
 	{
diff --git a/src/pglogical_output.c b/src/pglogical_output.c
diff --git a/src/state.c b/src/state.c

Original file line number	Diff line number	Diff line change
`@@ -141,6 +141,7 @@ sub init`
`141`	`141`	`{`
`142`	`142`	`$node->init(allows_streaming => 'logical');`
`143`	`143`	`$node->append_conf('postgresql.conf', qq{`
	`144`	`+ enable_self_join_removal = off`
`144`	`145`	`max_connections = 50`
`145`	`146`	`log_line_prefix = '%m [%p] [xid%x] %i '`
`146`	`147`	`log_statement = all`
Original file line number	Diff line number	Diff line change
`@@ -353,7 +353,7 @@ dmq_shmem_startup_hook(void)`
`353`	`353`	`DMQ_MAX_SUBS_PER_BACKEND * MaxBackends,`
`354`	`354`	`DMQ_MAX_SUBS_PER_BACKEND * MaxBackends,`
`355`	`355`	`&hash_info,`
`356`		`- HASH_ELEM);`
	`356`	`+ HASH_ELEM \| HASH_STRINGS);`
`357`	`357`
`358`	`358`	`LWLockRelease(AddinShmemInitLock);`
`359`	`359`	`}`
`@@ -1443,7 +1443,7 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)`
`1443`	`1443`	`extra = dmq_receiver_start_hook(sender_name);`
`1444`	`1444`
`1445`	`1445`	`/* do not hold globalxmin. XXX: try to carefully release snaps */`
`1446`		`- MyPgXact->xmin = InvalidTransactionId;`
	`1446`	`+ MyProc->xmin = InvalidTransactionId;`
`1447`	`1447`
`1448`	`1448`	`for (;;)`
`1449`	`1449`	`{`
Original file line number	Diff line number	Diff line change
`@@ -185,7 +185,7 @@ MtmGlobalTxShmemStartup(void)`
`185`	`185`	`gtx_shared->lock = &(GetNamedLWLockTranche("mtm-gtx-lock"))->lock;`
`186`	`186`
`187`	`187`	`gtx_shared->gid2gtx = ShmemInitHash("gid2gtx", 2MaxConnections, 2MaxConnections,`
`188`		`- &info, HASH_ELEM);`
	`188`	`+ &info, HASH_ELEM \| HASH_STRINGS);`
`189`	`189`
`190`	`190`	`LWLockRelease(AddinShmemInitLock);`
`191`	`191`	`}`