v53-0002-Row-filter-updates-based-on-old-new-tuples.patch

From 4810cb7a3926ce50ca833b656bf812ea2f51cb01 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <ajinc@fast.au.fujitsu.com>
Date: Wed, 22 Dec 2021 05:56:03 -0500
Subject: [PATCH v53 2/3] Row filter updates based on old/new tuples.

When applying row filter on updates, check both old_tuple and new_tuple
to decide how an update needs to be transformed.

UPDATE
old-row (match)       new-row (no match)    -> DELETE
old-row (no match)    new row (match)       -> INSERT
old-row (match)       new row (match)       -> UPDATE
old-row (no match)    new-row (no match)    -> (drop change)

Also tuples that have been deformed will be cached in slots to avoid
multiple deforming of tuples.

Author: Ajin Cherian
---
 src/backend/replication/logical/proto.c     |  37 ++--
 src/backend/replication/pgoutput/pgoutput.c | 276 +++++++++++++++++++++++-----
 src/include/replication/logicalproto.h      |   7 +-
 src/include/replication/reorderbuffer.h     |   6 +-
 src/test/subscription/t/027_row_filter.pl   |   4 +-
 src/tools/pgindent/typedefs.list            |   1 +
 6 files changed, 268 insertions(+), 63 deletions(-)

diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 9f5bf4b..4d0c387 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -15,6 +15,7 @@
 #include "access/sysattr.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_type.h"
+#include "executor/executor.h"
 #include "libpq/pqformat.h"
 #include "replication/logicalproto.h"
 #include "utils/lsyscache.h"
@@ -31,8 +32,8 @@
 
 static void logicalrep_write_attrs(StringInfo out, Relation rel);
 static void logicalrep_write_tuple(StringInfo out, Relation rel,
-								   HeapTuple tuple, bool binary);
-
+								   HeapTuple tuple, TupleTableSlot *slot,
+								   bool binary);
 static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel);
 static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
 
@@ -410,7 +411,7 @@ logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
 	pq_sendint32(out, RelationGetRelid(rel));
 
 	pq_sendbyte(out, 'N');		/* new tuple follows */
-	logicalrep_write_tuple(out, rel, newtuple, binary);
+	logicalrep_write_tuple(out, rel, newtuple, NULL, binary);
 }
 
 /*
@@ -442,7 +443,8 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
  */
 void
 logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
-						HeapTuple oldtuple, HeapTuple newtuple, bool binary)
+						HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *oldslot,
+						TupleTableSlot *newslot, bool binary)
 {
 	pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE);
 
@@ -463,11 +465,11 @@ logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
 			pq_sendbyte(out, 'O');	/* old tuple follows */
 		else
 			pq_sendbyte(out, 'K');	/* old key follows */
-		logicalrep_write_tuple(out, rel, oldtuple, binary);
+		logicalrep_write_tuple(out, rel, oldtuple, oldslot, binary);
 	}
 
 	pq_sendbyte(out, 'N');		/* new tuple follows */
-	logicalrep_write_tuple(out, rel, newtuple, binary);
+	logicalrep_write_tuple(out, rel, newtuple, newslot, binary);
 }
 
 /*
@@ -536,7 +538,7 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel,
 	else
 		pq_sendbyte(out, 'K');	/* old key follows */
 
-	logicalrep_write_tuple(out, rel, oldtuple, binary);
+	logicalrep_write_tuple(out, rel, oldtuple, NULL, binary);
 }
 
 /*
@@ -749,13 +751,16 @@ logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
  * Write a tuple to the outputstream, in the most efficient format possible.
  */
 static void
-logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary)
+logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, TupleTableSlot *slot,
+bool binary)
 {
 	TupleDesc	desc;
-	Datum		values[MaxTupleAttributeNumber];
-	bool		isnull[MaxTupleAttributeNumber];
+	Datum		*values;
+	bool		*isnull;
 	int			i;
 	uint16		nliveatts = 0;
+	Datum		attr_values[MaxTupleAttributeNumber];
+	bool		attr_isnull[MaxTupleAttributeNumber];
 
 	desc = RelationGetDescr(rel);
 
@@ -771,7 +776,17 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binar
 	enlargeStringInfo(out, tuple->t_len +
 					  nliveatts * (1 + 4));
 
-	heap_deform_tuple(tuple, desc, values, isnull);
+	if (TupIsNull(slot))
+	{
+		values = attr_values;
+		isnull = attr_isnull;
+		heap_deform_tuple(tuple, desc, values, isnull);
+	}
+	else
+	{
+		values = slot->tts_values;
+		isnull = slot->tts_isnull;
+	}
 
 	/* Write the values */
 	for (i = 0; i < desc->natts; i++)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index d9bb0f6..0e4c736 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -13,6 +13,7 @@
 #include "postgres.h"
 
 #include "access/tupconvert.h"
+#include "access/xact.h"
 #include "catalog/partition.h"
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_rel.h"
@@ -25,6 +26,7 @@
 #include "parser/parse_coerce.h"
 #include "replication/logical.h"
 #include "replication/logicalproto.h"
+#include "replication/logicalrelation.h"
 #include "replication/origin.h"
 #include "replication/pgoutput.h"
 #include "utils/builtins.h"
@@ -139,7 +141,9 @@ typedef struct RelationSyncEntry
 #define NUM_ROWFILTER_PUBACTIONS	3
 	/* ExprState array for row filter. One per publication action. */
 	ExprState	   *exprstate[NUM_ROWFILTER_PUBACTIONS];
-	TupleTableSlot *scantuple;		/* tuple table slot for row filter */
+	TupleTableSlot *scan_slot;		/* tuple table slot for row filter */
+	TupleTableSlot *new_slot;		/* slot for storing deformed new tuple during updates */
+	TupleTableSlot *old_slot;		/* slot for storing deformed old tuple during updates */
 
 	/*
 	 * OID of the relation to publish changes as.  For a partition, this may
@@ -174,11 +178,15 @@ static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
 
 /* row filter routines */
 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(enum ReorderBufferChangeType changetype, PGOutputData *data,
-								Relation relation, HeapTuple oldtuple,
-								HeapTuple newtuple, RelationSyncEntry *entry);
+static bool pgoutput_row_filter(enum ReorderBufferChangeType changetype, EState *relation, Oid relid,
+								HeapTuple oldtuple, HeapTuple newtuple,
+								TupleTableSlot *slot, RelationSyncEntry *entry);
+static bool pgoutput_row_filter_update_check(enum ReorderBufferChangeType changetype, Relation relation,
+											 HeapTuple oldtuple, HeapTuple newtuple,
+											 RelationSyncEntry *entry, ReorderBufferChangeType *action);
 
 /*
  * Specify output plugin callbacks
@@ -742,26 +750,141 @@ pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
 }
 
 /*
- * Change is checked against the row filter, if any.
- *
- * If it returns true, the change is replicated, otherwise, it is not.
+ * Update is checked against the row filter, if any.
+ * Updates are transformed to inserts and deletes based on the
+ * old_tuple and new_tuple. The new action is updated in the
+ * action parameter. If not updated, action remains as update.
+ * old-row (no match)    new-row (no match)  -> (drop change)
+ * old-row (no match)    new row (match)     -> INSERT
+ * old-row (match)       new-row (no match)  -> DELETE
+ * old-row (match)       new row (match)     -> UPDATE
+ * If the change is to be replicated returns true, else false.
  */
 static bool
-pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
-					Relation relation, HeapTuple oldtuple, HeapTuple newtuple,
-					RelationSyncEntry *entry)
+pgoutput_row_filter_update_check(enum ReorderBufferChangeType changetype, Relation relation,
+								 HeapTuple oldtuple, HeapTuple newtuple,
+								 RelationSyncEntry *entry, ReorderBufferChangeType *action)
 {
-	EState	   *estate;
-	ExprContext *ecxt;
-	ListCell   *lc;
-	bool		result = true;
-	Oid			relid = RelationGetRelid(relation);
-	List	   *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */
-	bool		no_filter[] = {false, false, false}; /* One per pubaction */
+	TupleDesc	desc = RelationGetDescr(relation);
+	int		i;
+	bool	old_matched, new_matched;
+	TupleTableSlot	*tmp_new_slot, *old_slot, *new_slot;
+	EState   *estate = NULL;
 
 	Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
 		   changetype == REORDER_BUFFER_CHANGE_UPDATE ||
 		   changetype == REORDER_BUFFER_CHANGE_DELETE);
