pg-11 compat

Author: kelvich

arbiter.c

@@ -27,7 +27,7 @@
 #include "postgres.h"
 #include "fmgr.h"
 #include "miscadmin.h"
-#include "common/pg_socket.h"
+// #include "common/pg_socket.h"
 #include "postmaster/postmaster.h"
 #include "postmaster/bgworker.h"
 #include "storage/s_lock.h"
@@ -66,6 +66,7 @@
 #include "common/ip.h"
 #include "pgstat.h"
 
+#include "compat.h"
 
 #ifndef USE_EPOLL
 #ifdef __linux__
@@ -119,6 +120,7 @@ char const* const MtmMessageKindMnem[] =
 
 static BackgroundWorker MtmSenderWorker = {
 	"mtm-sender",
+	"mtm",
 	BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION, 
 	BgWorkerStart_ConsistentState,
 	MULTIMASTER_BGW_RESTART_TIMEOUT,
@@ -132,6 +134,7 @@ static BackgroundWorker MtmSenderWorker = {
 
 static BackgroundWorker MtmRecevierWorker = {
 	"mtm-receiver",
+	"mtm",
 	BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION, 
 	BgWorkerStart_ConsistentState,
 	MULTIMASTER_BGW_RESTART_TIMEOUT,
@@ -145,6 +148,7 @@ static BackgroundWorker MtmRecevierWorker = {
 
 static BackgroundWorker MtmMonitorWorker = {
 	"mtm-monitor",
+	"mtm",
 	BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION, 
 	BgWorkerStart_ConsistentState,
 	MULTIMASTER_BGW_RESTART_TIMEOUT,
@@ -763,7 +767,7 @@ void MtmSender(Datum arg)
 	BackgroundWorkerUnblockSignals();
 
 	/* Connect to a database */
-	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL);
+	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL, 0);
 
 	/* Start heartbeat times */
 	heartbeat_timer = RegisterTimeout(USER_TIMEOUT, MtmScheduleHeartbeat);
@@ -851,7 +855,7 @@ void MtmMonitor(Datum arg)
 	BackgroundWorkerUnblockSignals();
 
 	/* Connect to a database */
-	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL);
+	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL, 0);
 
 	while (!stop) {
 		int rc = WaitLatch(&MyProc->procLatch, WL_TIMEOUT | WL_POSTMASTER_DEATH, MtmHeartbeatRecvTimeout, PG_WAIT_EXTENSION);
@@ -898,7 +902,7 @@ void MtmReceiver(Datum arg)
 	BackgroundWorkerUnblockSignals();
 
 	/* Connect to a database */
-	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL);
+	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL, 0);
 
 	MtmAcceptIncomingConnections();
 

bgwpool.c

@@ -50,7 +50,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
 	pqsignal(SIGHUP, PostgresSigHupHandler);
 
     BackgroundWorkerUnblockSignals();
-	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser);
+	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser, 0);
 	ActivePortal = &fakePortal;
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";

compat.h

@@ -0,0 +1,22 @@
+
+#ifndef MTMCOMPAT_H
+#define MTMCOMPAT_H
+
+/* Allow use of rdma functions on non-rdma enabled postgres */
+#define pg_socket(a, b, c, d) socket(a, b, c)
+#define pg_setsockopt(a, b, c, d, e, f) setsockopt(a, b, c, d, e)
+#define pg_getsockopt(a, b, c, d, e, f) getsockopt(a, b, c, d, e)
+#define pg_set_noblock(a, b) pg_set_noblock(a)
+#define pg_bind(a, b, c, d) bind(a, b, c)
+#define pg_listen(a, b, c) listen(a, b)
+#define pg_connect(a, b, c, d) connect(a, b, c)
+#define pg_accept(a, b, c, d) accept(a, b, c)
+#define pg_select(a, b, c, d, e, f) select(a, b, c, d, e)
+#define pg_send(a, b, c, d, e) send(a, b, c, d)
+#define pg_recv(a, b, c, d, e) recv(a, b, c, d)
+#define pg_closesocket(a, b) closesocket(a)
+
+#define PQselect(a, b, c, d, e, f) select(a, b, c, d, e)
+#define PQisRsocket(a) false
+
+#endif							/* MTMCOMPAT_H */

