Trace logical decoding

Author: knizhnik

arbiter.c

@@ -129,15 +129,15 @@ static char const* const messageText[] =
 
 static BackgroundWorker MtmSender = {
 	"mtm-sender",
-	BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION, 
+	BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION, 
 	BgWorkerStart_ConsistentState,
 	MULTIMASTER_BGW_RESTART_TIMEOUT,
 	MtmTransSender
 };
 
 static BackgroundWorker MtmRecevier = {
 	"mtm-receiver",
-	BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION, 
+	BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION, 
 	BgWorkerStart_ConsistentState,
 	MULTIMASTER_BGW_RESTART_TIMEOUT,
 	MtmTransReceiver

multimaster.c

@@ -990,7 +990,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 				/* Assert(ts->status == TRANSACTION_STATUS_UNKNOWN); */
 				Assert(ts->status == TRANSACTION_STATUS_UNKNOWN 
 					   || (ts->status == TRANSACTION_STATUS_IN_PROGRESS && Mtm->status == MTM_RECOVERY)); /* ??? Why there is commit without prepare */
-				if (x->csn > ts->csn) {
+				if (x->csn > ts->csn || Mtm->status == MTM_RECOVERY) {
 					ts->csn = x->csn;
 					MtmSyncClock(ts->csn);
 				}
@@ -1514,7 +1514,7 @@ bool MtmRefreshClusterStatus(bool nowait)
 					MtmWakeUpBackend(ts);
 				}
 #if 0
-			} else if (TransactionIdIsValid(ts->gtid.xid) && BIT_CHECK(disabled, ts->gtid.node-1)) { // coordinator of transaction is on disabled node
+			} else if (TransactionIdIsValid(ts->gtid.xid) && BIT_CHECK(disabled, ts->gtid.node-1)) { /* coordinator of transaction is on disabled node */
 				if (ts->gid[0]) { 
 					if (ts->status == TRANSACTION_STATUS_UNKNOWN || ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
 						MTM_LOG1("%d: Abort trasaction %s because its coordinator is at disabled node %d", MyProcPid, ts->gid, ts->gtid.node);
@@ -1604,12 +1604,14 @@ void MtmOnNodeDisconnect(int nodeId)
 						MtmAbortTransaction(ts);
 						MtmWakeUpBackend(ts);
 					}
-				} else if (TransactionIdIsValid(ts->gtid.xid) && ts->gtid.node == nodeId) { //coordinator of transaction is on disabled node
+#if 0
+				} else if (TransactionIdIsValid(ts->gtid.xid) && ts->gtid.node == nodeId) { /* coordinator of transaction is on disabled node */
 					if (ts->gid[0] && ts->status != TRANSACTION_STATUS_ABORTED) {
 						MtmAbortTransaction(ts);
 						MtmTx.status = TRANSACTION_STATUS_ABORTED; /* prevent recursive invocation of MtmAbortPreparedTransaction */
 						FinishPreparedTransaction(ts->gid, false);
 					}
+#endif
 				}
 			}
 		}
@@ -2284,11 +2286,11 @@ _PG_init(void)
 		NULL,
 		&MtmConnStrs,
 		"",
