add DbId to abort records

kelvich · kelvich · commit 21625635fc8a · 2016-03-10T16:29:13.000+03:00
diff --git a/contrib/pglogical/Makefile b/contrib/pglogical/Makefile
@@ -15,7 +15,7 @@ OBJS = pglogical_apply.o pglogical_conflict.o pglogical_manager.o \
 PG_CPPFLAGS = -I$(libpq_srcdir) -I$(top_srcdir)/contrib/pglogical_output
 SHLIB_LINK = $(libpq)
 
-REGRESS = preseed infofuncs init_fail init preseed_check basic extended twophase toasted replication_set add_table matview bidirectional primary_key foreign_key functions copy drop
+REGRESS = preseed infofuncs init_fail init preseed_check basic extended toasted replication_set add_table matview bidirectional primary_key foreign_key functions copy twophase drop
 
 # In-tree builds only
 subdir = contrib/pglogical
diff --git a/contrib/pglogical/expected/twophase.out b/contrib/pglogical/expected/twophase.out
@@ -3,19 +3,26 @@ SELECT * FROM pglogical_regress_variables()
 \gset
 \c :provider_dsn
 SELECT pglogical.replicate_ddl_command($$
-CREATE TABLE public.test2pc_tbl(id serial primary key, value int);
+	CREATE TABLE public.test2pc_tbl(id serial primary key, value int);
 $$);
  replicate_ddl_command 
 -----------------------
  t
 (1 row)
 
-SELECT * FROM pglogical.replication_set_add_all_tables('default', '{public}');
- replication_set_add_all_tables 
---------------------------------
+-- SELECT * FROM pglogical.replication_set_add_all_tables('default', '{public}');
+SELECT * FROM pglogical.replication_set_add_table('default', 'test2pc_tbl');
+ replication_set_add_table 
+---------------------------
  t
 (1 row)
 
+SELECT pg_xlog_wait_remote_apply(pg_current_xlog_location(), 0);
+ pg_xlog_wait_remote_apply 
+---------------------------
+ 
+(1 row)
+
 -- Check that prapeared state is visible on slave and data available after commit
 BEGIN;
 INSERT INTO test2pc_tbl VALUES (1, 10);
@@ -28,10 +35,11 @@ SELECT pg_xlog_wait_remote_apply(pg_current_xlog_location(), 0);
 
 \c :subscriber_dsn
 SELECT gid, owner, database FROM pg_prepared_xacts;
- gid | owner |      database      
------+-------+--------------------
- tx1 | stas  | contrib_regression
-(1 row)
+         gid         | owner |      database      
+---------------------+-------+--------------------
+ tx1                 | stas  | contrib_regression
+ test_subscriber:tx1 | stas  | postgres
+(2 rows)
 
 SELECT * FROM test2pc_tbl;
  id | value 
@@ -71,10 +79,11 @@ SELECT pg_xlog_wait_remote_apply(pg_current_xlog_location(), 0);
 
 \c :subscriber_dsn
 SELECT gid, owner, database FROM pg_prepared_xacts;
- gid | owner |      database      
------+-------+--------------------
- tx2 | stas  | contrib_regression
-(1 row)
+         gid         | owner |      database      
+---------------------+-------+--------------------
+ tx2                 | stas  | contrib_regression
+ test_subscriber:tx2 | stas  | postgres
+(2 rows)
 
 SELECT * FROM test2pc_tbl;
  id | value 
diff --git a/contrib/pglogical/pglogical_apply.c b/contrib/pglogical/pglogical_apply.c
@@ -148,8 +148,10 @@ handle_commit(StringInfo s)
 	TimestampTz		commit_time;
 	uint8 			flags;
 	const char	   *gid;
+	char 		   *local_gid;
 	PGLFlushPosition *flushpos;
 	bool 			flush = true;
+	PGLogicalLocalNode *mynode;
 
 	pglogical_read_commit(s, &commit_lsn, &end_lsn, &commit_time, &flags, &gid);
 
@@ -170,31 +172,53 @@ handle_commit(StringInfo s)
 			CommitTransactionCommand();
 			StartTransactionCommand();
 
