Trace logical decoding

Author: knizhnik

contrib/mmts/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

contrib/mmts/multimaster.c

@@ -981,7 +981,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);
 				}
@@ -1505,7 +1505,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);
@@ -1595,12 +1595,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
 				}
 			}
 		}
@@ -2213,11 +2215,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(
@@ -3249,15 +3251,19 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			break;
 		case T_DiscardStmt:
 			{
-				//DiscardStmt *stmt = (DiscardStmt *) parsetree;
-				//skipCommand = stmt->target == DISCARD_TEMP;
+				/*
+				 * DiscardStmt *stmt = (DiscardStmt *) parsetree;
+				 * skipCommand = stmt->target == DISCARD_TEMP;
+				 */
 
 				skipCommand = true;
 
 				if (MtmGUCBufferAllocated)
 				{
-					// XXX: move allocation somewhere to backend startup and check
-					// where buffer is empty in send routines.
+					/*
+					 * XXX: move allocation somewhere to backend startup and check
+					 * where buffer is empty in send routines.
+					 */
 					MtmGUCBufferAllocated = false;
 					pfree(MtmGUCBuffer);
 				}
@@ -3286,7 +3292,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 
 				appendStringInfoString(MtmGUCBuffer, queryString);
 
-				// sometimes there is no ';' char at the end.
+				/* sometimes there is no ';' char at the end. */
 				appendStringInfoString(MtmGUCBuffer, ";");
 			}
 			break;

contrib/mmts/pglogical_apply.c

@@ -59,6 +59,8 @@ typedef struct TupleData
 	bool		changed[MaxTupleAttributeNumber];
 } TupleData;
 
+static int MtmTransactionRecords;
+
 static Relation read_rel(StringInfo s, LOCKMODE mode);
 static void read_tuple_parts(StringInfo s, Relation rel, TupleData *tup);
 static EState* create_rel_estate(Relation rel);
@@ -509,11 +511,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 */
@@ -620,6 +627,8 @@ process_remote_insert(StringInfo s, Relation rel)
 	char* relname = RelationGetRelationName(rel);
 	int	i;
 
+	MtmTransactionRecords += 1;
+
 	estate = create_rel_estate(rel);
 	newslot = ExecInitExtraTupleSlot(estate);
 	oldslot = ExecInitExtraTupleSlot(estate);
@@ -734,6 +743,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 */
@@ -851,6 +862,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));
@@ -938,6 +951,7 @@ void MtmExecutor(int id, void* work, size_t size)
     }
     MemoryContextSwitchTo(ApplyContext);
 	replorigin_session_origin = InvalidRepOriginId;
+	MtmTransactionRecords = 0;
     PG_TRY();
     {    
         while (true) { 

contrib/mmts/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;
 	}
 }
 
@@ -128,6 +131,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)
@@ -141,6 +148,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 
 	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) { 
 		if (MtmIsFilteredTxn) { 
+			Assert(MtmTransactionRecords == 0);
 			return;
 		}
 	} else { 
@@ -152,6 +160,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 		 */
 		if (csn == INVALID_CSN && !isRecovery) 
 		{
+			Assert(MtmTransactionRecords == 0);
 			return;
 		}
 		if (MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn)) { 
@@ -167,18 +176,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");
 }
 
@@ -190,6 +204,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);
 	}
@@ -203,6 +218,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)
@@ -224,6 +244,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);
 	}

contrib/mmts/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"

contrib/mmts/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)
 

src/backend/replication/logical/decode.c

@@ -28,6 +28,7 @@
 #include <time.h>
 
 #include "postgres.h"
+#include "miscadmin.h"
 
 #include "access/heapam.h"
 #include "access/heapam_xlog.h"
@@ -423,6 +424,8 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 	ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
 
+	elog(LOG, "%d: DecodeHeapOp XID=%d, info=%d", MyProcPid, xid, info);
+
 	/* no point in doing anything yet */
 	if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
 		return;
@@ -813,16 +816,20 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	char	   *data;
 	RelFileNode target_node;
 
+	elog(LOG, "%d: DecodeUpdate XID=%d", MyProcPid);
+
 	xlrec = (xl_heap_update *) XLogRecGetData(r);
 
 	/* only interested in our database */
 	XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
-	if (target_node.dbNode != ctx->slot->data.database)
+	if (target_node.dbNode != ctx->slot->data.database) 
 		return;
 
 	/* output plugin doesn't look for this origin, no need to queue */
-	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
+	if (FilterByOrigin(ctx, XLogRecGetOrigin(r))) {
+		elog(LOG, "%d: DecodeUpdate XID=%d filtered by origin %lx", MyProcPid, XLogRecGetOrigin(r));
 		return;
+	}
 
 	change = ReorderBufferGetChange(ctx->reorder);
 	change->action = REORDER_BUFFER_CHANGE_UPDATE;