-		PGC_BACKEND, // context
-		0, // flags,
-		NULL, // GucStringCheckHook check_hook,
-		NULL, // GucStringAssignHook assign_hook,
-		NULL // GucShowHook show_hook
+		PGC_BACKEND, /* context */
+		0,           /* flags */
+		NULL,        /* GucStringCheckHook check_hook */
+		NULL,        /* GucStringAssignHook assign_hook */
+		NULL         /* GucShowHook show_hook */
 	);
 
 	DefineCustomIntVariable(
@@ -3348,13 +3350,19 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		/* Save GUC context for consequent DDL execution */
 		case T_DiscardStmt:
 			{
-				DiscardStmt *stmt = (DiscardStmt *) parsetree;
-				skipCommand = stmt->target == DISCARD_TEMP;
+				/*
+				 * DiscardStmt *stmt = (DiscardStmt *) parsetree;
+				 * skipCommand = stmt->target == DISCARD_TEMP;
+				 */
 
 				if (!IsTransactionBlock())
 				{
-					skipCommand = true;
-					MtmGUCBufferAppend(queryString);
+					/*
+					 * XXX: move allocation somewhere to backend startup and check
+					 * where buffer is empty in send routines.
+					 */
+					MtmGUCBufferAllocated = false;
+					pfree(MtmGUCBuffer);
 				}
 			}
 			break;
@@ -3364,7 +3372,38 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 
 				/* Prevent SET TRANSACTION from replication */
 				if (stmt->kind == VAR_SET_MULTI)
-					skipCommand = true;
+					break;
+
+				if (!MtmGUCBufferAllocated)
+				{
+					MemoryContext oldcontext;
+
+					oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+					MtmGUCBuffer = makeStringInfo();
+					MemoryContextSwitchTo(oldcontext);
+					MtmGUCBufferAllocated = true;
+				}
+
+				appendStringInfoString(MtmGUCBuffer, queryString);
+
+				/* sometimes there is no ';' char at the end. */
+				appendStringInfoString(MtmGUCBuffer, ";");
+			}
+			break;
+		case T_CreateStmt:
+			{
+				/* Do not replicate temp tables */
+				CreateStmt *stmt = (CreateStmt *) parsetree;
+				skipCommand = stmt->relation->relpersistence == RELPERSISTENCE_TEMP ||
+					(stmt->relation->schemaname && strcmp(stmt->relation->schemaname, "pg_temp") == 0);
+			}
+			break;
+		case T_IndexStmt:
+			{
+				Oid			relid;
+				Relation	rel;
+				IndexStmt *stmt = (IndexStmt *) parsetree;
+				bool		isTopLevel = (context == PROCESS_UTILITY_TOPLEVEL);
 
 				if (stmt->kind == VAR_RESET && strcmp(stmt->name, "session_authorization") == 0)
 					MtmGUCBufferClear();

pglogical_apply.c

@@ -59,7 +59,7 @@ typedef struct TupleData
 	bool		changed[MaxTupleAttributeNumber];
 } TupleData;
 
-static bool inside_tx = false;
+static int MtmTransactionRecords;
 
 static Relation read_rel(StringInfo s, LOCKMODE mode);
 static void read_tuple_parts(StringInfo s, Relation rel, TupleData *tup);
@@ -543,11 +543,16 @@ process_remote_commit(StringInfo in)
 	csn_t       csn;
 	const char *gid = NULL;	
 	XLogRecPtr  end_lsn;
-
+	int         n_records;
 	/* read flags */
 	flags = pq_getmsgbyte(in);
 	MtmReplicationNodeId = pq_getmsgbyte(in);
 
+	n_records = pq_getmsgint(in, 4);
+	if (MtmTransactionRecords != n_records) { 
+		elog(ERROR, "Transaction %d flags %d contains %d records instead of %d", MtmGetCurrentTransactionId(), flags, MtmTransactionRecords, n_records);
+	}
+
 	/* read fields */
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
 	end_lsn = pq_getmsgint64(in); /* end_lsn */
@@ -654,6 +659,8 @@ process_remote_insert(StringInfo s, Relation rel)
 	ScanKey	*index_keys;
 	int	i;
 
+	MtmTransactionRecords += 1;
+
 	estate = create_rel_estate(rel);
 	newslot = ExecInitExtraTupleSlot(estate);
 	oldslot = ExecInitExtraTupleSlot(estate);
@@ -752,6 +759,8 @@ process_remote_update(StringInfo s, Relation rel)
 	ScanKeyData skey[INDEX_MAX_KEYS];
 	HeapTuple	remote_tuple = NULL;
 
+	MtmTransactionRecords += 1;
+
 	action = pq_getmsgbyte(s);
 
 	/* old key present, identifying key changed */
@@ -869,6 +878,8 @@ process_remote_delete(StringInfo s, Relation rel)
 	ScanKeyData skey[INDEX_MAX_KEYS];
 	bool		found_old;
 
+	MtmTransactionRecords += 1;
+
 	estate = create_rel_estate(rel);
 	oldslot = ExecInitExtraTupleSlot(estate);
 	ExecSetSlotDescriptor(oldslot, RelationGetDescr(rel));
@@ -956,6 +967,7 @@ void MtmExecutor(int id, void* work, size_t size)
     }
     MemoryContextSwitchTo(ApplyContext);
 	replorigin_session_origin = InvalidRepOriginId;
