v50-0005-UPDATEs-might-require-transformation.patch

From 97cfd82888e74f261e7cf446444dd75c20196abd Mon Sep 17 00:00:00 2001
From: Euler Taveira <euler.taveira@enterprisedb.com>
Date: Wed, 8 Dec 2021 17:57:14 -0300
Subject: [PATCH v50 5/8] UPDATEs might require transformation

UPDATE should evaluate the row filter for old and new tuple. If both
evaluations are true, it sends the UPDATE. If both evaluations are
false, it doesn't send the UPDATE. If only one of the tuples matches the
row filter expression, there is a data consistency issue. Fix this issue
requires a transformation.

Let's say the old tuple satisfies the row filter but the new tuple
doesn't. Since the old tuple satisfies, the initial table
synchronization copied this row (or another method was used to guarantee
that there is data consistency).  However, after the UPDATE the new
tuple doesn't satisfy the row filter then, from the data consistency
perspective, that row should be removed on the subscriber. The UPDATE
should be transformed into a DELETE statement and be sent to the
subscriber. Keep this row on the subscriber is undesirable because it
doesn't reflect what was defined in the row filter expression on the
publisher. This row on the subscriber would likely not be modified by
replication again. If someone inserted a new row with the same old
identifier, replication could stop due to a constraint violation.

Let's say the old tuple doesn't match the row filter but the new tuple
does. Since the old tuple doesn't satisfy, the initial table
synchronization probably didn't copy this row. However, after the UPDATE
the new tuple does satisfies the row filter then, from the data
consistency perspective, that row should inserted on the subscriber. The
UPDATE should be transformed into a INSERT statement and be sent to the
subscriber. Subsequent UPDATE or DELETE statements have no effect.
However, this might surprise someone who expects the data set to satisfy
the row filter expression on the provider.
---
 src/backend/replication/pgoutput/pgoutput.c | 279 ++++++++++++++++++--
 1 file changed, 253 insertions(+), 26 deletions(-)

diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 7192948399..7e5b93a135 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -94,6 +94,14 @@ static void send_repl_origin(LogicalDecodingContext *ctx,
 							 RepOriginId origin_id, XLogRecPtr origin_lsn,
 							 bool send_origin);
 
+/* index for exprstate array in RelationSyncEntry */
+enum PublishAction
+{
+	PUBLISH_ACTION_INSERT,
+	PUBLISH_ACTION_UPDATE,
+	PUBLISH_ACTION_DELETE
+};
+
 /*
  * Entry in the map used to remember which relation schemas we sent.
  *
@@ -176,9 +184,11 @@ static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
 static EState *create_estate_for_relation(Relation rel);
 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 void pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry);
+static bool pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple,
+								HeapTuple newtuple, RelationSyncEntry *entry, int action);
+static bool pgoutput_row_filter_for_update(PGOutputData *data, Relation relation, HeapTuple oldtuple,
+								HeapTuple newtuple, RelationSyncEntry *entry, enum PublishAction *action, HeapTuple transformedtuple);
 
 /*
  * Specify output plugin callbacks
@@ -742,26 +752,15 @@ 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.
+ * Initialize the row filter, the first time.
  */
-static bool
-pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
-					Relation relation, HeapTuple oldtuple, HeapTuple newtuple,
-					RelationSyncEntry *entry)
+static void
+pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry)
 {
-	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 */
 
-	Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
-		   changetype == REORDER_BUFFER_CHANGE_UPDATE ||
-		   changetype == REORDER_BUFFER_CHANGE_DELETE);
 	/*
 	 * 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.
@@ -926,9 +925,31 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
 
 		entry->exprstate_valid = true;
 	}
+}
 
-	/* Bail out if there is no row filter */
-	if (!entry->exprstate[changetype])
+/*
+ * 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(PGOutputData *data, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, int action)
+{
+	EState	   *estate;
+	ExprContext *ecxt;
+	bool		result = false;
+	Oid			relid = RelationGetRelid(relation);
+
+	Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
+		   changetype == REORDER_BUFFER_CHANGE_UPDATE ||
+		   changetype == REORDER_BUFFER_CHANGE_DELETE);
+
+	/*
+	 * Bail out if for a certain operation there is no row filter to process.
+	 * This is a fast path optimization. Read the explanation above about
+	 * rf_in_all_pubs.
+	 */
+	if (entry->exprstate[action] == NULL)
 		return true;
 
 	if (message_level_is_interesting(DEBUG3))
@@ -944,16 +965,22 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
 	ecxt = GetPerTupleExprContext(estate);
 	ecxt->ecxt_scantuple = entry->scantuple;
 
+	/*
+	 * The default behavior for UPDATEs is to use the new tuple for row
+	 * filtering.
+	 * If the UPDATE requires a transformation, the new tuple will be replaced
+	 * by the transformed tuple before calling this routine.
+	 */
 	ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
 
 	/*
 	 * NOTE: Multiple publication row filters have already been combined to a
 	 * single exprstate (for this pubaction).
 	 */
