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

Author: alvherre

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. */

src/test/subscription/t/002_types.pl

@@ -4,7 +4,7 @@
 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 @@
 	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 @@
 		(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 @@
 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');