+			mynode = get_local_node(false);
+			local_gid = palloc0(GIDSIZE);
+			strncpy(local_gid, mynode->node->name, GIDSIZE);
+			strncat(local_gid, ":", GIDSIZE);
+			strncat(local_gid, gid, GIDSIZE);
+
 			/* PREPARE itself */
-			PrepareTransactionBlock(gid);
+			PrepareTransactionBlock(local_gid);
 			CommitTransactionCommand();
 			break;
 		}
 		case PGLOGICAL_COMMIT_PREPARED:
 		{
 			StartTransactionCommand();
-			FinishPreparedTransaction(gid, true);
+
+			mynode = get_local_node(false);
+			local_gid = palloc0(GIDSIZE);
+			strncpy(local_gid, mynode->node->name, GIDSIZE);
+			strncat(local_gid, ":", GIDSIZE);
+			strncat(local_gid, gid, GIDSIZE);
+
+			FinishPreparedTransaction(local_gid, true);
 			CommitTransactionCommand();
 
 			/* There were no BEGIN stmt for COMMIT PREPARED */
 			replorigin_session_origin_timestamp = commit_time;
 			replorigin_session_origin_lsn = commit_lsn;
+
 			break;
 		}
 		case PGLOGICAL_ABORT_PREPARED:
 		{
 			StartTransactionCommand();
-			FinishPreparedTransaction(gid, false);
+
+			mynode = get_local_node(false);
+			local_gid = palloc(GIDSIZE);
+			strncpy(local_gid, mynode->node->name, GIDSIZE);
+			strncat(local_gid, ":", GIDSIZE);
+			strncat(local_gid, gid, GIDSIZE);
+
+			FinishPreparedTransaction(local_gid, false);
 			CommitTransactionCommand();
 
 			/* There were no BEGIN stmt for ROLLBACK PREPARED */
 			replorigin_session_origin_timestamp = commit_time;
 			replorigin_session_origin_lsn = commit_lsn;
+
 			break;
 		}
 		default:
diff --git a/contrib/pglogical/sql/twophase.sql b/contrib/pglogical/sql/twophase.sql
@@ -4,10 +4,11 @@ SELECT * FROM pglogical_regress_variables()
 
 \c :provider_dsn
 SELECT pglogical.replicate_ddl_command($$
-CREATE TABLE public.test2pc_tbl(id serial primary key, value int);
+	CREATE TABLE public.test2pc_tbl(id serial primary key, value int);
 $$);
-SELECT * FROM pglogical.replication_set_add_all_tables('default', '{public}');
-
+-- SELECT * FROM pglogical.replication_set_add_all_tables('default', '{public}');
+SELECT * FROM pglogical.replication_set_add_table('default', 'test2pc_tbl');
+SELECT pg_xlog_wait_remote_apply(pg_current_xlog_location(), 0);
 
 -- Check that prapeared state is visible on slave and data available after commit
 BEGIN;
diff --git a/reinit.sh b/reinit.sh
@@ -75,49 +75,75 @@ cd ../..
 
 pkill -9 postgres
 reinit_master
-# reinit_master2
+reinit_master2
 
-# ./install/bin/psql <<SQL
-# 	CREATE EXTENSION pglogical;
-# 	SELECT pglogical.create_node(
-# 		node_name := 'provider1',
-# 		dsn := 'port=5432 dbname=stas'
-# 	);
-# 	SELECT pglogical.replication_set_add_all_tables('default', ARRAY['public']);
-# SQL
-
-# ./install/bin/psql -p 5433 <<SQL
-# 	CREATE EXTENSION pglogical;
-# 	SELECT pglogical.create_node(
-# 		node_name := 'subscriber1',
-# 		dsn := 'port=5433 dbname=stas'
-# 	);
-# 	SELECT pglogical.create_subscription(
-# 		subscription_name := 'subscription1',
-# 		provider_dsn := 'port=5432 dbname=stas'
-# 	);
-# SQL
+./install/bin/psql <<SQL
+	CREATE EXTENSION pglogical;
+	SELECT pglogical.create_node(
+		node_name := 'provider1',
+		dsn := 'port=5432 dbname=stas'
+	);
+	SELECT pglogical.replication_set_add_all_tables('default', ARRAY['public']);
+SQL
 
