Fix MtmAdjustOldestXid

Author: knizhnik

multimaster.c

@@ -517,9 +517,16 @@ MtmAdjustOldestXid(TransactionId xid)
 	MTM_LOG2("%d: MtmAdjustOldestXid(%d): snapshot=%ld, csn=%ld, status=%d", MyProcPid, xid, ts != NULL ? ts->snapshot : 0, ts != NULL ? ts->csn : 0, ts != NULL ? ts->status : -1);
 	Mtm->gcCount = 0;
 
+	//return FirstNormalTransactionId;
+
 	if (ts != NULL) { 
 		oldestSnapshot = ts->snapshot;
-		Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
+		Assert(oldestSnapshot != INVALID_CSN);
+		if (Mtm->nodes[MtmNodeId-1].oldestSnapshot < oldestSnapshot) { 
+			Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
+		} else {
+			oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
+		}
 		for (i = 0; i < Mtm->nAllNodes; i++) { 
 			if (!BIT_CHECK(Mtm->disabledNodeMask, i)
 				&& Mtm->nodes[i].oldestSnapshot < oldestSnapshot) 
@@ -532,9 +539,11 @@ MtmAdjustOldestXid(TransactionId xid)
 		for (ts = Mtm->transListHead; 
 			 ts != NULL 
 				 && ts->csn < oldestSnapshot
-				 && TransactionIdPrecedes(ts->xid, xid)
+				 && TransactionIdPrecedes(ts->xid, xid);
+/*
 				 && (ts->status == TRANSACTION_STATUS_COMMITTED ||
 					 ts->status == TRANSACTION_STATUS_ABORTED);
+*/
 			 prev = ts, ts = ts->next) 
 		{ 
 			if (prev != NULL) { 
@@ -547,9 +556,10 @@ MtmAdjustOldestXid(TransactionId xid)
 	if (MtmUseDtm) 
 	{ 
 		if (prev != NULL) { 
+			MTM_LOG1("%d: MtmAdjustOldestXid: oldestXid=%d, prev->xid=%d, prev->status=%d, prev->snapshot=%ld, ts->xid=%d, ts->status=%d, ts->snapshot=%ld, oldestSnapshot=%ld", 
+					 MyProcPid, xid, prev->xid, prev->status, prev->snapshot, (ts ? ts->xid : 0), (ts ? ts->status : -1), (ts ? ts->snapshot : -1), oldestSnapshot);
 			Mtm->transListHead = prev;
 			Mtm->oldestXid = xid = prev->xid;            
-			MTM_LOG2("%d: MtmAdjustOldestXid: oldestXid=%d, oldestSnapshot=%ld", MyProcPid, xid, oldestSnapshot);
 		} else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) {  
 			xid = Mtm->oldestXid;
 		}

pglogical_output.c

@@ -384,30 +384,31 @@ pg_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
 	send_replication_origin &= txn->origin_id != InvalidRepOriginId;
 
 	OutputPluginPrepareWrite(ctx, !send_replication_origin);
-	data->api->write_begin(ctx->out, data, txn);
+	if (data->api) { 
+		data->api->write_begin(ctx->out, data, txn);
 
-	if (send_replication_origin)
-	{
-		char *origin;
-
-		/* Message boundary */
-		OutputPluginWrite(ctx, false);
-		OutputPluginPrepareWrite(ctx, true);
-
-		/*
-		 * XXX: which behaviour we want here?
-		 *
-		 * Alternatives:
-		 *  - don't send origin message if origin name not found
-		 *    (that's what we do now)
-		 *  - throw error - that will break replication, not good
-		 *  - send some special "unknown" origin
-		 */
-		if (data->api->write_origin &&
-			replorigin_by_oid(txn->origin_id, true, &origin))
+		if (send_replication_origin)
+		{
+			char *origin;
+			
+			/* Message boundary */
+			OutputPluginWrite(ctx, false);
+			OutputPluginPrepareWrite(ctx, true);
+			
+			/*
+			 * XXX: which behaviour we want here?
+			 *
+			 * Alternatives:
+			 *  - don't send origin message if origin name not found
+			 *    (that's what we do now)
+			 *  - throw error - that will break replication, not good
+			 *  - send some special "unknown" origin
+			 */
+			if (data->api->write_origin &&
+				replorigin_by_oid(txn->origin_id, true, &origin))
 			data->api->write_origin(ctx->out, origin, txn->origin_lsn);
+		}
 	}
-
 	OutputPluginWrite(ctx, true);
 }
 
@@ -421,7 +422,9 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	PGLogicalOutputData* data = (PGLogicalOutputData*)ctx->output_plugin_private;
 
 	OutputPluginPrepareWrite(ctx, true);
-	data->api->write_commit(ctx->out, data, txn, commit_lsn);
+	if (data->api) { 
+		data->api->write_commit(ctx->out, data, txn, commit_lsn);
+	}
 	OutputPluginWrite(ctx, true);
 }
 
@@ -433,7 +436,7 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	MemoryContext old;
 
 	/* First check the table filter */
-	if (!call_row_filter_hook(data, txn, relation, change))
+	if (!call_row_filter_hook(data, txn, relation, change) || data->api == NULL)
 		return;
 
 	/* Avoid leaking memory by using and resetting our own context */
@@ -539,7 +542,9 @@ send_startup_message(LogicalDecodingContext *ctx,
 	 */
 
 	OutputPluginPrepareWrite(ctx, last_message);
-	data->api->write_startup_message(ctx->out, msg);
+	if (data->api) {
+		data->api->write_startup_message(ctx->out, msg);
+	}
 	OutputPluginWrite(ctx, last_message);
 
 	pfree(msg);

tests2/lib/bank_client.py

@@ -153,10 +153,13 @@ def check_total(self):
         def tx(conn, cur):
             cur.execute('select sum(amount) from bank_test')
             res = cur.fetchone()
+            total = res[0]
+            if total != 0:
+                cur.execute('select mtm.get_snapshot()')
+                res = cur.fetchone()
+                print("Isolation error, total = %d, node = %d, snapshot = %d" % (total,self.node_id,res[0]))
+                #raise BaseException
             conn.commit()
-            if res[0] != 0:
-                print("Isolation error, total = %d, node = %d" % (res[0],self.node_id))
-                raise BaseException
 
         self.exec_tx('total', tx)
 
@@ -175,10 +178,14 @@ def tx(conn, cur):
                 set amount = amount - %s
                 where uid = %s''',
                 (amount, from_uid))
+            if (cur.rowcount != 1):
+                raise BaseException
             cur.execute('''update bank_test
                 set amount = amount + %s
                 where uid = %s''',
                 (amount, to_uid))
+            if (cur.rowcount != 1):
+                raise BaseException
             conn.commit()
 
         self.exec_tx('transfer', tx)