Set ActiveSnapshot when logically replaying inserts

Input functions for the inserted tuples may require a snapshot, when
they are replayed by native logical replication.  An example is a domain
with a constraint using a SQL-language function, which prior to this
commit failed to apply on the subscriber side.

Reported-by: Mai Peng 
Co-authored-by: Minh-Quan TRAN 
Co-authored-by: Álvaro Herrera 
Discussion: https://postgr.es/m/4EB4BD78-BFC3-4D04-B8DA-D53DF7160354@webedia-group.com
Discussion: https://postgr.es/m/153211336163.1404.11721804383024050689@wrigleys.postgresql.org

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6ca6cdcacfaf0d421ec50f06471e1c8ad08d9698..2054abe6532d6c8ad41ed6461f2b32364bed4e1b 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -610,13 +610,15 @@ apply_handle_insert(StringInfo s)
    remoteslot = ExecInitExtraTupleSlot(estate,
                                        RelationGetDescr(rel->localrel));
 
+   /* Input functions may need an active snapshot, so get one */
+   PushActiveSnapshot(GetTransactionSnapshot());
+
    /* Process and store remote tuple in the slot */
    oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
    slot_store_cstrings(remoteslot, rel, newtup.values);
    slot_fill_defaults(rel, estate, remoteslot);
    MemoryContextSwitchTo(oldctx);
 
-   PushActiveSnapshot(GetTransactionSnapshot());
    ExecOpenIndices(estate->es_result_relation_info, false);
 
    /* Do the insert. */

diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
index a49e56f6303047e6b03f10c92b537da6949d75bb..4449470f7657d76a6afb6a5ac45990752b66df70 100644 (file)
--- a/src/test/subscription/t/002_types.pl
+++ b/src/test/subscription/t/002_types.pl
@@ -4,7 +4,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 3;
+use Test::More tests => 4;
 
 # Initialize publisher node
 my $node_publisher = get_new_node('publisher');
@@ -90,7 +90,13 @@ my $ddl = qq(
    CREATE TABLE public.tst_hstore (
        a INTEGER PRIMARY KEY,
        b public.hstore
-   ););
+   );
+
+   SET check_function_bodies=off;
+   CREATE FUNCTION public.monot_incr(int) RETURNS bool LANGUAGE sql
+       AS ' select \$1 > max(a) from public.tst_dom_constr; ';
+   CREATE DOMAIN monot_int AS int CHECK (monot_incr(VALUE));
+   CREATE TABLE public.tst_dom_constr (a monot_int););
 
 # Setup structure on both nodes
 $node_publisher->safe_psql('postgres', $ddl);
@@ -240,6 +246,9 @@ $node_publisher->safe_psql(
        (2, '"zzz"=>"foo"'),
        (3, '"123"=>"321"'),
        (4, '"yellow horse"=>"moaned"');
+
+   -- tst_dom_constr
+   INSERT INTO tst_dom_constr VALUES (10);
 ));
 
 $node_publisher->wait_for_catchup($appname);
@@ -541,5 +550,16 @@ e|{e,d}
 4|"yellow horse"=>"moaned"',
    'check replicated deletes on subscriber');
 
+# Test a domain with a constraint backed by a SQL-language function,
+# which needs an active snapshot in order to operate.
+$node_publisher->safe_psql('postgres', "INSERT INTO tst_dom_constr VALUES (11)");
+
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT sum(a) FROM tst_dom_constr");
+is($result, '21', 'sql-function constraint on domain');
+
 $node_subscriber->stop('fast');
 $node_publisher->stop('fast');