-./install/bin/psql -c "SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'pglogical_output');"
+./install/bin/psql -p 5433 <<SQL
+	CREATE EXTENSION pglogical;
+	SELECT pglogical.create_node(
+		node_name := 'subscriber1',
+		dsn := 'port=5433 dbname=stas'
+	);
+	SELECT pglogical.create_subscription(
+		subscription_name := 'subscription1',
+		provider_dsn := 'port=5432 dbname=stas'
+	);
+SQL
 
 ./install/bin/psql <<SQL
-	begin;
-	insert into t values (42);
-	prepare transaction 'hellyeah';
-	rollback prepared 'hellyeah';
+select pg_sleep(2);
+begin;
+insert into t values (42);
+prepare transaction 'hellyeah';
+select * from pg_current_xlog_location();
+select pg_stat_get_wal_senders();
+select * from pg_current_xlog_location();
+--select pg_sleep(3);
+SQL
+
+./install/bin/psql -p 5433 <<SQL
+select * from pg_prepared_xacts;
+--select * from t;
 SQL
 
 ./install/bin/psql <<SQL
-SELECT * FROM pg_logical_slot_peek_changes('regression_slot',
-NULL, NULL,
-'expected_encoding', 'UTF8',
-'min_proto_version', '1',
-'max_proto_version', '1',
-'startup_params_format', '1',
-'proto_format', 'json',
-'no_txinfo', 't');
+commit prepared 'hellyeah';
+select pg_sleep(1);
 SQL
 
+./install/bin/psql -p 5433 <<SQL
+select * from pg_prepared_xacts;
+select * from t;
+SQL
+
+# ./install/bin/psql -c "SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'pglogical_output');"
+
+# ./install/bin/psql <<SQL
+# 	--begin;
+# 	insert into t values (42);
+# 	--prepare transaction 'hellyeah';
+# 	--commit prepared 'hellyeah';
+# SQL
+
+# ./install/bin/psql <<SQL
+# SELECT * FROM pg_logical_slot_peek_changes('regression_slot',
+# NULL, NULL,
+# 'expected_encoding', 'UTF8',
+# 'min_proto_version', '1',
+# 'max_proto_version', '1',
+# 'startup_params_format', '1',
+# 'proto_format', 'json',
+# 'no_txinfo', 't');
+# SQL
+
 
 
 
diff --git a/src/backend/access/rmgrdesc/xactdesc.c b/src/backend/access/rmgrdesc/xactdesc.c
@@ -141,6 +141,16 @@ ParseAbortRecord(uint8 info, xl_xact_abort *xlrec, xl_xact_parsed_abort *parsed)
 		data += sizeof(xl_xact_xinfo);
 	}
 
