From 8aff7339a51395bcfb3e8f52d9e4f94a648ed915 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Wed, 8 Dec 2021 05:37:36 -0500 Subject: [PATCH v45 5/5] Cache ExprState per pubaction. If a subscriber has multiple publications and these publications include the same table then there can be multiple filters that apply to that table. These filters are stored per-pubactions of the publications. There are 4 kinds of pubaction ("insert", "update", "delete", "truncate"), but row-filters are not applied for "truncate". Filters for the same pubaction are all combined (OR'ed) and cached as one, so at the end there are at most 3 cached filters per table. The appropriate (pubaction) filter is executed according to the DML operation. Author: Peter Smith Discussion: https://www.postgresql.org/message-id/CAA4eK1%2BhVXfOSScbf5LUB%3D5is%3DwYaC6NBhLxuvetbWQnZRnsVQ%40mail.gmail.com --- src/backend/replication/pgoutput/pgoutput.c | 176 ++++++++++++++++++---------- 1 file changed, 117 insertions(+), 59 deletions(-) diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 0ccffa7..c244961 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -130,10 +130,16 @@ typedef struct RelationSyncEntry * The flag 'rowfilter_valid' indicates if the exprstate has been assigned * yet or not. We cannot just use the exprstate value for this purpose * because there might be no filter at all for the current relid (e.g. - * exprstate is NULL). + * every exprstate is NULL). + * The row-filter exprstate is stored per pubaction type (row-filters are + * not applied for "truncate" pubaction). */ bool rowfilter_valid; - ExprState *exprstate; /* ExprState for row filter(s) */ +#define IDX_PUBACTION_INSERT 0 +#define IDX_PUBACTION_UPDATE 1 +#define IDX_PUBACTION_DELETE 2 +#define IDX_PUBACTION_n 3 + ExprState *exprstate[IDX_PUBACTION_n]; /* ExprState for row filter(s). One per pubaction. */ TupleTableSlot *scantuple; /* tuple table slot for row filter */ TupleTableSlot *new_tuple; /* slot for storing deformed new tuple during updates */ TupleTableSlot *old_tuple; /* slot for storing deformed old tuple during updates */ @@ -175,10 +181,10 @@ static EState *create_estate_for_relation(Relation rel); static void pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry); static ExprState *pgoutput_row_filter_init_expr(Node *rfnode); static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext); -static bool pgoutput_row_filter(Relation relation, HeapTuple oldtuple, +static bool pgoutput_row_filter(int idx_pubaction, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *slot, RelationSyncEntry *entry); -static bool pgoutput_row_filter_update_check(Relation relation, HeapTuple oldtuple, +static bool pgoutput_row_filter_update_check(int idx_pubaction, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, ReorderBufferChangeType *action); @@ -755,7 +761,7 @@ pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext) * If the change is to be replicated returns true, else false. */ static bool -pgoutput_row_filter_update_check(Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, ReorderBufferChangeType *action) +pgoutput_row_filter_update_check(int idx_pubaction, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, ReorderBufferChangeType *action) { TupleDesc desc = RelationGetDescr(relation); int i; @@ -763,7 +769,7 @@ pgoutput_row_filter_update_check(Relation relation, HeapTuple oldtuple, HeapTupl TupleTableSlot *tmp_new_slot, *old_slot, *new_slot; /* Bail out if there is no row filter */ - if (!entry->exprstate) + if (!entry->exprstate[idx_pubaction]) return true; /* update requires a new tuple */ @@ -780,7 +786,7 @@ pgoutput_row_filter_update_check(Relation relation, HeapTuple oldtuple, HeapTupl if (!oldtuple) { *action = REORDER_BUFFER_CHANGE_UPDATE; - return pgoutput_row_filter(relation, NULL, newtuple, NULL, entry); + return pgoutput_row_filter(idx_pubaction, relation, NULL, newtuple, NULL, entry); } old_slot = entry->old_tuple; @@ -827,8 +833,8 @@ pgoutput_row_filter_update_check(Relation relation, HeapTuple oldtuple, HeapTupl } - old_matched = pgoutput_row_filter(relation, NULL, NULL, old_slot, entry); - new_matched = pgoutput_row_filter(relation, NULL, NULL, tmp_new_slot, entry); + old_matched = pgoutput_row_filter(idx_pubaction, relation, NULL, NULL, old_slot, entry); + new_matched = pgoutput_row_filter(idx_pubaction, relation, NULL, NULL, tmp_new_slot, entry); if (!old_matched && !new_matched) return false; @@ -850,8 +856,8 @@ static void pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry) { ListCell *lc; - List *rfnodes = NIL; - int n_filters; + List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */ + bool no_filter[] = {false, false, false}; /* One per pubaction */ /* * If the row filter caching is currently flagged "invalid" then it means we @@ -907,7 +913,7 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr bool rfisnull; /* - * Lookup if there is a row-filter, and if yes remember it in a list. + * Lookup if there is a row-filter, and if yes remember it in a list (per pubaction). * In code following this 'publications' loop we will combine all filters. */ rftuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(entry->publish_as_relid), ObjectIdGetDatum(pub->oid)); @@ -920,56 +926,101 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr Node *rfnode; oldctx = MemoryContextSwitchTo(CacheMemoryContext); - rfnode = stringToNode(TextDatumGetCString(rfdatum)); - rfnodes = lappend(rfnodes, rfnode); + /* Gather the rfnodes per pubaction of this publiaction. */ + if (pub->pubactions.pubinsert) + { + rfnode = stringToNode(TextDatumGetCString(rfdatum)); + rfnodes[IDX_PUBACTION_INSERT] = lappend(rfnodes[IDX_PUBACTION_INSERT], rfnode); + } + if (pub->pubactions.pubupdate) + { + rfnode = stringToNode(TextDatumGetCString(rfdatum)); + rfnodes[IDX_PUBACTION_UPDATE] = lappend(rfnodes[IDX_PUBACTION_UPDATE], rfnode); + } + if (pub->pubactions.pubdelete) + { + rfnode = stringToNode(TextDatumGetCString(rfdatum)); + rfnodes[IDX_PUBACTION_DELETE] = lappend(rfnodes[IDX_PUBACTION_DELETE], rfnode); + } MemoryContextSwitchTo(oldctx); - - ReleaseSysCache(rftuple); } else { - /* - * If there is no row-filter, then any other row-filters for this table - * also have no effect (because filters get OR-ed together) so we can - * just discard anything found so far and exit early from the publications - * loop. - */ - if (rfnodes) - { - list_free_deep(rfnodes); - rfnodes = NIL; - } - ReleaseSysCache(rftuple); - break; + /* Remember which pubactions have no row-filter. */ + if (pub->pubactions.pubinsert) + no_filter[IDX_PUBACTION_INSERT] = true; + if (pub->pubactions.pubupdate) + no_filter[IDX_PUBACTION_UPDATE] = true; + if (pub->pubactions.pubdelete) + no_filter[IDX_PUBACTION_DELETE] = true; } + ReleaseSysCache(rftuple); } } /* loop all subscribed publications */ /* - * Combine using all the row-filters (if any) into a single filter, and then build the ExprState for it + * Now all the filters for all pubactions are known, let's try to combine them + * when their pubactions are same. */ - n_filters = list_length(rfnodes); - if (n_filters > 0) { - Node *rfnode; - TupleDesc tupdesc = RelationGetDescr(relation); + int idx; + bool found_filters = false; - oldctx = MemoryContextSwitchTo(CacheMemoryContext); - rfnode = n_filters > 1 ? makeBoolExpr(OR_EXPR, rfnodes, -1) : linitial(rfnodes); - entry->exprstate = pgoutput_row_filter_init_expr(rfnode); + /* For each pubaction... */ + for (idx = 0; idx < IDX_PUBACTION_n; idx++) + { + int n_filters; - /* - * Create tuple table slots for row filter. Create a copy of the - * TupleDesc as it needs to live as long as the cache remains. - */ - tupdesc = CreateTupleDescCopy(tupdesc); - entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple); - entry->old_tuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); - entry->new_tuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); - entry->tmp_new_tuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); - MemoryContextSwitchTo(oldctx); + /* + * If one or more publications with this pubaction had no filter at all, + * then that nullifies the effect of all other filters for the same + * pubaction (because filters get OR'ed together). + */ + if (no_filter[idx]) + { + if (rfnodes[idx]) + { + list_free_deep(rfnodes[idx]); + rfnodes[idx] = NIL; + } + } + + /* + * If there was one or more filter for this pubaction then combine them + * (if necessary) and cache the ExprState. + */ + n_filters = list_length(rfnodes[idx]); + if (n_filters > 0) + { + Node *rfnode; + + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + rfnode = n_filters > 1 ? makeBoolExpr(OR_EXPR, rfnodes[idx], -1) : linitial(rfnodes[idx]); + entry->exprstate[idx] = pgoutput_row_filter_init_expr(rfnode); + MemoryContextSwitchTo(oldctx); + + found_filters = true; /* flag that we will need slots made */ + } + } /* for each pubaction */ + + if (found_filters) + { + TupleDesc tupdesc = RelationGetDescr(relation); + + /* + * Create tuple table slots for row filter. Create a copy of the + * TupleDesc as it needs to live as long as the cache remains. + */ + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + tupdesc = CreateTupleDescCopy(tupdesc); + entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple); + entry->old_tuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); + entry->new_tuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); + entry->tmp_new_tuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); + MemoryContextSwitchTo(oldctx); + } } entry->rowfilter_valid = true; @@ -982,7 +1033,7 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr * If it returns true, the change is replicated, otherwise, it is not. */ static bool -pgoutput_row_filter(Relation relation, HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *slot, +pgoutput_row_filter(int idx_pubaction, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *slot, RelationSyncEntry *entry) { EState *estate; @@ -991,7 +1042,7 @@ RelationSyncEntry *entry) Oid relid = RelationGetRelid(relation); /* Bail out if there is no row filter */ - if (!entry->exprstate) + if (!entry->exprstate[idx_pubaction]) return true; if (message_level_is_interesting(DEBUG3)) @@ -1016,12 +1067,12 @@ RelationSyncEntry *entry) /* * NOTE: Multiple publication row-filters have already been combined to a - * single exprstate. + * single exprstate (for this pubaction). */ - if (entry->exprstate) + if (entry->exprstate[idx_pubaction]) { /* Evaluates row filter */ - result = pgoutput_row_filter_exec_expr(entry->exprstate, ecxt); + result = pgoutput_row_filter_exec_expr(entry->exprstate[idx_pubaction], ecxt); } /* Cleanup allocated resources */ @@ -1093,7 +1144,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, HeapTuple tuple = &change->data.tp.newtuple->tuple; /* Check row filter. */ - if (!pgoutput_row_filter(relation, NULL, tuple, NULL, relentry)) + if (!pgoutput_row_filter(IDX_PUBACTION_INSERT, relation, NULL, tuple, NULL, relentry)) break; /* @@ -1126,7 +1177,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, HeapTuple newtuple = &change->data.tp.newtuple->tuple; ReorderBufferChangeType modified_action = REORDER_BUFFER_CHANGE_UPDATE; - if (!pgoutput_row_filter_update_check(relation, oldtuple, newtuple, relentry, + if (!pgoutput_row_filter_update_check(IDX_PUBACTION_UPDATE, relation, oldtuple, newtuple, relentry, &modified_action)) break; @@ -1180,7 +1231,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, HeapTuple oldtuple = &change->data.tp.oldtuple->tuple; /* Check row filter. */ - if (!pgoutput_row_filter(relation, oldtuple, NULL, NULL, relentry)) + if (!pgoutput_row_filter(IDX_PUBACTION_DELETE, relation, oldtuple, NULL, NULL, relentry)) break; maybe_send_schema(ctx, change, relation, relentry); @@ -1601,7 +1652,9 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) entry->new_tuple = NULL; entry->old_tuple = NULL; entry->tmp_new_tuple = NULL; - entry->exprstate = NULL; + entry->exprstate[IDX_PUBACTION_INSERT] = NULL; + entry->exprstate[IDX_PUBACTION_UPDATE] = NULL; + entry->exprstate[IDX_PUBACTION_DELETE] = NULL; entry->publish_as_relid = InvalidOid; entry->map = NULL; /* will be set by maybe_send_schema() if * needed */ @@ -1768,6 +1821,7 @@ static void rel_sync_cache_relation_cb(Datum arg, Oid relid) { RelationSyncEntry *entry; + int idx; /* * We can get here if the plugin was used in SQL interface as the @@ -1822,10 +1876,14 @@ rel_sync_cache_relation_cb(Datum arg, Oid relid) ExecDropSingleTupleTableSlot(entry->scantuple); entry->scantuple = NULL; } - if (entry->exprstate != NULL) + /* Cleanup the ExprState for each of the pubactions. */ + for (idx = 0; idx < IDX_PUBACTION_n; idx++) { - pfree(entry->exprstate); - entry->exprstate = NULL; + if (entry->exprstate[idx] != NULL) + { + pfree(entry->exprstate[idx]); + entry->exprstate[idx] = NULL; + } } } } -- 1.8.3.1