PG_VERSION_NUM test macro

Author: Andrei Krichinin

src/dmq.c

@@ -1119,7 +1119,11 @@ dmq_handle_message(StringInfo msg, DmqReceiverSlot *my_slot,
 	}
 	else
 	{
+#if PG_VERSION_NUM < 150000
+		res = shm_mq_send(mq_handles[sub.procno], body_len, body, false);
+#else
 		res = shm_mq_send(mq_handles[sub.procno], body_len, body, false, true);
+#endif
 		if (res == SHM_MQ_DETACHED)
 			mtm_log(COMMERROR, "[DMQ] queue %d is detached, dropping message (stream=%s)",
 					sub.procno, stream_name);
@@ -1623,7 +1627,11 @@ dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg)
 			buf.len, buf.len, buf.data);
 
 	/* XXX: use sendv instead */
+#if PG_VERSION_NUM < 150000
+	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, false);
+#else
 	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, false, true);
+#endif
 	pfree(buf.data);
 	if (res != SHM_MQ_SUCCESS)
 		mtm_log(ERROR, "[DMQ] dmq_push: can't send to queue");
@@ -1648,7 +1656,11 @@ dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *payload
 			buf.len, buf.len, buf.data);
 
 	/* XXX: use sendv instead */
+#if PG_VERSION_NUM < 150000
+	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, false);
+#else
 	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, false, true);
+#endif
 	pfree(buf.data);
 	if (res != SHM_MQ_SUCCESS)
 		mtm_log(ERROR, "[DMQ] dmq_push: can't send to queue, status = %d", res);

src/multimaster.c

@@ -1097,7 +1097,9 @@ mtm_after_node_create(PG_FUNCTION_ARGS)
 	bool		conninfo_isnull;
 	int			n_nodes;
 	int			rc;
+#if PG_VERSION_NUM >= 150000
 	ParseState *pstate;
+#endif
 
 	Assert(CALLED_AS_TRIGGER(fcinfo));
 	Assert(TRIGGER_FIRED_FOR_ROW(trigdata->tg_event));
@@ -1138,7 +1140,9 @@ mtm_after_node_create(PG_FUNCTION_ARGS)
 
 	mtm_log(NodeMgmt, "mtm_after_node_create %d", node_id);
 
+#if PG_VERSION_NUM >= 150000
 	pstate = make_parsestate(NULL);
+#endif
 
 	if (is_self)
 	{
@@ -1148,11 +1152,19 @@ mtm_after_node_create(PG_FUNCTION_ARGS)
 		 */
 		pub_stmt->pubname = MULTIMASTER_NAME;
 		pub_stmt->for_all_tables = false;
+#if PG_VERSION_NUM < 150000
+		pub_stmt->tables = NIL;
+#else
 		pub_stmt->pubobjects = NIL;
+#endif
 		pub_stmt->options = list_make1(
 									   makeDefElem("publish", (Node *) makeString(pstrdup("insert, truncate")), -1)
 			);
+#if PG_VERSION_NUM < 150000
+		CreatePublication(pub_stmt);
+#else
 		CreatePublication(pstate, pub_stmt);
+#endif
 
 		/* liftoff */
 		MtmMonitorStart(MyDatabaseId, GetUserId());
@@ -1191,7 +1203,11 @@ mtm_after_node_create(PG_FUNCTION_ARGS)
 		client_min_messages = ERROR;
 		log_min_messages = ERROR;
 
+#if PG_VERSION_NUM < 150000
+		CreateSubscription(cs_stmt, true);
+#else
 		CreateSubscription(pstate, cs_stmt, true);
+#endif
 
 		/* restore log_level's */
 		client_min_messages = saved_client_min_messages;
@@ -1209,7 +1225,9 @@ mtm_after_node_create(PG_FUNCTION_ARGS)
 		origin_name = psprintf(MULTIMASTER_SLOT_PATTERN, node_id);
 		replorigin_create(origin_name);
 	}
+#if PG_VERSION_NUM >= 150000
 	free_parsestate(pstate);
+#endif
 
 	PG_RETURN_VOID();
 }

src/pglogical_proto.c