multimaster.c

@@ -13,7 +13,7 @@
 #include "funcapi.h"
 #include "fmgr.h"
 #include "miscadmin.h"
-#include "common/pg_socket.h"
+// #include "common/pg_socket.h"
 #include "pgstat.h"
 #include "utils/regproc.h"
 
@@ -67,7 +67,7 @@
 #include "access/htup_details.h"
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
-#include "catalog/pg_constraint_fn.h"
+#include "catalog/pg_constraint.h"
 #include "catalog/pg_proc.h"
 #include "pglogical_output/hooks.h"
 #include "parser/analyze.h"
@@ -83,6 +83,8 @@
 #include "ddd.h"
 #include "state.h"
 
+#include "compat.h"
+
 typedef struct {
 	TransactionId xid;	  /* local transaction ID	*/
 	GlobalTransactionId gtid; /* global transaction ID assigned by coordinator of transaction */
@@ -206,16 +208,7 @@ static dlist_head MtmLsnMapping = DLIST_STATIC_INIT(MtmLsnMapping);
 
 static TransactionManager MtmTM =
 {
-	PgTransactionIdGetStatus,
-	PgTransactionIdSetTreeStatus,
-	MtmGetSnapshot,
-	PgGetNewTransactionId,
-	MtmGetOldestXmin,
-	PgTransactionIdIsInProgress,
-	PgGetGlobalTransactionId,
-	PgXidInMVCCSnapshot,
 	MtmDetectGlobalDeadLock,
-	MtmGetName,
 	MtmGetTransactionStateSize,
 	MtmSerializeTransactionState,
 	MtmDeserializeTransactionState,
@@ -630,7 +623,7 @@ void MtmSetSnapshot(csn_t globalSnapshot)
 
 Snapshot MtmGetSnapshot(Snapshot snapshot)
 {
-	snapshot = PgGetSnapshotData(snapshot);
+	snapshot = GetSnapshotData(snapshot);
 	if (XactIsoLevel == XACT_READ_COMMITTED && MtmTx.snapshot != INVALID_CSN) {
 		MtmTx.snapshot = MtmGetCurrentTime();
 		if (TransactionIdIsValid(GetCurrentTransactionIdIfAny())) {
@@ -644,7 +637,7 @@ Snapshot MtmGetSnapshot(Snapshot snapshot)
 
 TransactionId MtmGetOldestXmin(Relation rel, int flags)
 {
-	TransactionId xmin = PgGetOldestXmin(rel, flags); /* consider all backends */
+	TransactionId xmin = GetOldestXmin(rel, flags); /* consider all backends */
 	// if (TransactionIdIsValid(xmin)) {
 	// 	MtmLock(LW_EXCLUSIVE);
 	// 	xmin = MtmAdjustOldestXid(xmin);
@@ -4802,7 +4795,7 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 		MtmGenerateGid(x->gid);
 
 		if (!x->isTransactionBlock) {
-			BeginTransactionBlock(false);
+			BeginTransactionBlock();
 			x->isTransactionBlock = true;
 			CommitTransactionCommand();
 			StartTransactionCommand();
@@ -5807,10 +5800,10 @@ MtmDetectGlobalDeadLockForXid(TransactionId xid)
 
 		if (!hasDeadlock)
 		{
-			TimestampTz start_time = get_timeout_start_time(DEADLOCK_TIMEOUT);
+			// TimestampTz start_time = get_timeout_start_time(DEADLOCK_TIMEOUT);
 			MTM_LOG1("Enable deadlock timeout in backend %d for transaction %llu", MyProcPid, (long64)xid);
 			enable_timeout_after(DEADLOCK_TIMEOUT, DeadlockTimeout);
-			set_timeout_start_time(DEADLOCK_TIMEOUT, start_time);
+			// set_timeout_start_time(DEADLOCK_TIMEOUT, start_time);
 		}
 	}
 	return hasDeadlock;

pglogical_apply.c

@@ -514,7 +514,7 @@ process_remote_message(StringInfo s)
 					ExecVacuum(MtmVacuumStmt, 1);
 				} else if (MtmIndexStmt != NULL) {
 					Oid relid =	RangeVarGetRelidExtended(MtmIndexStmt->relation, ShareUpdateExclusiveLock,
-														 false, false,
+														 0,
 														 NULL,
 														 NULL);
 					/* Run parse analysis ... */
@@ -523,12 +523,14 @@ process_remote_message(StringInfo s)
 					DefineIndex(relid,		/* OID of heap relation */
 								MtmIndexStmt,
 								InvalidOid, /* no predefined OID */
+								InvalidOid, /* no parent index */
+								InvalidOid, /* no parent constraint */
 								false,		/* is_alter_table */
 								true,		/* check_rights */
 								true,		/* check_not_in_use */
 								false,		/* skip_build */
 								false);		/* quiet */
-					
+
 				}
 				else if (MtmDropStmt != NULL)
 				{
@@ -631,7 +633,7 @@ read_tuple_parts(StringInfo s, Relation rel, TupleData *tup)
 
 	for (i = 0; i < desc->natts; i++)
 	{
-		Form_pg_attribute att = desc->attrs[i];
+		Form_pg_attribute att = TupleDescAttr(desc, i);
 		char		kind;
 		const char *data;
 		int			len;
@@ -745,7 +747,7 @@ read_rel(StringInfo s, LOCKMODE mode)
 		relnamelen = pq_getmsgbyte(s);
 		rv->relname = (char *) pq_getmsgbytes(s, relnamelen);
 
-		local_relid = RangeVarGetRelidExtended(rv, mode, false, false, NULL, NULL);
+		local_relid = RangeVarGetRelidExtended(rv, mode, 0, NULL, NULL);
 		old_context = MemoryContextSwitchTo(TopMemoryContext);
 		pglogical_relid_map_put(remote_relid, local_relid);
 		MemoryContextSwitchTo(old_context);
@@ -820,7 +822,7 @@ process_remote_commit(StringInfo in)
 				AbortCurrentTransaction();
 			} else { 				
 				/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
-				BeginTransactionBlock(false);
+				BeginTransactionBlock();
 				CommitTransactionCommand();
 				StartTransactionCommand();
 
@@ -915,8 +917,8 @@ process_remote_insert(StringInfo s, Relation rel)
 	PushActiveSnapshot(GetTransactionSnapshot());
 
 	estate = create_rel_estate(rel);
-	newslot = ExecInitExtraTupleSlot(estate);
-	oldslot = ExecInitExtraTupleSlot(estate);
+	newslot = ExecInitExtraTupleSlot(estate, NULL);
+	oldslot = ExecInitExtraTupleSlot(estate, NULL);
 	ExecSetSlotDescriptor(newslot, tupDesc);
 	ExecSetSlotDescriptor(oldslot, tupDesc);
 
@@ -1086,9 +1088,9 @@ process_remote_update(StringInfo s, Relation rel)
 			 action);
 
 	estate = create_rel_estate(rel);
-	oldslot = ExecInitExtraTupleSlot(estate);
+	oldslot = ExecInitExtraTupleSlot(estate, NULL);
 	ExecSetSlotDescriptor(oldslot, tupDesc);
-	newslot = ExecInitExtraTupleSlot(estate);
+	newslot = ExecInitExtraTupleSlot(estate, NULL);
 	ExecSetSlotDescriptor(newslot, tupDesc);
 
 	if (action == 'K')
@@ -1196,7 +1198,7 @@ process_remote_delete(StringInfo s, Relation rel)
 	bool		found_old;
 
 	estate = create_rel_estate(rel);
-	oldslot = ExecInitExtraTupleSlot(estate);
+	oldslot = ExecInitExtraTupleSlot(estate, NULL);
 	ExecSetSlotDescriptor(oldslot, tupDesc);
 
 	read_tuple_parts(s, rel, &oldtup);
@@ -1277,9 +1279,7 @@ void MtmExecutor(void* work, size_t size)
     if (MtmApplyContext == NULL) {
         MtmApplyContext = AllocSetContextCreate(TopMemoryContext,
 												"ApplyContext",
-												ALLOCSET_DEFAULT_MINSIZE,
-												ALLOCSET_DEFAULT_INITSIZE,
-												ALLOCSET_DEFAULT_MAXSIZE);
+												ALLOCSET_DEFAULT_SIZES);
     }
 	top_context = MemoryContextSwitchTo(MtmApplyContext);
 	replorigin_session_origin = InvalidRepOriginId;

pglogical_output.c

@@ -209,9 +209,7 @@ pg_decode_startup(LogicalDecodingContext * ctx, OutputPluginOptions *opt,
 
 	data->context = AllocSetContextCreate(TopMemoryContext,
 										  "pglogical conversion context",
-										  ALLOCSET_DEFAULT_MINSIZE,
-										  ALLOCSET_DEFAULT_INITSIZE,
-										  ALLOCSET_DEFAULT_MAXSIZE);
+										  ALLOCSET_DEFAULT_SIZES);
 	data->allow_internal_basetypes = true;
 	data->allow_binary_basetypes = true;
 
@@ -407,9 +405,7 @@ pg_decode_startup(LogicalDecodingContext * ctx, OutputPluginOptions *opt,
 
 			data->hooks_mctxt = AllocSetContextCreate(ctx->context,
 					"pglogical_output hooks context",
-					ALLOCSET_SMALL_MINSIZE,
-					ALLOCSET_SMALL_INITSIZE,
-					ALLOCSET_SMALL_MAXSIZE);
+					ALLOCSET_DEFAULT_SIZES);
 
 			load_hooks(data);
 			call_startup_hook(data, ctx->output_plugin_options);

pglogical_proto.c

@@ -631,7 +631,7 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 
 	for (i = 0; i < desc->natts; i++)
 	{
-		if (desc->attrs[i]->attisdropped)
+		if (TupleDescAttr(desc, i)->attisdropped)
 			continue;
 		nliveatts++;
 	}
@@ -652,7 +652,7 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 	{
 		HeapTuple	typtup;
 		Form_pg_type typclass;
-		Form_pg_attribute att = desc->attrs[i];
+		Form_pg_attribute att = TupleDescAttr(desc, i);
 		char		transfer_type;
 
 		/* skip dropped columns */

pglogical_receiver.c

@@ -16,10 +16,12 @@
 /* Some general headers for custom bgworker facility */
 
 #include <unistd.h>
+#include <sys/time.h>
+
 #include "postgres.h"
 #include "fmgr.h"
 #include "miscadmin.h"
-#include "common/pg_socket.h"
+// #include "common/pg_socket.h"
 #include "pqexpbuffer.h"
 #include "access/xact.h"
 #include "access/clog.h"
@@ -47,6 +49,7 @@
 #include "spill.h"
 #include "state.h"
 #include "bgwpool.h"
+#include "compat.h"
 
 #define ERRCODE_DUPLICATE_OBJECT_STR  "42710"
 #define RECEIVER_SUSPEND_TIMEOUT (1*USECS_PER_SEC)
@@ -269,7 +272,7 @@ pglogical_receiver_main(Datum main_arg)
 	BackgroundWorkerUnblockSignals();
 
 	/* Connect to a database */
-	BackgroundWorkerInitializeConnection(MtmDatabaseName, MtmDatabaseUser);
+	BackgroundWorkerInitializeConnection(MtmDatabaseName, MtmDatabaseUser, 0);
 	ActivePortal = &fakePortal;
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";

spill.c

@@ -67,8 +67,7 @@ int MtmCreateSpillFile(int node_id, int* file_id)
 	sprintf(path, "pg_mtm/%d/txn-%d.snap", 
 			node_id, ++spill_file_id);
 	fd = OpenTransientFile(path,
-						   O_CREAT | O_TRUNC | O_WRONLY | O_APPEND | PG_BINARY,
-						   S_IRUSR | S_IWUSR);
+						   O_CREAT | O_TRUNC | O_WRONLY | O_APPEND | PG_BINARY);
 	if (fd < 0) { 
 		ereport(PANIC,
 				(errcode_for_file_access(),
@@ -87,8 +86,7 @@ int MtmOpenSpillFile(int node_id, int file_id)
 	sprintf(path, "pg_mtm/%d/txn-%d.snap", 
 			node_id, file_id);
 	fd = OpenTransientFile(path,
-						   O_RDONLY | PG_BINARY,
-						   S_IRUSR | S_IWUSR);
+						   O_RDONLY | PG_BINARY);
 	if (fd < 0) { 
 		ereport(PANIC,
 				(errcode_for_file_access(),