+	MtmTransactionRecords = 0;
     PG_TRY();
     {    
         while (true) { 

pglogical_proto.c

@@ -39,6 +39,7 @@
 #include "pglogical_relid_map.h"
 
 static bool MtmIsFilteredTxn;
+static int  MtmTransactionRecords;
 
 static void pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel);
 
@@ -106,7 +107,8 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 {
 	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
-	MTM_LOG2("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d", MyProcPid, txn->xid, MtmReplicationNodeId, csn, isRecovery);
+	MTM_LOG1("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
+			 MyProcPid, txn->xid, MtmReplicationNodeId, csn, isRecovery, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
 
 	if (csn == INVALID_CSN && !isRecovery) { 
 		MtmIsFilteredTxn = true;
@@ -116,6 +118,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
 		pq_sendint64(out, csn);
 		MtmIsFilteredTxn = false;
+		MtmTransactionRecords = 0;
 	}
 }
 
@@ -137,6 +140,10 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 {
     uint8 flags = 0;
 
+	MTM_LOG1("%d: pglogical_write_commit XID=%d node=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
+			 MyProcPid, txn->xid, MtmReplicationNodeId, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
+
+
     if (txn->xact_action == XLOG_XACT_COMMIT) 
     	flags = PGLOGICAL_COMMIT;
 	else if (txn->xact_action == XLOG_XACT_PREPARE)
@@ -150,6 +157,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 
 	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) { 
 		if (MtmIsFilteredTxn) { 
+			Assert(MtmTransactionRecords == 0);
 			return;
 		}
 	} else { 
@@ -161,6 +169,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 		 */
 		if (csn == INVALID_CSN && !isRecovery) 
 		{
+			Assert(MtmTransactionRecords == 0);
 			return;
 		}
 		if (MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn)) { 
@@ -176,18 +185,23 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     pq_sendbyte(out, flags);
     pq_sendbyte(out, MtmNodeId);
 
+	Assert(txn->xact_action != XLOG_XACT_PREPARE || txn->xid < 1000 || MtmTransactionRecords >= 2);
+	pq_sendint(out, MtmTransactionRecords, 4);
+
     /* send fixed fields */
     pq_sendint64(out, commit_lsn);
     pq_sendint64(out, txn->end_lsn);
     pq_sendint64(out, txn->commit_time);
 
 	if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED) { 
+		Assert(MtmTransactionRecords == 0);
 		pq_sendint64(out, MtmGetTransactionCSN(txn->xid));
 	}
     if (txn->xact_action != XLOG_XACT_COMMIT) { 
     	pq_sendstring(out, txn->gid);
 	}
 
+	MtmTransactionRecords = 0;
 	MTM_TXTRACE(txn, "pglogical_write_commit Finish");
 }
 
@@ -199,6 +213,7 @@ pglogical_write_insert(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple newtuple)
 {
     if (!MtmIsFilteredTxn) { 
+		MtmTransactionRecords += 1;
 		pq_sendbyte(out, 'I');		/* action INSERT */
 		pglogical_write_tuple(out, data, rel, newtuple);
 	}
@@ -212,6 +227,11 @@ pglogical_write_update(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
 {
     if (!MtmIsFilteredTxn) { 
+		MtmTransactionRecords += 1;
+
+		MTM_LOG1("%d: pglogical_write_update confirmed_flush=%lx", MyProcPid, MyReplicationSlot->data.confirmed_flush);
+
+
 		pq_sendbyte(out, 'U');		/* action UPDATE */
 		/* FIXME support whole tuple (O tuple type) */
 		if (oldtuple != NULL)
@@ -233,6 +253,7 @@ pglogical_write_delete(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple oldtuple)
 {
     if (!MtmIsFilteredTxn) {
+		MtmTransactionRecords += 1;
 		pq_sendbyte(out, 'D');		/* action DELETE */
 		pglogical_write_tuple(out, data, rel, oldtuple);
 	}

tests2/docker-entrypoint.sh

@@ -6,7 +6,7 @@ if [ "${1:0:1}" = '-' ]; then
 	set -- postgres "$@"
 fi
 
-#echo "/pg/%p.%s.%c.%P.core" | sudo tee /proc/sys/kernel/core_pattern
+echo "/pg/%p.%s.%c.%P.core" | sudo tee /proc/sys/kernel/core_pattern
 
 if [ "$1" = 'postgres' ]; then
 	mkdir -p "$PGDATA"

tests2/test_recovery.py

@@ -12,7 +12,7 @@ def setUpClass(self):
             "dbname=postgres user=postgres host=127.0.0.1",
             "dbname=postgres user=postgres host=127.0.0.1 port=5433",
             "dbname=postgres user=postgres host=127.0.0.1 port=5434"
-        ], n_accounts=100000)
+        ], n_accounts=1000)
         self.client.bgrun()
         time.sleep(5)