@@ -468,7 +468,11 @@ pglogical_write_prepare(StringInfo out, PGLogicalOutputData *data,
 	/* send fixed fields */
 	pq_sendint64(out, lsn);
 	pq_sendint64(out, txn->end_lsn);
+#if PG_VERSION_NUM < 150000
+	pq_sendint64(out, txn->commit_time);
+#else
 	pq_sendint64(out, txn->xact_time.commit_time);
+#endif
 
 	send_node_id(out, txn, hooks_data);
 	pq_sendint64(out, txn->origin_lsn);
@@ -499,7 +503,11 @@ pglogical_write_commit_prepared(StringInfo out, PGLogicalOutputData *data,
 	/* send fixed fields */
 	pq_sendint64(out, lsn);
 	pq_sendint64(out, txn->end_lsn);
+#if PG_VERSION_NUM < 150000
+	pq_sendint64(out, txn->commit_time);
+#else
 	pq_sendint64(out, txn->xact_time.commit_time);
+#endif
 
 	send_node_id(out, txn, hooks_data);
 	pq_sendint64(out, txn->origin_lsn);
@@ -532,7 +540,11 @@ pglogical_write_abort_prepared(StringInfo out, PGLogicalOutputData *data,
 	/* send fixed fields */
 	pq_sendint64(out, lsn);
 	pq_sendint64(out, txn->end_lsn);
+#if PG_VERSION_NUM < 150000
+	pq_sendint64(out, txn->commit_time);
+#else
 	pq_sendint64(out, txn->xact_time.commit_time);
+#endif
 
 	send_node_id(out, txn, hooks_data);
 	pq_sendint64(out, txn->origin_lsn);
@@ -560,7 +572,11 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	/* send fixed fields */
 	pq_sendint64(out, lsn);
 	pq_sendint64(out, txn->end_lsn);
+#if PG_VERSION_NUM < 150000
+	pq_sendint64(out, txn->commit_time);
+#else
 	pq_sendint64(out, txn->xact_time.commit_time);
+#endif
 
 	send_node_id(out, txn, hooks_data);
 	pq_sendint64(out, txn->origin_lsn);
@@ -583,7 +599,11 @@ pglogical_write_abort(StringInfo out, PGLogicalOutputData *data,
 	/* send fixed fields */
 	pq_sendint64(out, lsn);
 	pq_sendint64(out, txn->end_lsn);
+#if PG_VERSION_NUM < 150000
+	pq_sendint64(out, txn->commit_time);
+#else
 	pq_sendint64(out, txn->xact_time.commit_time);
+#endif
 
 	send_node_id(out, txn, hooks_data);
 	pq_sendint64(out, txn->origin_lsn);

src/state.c

@@ -12,7 +12,9 @@
 #include <sys/types.h>
 #include <unistd.h>
 
+#if PG_VERSION_NUM >= 150000
 #include "common/pg_prng.h"
+#endif
 #include "access/twophase.h"
 #include "access/xlogutils.h"
 #include "access/xlog_internal.h"
@@ -1678,7 +1680,11 @@ CampaignerMain(Datum main_arg)
 	MemoryContext campaigner_ctx =	AllocSetContextCreate(TopMemoryContext,
 														  "CampaignerContext",
 														  ALLOCSET_DEFAULT_SIZES);
+#if PG_VERSION_NUM < 150000
+	static unsigned short drandom_seed[3] = {0, 0, 0};
+#else
 	static pg_prng_state drandom_seed = {0, 0};
+#endif
 	TimestampTz last_campaign_at = 0;
 	int			rc = WL_TIMEOUT;
 
@@ -1720,7 +1726,13 @@ CampaignerMain(Datum main_arg)
 
 		/* Mix the PID with the most predictable bits of the timestamp */
 		iseed = (uint64) now ^ ((uint64) MyProcPid << 32);
+#if PG_VERSION_NUM < 150000
+		drandom_seed[0] = (unsigned short) iseed;
+		drandom_seed[1] = (unsigned short) (iseed >> 16);
+		drandom_seed[2] = (unsigned short) (iseed >> 32);
+#else
 		pg_prng_seed(&drandom_seed, iseed);
+#endif
 	}
 
 	/*
@@ -1800,10 +1812,17 @@ CampaignerMain(Datum main_arg)
 		 * here nodes will mostly propose the same set of candidates,
 		 * supporting each other)
 		 */
+#if PG_VERSION_NUM < 150000
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+					   campaign_retry_interval * pg_erand48(drandom_seed),
+					   PG_WAIT_EXTENSION);
+#else
 		rc = WaitLatch(MyLatch,
 					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
 					   campaign_retry_interval * pg_prng_double(&drandom_seed),
 					   PG_WAIT_EXTENSION);
+#endif
 
 		if (rc & WL_LATCH_SET)
 			ResetLatch(MyLatch);
@@ -4239,7 +4258,11 @@ GetLoggedPreparedXactState(HTAB *txset)
 	XLogRecPtr start_lsn;
 	XLogRecPtr lsn;
 	TimeLineID timeline;
+#if PG_VERSION_NUM < 150000
+	XLogRecPtr end_wal_lsn = GetFlushRecPtr();
+#else
 	XLogRecPtr end_wal_lsn = GetFlushRecPtr(NULL);
+#endif
 	XLogRecPtr end_lsn = end_wal_lsn;
 	int		   n_trans = hash_get_num_entries(txset);
 

src/syncpoint.c

@@ -477,7 +477,11 @@ RecoveryFilterLoad(int filter_node_id, Syncpoint *spvector, MtmConfig *mtm_cfg)
 
 	/* ensure we will scan everything written up to this point, just in case */
 	XLogFlush(GetXLogWriteRecPtr());
+#if PG_VERSION_NUM < 150000
+	current_last_lsn = GetFlushRecPtr();
+#else
 	current_last_lsn = GetFlushRecPtr(NULL);
+#endif
 
 	Assert(current_last_lsn != InvalidXLogRecPtr);
 
@@ -490,7 +494,11 @@ RecoveryFilterLoad(int filter_node_id, Syncpoint *spvector, MtmConfig *mtm_cfg)
 	}
 
 	/* create hash */
+#if PG_VERSION_NUM < 150000
+	estimate_size = (GetFlushRecPtr() - start_lsn) / 100;
+#else
 	estimate_size = (GetFlushRecPtr(NULL) - start_lsn) / 100;
+#endif
 	estimate_size = Min(Max(estimate_size, 1000), 100000);
 	MemSet(&hash_ctl, 0, sizeof(hash_ctl));
 	hash_ctl.keysize = hash_ctl.entrysize = sizeof(FilterEntry);