filter XLOG records produced by DLL statements fixes #16

Author: kelvich

contrib/mmts/multimaster.c

@@ -3747,6 +3747,11 @@ static bool MtmProcessDDLCommand(char const* queryString)
 	return false;
 }
 
+static void MtmFinishDDLCommand()
+{
+	LogLogicalMessage("E", "", 1, true);
+}
+
 void MtmUpdateLockGraph(int nodeId, void const* messageBody, int messageSize)
 {
 	int allocated;
@@ -3768,6 +3773,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 DestReceiver *dest, char *completionTag)
 {
 	bool skipCommand = false;
+	bool executed = false;
 
 	MTM_LOG3("%d: Process utility statement %s", MyProcPid, queryString);
 	switch (nodeTag(parsetree))
@@ -3912,6 +3918,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		if (!skipCommand && !MtmTx.isReplicated && (MtmUtilityProcessedInXid == InvalidTransactionId)) {
 			MtmUtilityProcessedInXid = GetCurrentTransactionId();
 			MtmProcessDDLCommand(queryString);
+			executed = true;
 		}
 	}
 
@@ -3933,6 +3940,10 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		MtmTx.snapshot = INVALID_CSN;
 	}
 
+	if (executed)
+	{
+		MtmFinishDDLCommand();
+	}
 }
 
 
@@ -3967,12 +3978,6 @@ MtmExecutorFinish(QueryDesc *queryDesc)
         }
     }
 
-	// if (MyXactAccessedRel)
-	// {
-	// 	MTM_LOG1("MtmTx.containsDML = true");
-	// 	MtmTx.containsDML = true;
-	// }
-
     if (PreviousExecutorFinishHook != NULL)
     {
         PreviousExecutorFinishHook(queryDesc);

contrib/mmts/pglogical_apply.c

@@ -59,8 +59,6 @@ typedef struct TupleData
 	bool		changed[MaxTupleAttributeNumber];
 } TupleData;
 
-static bool inside_tx = false;
-
 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);
@@ -353,8 +351,6 @@ process_remote_begin(StringInfo s)
 	StartTransactionCommand();
     MtmJoinTransaction(&gtid, snapshot);
 
-	inside_tx = true;
-
 	return true;
 }
 
@@ -364,11 +360,6 @@ process_remote_transactional_message(StringInfo s)
 	int rc;
 	int messageSize = pq_getmsgint(s, 4);
 	char const* stmt = pq_getmsgbytes(s, messageSize);
-	if (!inside_tx)
-	{
-		MTM_LOG1("%d: Ignoring utility statement %s", MyProcPid, stmt);
-		return;
-	}
 
 	MTM_LOG1("%d: Executing utility statement %s", MyProcPid, stmt);
 	SPI_connect();	
@@ -667,7 +658,6 @@ process_remote_commit(StringInfo in)
 	if (flags & PGLOGICAL_CAUGHT_UP) {
 		MtmRecoveryCompleted();
 	}
-	inside_tx = false;
 }
 
 static void
@@ -992,7 +982,7 @@ void MtmExecutor(void* work, size_t size)
     {    
         while (true) { 
             char action = pq_getmsgbyte(&s);
-            MTM_LOG3("%d: REMOTE process action %c", MyProcPid, action);
+            MTM_LOG1("%d: REMOTE process action %c", MyProcPid, action);
 #if 0
 			if (Mtm->status == MTM_RECOVERY) { 
 				MTM_LOG1("Replay action %c[%x]",   action, s.data[s.cursor]);
@@ -1054,6 +1044,7 @@ void MtmExecutor(void* work, size_t size)
 				continue;
 			}
 			case 'G':
+			case 'E':
 			{
 				process_remote_transactional_message(&s);
 				continue;

contrib/mmts/pglogical_proto.c

@@ -38,8 +38,9 @@
 #include "multimaster.h"
 #include "pglogical_relid_map.h"
 
-static bool MtmIsFilteredTxn;
-static int  MtmTransactionRecords;
+static int MtmTransactionRecords;
+static TransactionId MtmCurrentXid;
+static bool DDLInProress = false;
 
 static void pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel);
 
@@ -74,10 +75,17 @@ pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel)
 	const char *relname;
 	uint8		relnamelen;
 	Oid         relid;
-    if (MtmIsFilteredTxn) { 
+
+	if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN) {
+		MTM_LOG1("%d: pglogical_write_message filtered", MyProcPid);
 		return;
 	}
-	
+
+	if (DDLInProress) {
+		MTM_LOG1("%d: pglogical_write_message filtered DDLInProress", MyProcPid);
+		return;
+	}
+
 	relid = RelationGetRelid(rel);
 	pq_sendbyte(out, 'R');		/* sending RELATION */	
 	pq_sendint(out, relid, sizeof relid); /* use Oid as relation identifier */
@@ -107,36 +115,42 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 {
 	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
+
+	MtmCurrentXid = txn->xid;
+
 	MTM_LOG3("%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 (!isRecovery && csn == INVALID_CSN) { 
-		MtmIsFilteredTxn = true;
-		MTM_LOG3("%d: pglogical_write_begin XID=%d filtered", MyProcPid, txn->xid);
-	} else {
-		MTM_LOG3("%d: pglogical_write_begin XID=%d sent", MyProcPid, txn->xid);
-		MtmIsFilteredTxn = false;
-		pq_sendbyte(out, 'B');		/* BEGIN */
-		pq_sendint(out, MtmNodeId, 4);
-		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
-		pq_sendint64(out, csn);
-		MtmTransactionRecords = 0;
-	}
+
+	MTM_LOG3("%d: pglogical_write_begin XID=%d sent", MyProcPid, txn->xid);
+	pq_sendbyte(out, 'B');		/* BEGIN */
+	pq_sendint(out, MtmNodeId, 4);
+	pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
+	pq_sendint64(out, csn);
+	MtmTransactionRecords = 0;
 }
 
 static void
 pglogical_write_message(StringInfo out,
 						const char *prefix, Size sz, const char *message)
 {
-	if (*prefix == 'L') { 
+	if (*prefix == 'L')
+	{
 		MTM_LOG1("Send deadlock message to node %d", MtmReplicationNodeId);
-	} else { 
-		if (MtmIsFilteredTxn)
+	}
+	else if (*prefix == 'G')
+	{
+		if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN)
 		{
-			MTM_LOG3("%d: pglogical_write_message filtered", MyProcPid);
+			MTM_LOG1("%d: pglogical_write_message filtered", MyProcPid);
 			return;
 		}
+		DDLInProress = true;
 	}
+	else if (*prefix == 'E')
+	{
+		DDLInProress = false;
+	}
+
 	pq_sendbyte(out, *prefix);
 	pq_sendint(out, sz, 4);
 	pq_sendbytes(out, message, sz);
@@ -169,10 +183,10 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	Assert(flags != PGLOGICAL_COMMIT_PREPARED || txn->xid < 1000 || MtmTransactionRecords != 1);
 
 	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) { 
-		if (MtmIsFilteredTxn) { 
-			Assert(MtmTransactionRecords == 0);
-			return;
-		}
+		// if (MtmIsFilteredTxn) { 
+			// Assert(MtmTransactionRecords == 0);
+			// return;
+		// }
 	} else { 
 		csn_t csn = MtmTransactionSnapshot(txn->xid);
 		bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
@@ -242,11 +256,20 @@ static void
 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);
+	if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
+		MTM_LOG1("%d: pglogical_write_insert filtered", MyProcPid);
+		return;
+	}
+
+	if (DDLInProress) {
+		MTM_LOG1("%d: pglogical_write_insert filtered DDLInProress", MyProcPid);
+		return;
 	}
