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).