v45-0005-Cache-ExprState-per-pubaction.patch

From 8aff7339a51395bcfb3e8f52d9e4f94a648ed915 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <ajinc@fast.au.fujitsu.com>
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