+
+	/* Bail out if there is no row filter */
+	if (!entry->exprstate[changetype])
+		return true;
+
+	/* update requires a new tuple */
+	Assert(newtuple);
+
+	elog(DEBUG3, "table \"%s.%s\" has row filter",
+			get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
+			get_rel_name(relation->rd_id));
+
+	/* Clear the tuples */
+	ExecClearTuple(entry->old_slot);
+	ExecClearTuple(entry->new_slot);
+
+	estate = create_estate_for_relation(relation);
+
+	/*
+	 * If no old_tuple, then none of the replica identity columns changed
+	 * and this would reduce to a simple update.
+	 */
+	if (!oldtuple)
+	{
+		bool	res;
+
+		*action = REORDER_BUFFER_CHANGE_UPDATE;
+		res = pgoutput_row_filter(changetype, estate,
+								  RelationGetRelid(relation), NULL, newtuple,
+								  NULL, entry);
+
+		FreeExecutorState(estate);
+		return res;
+	}
+
+	old_slot = entry->old_slot;
+	new_slot = entry->new_slot;
+	tmp_new_slot = new_slot;
+
+	heap_deform_tuple(newtuple, desc, new_slot->tts_values, new_slot->tts_isnull);
+	heap_deform_tuple(oldtuple, desc, old_slot->tts_values, old_slot->tts_isnull);
+
+	ExecStoreVirtualTuple(old_slot);
+	ExecStoreVirtualTuple(new_slot);
+
+	/*
+	 * For updates, both the newtuple and oldtuple needs to be checked
+	 * against the row-filter. The newtuple might not have all the
+	 * replica identity columns, in which case it needs to be
+	 * copied over from the oldtuple.
+	 */
+	for (i = 0; i < desc->natts; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(desc, i);
+
+		/* if the column in the new_tuple is null, nothing to do */
+		if (tmp_new_slot->tts_isnull[i])
+			continue;
+
+		/*
+		 * Unchanged toasted replica identity columns are
+		 * only detoasted in the old tuple, copy this over to the newtuple.
+		 */
+		if ((att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(tmp_new_slot->tts_values[i])) &&
+				(!old_slot->tts_isnull[i] &&
+					!(VARATT_IS_EXTERNAL_ONDISK(old_slot->tts_values[i]))))
+		{
+			if (tmp_new_slot == new_slot)
+			{
+				tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);;
+				ExecClearTuple(tmp_new_slot);
+				ExecCopySlot(tmp_new_slot, new_slot);
+			}
+
+			tmp_new_slot->tts_values[i] = old_slot->tts_values[i];
+		}
+
+	}
+
+	old_matched = pgoutput_row_filter(changetype, estate,
+									  RelationGetRelid(relation), NULL, NULL,
+									  old_slot, entry);
+	new_matched = pgoutput_row_filter(changetype, estate,
+									  RelationGetRelid(relation), NULL, NULL,
+									  tmp_new_slot, entry);
+
+	FreeExecutorState(estate);
+
+	if (!old_matched && !new_matched)
+		return false;
+
+	if (!old_matched && new_matched)
+		*action = REORDER_BUFFER_CHANGE_INSERT;
+	else if (old_matched && !new_matched)
+		*action = REORDER_BUFFER_CHANGE_DELETE;
+	else if (new_matched && old_matched)
+		*action = REORDER_BUFFER_CHANGE_UPDATE;
+
+	return true;
+}
+
+/*
+ * Initialize the row filter, the first time.
+ */
+static void
+pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry)
+{
+	ListCell   *lc;
+	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
 	 * don't know yet if there is/isn't any row filters for this relation.
@@ -971,16 +1094,32 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
 			 */
 			oldctx = MemoryContextSwitchTo(CacheMemoryContext);
 			tupdesc = CreateTupleDescCopy(tupdesc);
-			entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+			entry->scan_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+			entry->old_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);
+			entry->new_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);
 			MemoryContextSwitchTo(oldctx);
 		}
 
 		entry->exprstate_valid = true;
 	}