-	if (entry->exprstate[changetype])
+	if (entry->exprstate[action])
 	{
 		/* Evaluates row filter */
-		result = pgoutput_row_filter_exec_expr(entry->exprstate[changetype], ecxt);
+		result = pgoutput_row_filter_exec_expr(entry->exprstate[action], ecxt);
 	}
 
 	/* Cleanup allocated resources */
@@ -964,6 +991,182 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
 	return result;
 }
 
+/*
+ * Evaluates the row filter for old and new tuple. If both evaluations are
+ * true, it sends the UPDATE. If both evaluations are false, it doesn't send
+ * the UPDATE. If only one of the tuples matches the row filter expression,
+ * there is a data consistency issue. Fix this issue requires a transformation.
+ *
+ * Let's say the old tuple satisfies the row filter but the new tuple doesn't.
+ * Since the old tuple satisfies, the initial table synchronization copied this
+ * row (or another method was used to guarantee that there is data
+ * consistency).  However, after the UPDATE the new tuple doesn't satisfy the
+ * row filter then, from the data consistency perspective, that row should be
+ * removed on the subscriber. The UPDATE should be transformed into a DELETE
+ * statement and be sent to the subscriber. Keep this row on the subscriber is
+ * undesirable because it doesn't reflect what was defined in the row filter
+ * expression on the publisher. This row on the subscriber would likely not be
+ * modified by replication again. If someone inserted a new row with the same
+ * old identifier, replication could stop due to a constraint violation.
+ *
+ * Let's say the old tuple doesn't match the row filter but the new tuple does.
+ * Since the old tuple doesn't satisfy, the initial table synchronization
+ * probably didn't copy this row. However, after the UPDATE the new tuple does
+ * satisfies the row filter then, from the data consistency perspective, that
+ * row should inserted on the subscriber. The UPDATE should be transformed into
+ * a INSERT statement and be sent to the subscriber. Subsequent UPDATE or
+ * DELETE statements have no effect (it matches no row -- see
+ * apply_handle_update_internal()). However, this might surprise someone who
+ * expects the data set to satisfy the row filter expression on the provider.
+ */
+static bool
+pgoutput_row_filter_for_update(PGOutputData *data, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, enum PublishAction *action, HeapTuple transformedtuple)
+{
+	MemoryContext	oldctx;
+	Oid			relid = RelationGetRelid(relation);
+	bool		oldtuple_match, newtuple_match;
+	Datum		*oldvalues, *newvalues;
+	bool		*oldnulls, *newnulls, *replaces;
+	TupleDesc	tupdesc;
+	int			nratts;
+	int			i;
+
+	/*
+	 * Bail out if for a certain operation there is no row filter to process.
+	 * This is a fast path optimization. Read the explanation in
+	 * pgoutput_row_filter_init().
+	 */
+	if (entry->exprstate[*action] == NULL)
+		return true;
+
+	/* Update requires a new tuple. */
+	Assert(newtuple);
+
+	elog(DEBUG3, "table \"%s.%s\" has row filter",
+		 get_namespace_name(get_rel_namespace(relid)),
+		 get_rel_name(relid));
+
+	/*
+	 * If there is no oldtuple available then none of the replica identity
+	 * columns were modified. No transformation is required in this case.
+	 */
+	if (!oldtuple)
+		return pgoutput_row_filter(data, relation, NULL, newtuple, entry, PUBLISH_ACTION_UPDATE);
+
+	/* Evaluates row filter in the old tuple. */
+	oldtuple_match = pgoutput_row_filter(data, relation, oldtuple, NULL, entry, PUBLISH_ACTION_UPDATE);
+
+	/*
+	 * Create a tuple table slot for row filter. TupleDesc must live as
+	 * long as the cache remains. Release the tuple table slot if it
+	 * already exists.
+	 */
+	if (entry->scantuple != NULL)
+	{
+		ExecDropSingleTupleTableSlot(entry->scantuple);
+		entry->scantuple = NULL;
+	}
+	tupdesc = RelationGetDescr(relation);
+	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+	entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+	MemoryContextSwitchTo(oldctx);
+
+	/*
+	 * Obtain a transformed tuple based on the new tuple. This step is
+	 * necessary because the new tuple might not have unchanged TOAST values
+	 * from the replica identity.  Copy them from the old tuple where it is
+	 * available.
+	 */
+	nratts = tupdesc->natts;
+
+	oldvalues = (Datum *) palloc(nratts * sizeof(Datum));
+	oldnulls = (bool *) palloc(nratts * sizeof(bool));
+	newvalues = (Datum *) palloc(nratts * sizeof(Datum));
+	newnulls = (bool *) palloc(nratts * sizeof(bool));
+	replaces = (bool *) palloc(nratts * sizeof(bool));
+
+	/* Indicates the modified values. */
+	memset(replaces, false, nratts * sizeof(bool));
+
+	/* Break down the tuples into fields. */
+	heap_deform_tuple(oldtuple, tupdesc, oldvalues, oldnulls);
+	heap_deform_tuple(newtuple, tupdesc, newvalues, newnulls);
+
+	/* Modify the transformed tuple if it does not have the value yet. */
+	for (i = 0; i < nratts; i++)
+	{
+		Form_pg_attribute att;
+
+		/* column in new tuple is null, nothing to do */
+		if (newnulls[i])
+			continue;
+
+		att = TupleDescAttr(tupdesc, i);
+
+		/*
+		 * Unchanged TOASTed replica identity columns are only detoasted in the
+		 * old tuple, copy this over to the new tuple.
+		 */
+		if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(newvalues[i]) &&
+				!oldnulls[i] && !VARATT_IS_EXTERNAL_ONDISK(oldvalues[i]))
+		{
+			newvalues[i] = oldvalues[i];
+			replaces[i] = true;
+		}
+	}
+
+	/* Transformed tuple is the new tuple with some possible modifications. */
+	transformedtuple = heap_modify_tuple(newtuple, tupdesc, newvalues, newnulls, replaces);
+
+	/*
+	 * Evaluates row filter in the transformed tuple. The new tuple isn't used
+	 * because it might not contain all values used in the row filter
+	 * expression.
+	 */
+	newtuple_match = pgoutput_row_filter(data, relation, NULL, transformedtuple, entry, PUBLISH_ACTION_UPDATE);
+
+	/*
+	 * Case 1: if both tuples matches the row filter, transformation isn't
+	 * required. Bail out. Send the UPDATE.
+	 */
+	if (oldtuple_match && newtuple_match)
+		return true;
+
+	/*
+	 * Case 2: if both tuples don't match the row filter, bail out. Send
+	 * nothing.
+	 */
+	if (!oldtuple_match && !newtuple_match)
+		return false;
+
+	/*
+	 * Case 3: if the old tuple satisfies the row filter but the new tuple
+	 * doesn't, transform the UPDATE into DELETE.
+	 *
+	 * This transformation does not require another tuple. Old tuple will be
+	 * used for DELETE.
+	 */
+	if (oldtuple_match && !newtuple_match)
+	{
+		*action = PUBLISH_ACTION_DELETE;
+		return true;
+	}
+
+	/*
+	 * Case 4: if the old tuple doesn't satisfy the row filter but the new
+	 * tuple does, transform the UPDATE into INSERT.
+	 *
+	 * This transformation requires another tuple. This transformed tuple will
+	 * be used for INSERT. The new tuple is the base for the transformed tuple.
+	 * However, the new tuple might not have column values from the replica
+	 * identity. In this case, copy these values from the old tuple.
+	 */
+	if (!oldtuple_match && newtuple_match)
+		*action = PUBLISH_ACTION_INSERT;
+
+	return true;
+}
+
 /*
  * Sends the decoded DML over wire.
  *
@@ -1015,6 +1218,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)
 	{
@@ -1023,7 +1229,7 @@ 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))
+				if (!pgoutput_row_filter(data, relation, NULL, tuple, relentry, PUBLISH_ACTION_INSERT))
 					break;
 
 				/*
@@ -1054,9 +1260,11 @@ 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;
+				HeapTuple	transformedtuple = NULL;
+				enum PublishAction transformedaction = PUBLISH_ACTION_UPDATE;
 
 				/* Check row filter. */
