Don't use on-disk snapshots for exported logical decoding snapshot.

Logical decoding stores historical snapshots on disk, so that logical
decoding can restart without having to reconstruct a snapshot from
scratch (for which the resources are not guaranteed to be present
anymore).  These serialized snapshots were also used when creating a
new slot via the walsender interface, which can export a "full"
snapshot (i.e. one that can read all tables, not just catalog ones).

The problem is that the serialized snapshots are only useful for
catalogs and not for normal user tables.  Thus the use of such a
serialized snapshot could result in an inconsistent snapshot being
exported, which could lead to queries returning wrong data.  This
would only happen if logical slots are created while another logical
slot already exists.

Author: Petr Jelinek
Reviewed-By: Andres Freund
Discussion: https://postgr.es/m/f37e975c-908f-858e-707f-058d3b1eb214@2ndquadrant.com
Backport: 9.4, where logical decoding was introduced.

diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index e798914ea3744a3a8c2580d099994418d9d10e20..6a570d0e81088861404a2a9adeadcbfa67d77c5f 100644 (file)
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -109,6 +109,7 @@ static LogicalDecodingContext *
 StartupDecodingContext(List *output_plugin_options,
                                           XLogRecPtr start_lsn,
                                           TransactionId xmin_horizon,
+                                          bool need_full_snapshot,
                                           XLogPageReadCB read_page,
                                           LogicalOutputPluginWriterPrepareWrite prepare_write,
                                           LogicalOutputPluginWriterWrite do_write)
@@ -168,7 +169,8 @@ StartupDecodingContext(List *output_plugin_options,
 
        ctx->reorder = ReorderBufferAllocate();
        ctx->snapshot_builder =
-               AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn);
+               AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn,
+                                                               need_full_snapshot);
 
        ctx->reorder->private_data = ctx;
 
@@ -340,7 +342,8 @@ CreateInitDecodingContext(char *plugin,
        ReplicationSlotSave();
 
        ctx = StartupDecodingContext(NIL, InvalidXLogRecPtr, xmin_horizon,
-                                                                read_page, prepare_write, do_write);
+                                                                need_full_snapshot, read_page, prepare_write,
+                                                                do_write);
 
        /* call output plugin initialization callback */
        old_context = MemoryContextSwitchTo(ctx->context);
@@ -420,7 +423,7 @@ CreateDecodingContext(XLogRecPtr start_lsn,
        }
 
        ctx = StartupDecodingContext(output_plugin_options,
-                                                                start_lsn, InvalidTransactionId,
+                                                                start_lsn, InvalidTransactionId, false,
                                                                 read_page, prepare_write, do_write);
 
        /* call output plugin initialization callback */

diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 9bffa53345e2649b485654ce56823a42e7900e09..94ea204ffdc4a7c0b6c13bbfa876d4d951703dfe 100644 (file)
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -165,6 +165,9 @@ struct SnapBuild
         */
        TransactionId initial_xmin_horizon;
 
+       /* Indicates if we are building full snapshot or just catalog one .*/
+       bool            building_full_snapshot;
+
        /*
         * Snapshot that's valid to see the catalog state seen at this moment.
         */
@@ -281,7 +284,8 @@ static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);
 SnapBuild *
 AllocateSnapshotBuilder(ReorderBuffer *reorder,
                                                TransactionId xmin_horizon,
-                                               XLogRecPtr start_lsn)
+                                               XLogRecPtr start_lsn,
+                                               bool need_full_snapshot)
 {
        MemoryContext context;
        MemoryContext oldcontext;
@@ -310,6 +314,7 @@ AllocateSnapshotBuilder(ReorderBuffer *reorder,
 
        builder->initial_xmin_horizon = xmin_horizon;
        builder->start_decoding_at = start_lsn;
+       builder->building_full_snapshot = need_full_snapshot;
 
        MemoryContextSwitchTo(oldcontext);
 
@@ -1210,7 +1215,7 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
         *
         * a) There were no running transactions when the xl_running_xacts record
         *        was inserted, jump to CONSISTENT immediately. We might find such a
-        *        state we were waiting for b) and c).
+        *        state we were waiting for b) or c).
         *
         * b) Wait for all toplevel transactions that were running to end. We
         *        simply track the number of in-progress toplevel transactions and
@@ -1225,7 +1230,10 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
         *        at all.
         *
         * c) This (in a previous run) or another decoding slot serialized a
-        *        snapshot to disk that we can use.
+        *        snapshot to disk that we can use.  Can't use this method for the
+        *        initial snapshot when slot is being created and needs full snapshot
+        *        for export or direct use, as that snapshot will only contain catalog
+        *        modifying transactions.
         * ---
         */
 
@@ -1280,8 +1288,9 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
 
                return false;
        }
-       /* c) valid on disk state */
-       else if (SnapBuildRestore(builder, lsn))
+       /* c) valid on disk state and not building full snapshot */
+       else if (!builder->building_full_snapshot &&
+                        SnapBuildRestore(builder, lsn))
        {
                /* there won't be any state to cleanup */
                return false;

diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index e5d61ff3c4b68879307e027b8d45589789c4cc3a..d1139ecf2cc05c4665d173c499feafad72f8fd0b 100644 (file)
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -54,7 +54,8 @@ struct xl_running_xacts;
 extern void CheckPointSnapBuild(void);
 
 extern SnapBuild *AllocateSnapshotBuilder(struct ReorderBuffer *cache,
-                                               TransactionId xmin_horizon, XLogRecPtr start_lsn);
+                                               TransactionId xmin_horizon, XLogRecPtr start_lsn,
+                                               bool need_full_snapshot);
 extern void FreeSnapshotBuilder(SnapBuild *cache);
 
 extern void SnapBuildSnapDecRefcount(Snapshot snap);