+}
 
-	/* Bail out if there is no row filter */
-	if (!entry->exprstate[changetype])
-		return true;
+/*
+ * Change is checked against the row filter, if any.
+ *
+ * If it returns true, the change is replicated, otherwise, it is not.
+ */
+static bool
+pgoutput_row_filter(enum ReorderBufferChangeType changetype, EState *estate, Oid relid,
+					HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *slot,
+					RelationSyncEntry *entry)
+{
+	ExprContext *ecxt;
+	bool		result = true;
+
+	Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
+		   changetype == REORDER_BUFFER_CHANGE_UPDATE ||
+		   changetype == REORDER_BUFFER_CHANGE_DELETE);
 
 	if (message_level_is_interesting(DEBUG3))
 		elog(DEBUG3, "table \"%s.%s\" has row filter",
@@ -989,13 +1128,16 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
 
 	PushActiveSnapshot(GetTransactionSnapshot());
 
-	estate = create_estate_for_relation(relation);
-
 	/* Prepare context per tuple */
 	ecxt = GetPerTupleExprContext(estate);
-	ecxt->ecxt_scantuple = entry->scantuple;
+	ecxt->ecxt_scantuple = entry->scan_slot;
 
-	ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
+	if (newtuple || oldtuple)
+		ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
+	else
+	{
+		ecxt->ecxt_scantuple = slot;
+	}
 
 	/*
 	 * NOTE: Multiple publication row filters have already been combined to a
@@ -1007,9 +1149,6 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
 		result = pgoutput_row_filter_exec_expr(entry->exprstate[changetype], ecxt);
 	}
 
-	/* Cleanup allocated resources */
-	ResetExprContext(ecxt);
-	FreeExecutorState(estate);
 	PopActiveSnapshot();
 
 	return result;
@@ -1029,6 +1168,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	RelationSyncEntry *relentry;
 	TransactionId xid = InvalidTransactionId;
 	Relation	ancestor = NULL;
+	EState	   *estate = NULL;
 
 	if (!is_publishable_relation(relation))
 		return;
@@ -1066,6 +1206,9 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	/* Avoid leaking memory by using and resetting our own context */
 	old = MemoryContextSwitchTo(data->context);
 
+	/* Initialize the row_filter */
+	pgoutput_row_filter_init(data, relation, relentry);
+
 	/* Send the data */
 	switch (change->action)
 	{
@@ -1073,10 +1216,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 			{
 				HeapTuple	tuple = &change->data.tp.newtuple->tuple;
 
-				/* Check row filter. */
-				if (!pgoutput_row_filter(change->action, data, relation, NULL, tuple, relentry))
-					break;
-
 				/*
 				 * Schema should be sent before the logic that replaces the
 				 * relation because it also sends the ancestor's relation.
@@ -1094,6 +1233,17 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 						tuple = execute_attr_map_tuple(tuple, relentry->map);
 				}
 
+				if (relentry->exprstate[change->action])
+				{
+					estate = create_estate_for_relation(relation);
+
+					/* Check row filter. */
+					if (!pgoutput_row_filter(change->action, estate,
+											 RelationGetRelid(relation), NULL,
+											 tuple, NULL, relentry))
+						break;
+				}
+
 				OutputPluginPrepareWrite(ctx, true);
 				logicalrep_write_insert(ctx->out, xid, relation, tuple,
 										data->binary);
@@ -1105,10 +1255,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				HeapTuple	oldtuple = change->data.tp.oldtuple ?
 				&change->data.tp.oldtuple->tuple : NULL;
 				HeapTuple	newtuple = &change->data.tp.newtuple->tuple;
-
-				/* Check row filter. */
-				if (!pgoutput_row_filter(change->action, data, relation, oldtuple, newtuple, relentry))
-					break;
+				ReorderBufferChangeType modified_action = REORDER_BUFFER_CHANGE_UPDATE;
 
 				maybe_send_schema(ctx, change, relation, relentry);
 
@@ -1129,9 +1276,34 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 					}
 				}
 
