diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index fb3ba5c415..dbfcfc2ce8 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -338,10 +338,13 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s) * Executor state preparation for evaluation of constraint expressions, * indexes and triggers. * - * This is based on similar code in copy.c + * A ResultRelInfo for the relation to be passed to executor routines is + * returned in *resultRelInfo. The caller must open/close any indexes to + * be updated as needed. */ static EState * -create_estate_for_relation(LogicalRepRelMapEntry *rel) +create_estate_for_relation(LogicalRepRelMapEntry *rel, + ResultRelInfo **resultRelInfo) { EState *estate; RangeTblEntry *rte; @@ -355,6 +358,24 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel) rte->rellockmode = AccessShareLock; ExecInitRangeTable(estate, list_make1(rte)); + *resultRelInfo = makeNode(ResultRelInfo); + + /* + * Use Relation opened by logicalrep_rel_open() instead of opening it + * again. + */ + InitResultRelInfo(*resultRelInfo, rel->localrel, 1, NULL, 0); + + /* + * We put the ResultRelInfos in the es_opened_result_relations list, even + * though we don't populate the es_result_relations array. That's a bit + * bogus, but it's enough to make ExecGetTriggerResultRel() find them. + * Also, because we didn't open the Relation ourselves here, there's no + * need to worry about closing it. + */ + estate->es_opened_result_relations = + lappend(estate->es_opened_result_relations, *resultRelInfo); + estate->es_output_cid = GetCurrentCommandId(true); /* Prepare to catch AFTER triggers. */ @@ -363,6 +384,18 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel) return estate; } +/* Winds down the executor created in create_estate_for_relation(). */ +static void +finish_estate(EState *estate) +{ + /* Handle any queued AFTER triggers. */ + AfterTriggerEndQuery(estate); + + /* Cleanup. */ + ExecResetTupleTable(estate->es_tupleTable, false); + FreeExecutorState(estate); +} + /* * Executes default values for columns for which we can't map to remote * relation columns. @@ -1168,12 +1201,10 @@ apply_handle_insert(StringInfo s) } /* Initialize the executor state. */ - estate = create_estate_for_relation(rel); + estate = create_estate_for_relation(rel, &resultRelInfo); remoteslot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel->localrel), &TTSOpsVirtual); - resultRelInfo = makeNode(ResultRelInfo); - InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0); /* Input functions may need an active snapshot, so get one */ PushActiveSnapshot(GetTransactionSnapshot()); @@ -1194,11 +1225,7 @@ apply_handle_insert(StringInfo s) PopActiveSnapshot(); - /* Handle queued AFTER triggers. */ - AfterTriggerEndQuery(estate); - - ExecResetTupleTable(estate->es_tupleTable, false); - FreeExecutorState(estate); + finish_estate(estate); logicalrep_rel_close(rel, NoLock); @@ -1293,12 +1320,10 @@ apply_handle_update(StringInfo s) check_relation_updatable(rel); /* Initialize the executor state. */ - estate = create_estate_for_relation(rel); + estate = create_estate_for_relation(rel, &resultRelInfo); remoteslot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel->localrel), &TTSOpsVirtual); - resultRelInfo = makeNode(ResultRelInfo); - InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0); /* * Populate updatedCols so that per-column triggers can fire, and so @@ -1345,11 +1370,7 @@ apply_handle_update(StringInfo s) PopActiveSnapshot(); - /* Handle queued AFTER triggers. */ - AfterTriggerEndQuery(estate); - - ExecResetTupleTable(estate->es_tupleTable, false); - FreeExecutorState(estate); + finish_estate(estate); logicalrep_rel_close(rel, NoLock); @@ -1450,12 +1471,10 @@ apply_handle_delete(StringInfo s) check_relation_updatable(rel); /* Initialize the executor state. */ - estate = create_estate_for_relation(rel); + estate = create_estate_for_relation(rel, &resultRelInfo); remoteslot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel->localrel), &TTSOpsVirtual); - resultRelInfo = makeNode(ResultRelInfo); - InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0); PushActiveSnapshot(GetTransactionSnapshot()); @@ -1474,11 +1493,7 @@ apply_handle_delete(StringInfo s) PopActiveSnapshot(); - /* Handle queued AFTER triggers. */ - AfterTriggerEndQuery(estate); - - ExecResetTupleTable(estate->es_tupleTable, false); - FreeExecutorState(estate); + finish_estate(estate); logicalrep_rel_close(rel, NoLock); diff --git a/src/test/subscription/t/003_constraints.pl b/src/test/subscription/t/003_constraints.pl index 9f140b552b..f7628710dd 100644 --- a/src/test/subscription/t/003_constraints.pl +++ b/src/test/subscription/t/003_constraints.pl @@ -3,7 +3,7 @@ use strict; use warnings; use PostgresNode; use TestLib; -use Test::More tests => 6; +use Test::More tests => 8; # Initialize publisher node my $node_publisher = get_new_node('publisher'); @@ -84,7 +84,11 @@ BEGIN RETURN NULL; END IF; ELSIF (TG_OP = 'UPDATE') THEN - RETURN NULL; + IF (NEW.bid = 4 AND NEW.id = OLD.id) THEN + RETURN NEW; + ELSE + RETURN NULL; + END IF; ELSE RAISE WARNING 'Unknown action'; RETURN NULL; @@ -95,6 +99,28 @@ CREATE TRIGGER filter_basic_dml_trg BEFORE INSERT OR UPDATE OF bid ON tab_fk_ref FOR EACH ROW EXECUTE PROCEDURE filter_basic_dml_fn(); ALTER TABLE tab_fk_ref ENABLE REPLICA TRIGGER filter_basic_dml_trg; +CREATE FUNCTION log_tab_fk_ref_ins() RETURNS TRIGGER AS \$\$ +BEGIN + CREATE TABLE IF NOT EXISTS public.tab_fk_ref_op_log (tgtab text, tgop text, tgwhen text, tglevel text, oldbid int, newbid int); + INSERT INTO public.tab_fk_ref_op_log SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.bid; + RETURN NULL; +END; +\$\$ LANGUAGE plpgsql; +CREATE FUNCTION log_tab_fk_ref_upd() RETURNS TRIGGER AS \$\$ +BEGIN + CREATE TABLE IF NOT EXISTS public.tab_fk_ref_op_log (tgtab text, tgop text, tgwhen text, tglevel text, oldbid int, newbid int); + INSERT INTO public.tab_fk_ref_op_log SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.bid, NEW.bid; + RETURN NULL; +END; +\$\$ LANGUAGE plpgsql; +CREATE TRIGGER tab_fk_ref_log_ins_after_trg + AFTER INSERT ON tab_fk_ref + FOR EACH ROW EXECUTE PROCEDURE log_tab_fk_ref_ins(); +ALTER TABLE tab_fk_ref ENABLE REPLICA TRIGGER tab_fk_ref_log_ins_after_trg; +CREATE TRIGGER tab_fk_ref_log_upd_after_trg + AFTER UPDATE ON tab_fk_ref + FOR EACH ROW EXECUTE PROCEDURE log_tab_fk_ref_upd(); +ALTER TABLE tab_fk_ref ENABLE REPLICA TRIGGER tab_fk_ref_log_upd_after_trg; }); # Insert data @@ -108,6 +134,17 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), min(bid), max(bid) FROM tab_fk_ref;"); is($result, qq(2|1|2), 'check replica insert trigger applied on subscriber'); +# Will be inserted on subscriber, causing the after trigger to execute +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_fk_ref (id, bid) VALUES (3, 3);"); + +$node_publisher->wait_for_catchup('tap_sub'); + +# The after trigger should have recorded the above insert in tab_fk_ref_op_log +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab_fk_ref_op_log ORDER BY tgop, newbid;"); +is($result, qq(tab_fk_ref|INSERT|AFTER|ROW||3), 'check replica insert after trigger applied on subscriber'); + # Update data $node_publisher->safe_psql('postgres', "UPDATE tab_fk_ref SET bid = 2 WHERE bid = 1;"); @@ -117,9 +154,21 @@ $node_publisher->wait_for_catchup('tap_sub'); # The trigger should cause the update to be skipped on subscriber $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), min(bid), max(bid) FROM tab_fk_ref;"); -is($result, qq(2|1|2), +is($result, qq(3|1|3), 'check replica update column trigger applied on subscriber'); +# Will be updated on subscriber, causing the after trigger to execute +$node_publisher->safe_psql('postgres', + "UPDATE tab_fk_ref SET bid = 4 WHERE bid = 3;"); + +$node_publisher->wait_for_catchup('tap_sub'); + +# The after trigger should have recorded the above update in tab_fk_ref_op_log +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab_fk_ref_op_log ORDER BY tgop, newbid;"); +is($result, qq(tab_fk_ref|INSERT|AFTER|ROW||3 +tab_fk_ref|UPDATE|AFTER|ROW|3|4), 'check replica update after trigger applied on subscriber'); + # Update on a column not specified in the trigger, but it will trigger # anyway because logical replication ships all columns in an update. $node_publisher->safe_psql('postgres', @@ -129,7 +178,7 @@ $node_publisher->wait_for_catchup('tap_sub'); $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), min(id), max(id) FROM tab_fk_ref;"); -is($result, qq(2|1|2), +is($result, qq(3|1|3), 'check column trigger applied even on update for other column'); $node_subscriber->stop('fast'); diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl index a04c03a7e2..7a0e7385d4 100644 --- a/src/test/subscription/t/013_partition.pl +++ b/src/test/subscription/t/013_partition.pl @@ -3,7 +3,7 @@ use strict; use warnings; use PostgresNode; use TestLib; -use Test::More tests => 51; +use Test::More tests => 54; # setup @@ -67,6 +67,33 @@ $node_subscriber1->safe_psql('postgres', "CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1" ); +# Add replica trigger +$node_subscriber1->safe_psql( + 'postgres', qq{ +CREATE FUNCTION log_tab1_ins() RETURNS TRIGGER AS \$\$ +BEGIN + CREATE TABLE IF NOT EXISTS public.tab1_op_log (tgtab text, tgop text, tgwhen text, tglevel text, olda int, newa int); + INSERT INTO public.tab1_op_log SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.a; + RETURN NULL; +END; +\$\$ LANGUAGE plpgsql; +CREATE FUNCTION log_tab1_upd() RETURNS TRIGGER AS \$\$ +BEGIN + CREATE TABLE IF NOT EXISTS public.tab1_op_log (tgtab text, tgop text, tgwhen text, tglevel text, olda int, newa int); + INSERT INTO public.tab1_op_log SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.a, NEW.a; + RETURN NULL; +END; +\$\$ LANGUAGE plpgsql; +CREATE TRIGGER tab1_log_ins_after_trg + AFTER INSERT ON tab1_2_2 + FOR EACH ROW EXECUTE PROCEDURE log_tab1_ins(); +ALTER TABLE tab1_2_2 ENABLE REPLICA TRIGGER tab1_log_ins_after_trg; +CREATE TRIGGER tab1_log_upd_after_trg + AFTER INSERT OR UPDATE ON tab1_2_2 + FOR EACH ROW EXECUTE PROCEDURE log_tab1_upd(); +ALTER TABLE tab1_2_2 ENABLE REPLICA TRIGGER tab1_log_upd_after_trg; +}); + # subscriber 2 # # This does not use partitioning. The tables match the leaf tables on @@ -87,6 +114,33 @@ $node_subscriber2->safe_psql('postgres', "CREATE SUBSCRIPTION sub2 CONNECTION '$publisher_connstr' PUBLICATION pub_all" ); +# Add replica trigger +$node_subscriber2->safe_psql( + 'postgres', qq{ +CREATE FUNCTION log_tab1_ins() RETURNS TRIGGER AS \$\$ +BEGIN + CREATE TABLE IF NOT EXISTS public.tab1_op_log (tgtab text, tgop text, tgwhen text, tglevel text, olda int, newa int); + INSERT INTO public.tab1_op_log SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.a; + RETURN NULL; +END; +\$\$ LANGUAGE plpgsql; +CREATE FUNCTION log_tab1_upd() RETURNS TRIGGER AS \$\$ +BEGIN + CREATE TABLE IF NOT EXISTS public.tab1_op_log (tgtab text, tgop text, tgwhen text, tglevel text, olda int, newa int); + INSERT INTO public.tab1_op_log SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.a, NEW.a; + RETURN NULL; +END; +\$\$ LANGUAGE plpgsql; +CREATE TRIGGER tab1_log_ins_after_trg + AFTER INSERT ON tab1_2 + FOR EACH ROW EXECUTE PROCEDURE log_tab1_ins(); +ALTER TABLE tab1_2 ENABLE REPLICA TRIGGER tab1_log_ins_after_trg; +CREATE TRIGGER tab1_log_upd_after_trg + AFTER UPDATE ON tab1_2 + FOR EACH ROW EXECUTE PROCEDURE log_tab1_upd(); +ALTER TABLE tab1_2 ENABLE REPLICA TRIGGER tab1_log_upd_after_trg; +}); + # Wait for initial sync of all subscriptions my $synced_query = "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');"; @@ -130,6 +184,11 @@ $result = $node_subscriber2->safe_psql('postgres', "SELECT c, a FROM tab1_2 ORDER BY 1, 2"); is($result, qq(sub2_tab1_2|5), 'inserts into tab1_2 replicated'); +# The after trigger of tab1_2 should have recorded the insert +$result = $node_subscriber2->safe_psql('postgres', + "SELECT * FROM tab1_op_log ORDER BY tgop, newa;"); +is($result, qq(tab1_2|INSERT|AFTER|ROW||5), 'check replica insert after trigger applied on subscriber'); + $result = $node_subscriber2->safe_psql('postgres', "SELECT c, a FROM tab1_def ORDER BY 1, 2"); is($result, qq(sub2_tab1_def|0), 'inserts into tab1_def replicated'); @@ -161,6 +220,19 @@ $result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1_2_2 ORDER BY 1"); is($result, qq(6), 'updates of tab1_2 replicated into tab1_2_2 correctly'); +# The after trigger should have recorded the updates of tab1_2 as follows: +# 1) updates of tab1_2 (a) where 5 is changed to 6 is applied as +# cross-partition update, that is, delete 5 from tab1_2_1 followed by +# insert 6 into tab1_2_2. +# 2) updates of tab1_2 (a) where 6 is changed to 4 back to 6 are applied as +# updates of tab1_2_2. +$result = $node_subscriber1->safe_psql('postgres', + "SELECT * FROM tab1_op_log ORDER BY tgop, newa;"); +is($result, qq(tab1_2_2|INSERT|AFTER|ROW||6 +tab1_2_2|INSERT|AFTER|ROW||6 +tab1_2_2|UPDATE|AFTER|ROW|6|4 +tab1_2_2|UPDATE|AFTER|ROW|4|6), 'check replica insert/update after trigger applied on subscriber'); + $result = $node_subscriber2->safe_psql('postgres', "SELECT c, a FROM tab1_1 ORDER BY 1, 2"); is( $result, qq(sub2_tab1_1|2 @@ -170,6 +242,14 @@ $result = $node_subscriber2->safe_psql('postgres', "SELECT c, a FROM tab1_2 ORDER BY 1, 2"); is($result, qq(sub2_tab1_2|6), 'tab1_2 updated'); +# The after trigger should have recorded the updates of tab1_2 +$result = $node_subscriber2->safe_psql('postgres', + "SELECT * FROM tab1_op_log ORDER BY tgop, newa;"); +is($result, qq(tab1_2|INSERT|AFTER|ROW||5 +tab1_2|UPDATE|AFTER|ROW|6|4 +tab1_2|UPDATE|AFTER|ROW|5|6 +tab1_2|UPDATE|AFTER|ROW|4|6), 'check replica update after trigger applied on subscriber'); + $result = $node_subscriber2->safe_psql('postgres', "SELECT c, a FROM tab1_def ORDER BY 1"); is($result, qq(sub2_tab1_def|0), 'tab1_def unchanged');