v67_changes_amit_1.patch

diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 393d9100e4..888a461068 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -447,10 +447,10 @@ IsRowFilterSimpleExpr(Node *node)
  * unpleasant results because a historic snapshot is used. That's why only
  * immutable built-in functions are allowed in row filter expressions.
  *
- * We also don't allow system columns because the tuple from reorderbuffer do
- * not have system columns in its header. We can change that if requested but
- * there seems no need to filter system columns as we do not replicate it to
- * subscriber side.
+ * We don't allow system columns because currently, we don't have that
+ * information in the tuple passed to downstream. Also, as we don't replicate
+ * those to subscribers, there doesn't seem to be a need for a filter on those
+ * columns.
  */
 static bool
 check_simple_rowfilter_expr_walker(Node *node, Relation relation)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 12c9775c40..9cf52b0a3d 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -134,22 +134,16 @@ typedef struct RelationSyncEntry
 	bool		replicate_valid;
 	PublicationActions pubactions;
 
+	/* indicates whether row filter expr cache is valid */
+	bool		exprstate_valid;
+
 	/*
-	 * ExprState cannot be used to indicate no cache, invalid cache and valid
-	 * cache, so the flag exprstate_valid indicates if the current cache is
-	 * valid.
-	 *
-	 * Multiple ExprState entries might be used if there are multiple
-	 * publications for a single table. Different publication actions don't
+	 * ExprState array for row filter. Different publication actions don't
 	 * allow multiple expressions to always be combined into one, because
 	 * updates or deletes restrict the column in expression to be part of the
 	 * replica identity index whereas inserts do not have this restriction, so
-	 * there is one ExprState per publication action. The exprstate array is
-	 * indexed by ReorderBufferChangeType.
+	 * there is one ExprState per publication action.
 	 */
-	bool		exprstate_valid;
-
-	/* ExprState array for row filter. One per publication action. */
 	ExprState  *exprstate[NUM_ROWFILTER_PUBACTIONS];
 	MemoryContext cache_expr_cxt;	/* private context for exprstate, if any */
 
@@ -754,8 +748,6 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 	 * 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.
 	 *
-	 * This code is usually one-time execution.
-	 *
 	 * NOTE: The ExprState cache could have been created up-front in the
 	 * function get_rel_sync_entry() instead of the deferred on-the-fly
 	 * assignment below. The reason for choosing to do it here is because
@@ -767,8 +759,8 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 	 * get_rel_sync_entry) but which don't need to build ExprState.
 	 * Furthermore, because the decision to publish or not is made AFTER the
 	 * call to get_rel_sync_entry it may be that the filter evaluation is not
-	 * necessary at all. So the decision was to defer this logic to last
-	 * moment when we know it will be needed.
+	 * necessary at all. This avoids us to consume memory and spend CPU cycles
+	 * when we don't need to.
 	 */
 	if (entry->exprstate_valid)
 		return;
@@ -833,8 +825,7 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 			bool		rfisnull;
 
 			/*
-			 * Lookup if there is a row filter, If no, then remember there
-			 * was no filter for this pubaction.
+			 * Check for the presence of a row filter in this publication.
 			 */
 			rftuple = SearchSysCache2(PUBLICATIONRELMAP,
 									  ObjectIdGetDatum(entry->publish_as_relid),
@@ -853,13 +844,17 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 				List	   *relids;
 
 				/*
-				 * If no record in publication and the table is a partition,
-				 * check if the table's parent table is published by this
-				 * publication. If not, it means the table is not published by
-				 * this publication and we should skip this publication instead
-				 * of setting pub_no_filter flag. If the parent table is
-				 * published by this publication, it means the table has no
-				 * filter.
+				 * It is possible that one of the parent tables for this
+				 * partition is published via this publication in which case we
+				 * can deduce that we don't need to use any filter for it,
+				 * otherwise, we skip this publication. This is because when we
+				 * don't publicize the change via root, we use the individual
+				 * partition's filter.
+				 *
+				 * XXX We can avoid the need to check for the parent table if
+				 * we cache the list of publications for each RelationSyncEntry
+				 * but this case will be rare and we have to do this only the
+				 * first time we build the row filter expression.
 				 */
 				schemarelids = GetAllSchemaPublicationRelations(pub->oid,
 																PUBLICATION_PART_LEAF);
@@ -892,7 +887,10 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 			no_filter[PUBACTION_UPDATE] |= pub->pubactions.pubupdate;
 			no_filter[PUBACTION_DELETE] |= pub->pubactions.pubdelete;
 
-			/* Quick exit loop if all pubactions have no row filter. */
+			/*
+			 * Quick exit if all the DML actions are publicized via this
+			 * publication.
+			 */
 			if (no_filter[PUBACTION_INSERT] &&
 				no_filter[PUBACTION_UPDATE] &&
 				no_filter[PUBACTION_DELETE])
@@ -905,11 +903,7 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 			continue;
 		}
 