+				/* Check row filter */
+				if (!pgoutput_row_filter_update_check(change->action, relation,
+													  oldtuple, newtuple, relentry,
+													  &modified_action))
+					break;
+
 				OutputPluginPrepareWrite(ctx, true);
-				logicalrep_write_update(ctx->out, xid, relation, oldtuple,
-										newtuple, data->binary);
+
+				switch (modified_action)
+				{
+					case REORDER_BUFFER_CHANGE_INSERT:
+						logicalrep_write_insert(ctx->out, xid, relation, newtuple,
+												data->binary);
+						break;
+					case REORDER_BUFFER_CHANGE_UPDATE:
+						logicalrep_write_update(ctx->out, xid, relation,
+												oldtuple, newtuple, relentry->old_slot,
+												relentry->new_slot,
+												data->binary);
+						break;
+					case REORDER_BUFFER_CHANGE_DELETE:
+						logicalrep_write_delete(ctx->out, xid, relation, oldtuple,
+												data->binary);
+						break;
+					default:
+						Assert(false);
+				}
+
 				OutputPluginWrite(ctx, true);
 				break;
 			}
@@ -1140,10 +1312,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 			{
 				HeapTuple	oldtuple = &change->data.tp.oldtuple->tuple;
 
-				/* Check row filter. */
-				if (!pgoutput_row_filter(change->action, data, relation, oldtuple, NULL, relentry))
-					break;
-
 				maybe_send_schema(ctx, change, relation, relentry);
 
 				/* Switch relation if publishing via root. */
@@ -1157,6 +1325,17 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 						oldtuple = execute_attr_map_tuple(oldtuple, relentry->map);
 				}
 
+				if (relentry->exprstate[change->action])
+				{
+					estate = create_estate_for_relation(relation);
+
+					/* Check row filter. */
+					if (!pgoutput_row_filter(change->action, estate,
+											 RelationGetRelid(relation), oldtuple,
+											 NULL, NULL, relentry))
+						break;
+				}
+
 				OutputPluginPrepareWrite(ctx, true);
 				logicalrep_write_delete(ctx->out, xid, relation, oldtuple,
 										data->binary);
@@ -1175,6 +1354,9 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 		ancestor = NULL;
 	}
 
+	if (estate)
+		FreeExecutorState(estate);
+
 	/* Cleanup */
 	MemoryContextSwitchTo(old);
 	MemoryContextReset(data->context);
@@ -1558,7 +1740,9 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 		entry->exprstate_valid = false;
 		entry->pubactions.pubinsert = entry->pubactions.pubupdate =
 			entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
-		entry->scantuple = NULL;
+		entry->scan_slot = NULL;
+		entry->new_slot = NULL;
+		entry->old_slot = NULL;
 		entry->exprstate[REORDER_BUFFER_CHANGE_INSERT] = NULL;
 		entry->exprstate[REORDER_BUFFER_CHANGE_UPDATE] = NULL;
 		entry->exprstate[REORDER_BUFFER_CHANGE_DELETE] = NULL;
@@ -1769,10 +1953,10 @@ rel_sync_cache_relation_cb(Datum arg, Oid relid)
 		 * Row filter cache cleanups. (Will be rebuilt later if needed).
 		 */
 		entry->exprstate_valid = false;
