[Callahan, Drew <callaan@amazon.com>]

> I think we can prepare an isolation test of the above steps and
> include it in contrib/test_decoding/specs.

Thanks for the suggestion. Included an isolation test for the
exact repro and a couple other tests for fully before and fully after.
These two tests don't add a whole lot of value since this
is already heavily tested elsewhere, so I'm good with removing.
However, thought it made sense to include from a perspective of
totality given the test name.

> Another simple way to fix this is that we always set
> need_full_snapshot to true when slot creation. But it would then make
> the snapshot builder include non-catalog-change transactions too while
> finding the initial startpoint. Which is not necessary.

Since we're not accumulating changes anyway and we never distribute snapshots
while inconsistent it wouldn't cause any memory pressure in the ReorderBuffer.
This seems pretty contained and wouldn't require the use of a global variable
and memory context callbacks which is a bit nicer from a readability standpoint.
It's a little inefficient with memory and CPU, especially in the presence of
long running transactions, but likely would not be noticeable to most users.

> As for the proposed patch, it uses a global variable, InCreate, and
> memory context callback to reset it. While I believe this is for not
> breaking any ABI as this bug exists on back branches too, I think it's
> better to consider how to fix it in HEAD and then how to backpatch it.
> We don't necessarily need to have the same fixes for all branches all
> the time.

Attached a patch that is ABI breaking, but removes the need for a global variable.
This relies on pushing the context of whether we are in a create mode to the
logical decoding context, which is cleaner, but not as clean as just pushing
all the way down to the snapbuilder. I would push down to the snap builder,
but the entire object is persisted to disk. This is problematic to me since the
create flag would be persisted causing exisiting slots to restore a snapbuild
that could have the create flag set to true. This is not problematic as is,
since the variable would only be leveraged in SnapBuildRestore() which is only
called while inconsistent, but seems like a future bug waiting to happen.

We could always clear the flag when persisting, but that seems a bit hacky.
We could also fork the snap builder into on-disk and in-memory structs, which
would be convenient since we would not need to invalidate exisiting snapshots.
However, not sure how people feel about the fork.

> IIUC we need to skip snapshot restore only when we're finding the
> initial start point. So probably we don't need to set InCreate when
> calling create_logical_replication_slot() with find_startpoint =
> false.

Yep, that's correct. We can safely skip setting InCreate. We could move
the setter function slotfuncs.c as one way of doing this. I lean towards
setting it no matter what since it is harmless if we never try to find
a decoding point and removes the need for future callers of
CreateInitDecodingContext() from having to determine if it's safe to be false.
Though I agree that InCreate may a bigger hammer than needed which may need
to be modified in the future.

Thanks for the quick response,
Drew Callahan



Attachments:

  [application/octet-stream] isolation_test.patch (5.3K, ../2444AA15-D21B-4CCE-8052-52C7C2DAFE5C@amazon.com/2-isolation_test.patch)
  download | inline diff:
diff --git a/contrib/test_decoding/expected/slot_creation_consistency.out b/contrib/test_decoding/expected/slot_creation_consistency.out
new file mode 100755
index 0000000000..4ee9ffea5d
--- /dev/null
+++ b/contrib/test_decoding/expected/slot_creation_consistency.out
@@ -0,0 +1,133 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s2_create s1_b s1_c s3_slot1_get s3_slot2_get
+s1: NOTICE:  table "test" does not exist, skipping
+?column?
+--------
+init
+(1 row)
+
+step s2_create: SELECT 'init' FROM pg_create_logical_replication_slot('slot2', 'test_decoding');
+?column?
+--------
+init
+(1 row)
+
+step s1_b:
+BEGIN;
+INSERT INTO test VALUES (1);
+
+step s1_c:
+INSERT INTO test VALUES (2);
+COMMIT;
+
+step s3_slot1_get: SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+---------------------------------------
+BEGIN
+table public.test: INSERT: a[integer]:1
+table public.test: INSERT: a[integer]:2
+COMMIT
+(4 rows)
+
+step s3_slot2_get: SELECT data FROM pg_logical_slot_get_changes('slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+---------------------------------------
+BEGIN
+table public.test: INSERT: a[integer]:1
+table public.test: INSERT: a[integer]:2
+COMMIT
+(4 rows)
+
+pg_drop_replication_slot
+------------------------
+
+(1 row)
+
+
+starting permutation: s1_b s2_create s3_checkpoint s3_slot1_get s1_c s3_slot1_get s3_slot2_get
+?column?
+--------
+init
+(1 row)
+
+step s1_b:
+BEGIN;
+INSERT INTO test VALUES (1);
+
+step s2_create: SELECT 'init' FROM pg_create_logical_replication_slot('slot2', 'test_decoding'); <waiting ...>
+step s3_checkpoint: CHECKPOINT;
+step s3_slot1_get: SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+----
+(0 rows)
+
+step s1_c:
+INSERT INTO test VALUES (2);
+COMMIT;
+
+step s2_create: <... completed>
+?column?
+--------
+init
+(1 row)
+
+step s3_slot1_get: SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+---------------------------------------
+BEGIN
+table public.test: INSERT: a[integer]:1
+table public.test: INSERT: a[integer]:2
+COMMIT
+(4 rows)
+
+step s3_slot2_get: SELECT data FROM pg_logical_slot_get_changes('slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+----
+(0 rows)
+
+pg_drop_replication_slot
+------------------------
+
+(1 row)
+
+
+starting permutation: s1_b s1_c s2_create s3_slot1_get s3_slot2_get
+?column?
+--------
+init
+(1 row)
+
+step s1_b:
+BEGIN;
+INSERT INTO test VALUES (1);
+
+step s1_c:
+INSERT INTO test VALUES (2);
+COMMIT;
+
+step s2_create: SELECT 'init' FROM pg_create_logical_replication_slot('slot2', 'test_decoding');
+?column?
+--------
+init
+(1 row)
+
+step s3_slot1_get: SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+---------------------------------------
+BEGIN
+table public.test: INSERT: a[integer]:1
+table public.test: INSERT: a[integer]:2
+COMMIT
+(4 rows)
+
+step s3_slot2_get: SELECT data FROM pg_logical_slot_get_changes('slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+----
+(0 rows)
+
+pg_drop_replication_slot
+------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/specs/slot_creation_consistency.spec b/contrib/test_decoding/specs/slot_creation_consistency.spec
new file mode 100755
index 0000000000..e988872e65
--- /dev/null
+++ b/contrib/test_decoding/specs/slot_creation_consistency.spec
@@ -0,0 +1,48 @@
+# Test that a slot will correctly replicate changes only from transactions that
+# started after the slot was created. This is primarily done to validate that a slot
+# cannot become consistent prematurely through a snapshot that was persisted by
+# a different slot leading to a transaction being partially replicated.
+
+session "s1"
+setup {
+SET synchronous_commit=on;
+BEGIN;
+DROP TABLE IF EXISTS test;
+CREATE TABLE test (a INT);
+COMMIT;
+SELECT 'init' FROM pg_create_logical_replication_slot('slot1', 'test_decoding');
+}
+
+step s1_b {
+BEGIN;
+INSERT INTO test VALUES (1);
+}
+step s1_c {
+INSERT INTO test VALUES (2);
+COMMIT;
+}
+
+teardown {
+SELECT pg_drop_replication_slot('slot1');
+SELECT pg_drop_replication_slot('slot2');
+}
+
+session s2
+setup { SET synchronous_commit=on; }
+step s2_create { SELECT 'init' FROM pg_create_logical_replication_slot('slot2', 'test_decoding'); }
+
+session s3
+setup { SET synchronous_commit=on; }
+step s3_checkpoint { CHECKPOINT; }
+step s3_slot1_get { SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); }
+step s3_slot2_get { SELECT data FROM pg_logical_slot_get_changes('slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); }
+
+# Create and slot and then make changes. Already done heavily elsewhere, but addded for totality
+permutation s2_create s1_b s1_c s3_slot1_get s3_slot2_get
+
+# Create a slot in the presence of an active transaction. The checkpoint and slot
+# consumption will persist a snapshot to disk ahead of the current read position of the slot.
+permutation s1_b s2_create s3_checkpoint s3_slot1_get s1_c s3_slot1_get s3_slot2_get
+
+# Create a slot after the completion of the active transaction
+permutation s1_b s1_c s2_create s3_slot1_get s3_slot2_get


  [application/octet-stream] ctx_skip_snap_restore.patch (6.0K, ../2444AA15-D21B-4CCE-8052-52C7C2DAFE5C@amazon.com/3-ctx_skip_snap_restore.patch)
  download | inline diff:
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index d91055a440..285cc30298 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -371,7 +371,7 @@ standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 			{
 				xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);

-				SnapBuildProcessRunningXacts(builder, buf->origptr, running);
+				SnapBuildProcessRunningXacts(builder, buf->origptr, ctx->in_create, running);

 				/*
 				 * Abort all transactions that we keep track of, that are
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 41243d0187..77400dc2d4 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -151,6 +151,7 @@ StartupDecodingContext(List *output_plugin_options,
 					   TransactionId xmin_horizon,
 					   bool need_full_snapshot,
 					   bool fast_forward,
+					   bool in_create,
 					   XLogReaderRoutine *xl_routine,
 					   LogicalOutputPluginWriterPrepareWrite prepare_write,
 					   LogicalOutputPluginWriterWrite do_write,
@@ -200,6 +201,7 @@ StartupDecodingContext(List *output_plugin_options,
 	}

 	ctx->slot = slot;
+	ctx->in_create = in_create;

 	ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx);
 	if (!ctx->reader)
@@ -437,9 +439,9 @@ CreateInitDecodingContext(const char *plugin,
 	ReplicationSlotSave();

 	ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
-								 need_full_snapshot, false,
-								 xl_routine, prepare_write, do_write,
-								 update_progress);
+								 need_full_snapshot, false /* fast_forward */,
+								 true /* in_create */, xl_routine,
+								 prepare_write, do_write, update_progress);

 	/* call output plugin initialization callback */
 	old_context = MemoryContextSwitchTo(ctx->context);
@@ -572,9 +574,10 @@ CreateDecodingContext(XLogRecPtr start_lsn,
 	}

 	ctx = StartupDecodingContext(output_plugin_options,
-								 start_lsn, InvalidTransactionId, false,
-								 fast_forward, xl_routine, prepare_write,
-								 do_write, update_progress);
+								 start_lsn, InvalidTransactionId,
+								 false /* need_full_snapshot */, fast_forward,
+								 false /* in_create */, xl_routine,
+								 prepare_write, do_write, update_progress);

 	/* call output plugin initialization callback */
 	old_context = MemoryContextSwitchTo(ctx->context);
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 7a7aba33e1..f846710843 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -294,7 +294,8 @@ static inline bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, Transaction
 												 uint32 xinfo);

 /* xlog reading helper functions for SnapBuildProcessRunningXacts */
-static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running);
+static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn,
+								  const bool in_create,  xl_running_xacts *running);
 static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff);

 /* serialization functions */