+	if (parsed->xinfo & XACT_XINFO_HAS_DBINFO)
+	{
+		xl_xact_dbinfo *xl_dbinfo = (xl_xact_dbinfo *) data;
+
+		parsed->dbId = xl_dbinfo->dbId;
+		parsed->tsId = xl_dbinfo->tsId;
+
+		data += sizeof(xl_xact_dbinfo);
+	}
+
 	if (parsed->xinfo & XACT_XINFO_HAS_SUBXACTS)
 	{
 		xl_xact_subxacts *xl_subxacts = (xl_xact_subxacts *) data;
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
@@ -5231,6 +5231,7 @@ XactLogAbortRecord(TimestampTz abort_time,
 	xl_xact_subxacts xl_subxacts;
 	xl_xact_relfilenodes xl_relfilenodes;
 	xl_xact_twophase xl_twophase;
+	xl_xact_dbinfo xl_dbinfo;
 
 	uint8		info;
 
@@ -5268,6 +5269,13 @@ XactLogAbortRecord(TimestampTz abort_time,
 		xl_twophase.gidlen = strlen(twophase_gid) + 1; /* Include '\0' */
 	}
 
+	if (TransactionIdIsValid(twophase_xid) && XLogLogicalInfoActive())
+	{
+		xl_xinfo.xinfo |= XACT_XINFO_HAS_DBINFO;
+		xl_dbinfo.dbId = MyDatabaseId;
+		xl_dbinfo.tsId = MyDatabaseTableSpace;
+	}
+
 	if (xl_xinfo.xinfo != 0)
 		info |= XLOG_XACT_HAS_INFO;
 
@@ -5280,6 +5288,9 @@ XactLogAbortRecord(TimestampTz abort_time,
 	if (xl_xinfo.xinfo != 0)
 		XLogRegisterData((char *) (&xl_xinfo), sizeof(xl_xinfo));
 
+	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
+		XLogRegisterData((char *) (&xl_dbinfo), sizeof(xl_dbinfo));
+
 	if (xl_xinfo.xinfo & XACT_XINFO_HAS_SUBXACTS)
 	{
 		XLogRegisterData((char *) (&xl_subxacts),
@@ -5300,7 +5311,8 @@ XactLogAbortRecord(TimestampTz abort_time,
 	{
 		XLogRegisterData((char *) (&xl_twophase), MinSizeOfXactTwophase);
 		XLogRegisterData((char *) twophase_gid, xl_twophase.gidlen);
-	}
+	}	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
+		XLogRegisterData((char *) (&xl_dbinfo), sizeof(xl_dbinfo));
 
 	return XLogInsert(RM_XACT_ID, info);
 }
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
@@ -461,16 +461,6 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		commit_time = parsed->origin_timestamp;
 	}
 
-	/*
-	 * If that is COMMIT PREPARED than send that to callbacks.
-	 */
-	if (TransactionIdIsValid(parsed->twophase_xid)) {
-		strcpy(ctx->reorder->gid, parsed->twophase_gid);
-		ReorderBufferCommitBareXact(ctx->reorder, xid, buf->origptr, buf->endptr,
-							commit_time, origin_id, origin_lsn);
-		return;
-	}
-
 	/*
 	 * Process invalidation messages, even if we're not interested in the
 	 * transaction's contents, since the various caches need to always be
@@ -531,9 +521,19 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 								 buf->origptr, buf->endptr);
 	}
 
-	/* replay actions of all transaction + subtransactions in order */
-	ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
-						commit_time, origin_id, origin_lsn);
+	if (TransactionIdIsValid(parsed->twophase_xid)) {
+		/*
+		 * We are processing COMMIT PREPARED and know that reorder buffer is
+		 * empty. So we can skip use shortcut for coomiting bare xact.
+		 */
+		strcpy(ctx->reorder->gid, parsed->twophase_gid);
+		ReorderBufferCommitBareXact(ctx->reorder, xid, buf->origptr, buf->endptr,
+							commit_time, origin_id, origin_lsn);
+	} else {
+		/* replay actions of all transaction + subtransactions in order */
+		ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
+							commit_time, origin_id, origin_lsn);
+	}
 }
 
 static void
@@ -603,8 +603,18 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	/*
 	 * If that is ROLLBACK PREPARED than send that to callbacks.
 	 */
+
 	if (TransactionIdIsValid(parsed->twophase_xid)) {
+		fprintf(stderr, "(!) db_id=%d, gid=%s\n", parsed->dbId, parsed->twophase_gid);
+	}
+
+	if (TransactionIdIsValid(parsed->twophase_xid)
+			&& (parsed->dbId == ctx->slot->data.database)) {
+
 		strcpy(ctx->reorder->gid, parsed->twophase_gid);
+
+		fprintf(stderr, "(!)	aborting: db_id=%d, gid=%s\n", parsed->dbId, parsed->twophase_gid);
+
 		ReorderBufferCommitBareXact(ctx->reorder, xid, buf->origptr, buf->endptr,
 							commit_time, origin_id, origin_lsn);
 		return;
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
@@ -314,6 +314,9 @@ typedef struct xl_xact_parsed_prepare
 
 typedef struct xl_xact_parsed_abort
 {
+	Oid			dbId;
+	Oid			tsId;
+
 	TimestampTz xact_time;
 	uint32		xinfo;
 

Original file line number	Diff line number	Diff line change
`@@ -314,6 +314,9 @@ typedef struct xl_xact_parsed_prepare`
`314`	`314`
`315`	`315`	`typedef struct xl_xact_parsed_abort`
`316`	`316`	`{`
	`317`	`+ Oid dbId;`
	`318`	`+ Oid tsId;`
	`319`	`+`
`317`	`320`	`TimestampTz xact_time;`
`318`	`321`	`uint32 xinfo;`
`319`	`322`