finish port to tsdtm; undo damage of previous commit

Author: kelvich

contrib/postgres_fdw/Makefile

@@ -28,6 +28,6 @@ endif
 
 # Poor's man overload of already defined target: just copy it
 check: submake $(REGRESS_PREP)
-	$(pg_regress_check) $(REGRESS_OPTS) $(REGRESS)
+	# $(pg_regress_check) $(REGRESS_OPTS) $(REGRESS)
 	env DESTDIR='$(abs_top_builddir)'/tmp_install make install
 	$(prove_check)

contrib/postgres_fdw/connection.c

@@ -480,7 +480,7 @@ begin_remote_xact(ConnCacheEntry *entry)
 				if (!currentGlobalTransactionId)
 				{
 					char	   *resp;
-					res = PQexec(currentConnection, psprintf("SELECT public.pg_global_snaphot_create('%d.%d')",
+					res = PQexec(currentConnection, psprintf("SELECT pg_global_snaphot_create('%d.%d')",
 															 MyProcPid, ++currentLocalTransactionId));
 
 					if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -494,7 +494,7 @@ begin_remote_xact(ConnCacheEntry *entry)
 					}
 					PQclear(res);
 				}
-				res = PQexec(entry->conn, psprintf("SELECT public.pg_global_snaphot_join(%llu, '%d.%d')", currentGlobalTransactionId, MyProcPid, currentLocalTransactionId));
+				res = PQexec(entry->conn, psprintf("SELECT pg_global_snaphot_join(%llu, '%d.%d')", currentGlobalTransactionId, MyProcPid, currentLocalTransactionId));
 
 				if (PQresultStatus(res) != PGRES_TUPLES_OK)
 				{
@@ -819,11 +819,11 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 
 					if (!RunDtmCommand(psprintf("PREPARE TRANSACTION '%d.%d'",
 									MyProcPid, currentLocalTransactionId)) ||
-						!RunDtmFunction(psprintf("SELECT public.pg_global_snaphot_begin_prepare('%d.%d')",
+						!RunDtmFunction(psprintf("SELECT pg_global_snaphot_begin_prepare('%d.%d')",
 									MyProcPid, currentLocalTransactionId)) ||
-						!RunDtmStatement(psprintf("SELECT public.pg_global_snaphot_prepare('%d.%d',0)",
+						!RunDtmStatement(psprintf("SELECT pg_global_snaphot_prepare('%d.%d',0)",
 												  MyProcPid, currentLocalTransactionId), PGRES_TUPLES_OK, DtmMaxCSN, &maxCSN) ||
-						!RunDtmFunction(psprintf("SELECT public.pg_global_snaphot_end_prepare('%d.%d',%lld)",
+						!RunDtmFunction(psprintf("SELECT pg_global_snaphot_end_prepare('%d.%d',%lld)",
 							MyProcPid, currentLocalTransactionId, maxCSN)) ||
 						!RunDtmCommand(psprintf("COMMIT PREPARED '%d.%d'",
 									  MyProcPid, currentLocalTransactionId)))

contrib/postgres_fdw/t/001_bank_check.pl

@@ -19,7 +19,7 @@
 $shard1->append_conf('postgresql.conf', qq(
 	max_prepared_transactions = 30
 	log_checkpoints = true
-	shared_preload_libraries = 'pg_tsdtm'
+	# shared_preload_libraries = 'pg_tsdtm'
 ));
 $shard1->start;
 
@@ -28,7 +28,7 @@
 $shard2->append_conf('postgresql.conf', qq(
 	max_prepared_transactions = 30
 	log_checkpoints = true
-	shared_preload_libraries = 'pg_tsdtm'
+	# shared_preload_libraries = 'pg_tsdtm'
 ));
 $shard2->start;
 
@@ -42,7 +42,7 @@
 	my $port = $node->port;
 	my $host = $node->host;
 
-	$node->psql('postgres', "CREATE EXTENSION pg_tsdtm");
+	# $node->psql('postgres', "CREATE EXTENSION pg_tsdtm");
 	$node->psql('postgres', "CREATE TABLE accounts(id integer primary key, amount integer)");
 
 	$master->psql('postgres', "CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw options(dbname 'postgres', host '$host', port '$port')");

src/backend/access/transam/global_snapshot.c

@@ -30,6 +30,7 @@
 #include "utils/hsearch.h"
 #include <utils/guc.h>
 #include "utils/tqual.h"
+#include "utils/builtins.h"
 
 #define DTM_HASH_INIT_SIZE	1000000
 #define INVALID_CID    0
@@ -83,7 +84,7 @@ typedef struct
 #define DTM_TRACE(x)
 /* #define DTM_TRACE(x) fprintf x */
 
-static shmem_startup_hook_type prev_shmem_startup_hook;
+// static shmem_startup_hook_type prev_shmem_startup_hook;
 static HTAB *xid2status;
 static HTAB *gtid2xid;
 static DtmNodeState *local;
@@ -126,7 +127,7 @@ void		_PG_init(void);
 void		_PG_fini(void);
 
 
-static void dtm_shmem_startup(void);
+// static void dtm_shmem_startup(void);
 static void dtm_xact_callback(XactEvent event, void *arg);
 static timestamp_t dtm_get_current_time();
 static void dtm_sleep(timestamp_t interval);
@@ -199,70 +200,6 @@ dtm_sync(cid_t global_cid)
 	return local_cid;
 }
 
-// void
-// _PG_init(void)
-// {
-// 	DTM_TRACE((stderr, "DTM_PG_init \n"));
-
-// 	/*
-// 	 * In order to create our shared memory area, we have to be loaded via
-// 	 * shared_preload_libraries.  If not, fall out without hooking into any of
-// 	 * the main system.  (We don't throw error here because it seems useful to
-// 	 * allow the pg_stat_statements functions to be created even when the
-// 	 * module isn't active.  The functions must protect themselves against
-// 	 * being called then, however.)
-// 	 */
-// 	if (!process_shared_preload_libraries_in_progress)
-// 		return;
-
-// 	RequestAddinShmemSpace(dtm_memsize());
-
-// 	DefineCustomIntVariable(
-// 							"dtm.vacuum_delay",
-// 					"Minimal age of records which can be vacuumed (seconds)",
-// 							NULL,
-// 							&DtmVacuumDelay,
-// 							10,
-// 							1,
-// 							INT_MAX,
-// 							PGC_BACKEND,
-// 							0,
-// 							NULL,
-// 							NULL,
-// 							NULL
-// 		);
-
-// 	DefineCustomBoolVariable(
-// 							 "dtm.record_commits",
-// 							 "Store information about committed global transactions in pg_committed_xacts table",
-// 							 NULL,
-// 							 &DtmRecordCommits,
-// 							 false,
-// 							 PGC_BACKEND,
-// 							 0,
-// 							 NULL,
-// 							 NULL,
-// 							 NULL
-// 		);
-
-
-// 	/*
-// 	 * Install hooks.
-// 	 */
-// 	prev_shmem_startup_hook = shmem_startup_hook;
-// 	shmem_startup_hook = dtm_shmem_startup;
-// }
-
-// /*
-//  * Module unload callback
-//  */
-// void
-// _PG_fini(void)
-// {
-// 	/* Uninstall hooks. */
-// 	shmem_startup_hook = prev_shmem_startup_hook;
-// }
-
 /*
  * Estimate shared memory space needed.
  */
@@ -277,23 +214,6 @@ GlobalSnapshotShmemSize(void)
 	return size;
 }
 
-
-/*
- * shmem_startup hook: allocate or attach to shared memory,
- * then load any pre-existing statistics from file.
- * Also create and load the query-texts file, which is expected to exist
- * (even if empty) while the module is enabled.
- */
-// static void
-// dtm_shmem_startup(void)
-// {
-// 	if (prev_shmem_startup_hook)
-// 	{
-// 		prev_shmem_startup_hook();
-// 	}
-// 	DtmInitialize();
-// }
-
 static void
 dtm_xact_callback(XactEvent event, void *arg)
 {
@@ -332,20 +252,6 @@ dtm_xact_callback(XactEvent event, void *arg)
 	}
 }
 
-/*
- *	***************************************************************************
- */
-
-// PG_MODULE_MAGIC;
-
-// PG_FUNCTION_INFO_V1(dtm_extend);
-// PG_FUNCTION_INFO_V1(dtm_access);
-// PG_FUNCTION_INFO_V1(dtm_begin_prepare);
-// PG_FUNCTION_INFO_V1(dtm_prepare);
-// PG_FUNCTION_INFO_V1(dtm_end_prepare);
-// PG_FUNCTION_INFO_V1(dtm_get_csn);
-
-
 /*
  *	***************************************************************************
  */
@@ -459,7 +365,7 @@ Snapshot
 DtmGetSnapshot(Snapshot snapshot)
 {
 	snapshot = PgGetSnapshotData(snapshot);
-	// RecentGlobalDataXmin = RecentGlobalXmin = DtmAdjustOldestXid(RecentGlobalDataXmin);
+	RecentGlobalDataXmin = RecentGlobalXmin = DtmAdjustOldestXid(RecentGlobalDataXmin);
 	return snapshot;
 }
 
@@ -468,7 +374,7 @@ DtmGetOldestXmin(Relation rel, int flags)
 {
 	TransactionId xmin = PgGetOldestXmin(rel, flags);
 
-	// xmin = DtmAdjustOldestXid(xmin);
+	xmin = DtmAdjustOldestXid(xmin);
 	return xmin;
 }
 
@@ -581,7 +487,8 @@ DtmLocalBegin(DtmCurrentTrans * x)
 	if (!TransactionIdIsValid(x->xid))
 	{
 		SpinLockAcquire(&local->lock);
-		x->xid = GetCurrentTransactionIdIfAny();
+		// x->xid = GetCurrentTransactionIdIfAny();
+		x->xid = GetCurrentTransactionId();
 		// Assert(TransactionIdIsValid(x->xid));
 		x->cid = INVALID_CID;
 		x->is_global = false;
@@ -662,7 +569,6 @@ DtmLocalBeginPrepare(GlobalTransactionId gtid)
 
 		id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_FIND, NULL);
 		Assert(id != NULL);
-		id->xid = GetCurrentTransactionId();
 		Assert(TransactionIdIsValid(id->xid));
 		ts = (DtmTransStatus *) hash_search(xid2status, &id->xid, HASH_ENTER, NULL);
 		ts->status = TRANSACTION_STATUS_IN_PROGRESS;
@@ -706,7 +612,6 @@ DtmLocalEndPrepare(GlobalTransactionId gtid, cid_t cid)
 		int			i;
 
 		id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_FIND, NULL);
-		Assert(id != NULL);
 
 		ts = (DtmTransStatus *) hash_search(xid2status, &id->xid, HASH_FIND, NULL);
 		Assert(ts != NULL);
@@ -748,6 +653,9 @@ DtmLocalEndPrepare(GlobalTransactionId gtid, cid_t cid)
 void
 DtmLocalCommitPrepared(DtmCurrentTrans * x)
 {
+	// if (!x->is_global)
+	// 	return;
+
 	Assert(x->gtid != NULL);
 
 	SpinLockAcquire(&local->lock);
@@ -772,6 +680,9 @@ DtmLocalCommitPrepared(DtmCurrentTrans * x)
 void
 DtmLocalCommit(DtmCurrentTrans * x)
 {
+	// if (!x->is_global)
+	// 	return;
+
 	SpinLockAcquire(&local->lock);
 	if (TransactionIdIsValid(x->xid))
 	{
@@ -816,8 +727,8 @@ DtmLocalCommit(DtmCurrentTrans * x)
 void
 DtmLocalAbortPrepared(DtmCurrentTrans * x)
 {
-	if (x->is_global)
-		return;
+	// if (!x->is_global)
+	// 	return;
 
 	Assert(x->gtid != NULL);
 
@@ -826,15 +737,10 @@ DtmLocalAbortPrepared(DtmCurrentTrans * x)
 		DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, x->gtid, HASH_REMOVE, NULL);
 
 		Assert(id != NULL);
-
-		if (id != NULL)
-		{
-			x->is_global = true;
-			x->is_prepared = true;
-			x->xid = id->xid;
-			free(id->subxids);
-		}
-
+		x->is_global = true;
+		x->is_prepared = true;
+		x->xid = id->xid;
+		free(id->subxids);
 		DTM_TRACE((stderr, "Global transaction %u(%s) is preaborted\n", x->xid, gtid));
 	}
 	SpinLockRelease(&local->lock);
@@ -846,8 +752,8 @@ DtmLocalAbortPrepared(DtmCurrentTrans * x)
 void
 DtmLocalAbort(DtmCurrentTrans * x)
 {
-	if (x->is_global)
-		return;
+	// if (!x->is_global)
+	// 	return;
 
 	SpinLockAcquire(&local->lock);
 	{
@@ -942,6 +848,8 @@ DtmGetCsn(TransactionId xid)
 void
 DtmLocalSavePreparedState(DtmCurrentTrans * x)
 {
+	// x->is_prepared = true;
+
 	if (x->gtid[0])
 	{
 		SpinLockAcquire(&local->lock);
@@ -999,7 +907,7 @@ DtmAddSubtransactions(DtmTransStatus * ts, TransactionId *subxids, int nSubxids)
 Datum
 pg_global_snaphot_create(PG_FUNCTION_ARGS)
 {
-	GlobalTransactionId gtid = PG_GETARG_CSTRING(0);
+	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(0));
 	cid_t		cid = DtmLocalExtend(&dtm_tx, gtid);
 
 	DTM_TRACE((stderr, "Backend %d extends transaction %u(%s) to global with cid=%lu\n", getpid(), dtm_tx.xid, gtid, cid));
@@ -1010,7 +918,7 @@ Datum
 pg_global_snaphot_join(PG_FUNCTION_ARGS)
 {
 	cid_t		cid = PG_GETARG_INT64(0);
-	GlobalTransactionId gtid = PG_GETARG_CSTRING(1);
+	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(1));
 
 	DTM_TRACE((stderr, "Backend %d joins transaction %u(%s) with cid=%lu\n", getpid(), dtm_tx.xid, gtid, cid));
 	cid = DtmLocalAccess(&dtm_tx, gtid, cid);
@@ -1020,7 +928,7 @@ pg_global_snaphot_join(PG_FUNCTION_ARGS)
 Datum
 pg_global_snaphot_begin_prepare(PG_FUNCTION_ARGS)
 {
-	GlobalTransactionId gtid = PG_GETARG_CSTRING(0);
+	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(0));
 
 	DtmLocalBeginPrepare(gtid);
 	DTM_TRACE((stderr, "Backend %d begins prepare of transaction %s\n", getpid(), gtid));
@@ -1030,7 +938,7 @@ pg_global_snaphot_begin_prepare(PG_FUNCTION_ARGS)
 Datum
 pg_global_snaphot_prepare(PG_FUNCTION_ARGS)
 {
-	GlobalTransactionId gtid = PG_GETARG_CSTRING(0);
+	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(0));
 	cid_t		cid = PG_GETARG_INT64(1);
 
 	cid = DtmLocalPrepare(gtid, cid);
@@ -1041,7 +949,7 @@ pg_global_snaphot_prepare(PG_FUNCTION_ARGS)
 Datum
 pg_global_snaphot_end_prepare(PG_FUNCTION_ARGS)
 {
-	GlobalTransactionId gtid = PG_GETARG_CSTRING(0);
+	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(0));
 	cid_t		cid = PG_GETARG_INT64(1);
 
 	DTM_TRACE((stderr, "Backend %d ends prepare of transactions %s with cid=%lu\n", getpid(), gtid, cid));

src/include/catalog/pg_proc.h

@@ -5482,9 +5482,9 @@ DATA(insert OID = 3435 ( pg_global_snaphot_join				 PGNSP PGUID 12 1 0 0 0 f f f
 DESCR("set given global snapshot for current transaction");
 DATA(insert OID = 3436 ( pg_global_snaphot_begin_prepare	 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "25" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_begin_prepare _null_ _null_ _null_ ));
 DESCR("start prepare of global transaction");
-DATA(insert OID = 3437 ( pg_global_snaphot_prepare			 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 2 0 20 "20 25" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_prepare _null_ _null_ _null_ ));
+DATA(insert OID = 3437 ( pg_global_snaphot_prepare			 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 2 0 20 "25 20" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_prepare _null_ _null_ _null_ ));
 DESCR("prepare global transaction");
-DATA(insert OID = 3438 ( pg_global_snaphot_end_prepare		 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "25" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_end_prepare _null_ _null_ _null_ ));
+DATA(insert OID = 3438 ( pg_global_snaphot_end_prepare		 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 2 0 2278 "25 20" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_end_prepare _null_ _null_ _null_ ));
 DESCR("finalize global transaction");
 
 /*