@@ -1210,7 +1211,8 @@ SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid,
  * anymore.
  */
 void
-SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
+SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn,
+								const bool in_create, xl_running_xacts *running)
 {
 	ReorderBufferTXN *txn;
 	TransactionId xmin;
@@ -1223,7 +1225,7 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
 	if (builder->state < SNAPBUILD_CONSISTENT)
 	{
 		/* returns false if there's no point in performing cleanup just yet */
-		if (!SnapBuildFindSnapshot(builder, lsn, running))
+		if (!SnapBuildFindSnapshot(builder, lsn, in_create, running))
 			return;
 	}
 	else
@@ -1312,7 +1314,8 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
  * using the xl_running_xacts record.
  */
 static bool
-SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
+SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, const bool in_create,
+					  xl_running_xacts *running)
 {
 	/* ---
 	 * Build catalog decoding snapshot incrementally using information about
@@ -1391,8 +1394,9 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn

 		return false;
 	}
-	/* b) valid on disk state and not building full snapshot */
+	/* b) valid on disk state, we are an exisiting slot, and not building full snapshot */
 	else if (!builder->building_full_snapshot &&
+			 !in_create &&
 			 SnapBuildRestore(builder, lsn))
 	{
 		/* there won't be any state to cleanup */
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 5f49554ea0..39a27f3db7 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -50,6 +50,12 @@ typedef struct LogicalDecodingContext
 	 */
 	bool		fast_forward;

+	/*
+	 * Marks the logical decoding context as being used for the initial creation
+	 * of a logical replication slot.
+	 */
+	bool            in_create;
+
 	OutputPluginCallbacks callbacks;
 	OutputPluginOptions options;

diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index f49b941b53..b84403f511 100644
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -88,7 +88,7 @@ extern void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
 								   XLogRecPtr lsn,
 								   struct xl_heap_new_cid *xlrec);
 extern void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn,
-										 struct xl_running_xacts *running);
+										 const bool in_create, struct xl_running_xacts *running);
 extern void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn);

 #endif							/* SNAPBUILD_H */