kinda lock in filter

kelvich · kelvich · commit f04783acdf4a · 2017-03-27T15:10:21.000+03:00
diff --git a/contrib/test_decoding/sql/prepared.sql b/contrib/test_decoding/sql/prepared.sql
@@ -45,6 +45,9 @@ DROP TABLE test_prepared1;
 DROP TABLE test_prepared2;
 
 -- show results
-SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+
+-- same but with twophase decoding
+SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1') ;
 
 SELECT pg_drop_replication_slot('regression_slot');
diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
@@ -24,6 +24,8 @@
 #include "replication/message.h"
 #include "replication/origin.h"
 
+#include "storage/procarray.h"
+
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -283,15 +285,35 @@ pg_filter_prepare(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 {
 	TestDecodingData *data = ctx->output_plugin_private;
 
-	// has_catalog_changes?
-	// LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	if (!data->twophase_decoding)
+		return true;
 
-	// OutputPluginPrepareWrite(ctx, true);
+	if (txn->has_catalog_changes)
+	{
+		LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+
+		if (TransactionIdIsInProgress(txn->xid))
+		{
+			/*
+			 * XXX
+			 */
+			LWLockRelease(TwoPhaseStateLock);
+			return true;
+		}
+		else if (TransactionIdDidAbort(txn->xid))
+		{
+			/*
+			 * Here we know that it is already aborted and should humble
+			 * ourselves.
+			 */
+			LWLockRelease(TwoPhaseStateLock);
+			return true;
+		}
 
-	// appendStringInfo(ctx->out, "pg_filter_prepare %s", gid);
+		LWLockRelease(TwoPhaseStateLock);
+	}
 
-	// OutputPluginWrite(ctx, true);
-	return true;
+	return false;
 }
 
 
@@ -307,7 +329,7 @@ pg_decode_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 
 	OutputPluginPrepareWrite(ctx, true);
 
-	appendStringInfo(ctx->out, "PREPARE! '%s'", txn->gid);
+	appendStringInfo(ctx->out, "PREPARE '%s'", txn->gid);
 
 	if (data->include_xids)
 		appendStringInfo(ctx->out, " %u", txn->xid);
