From e49c8d1393333e0e4d17e98f61b9a322fe543f75 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Thu, 20 Aug 2026 21:08:55 +0800 Subject: [PATCH v3] Log unchanged out-of-line columns for row-filtered UPDATEs An UPDATE that crosses a publication row filter's boundary is transformed into an INSERT by the output plugin during logical decoding, and the INSERT needs the complete new row. However, a column that is stored out-of-line, is not part of the replica identity, and is left unchanged by the update appears nowhere in the update's WAL record: the new tuple image only carries its on-disk TOAST pointer and no new toast chunks are written for it. The transformed INSERT therefore lost the value, and the subscriber silently stored NULL instead. Fix that by having heap_update() log such values with the old-key tuple: when the table is published with a row filter and its replica identity is not FULL, BuildOldKeyTuple() (formerly ExtractReplicaIdentity(), renamed since it now builds more than the bare identity) additionally keeps unchanged, non-replica-identity, out-of-line columns, which the existing flattening step then inlines into the WAL record. pgoutput's row filter code already copies non-external values from the old tuple over external pointers in the new tuple (added for unchanged toasted replica identity columns), so no output plugin change is needed. The extra logging only kicks in when the old-key tuple is logged, which requires a change of the replica identity key. That is a necessary condition for the transformation anyway, since row filters may only reference replica identity columns. To let heap_update() know whether the table is published with a row filter, pub_rf_contains_invalid_column() now also reports whether a row filter exists, and the result is cached in the relcache publication descriptor. --- doc/src/sgml/logical-replication.sgml | 17 +++ src/backend/access/heap/heapam.c | 131 ++++++++++++++++++---- src/backend/commands/publicationcmds.c | 12 +- src/backend/utils/cache/relcache.c | 40 +++++-- src/include/catalog/pg_publication.h | 13 +++ src/include/commands/publicationcmds.h | 3 +- src/include/utils/relcache.h | 1 + src/test/subscription/t/028_row_filter.pl | 49 ++++++++ 8 files changed, 236 insertions(+), 30 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 3a61e2d6889..6adec074509 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -927,6 +927,23 @@ HINT: To initiate replication, you must manually create the replication slot, e So the UPDATE is transformed into an INSERT. + + + An INSERT resulting from this transformation needs the + complete new row, but the value of a column that is stored out-of-line + (see ) is normally not present in the WAL + for an UPDATE that did not modify the column, unless + the column is part of the replica identity. To keep such values + available to the transformation, the publisher logs them with the old + row's key when the table is published with a row filter and the update + changes a replica identity column. This slightly increases the WAL + volume for such updates; it does not happen with + REPLICA IDENTITY FULL (see + ), where the complete + old row is logged anyway. + + + summarizes the applied transformations. diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index d0ffad19a16..01103013f94 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -42,6 +42,7 @@ #include "access/xloginsert.h" #include "catalog/pg_database.h" #include "catalog/pg_database_d.h" +#include "catalog/pg_publication.h" #include "commands/vacuum.h" #include "executor/instrument_node.h" #include "pgstat.h" @@ -110,8 +111,9 @@ static bool ConditionalMultiXactIdWait(MultiXactId multi, MultiXactStatus status static void index_delete_sort(TM_IndexDeleteOp *delstate); static int bottomup_sort_and_shrink(TM_IndexDeleteOp *delstate); static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup); -static HeapTuple ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_required, - bool *copy); +static HeapTuple BuildOldKeyTuple(Relation relation, HeapTuple tp, HeapTuple newtp, + bool key_required, bool log_unchanged_external, + bool *copy); /* @@ -3009,7 +3011,7 @@ l1: * we don't PANIC upon a memory allocation failure. */ old_key_tuple = walLogical ? - ExtractReplicaIdentity(relation, &tp, true, &old_key_copied) : NULL; + BuildOldKeyTuple(relation, &tp, NULL, true, false, &old_key_copied) : NULL; /* * If this is the first possibly-multixact-able operation in the current @@ -3308,6 +3310,7 @@ heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, bool checked_lockers; bool locker_remains; bool id_has_external = false; + bool log_unchanged_external = false; TransactionId xmax_new_tuple, xmax_old_tuple; uint16 infomask_old_tuple, @@ -3366,6 +3369,20 @@ heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, interesting_attrs = bms_add_members(interesting_attrs, key_attrs); interesting_attrs = bms_add_members(interesting_attrs, id_attrs); + /* + * Determine whether this update must WAL-log the values of unchanged, + * out-of-line, non-replica-identity columns, so that a publication row + * filter's UPDATE-to-INSERT transformation can still reconstruct them + * during decoding; see BuildOldKeyTuple. + * + * The relcache lookup for row filter existence requires catalog access, + * so we perform it before acquiring the buffer lock. + */ + if (walLogical && RelationIsLogicallyLogged(relation) && + relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL && + RelationHasPubRowFilterForUpdate(relation)) + log_unchanged_external = true; + block = ItemPointerGetBlockNumber(otid); INJECTION_POINT("heap_update-before-pin", NULL); buffer = ReadBuffer(relation, block); @@ -3447,7 +3464,7 @@ heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, * old tuple is externally stored or not. This is required because for * such attributes the flattened value won't be WAL logged as part of the * new tuple so we must include it as part of the old_key_tuple. See - * ExtractReplicaIdentity. + * BuildOldKeyTuple. */ modified_attrs = HeapDetermineColumnsInfo(relation, interesting_attrs, id_attrs, &oldtup, @@ -4090,15 +4107,23 @@ l2: /* * Compute replica identity tuple before entering the critical section so - * we don't PANIC upon a memory allocation failure. - * ExtractReplicaIdentity() will return NULL if nothing needs to be - * logged. Pass old key required as true only if the replica identity key - * columns are modified or it has external data. - */ - old_key_tuple = ExtractReplicaIdentity(relation, &oldtup, - bms_overlap(modified_attrs, id_attrs) || - id_has_external, - &old_key_copied); + * we don't PANIC upon a memory allocation failure. BuildOldKeyTuple() + * will return NULL if nothing needs to be logged. Pass old key required + * as true only if the replica identity key columns are modified or it has + * external data. + * + * Ask for unchanged, non-replica-identity, out-of-line column values to + * be logged along with the key when a publication row filter could + * transform this update into an insert during decoding; without the old + * tuple having external data there is nothing to preserve. + */ + old_key_tuple = BuildOldKeyTuple(relation, &oldtup, newtup, + bms_overlap(modified_attrs, id_attrs) || + id_has_external, + log_unchanged_external && + HeapTupleHasExternal(&oldtup) && + HeapTupleHasExternal(newtup), + &old_key_copied); clear_all_visible = PageIsAllVisible(page); clear_all_visible_new = newbuf != buffer && PageIsAllVisible(newpage); @@ -9322,8 +9347,9 @@ log_heap_new_cid(Relation relation, HeapTuple tup) } /* - * Build a heap tuple representing the configured REPLICA IDENTITY to represent - * the old tuple in an UPDATE or DELETE. + * Build the old-key tuple to be WAL-logged for an UPDATE or DELETE, holding + * the old tuple's replica identity column values (or the whole old tuple, + * flattened, for REPLICA IDENTITY FULL). * * Returns NULL if there's no need to log an identity or if there's no suitable * key defined. @@ -9331,12 +9357,24 @@ log_heap_new_cid(Relation relation, HeapTuple tup) * Pass key_required true if any replica identity columns changed value, or if * any of them have any external data. Delete must always pass true. * + * For updates (newtp non-NULL), log_unchanged_external asks for the values of + * unchanged, non-replica-identity columns that are stored out-of-line to be + * included in the old-key tuple as well. Such values normally appear nowhere + * in the update's WAL record, but a publication row filter can transform the + * update into an insert during decoding, and the insert would otherwise have no + * way to reconstruct them (the output plugin copies them over from the old + * tuple, see pgoutput_row_filter()). heap_update() requests this when the + * table is published with a row filter; the extra columns only matter when the + * key is logged too, because a row filter can only reference replica identity + * columns, so the transformation requires a key change. + * * *copy is set to true if the returned tuple is a modified copy rather than * the same tuple that was passed in. */ static HeapTuple -ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_required, - bool *copy) +BuildOldKeyTuple(Relation relation, HeapTuple tp, HeapTuple newtp, + bool key_required, bool log_unchanged_external, + bool *copy) { TupleDesc desc = RelationGetDescr(relation); char replident = relation->rd_rel->relreplident; @@ -9391,13 +9429,62 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_required, */ heap_deform_tuple(tp, desc, values, nulls); + Assert(newtp != NULL || !log_unchanged_external); + for (int i = 0; i < desc->natts; i++) { + CompactAttribute *att; + Datum new_value; + bool new_isnull; + bool old_isnull = nulls[i]; + if (bms_is_member(i + 1 - FirstLowInvalidHeapAttributeNumber, idattrs)) + { Assert(!nulls[i]); - else - nulls[i] = true; + continue; + } + + /* do not log non-replica-identity columns by default */ + nulls[i] = true; + + if (!log_unchanged_external) + continue; + + /* + * When requested, keep a column whose value is stored out-of-line and + * is unchanged by the update, so that a row filter's UPDATE-to-INSERT + * transformation can reconstruct it during decoding. The value kept + * here is still an on-disk TOAST pointer; it is flattened into the + * tuple below. + * + * Both the old and the new value must be on-disk TOAST pointers to + * the same value. If the new value is not external, the WAL record + * carries it (inline, or as newly inserted toast chunks that decoding + * can reassemble), so there is nothing to preserve here. + */ + att = TupleDescCompactAttr(desc, i); + + /* only varlena columns can be stored out-of-line */ + if (att->attlen != -1 || att->attisdropped) + continue; + + /* the old value must be stored externally on-disk */ + if (old_isnull || + !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(values[i]))) + continue; + + /* the new value must be stored externally on-disk as well */ + new_value = heap_getattr(newtp, i + 1, desc, &new_isnull); + + if (new_isnull || + !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(new_value))) + continue; + + /* keep the column if the update left the toast value unchanged */ + if (heap_attr_equals(desc, i + 1, values[i], new_value, + old_isnull, new_isnull)) + nulls[i] = false; } key_tuple = heap_form_tuple(desc, values, nulls); @@ -9406,8 +9493,10 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_required, bms_free(idattrs); /* - * If the tuple, which by here only contains indexed columns, still has - * toasted columns, force them to be inlined. This is somewhat unlikely + * If the tuple, which by here only contains replica identity columns plus + * any unchanged out-of-line columns kept above, still has toasted + * columns, force them to be inlined so that the WAL record contains the + * actual data. For replica identity columns this is somewhat unlikely * since there's limits on the size of indexed columns, so we don't * duplicate toast_flatten_tuple()s functionality in the above loop over * the indexed columns, even if it would be more efficient. diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 96838730fe1..ebbde79dfa2 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -275,10 +275,18 @@ contain_invalid_rfcolumn_walker(Node *node, rf_context *context) * REPLICA IDENTITY index or not. * * Returns true if any invalid column is found. + * + * On return, *rf_exists is set to true iff this publication has a row + * filter defined for the relation, independently of whether that filter is + * valid. It is left untouched (and so should be initialized by the caller) + * when REPLICA IDENTITY FULL makes the question moot: in that case the old + * tuple's full contents are always available, so callers that use + * *rf_exists to decide whether extra protection is needed don't need to + * know whether a filter exists. */ bool pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors, - bool pubviaroot) + bool pubviaroot, bool *rf_exists) { HeapTuple rftuple; Oid relid = RelationGetRelid(relation); @@ -328,6 +336,8 @@ pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors, Node *rfnode; Bitmapset *bms = NULL; + *rf_exists = true; + context.pubviaroot = pubviaroot; context.parentid = publish_as_relid; context.relid = relid; diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 9abbaeab4a9..fa75d33e815 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5916,14 +5916,22 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) * row filters and we can skip the validation. */ if (!pubform->puballtables && - (pubform->pubupdate || pubform->pubdelete) && - pub_rf_contains_invalid_column(pubid, relation, ancestors, - pubform->pubviaroot)) + (pubform->pubupdate || pubform->pubdelete)) { - if (pubform->pubupdate) - pubdesc->rf_valid_for_update = false; - if (pubform->pubdelete) - pubdesc->rf_valid_for_delete = false; + bool rf_exists = false; + + if (pub_rf_contains_invalid_column(pubid, relation, ancestors, + pubform->pubviaroot, + &rf_exists)) + { + if (pubform->pubupdate) + pubdesc->rf_valid_for_update = false; + if (pubform->pubdelete) + pubdesc->rf_valid_for_delete = false; + } + + if (rf_exists && pubform->pubupdate) + pubdesc->rf_exists_for_update = true; } /* @@ -5999,6 +6007,24 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) MemoryContextSwitchTo(oldcxt); } +/* + * Check whether the table is published with a row filter for UPDATEs. + */ +bool +RelationHasPubRowFilterForUpdate(Relation relation) +{ + if (!relation->rd_pubdesc) + { + PublicationDesc pubdesc; + + RelationBuildPublicationDesc(relation, &pubdesc); + + return pubdesc.rf_exists_for_update; + } + + return relation->rd_pubdesc->rf_exists_for_update; +} + static bytea ** CopyIndexAttOptions(bytea **srcopts, int natts) { diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 89b4bb14f62..ffa2b71414c 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -118,6 +118,19 @@ typedef struct PublicationDesc */ bool gencols_valid_for_update; bool gencols_valid_for_delete; + + /* + * true if some publication which publishes UPDATEs has a row filter on + * this relation. heap_update() consults this to decide whether it needs + * to protect unchanged, non-replica-identity, TOASTed column values that + * a row filter's UPDATE-to-INSERT transformation may need but cannot find + * in WAL. Note that when an UPDATE reaches heap_update(), any such row + * filter is known to reference only replica identity columns (otherwise + * CheckCmdReplicaIdentity() would have rejected the statement), which is + * what makes a replica identity change a necessary condition for the + * transformation. + */ + bool rf_exists_for_update; } PublicationDesc; #ifdef EXPOSE_TO_CLIENT_CODE diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h index 4cf45c17cc5..9238aade1ac 100644 --- a/src/include/commands/publicationcmds.h +++ b/src/include/commands/publicationcmds.h @@ -32,7 +32,8 @@ extern ObjectAddress AlterPublicationOwner(const char *name, Oid newOwnerId); extern void AlterPublicationOwner_oid(Oid pubid, Oid newOwnerId); extern void InvalidatePublicationRels(List *relids); extern bool pub_rf_contains_invalid_column(Oid pubid, Relation relation, - List *ancestors, bool pubviaroot); + List *ancestors, bool pubviaroot, + bool *rf_exists); extern bool pub_contains_invalid_column(Oid pubid, Relation relation, List *ancestors, bool pubviaroot, char pubgencols_type, diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h index 89c27aa1529..3618d7a6b4b 100644 --- a/src/include/utils/relcache.h +++ b/src/include/utils/relcache.h @@ -90,6 +90,7 @@ extern void RelationInitIndexAccessInfo(Relation relation); struct PublicationDesc; extern void RelationBuildPublicationDesc(Relation relation, struct PublicationDesc *pubdesc); +extern bool RelationHasPubRowFilterForUpdate(Relation relation); extern void RelationInitTableAccessMethod(Relation relation); diff --git a/src/test/subscription/t/028_row_filter.pl b/src/test/subscription/t/028_row_filter.pl index c666ae00483..d88425d9457 100644 --- a/src/test/subscription/t/028_row_filter.pl +++ b/src/test/subscription/t/028_row_filter.pl @@ -799,6 +799,55 @@ is($result, qq(), 'check replicated rows to tab_rowfilter_viaroot_part_1'); # Testcase end: FOR TABLE with row filter publications # ====================================================== +# ==================================================================== +# Testcase start: UPDATE-to-INSERT transformation of a row with an +# unchanged out-of-line column +# +# Decoding can transform an UPDATE into an INSERT when the row moves into +# the set of rows satisfying the row filter. An unchanged externally +# stored value of a column that is not part of the replica identity is +# not present in the update's WAL record, so heap_update() logs it along +# with the old-key tuple, letting the transformation send the complete +# row to the subscriber. + +# create tables pub and sub +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab_rowfilter_toast_loss (id int PRIMARY KEY, val text)"); +$node_subscriber->safe_psql('postgres', + "CREATE TABLE tab_rowfilter_toast_loss (id int PRIMARY KEY, val text)"); +$node_publisher->safe_psql('postgres', + "ALTER TABLE tab_rowfilter_toast_loss ALTER COLUMN val SET STORAGE EXTERNAL"); +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION tap_pub_toast_loss FOR TABLE tab_rowfilter_toast_loss WHERE (id = 7)"); +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION tap_sub ADD PUBLICATION tap_pub_toast_loss"); + +# wait for the sync of the newly added publication's table to finish +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname); + +# This row does not satisfy the row filter, so it is not replicated. +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_rowfilter_toast_loss VALUES (3, repeat('a', 5000))"); +$node_publisher->wait_for_catchup($appname); +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*) FROM tab_rowfilter_toast_loss"); +is($result, qq(0), 'row not satisfying the row filter is not replicated'); + +# Move the row into the filter's set, leaving the out-of-line column +# unchanged: the UPDATE is transformed into an INSERT, which must still +# carry the unchanged externally stored value. +$node_publisher->safe_psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET id = 7"); +$node_publisher->wait_for_catchup($appname); +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*) FROM tab_rowfilter_toast_loss WHERE id = 7 AND val = repeat('a', 5000)"); +is($result, qq(1), + 'transformed INSERT carries the unchanged out-of-line value'); + +# Testcase end: UPDATE-to-INSERT transformation of a row with an +# unchanged out-of-line column +# ==================================================================== + $node_subscriber->stop('fast'); $node_publisher->stop('fast'); -- 2.47.3