-		/*
-		 * If row filter exists remember it in a list (per pubaction).
-		 * Code following this 'publications' loop will combine all
-		 * filters.
-		 */
+		/* Form the per pubaction row filter lists. */
 		if (pub->pubactions.pubinsert && !no_filter[PUBACTION_INSERT])
 			rfnodes[PUBACTION_INSERT] = lappend(rfnodes[PUBACTION_INSERT],
 												TextDatumGetCString(rfdatum));
@@ -935,12 +929,9 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 		}
 	}
 
-	/* If no filter found, clean up the memory and return */
+	/* We are done if there are no applicable row filters */
 	if (!has_filter)
 	{
-		if (entry->cache_expr_cxt != NULL)
-			MemoryContextDelete(entry->cache_expr_cxt);
-
 		entry->exprstate_valid = true;
 		return;
 	}
@@ -960,8 +951,7 @@ pgoutput_row_filter_init(PGOutputData *data, RelationSyncEntry *entry)
 	 * All row filter expressions will be discarded if there is one
 	 * publication-relation entry without a row filter. That's because all
 	 * expressions are aggregated by the OR operator. The row filter
-	 * absence means replicate all rows so a single valid expression means
-	 * publish this row.
+	 * absence means replicate all rows.
 	 */
 	oldctx = MemoryContextSwitchTo(entry->cache_expr_cxt);
 	for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
@@ -1016,22 +1006,21 @@ pgoutput_tuple_slot_init(Relation relation, RelationSyncEntry *entry)
  *
  * For inserts: evaluates the row filter for new tuple.
  * For deletes: evaluates the row filter for old tuple.
- * For updates: 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. Fixing this issue requires a
- * transformation.
+ * For updates: evaluates the row filter for old and new tuple.
  *
- * Transformations:
- * 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.
+ * For updates, if both evaluations are true, we allow to send the UPDATE and
+ * if both the evaluations are false, it doesn't replicate the UPDATE. Now, if
+ * only one of the tuples matches the row filter expression, we transform
+ * UPDATE to DELETE or INSERT to avoid any data inconsistency based on the
+ * following rules:
  *
  * Case 1: old-row (no match)    new-row (no match)  -> (drop change)
  * Case 2: old-row (no match)    new row (match)     -> INSERT
  * Case 3: old-row (match)       new-row (no match)  -> DELETE
  * Case 4: old-row (match)       new row (match)     -> UPDATE
  *
+ * The new action is updated in the action parameter.
+ *
  * Examples:
  * 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
@@ -1039,8 +1028,8 @@ pgoutput_tuple_slot_init(Relation relation, RelationSyncEntry *entry)
  * consistency).  However, after the UPDATE the new tuple doesn't satisfy the
  * row filter, so from a 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
+ * statement and be sent to the subscriber. Keeping 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.
@@ -1071,6 +1060,10 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
 	ExprContext	   *ecxt;
 	ExprState	   *filter_exprstate;
 
+	/*
+	 * We need this map  to avoid relying on changes in ReorderBufferChangeType
+	 * enum.
+	 */
 	static int map_changetype_pubaction[] = {
 		[REORDER_BUFFER_CHANGE_INSERT] = PUBACTION_INSERT,
 		[REORDER_BUFFER_CHANGE_UPDATE] = PUBACTION_UPDATE,
@@ -1103,13 +1096,15 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
 	 * For the following occasions where there is only one tuple, we can
 	 * evaluates the row filter for that tuple and return.
 	 *
-	 * For inserts we only have the new tuple.
+	 * For inserts, we only have the new tuple.
 	 *
-	 * For updates if no old tuple, it means none of the replica identity
-	 * columns changed and this would reduce to a simple update. We only need
-	 * to evaluate the row filter for the new tuple.
+	 * For updates, we can have only a new tuple when none of the replica
+	 * identity columns changed. Ideally, we don't need to evaluate the row
+	 * filter in this case as for updates only replica identity columns are
+	 * allowed but users can use constant expressions in the row filter. So, we
+	 * evaluate the row filter for the new tuple in this case.
 	 *
-	 * For deletes we only have the old tuple.
+	 * For deletes, we only have the old tuple.
 	 */
 	if (!new_slot || !old_slot)
 	{
@@ -2056,6 +2051,8 @@ 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->cache_expr_cxt != NULL)
+			MemoryContextDelete(entry->cache_expr_cxt);
 
 		/*
 		 * Tuple slots cleanups. (Will be rebuilt later if needed).