+
+	MtmTransactionRecords += 1;
+	pq_sendbyte(out, 'I');		/* action INSERT */
+	pglogical_write_tuple(out, data, rel, newtuple);
+
 }
 
 /*
@@ -256,23 +279,30 @@ static void
 pglogical_write_update(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
 {
-    if (!MtmIsFilteredTxn) { 
-		MtmTransactionRecords += 1;
+	if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
+		MTM_LOG1("%d: pglogical_write_update filtered", MyProcPid);
+		return;
+	}
 
-		MTM_LOG3("%d: pglogical_write_update confirmed_flush=%lx", MyProcPid, MyReplicationSlot->data.confirmed_flush);
+	if (DDLInProress) {
+		MTM_LOG1("%d: pglogical_write_update filtered DDLInProress", MyProcPid);
+		return;
+	}
 
+	MtmTransactionRecords += 1;
 
-		pq_sendbyte(out, 'U');		/* action UPDATE */
-		/* FIXME support whole tuple (O tuple type) */
-		if (oldtuple != NULL)
-		{
-			pq_sendbyte(out, 'K');	/* old key follows */
-			pglogical_write_tuple(out, data, rel, oldtuple);
-		}
-		
-		pq_sendbyte(out, 'N');		/* new tuple follows */
-		pglogical_write_tuple(out, data, rel, newtuple);
+	MTM_LOG3("%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)
+	{
+		pq_sendbyte(out, 'K');	/* old key follows */
+		pglogical_write_tuple(out, data, rel, oldtuple);
 	}
+
+	pq_sendbyte(out, 'N');		/* new tuple follows */
+	pglogical_write_tuple(out, data, rel, newtuple);
 }
 
 /*
@@ -282,11 +312,19 @@ static void
 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);
+	if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
+		MTM_LOG1("%d: pglogical_write_delete filtered", MyProcPid);
+		return;
+	}
+
+	if (DDLInProress) {
+		MTM_LOG1("%d: pglogical_write_delete filtered DDLInProress", MyProcPid);
+		return;
 	}
+
+	MtmTransactionRecords += 1;
+	pq_sendbyte(out, 'D');		/* action DELETE */
+	pglogical_write_tuple(out, data, rel, oldtuple);
 }
 
 /*
@@ -311,6 +349,16 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 	int			i;
 	uint16		nliveatts = 0;
 
+	if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
+		MTM_LOG1("%d: pglogical_write_tuple filtered", MyProcPid);
+		return;
+	}
+
+	if (DDLInProress) {
+		MTM_LOG1("%d: pglogical_write_tuple filtered DDLInProress", MyProcPid);
+		return;
+	}
+
 	desc = RelationGetDescr(rel);
 
 	pq_sendbyte(out, 'T');			/* sending TUPLE */

src/test/regress/serial_schedule

@@ -50,7 +50,7 @@ test: oidjoins
 test: type_sanity
 test: opr_sanity
 test: insert
-# test: insert_conflict # issue#17
+# test: insert_conflict
 test: create_function_1
 test: create_type
 test: create_table
@@ -108,7 +108,7 @@ test: privileges
 test: init_privs
 test: security_label
 test: collate
-# test: matview
+test: matview
 test: lock
 test: replica_identity
 # test: rowsecurity  # issue#20