-				if (!pgoutput_row_filter(change->action, data, relation, oldtuple, newtuple, relentry))
+				if (!pgoutput_row_filter_for_update(data, relation, oldtuple, newtuple, relentry, &transformedaction, transformedtuple))
 					break;
 
 				maybe_send_schema(ctx, change, relation, relentry);
@@ -1075,12 +1283,31 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 															  relentry->map);
 						newtuple = execute_attr_map_tuple(newtuple,
 														  relentry->map);
+						/* INSERT uses transformed tuple. */
+						if (transformedaction == PUBLISH_ACTION_INSERT)
+							transformedtuple = execute_attr_map_tuple(transformedtuple, relentry->map);
 					}
 				}
 
 				OutputPluginPrepareWrite(ctx, true);
-				logicalrep_write_update(ctx->out, xid, relation, oldtuple,
-										newtuple, data->binary);
+				switch (transformedaction)
+				{
+					case PUBLISH_ACTION_INSERT:
+						logicalrep_write_insert(ctx->out, xid, relation, transformedtuple,
+												data->binary);
+						break;
+					case PUBLISH_ACTION_UPDATE:
+						logicalrep_write_update(ctx->out, xid, relation,
+												oldtuple, newtuple,
+												data->binary);
+						break;
+					case PUBLISH_ACTION_DELETE:
+						logicalrep_write_delete(ctx->out, xid, relation, oldtuple,
+												data->binary);
+						break;
+					default:
+						Assert(false);
+				}
 				OutputPluginWrite(ctx, true);
 				break;
 			}
@@ -1090,7 +1317,7 @@ 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))
+				if (!pgoutput_row_filter(data, relation, oldtuple, NULL, relentry, PUBLISH_ACTION_DELETE))
 					break;
 
 				maybe_send_schema(ctx, change, relation, relentry);
-- 
2.20.1