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 <maily.peng@webedia-group.com>
Co-authored-by: Minh-Quan TRAN <qtran@itscaro.me>
Co-authored-by: Álvaro Herrera <alvherre@alvh.no-ip.org>
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 0d2b795e392d1952e361688a2876e7d88dbee63d..ea75cdd3fc332c39f506a58801e65ca0ae65738d 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');