-		if (entry->scantuple != NULL)
+		if (entry->scan_slot != NULL)
 		{
-			ExecDropSingleTupleTableSlot(entry->scantuple);
-			entry->scantuple = NULL;
+			ExecDropSingleTupleTableSlot(entry->scan_slot);
+			entry->scan_slot = NULL;
 		}
 		/* Cleanup the ExprState for each of the pubactions. */
 		for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 83741dc..427c40a 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -16,6 +16,7 @@
 #include "access/xact.h"
 #include "replication/reorderbuffer.h"
 #include "utils/rel.h"
+#include "executor/executor.h"
 
 /*
  * Protocol capabilities
@@ -211,7 +212,11 @@ extern void logicalrep_write_insert(StringInfo out, TransactionId xid,
 extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
 extern void logicalrep_write_update(StringInfo out, TransactionId xid,
 									Relation rel, HeapTuple oldtuple,
-									HeapTuple newtuple, bool binary);
+									HeapTuple newtuple, TupleTableSlot *oldslot,
+									TupleTableSlot *newslot, bool binary);
+extern void logicalrep_write_update_cached(StringInfo out, TransactionId xid, Relation rel,
+											TupleTableSlot *oldtuple, TupleTableSlot *newtuple,
+										   bool binary);
 extern LogicalRepRelId logicalrep_read_update(StringInfo in,
 											  bool *has_oldtuple, LogicalRepTupleData *oldtup,
 											  LogicalRepTupleData *newtup);
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 5b40ff7..aec0059 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -51,7 +51,7 @@ typedef struct ReorderBufferTupleBuf
  * respectively.  They're used by INSERT .. ON CONFLICT .. UPDATE.  Users of
  * logical decoding don't have to care about these.
  */
-enum ReorderBufferChangeType
+typedef enum ReorderBufferChangeType
 {
 	REORDER_BUFFER_CHANGE_INSERT,
 	REORDER_BUFFER_CHANGE_UPDATE,
@@ -65,7 +65,7 @@ enum ReorderBufferChangeType
 	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
 	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
 	REORDER_BUFFER_CHANGE_TRUNCATE
-};
+} ReorderBufferChangeType;
 
 /* forward declaration */
 struct ReorderBufferTXN;
@@ -82,7 +82,7 @@ typedef struct ReorderBufferChange
 	XLogRecPtr	lsn;
 
 	/* The type of change. */
-	enum ReorderBufferChangeType action;
+	ReorderBufferChangeType action;
 
 	/* Transaction this change belongs to. */
 	struct ReorderBufferTXN *txn;
diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl
index abeaf76..73add45 100644
--- a/src/test/subscription/t/027_row_filter.pl
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -383,7 +383,8 @@ is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4');
 #
 # - 1001, 1002, 1980 already exist from initial data copy
 # - INSERT (800, 'test 800')   NO, because 800 is not > 1000
-# - INSERT (1600, 'test 1600') YES, because 1600 > 1000 and 'test 1600' <> 'filtered'
+# - INSERT (1600, 'test 1600') YES, because 1600 > 1000 and 'test 1600' <> 'filtered',
+#								    but row deleted after the update below.
 # - INSERT (1601, 'test 1601') YES, because 1601 > 1000 and 'test 1601' <> 'filtered'
 # - INSERT (1700, 'test 1700') YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
 # - UPDATE (1600, NULL)        NO, row filter evaluates to false because NULL is not <> 'filtered'
@@ -395,7 +396,6 @@ $result =
 	"SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
 is($result, qq(1001|test 1001
 1002|test 1002
-1600|test 1600
 1601|test 1601 updated
 1980|not filtered), 'check replicated rows to table tab_rowfilter_1');
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 915386e..982df27 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2199,6 +2199,7 @@ ReorderBufferApplyChangeCB
 ReorderBufferApplyTruncateCB
 ReorderBufferBeginCB
 ReorderBufferChange
+ReorderBufferChangeType
 ReorderBufferCommitCB
 ReorderBufferCommitPreparedCB
 ReorderBufferDiskChange
-- 
1.8.3.1