From fc935f1a648148987674ceafe022b9e2d958253d Mon Sep 17 00:00:00 2001 From: B1 Implementer Date: Mon, 17 Aug 2026 23:03:00 +0800 Subject: [PATCH v3 18] Reject UPDATEs that would silently lose a row-filtered TOASTed column. Fix logical replication row filters silently dropping unchanged TOASTed non-key columns when an UPDATE crosses a filter's boundary and gets transformed into an INSERT by the output plugin, since only the replica identity is guaranteed to be present in WAL for such columns. heap_update() now detects this case directly: when a table has a row filter and the replica identity key changes, it checks whether any unchanged, non-replica-identity column is stored out of line, and raises an error naming the column instead of silently producing an incomplete row on the subscriber. Back-patch to 18: instead of extending pub_rf_contains_invalid_column() with an rf_exists parameter as on master, add a separate pub_has_row_filter() helper, to keep the back-branch change smaller. --- doc/src/sgml/logical-replication.sgml | 21 ++++ src/backend/access/heap/heapam.c | 123 +++++++++++++++++++++- src/backend/commands/publicationcmds.c | 41 ++++++++ src/backend/utils/cache/relcache.c | 37 +++++++ src/include/catalog/pg_publication.h | 15 +++ src/include/commands/publicationcmds.h | 2 + src/include/utils/relcache.h | 1 + src/test/subscription/t/028_row_filter.pl | 56 ++++++++++ 8 files changed, 294 insertions(+), 2 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 2c032bafa2a..6f137d53939 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -901,6 +901,27 @@ 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 cannot + carry the value of a column that is stored out-of-line (see + ) if the update did not modify the + column, unless the column is part of the replica identity. Such values + are not present in the WAL for the change, so they cannot be + reconstructed when decoding. To prevent silent data loss on the + subscriber, the publisher rejects with an error any + UPDATE that changes a replica identity column of a + table published with a row filter while leaving an out-of-line column + that is not part of the replica identity unchanged. The error can also + be raised when the transformation would not actually happen, because + the row filter expressions are not evaluated when the decision is made. + To allow such updates, set the table's replica identity to + FULL (see + ), or remove the row + filter from the publication. + + + summarizes the applied transformations. diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index d72b41ef92f..61115298e05 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 "pgstat.h" #include "port/pg_bitutils.h" @@ -73,6 +74,10 @@ static Bitmapset *HeapDetermineColumnsInfo(Relation relation, Bitmapset *external_cols, HeapTuple oldtup, HeapTuple newtup, bool *has_external); +static AttrNumber HeapCheckRowFilterUnchangedExternal(Relation relation, + HeapTuple oldtup, + HeapTuple newtup, + Bitmapset *id_attrs); static bool heap_acquire_tuplock(Relation relation, ItemPointer tid, LockTupleMode mode, LockWaitPolicy wait_policy, bool *have_tuple_lock); @@ -3361,6 +3366,8 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup, bool checked_lockers; bool locker_remains; bool id_has_external = false; + bool id_changed; + bool check_unchanged_external = false; TransactionId xmax_new_tuple, xmax_old_tuple; uint16 infomask_old_tuple, @@ -3419,6 +3426,19 @@ heap_update(Relation relation, ItemPointer 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 be checked for unchanged out-of-line + * column values that a row filter's UPDATE-to-INSERT transformation would + * lose during decoding; see HeapCheckRowFilterUnchangedExternal. + * + * The relcache lookup for row filter existence requires catalog access, + * so we perform it before acquiring the buffer lock. + */ + if (RelationIsLogicallyLogged(relation) && + relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL && + RelationHasPubRowFilterForUpdate(relation)) + check_unchanged_external = true; + block = ItemPointerGetBlockNumber(otid); INJECTION_POINT("heap_update-before-pin", NULL); buffer = ReadBuffer(relation, block); @@ -3800,6 +3820,39 @@ l2: goto l2; } + id_changed = bms_overlap(modified_attrs, id_attrs); + + /* + * If the update could be transformed into an insert by a publication row + * filter during decoding, reject it when it would lose an unchanged + * out-of-line value of a column that is not part of the replica identity. + * + * This check should be done when the update is about to be performed, + * after all the locking and visibility checks, so that we don't reject + * updates that are not going to succeed anyway. + */ + if (check_unchanged_external && id_changed && + HeapTupleHasExternal(newtup) && HeapTupleHasExternal(&oldtup)) + { + AttrNumber attnum; + + attnum = HeapCheckRowFilterUnchangedExternal(relation, &oldtup, + newtup, id_attrs); + + if (AttributeNumberIsValid(attnum)) + { + UnlockReleaseBuffer(buffer); + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot update table \"%s\" because it is published with a row filter and column \"%s\" has an unchanged externally stored value", + RelationGetRelationName(relation), + NameStr(TupleDescAttr(RelationGetDescr(relation), + attnum - 1)->attname)), + errdetail("The row filter can transform the update into an insert during logical replication, and an insert cannot carry unchanged externally stored values."), + errhint("To enable updating the table, set REPLICA IDENTITY FULL or remove the row filter from the publication."))); + } + } + /* Fill in transaction status data */ /* @@ -4149,8 +4202,7 @@ l2: * columns are modified or it has external data. */ old_key_tuple = ExtractReplicaIdentity(relation, &oldtup, - bms_overlap(modified_attrs, id_attrs) || - id_has_external, + id_changed || id_has_external, &old_key_copied); clear_all_visible = PageIsAllVisible(page); @@ -4676,6 +4728,73 @@ HeapDetermineColumnsInfo(Relation relation, return modified; } +/* + * Check for a non-replica-identity, TOASTed column whose value is left + * unchanged by this update. + * + * Caller has already established that this relation is published by some + * publication with a row filter, and that the replica identity key changed + * value in this update. Since a row filter is only allowed to reference + * replica identity columns (see pub_rf_contains_invalid_column and + * CheckCmdReplicaIdentity), a changing key is the only way this update could + * flip the filter's old/new match status and trigger the output plugin's + * UPDATE-to-INSERT transformation. That transformation needs the complete + * new row, but an unchanged out-of-line column's value is not present + * anywhere in this transaction's WAL: heap_update() leaves such a column's + * on-disk TOAST pointer as-is instead of rewriting it, and unlike replica + * identity columns (see ExtractReplicaIdentity), nothing flattens it into + * the record for any other column. + * + * Returns the attribute number of the first such column found, or + * InvalidAttrNumber if none exists. Note that this looks only at what the WAL + * can represent; it does not know which specific publications' column lists or + * generated-column settings would actually cause the column to be sent, so it + * can conservatively flag a column that some particular subscriber's + * publication would never have published anyway. + */ +static AttrNumber +HeapCheckRowFilterUnchangedExternal(Relation relation, HeapTuple oldtup, + HeapTuple newtup, Bitmapset *id_attrs) +{ + TupleDesc desc = RelationGetDescr(relation); + + for (AttrNumber attnum = 1; attnum <= desc->natts; attnum++) + { + CompactAttribute *att = TupleDescCompactAttr(desc, attnum - 1); + int attidx = attnum - FirstLowInvalidHeapAttributeNumber; + Datum new_value; + Datum old_value; + bool new_isnull; + bool old_isnull; + + /* only varlena columns that can be stored out-of-line */ + if (att->attlen != -1 || !att->attispackable || att->attisdropped) + continue; + + /* replica identity columns are logged with the old tuple image */ + if (bms_is_member(attidx, id_attrs)) + continue; + + new_value = heap_getattr(newtup, attnum, desc, &new_isnull); + + if (new_isnull || + !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(new_value))) + continue; + + old_value = heap_getattr(oldtup, attnum, desc, &old_isnull); + + if (old_isnull || + !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(old_value))) + continue; + + if (heap_attr_equals(desc, attnum, old_value, new_value, + old_isnull, new_isnull)) + return attnum; + } + + return InvalidAttrNumber; +} + /* * simple_heap_update - replace a tuple * diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 0feec8b765a..95634c283ec 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -342,6 +342,47 @@ pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors, return result; } +/* + * Check if the publication has a row filter defined for the given relation. + * + * For a partition, if pubviaroot is true, the row filter of the topmost + * ancestor published via this publication is what applies, so that ancestor + * is consulted instead. + */ +bool +pub_has_row_filter(Oid pubid, Relation relation, List *ancestors, + bool pubviaroot) +{ + HeapTuple rftuple; + Oid relid = RelationGetRelid(relation); + Oid publish_as_relid = relid; + bool rfisnull; + + if (pubviaroot && relation->rd_rel->relispartition) + { + publish_as_relid + = GetTopMostAncestorInPublication(pubid, ancestors, NULL); + + if (!OidIsValid(publish_as_relid)) + publish_as_relid = relid; + } + + rftuple = SearchSysCache2(PUBLICATIONRELMAP, + ObjectIdGetDatum(publish_as_relid), + ObjectIdGetDatum(pubid)); + + if (!HeapTupleIsValid(rftuple)) + return false; + + SysCacheGetAttr(PUBLICATIONRELMAP, rftuple, + Anum_pg_publication_rel_prqual, + &rfisnull); + + ReleaseSysCache(rftuple); + + return !rfisnull; +} + /* * Check for invalid columns in the publication table definition. * diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 2186a6ffeba..d76a40f4ba9 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5893,6 +5893,25 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) pubdesc->rf_valid_for_delete = false; } + /* + * Remember if the publication has a row filter for the relation and + * publishes updates. During replication, such a row filter can + * transform an UPDATE into an INSERT, which heap_update() needs to + * know about to guard against losing unchanged externally stored + * column values. (FOR ALL TABLES publications cannot have row + * filters.) + * + * REPLICA IDENTITY FULL makes the question moot, since the old + * tuple's full contents are always available then, so no extra + * protection is needed. + */ + if (!pubform->puballtables && + pubform->pubupdate && + relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL && + pub_has_row_filter(pubid, relation, ancestors, + pubform->pubviaroot)) + pubdesc->rf_exists_for_update = true; + /* * Check if all columns are part of the REPLICA IDENTITY index or not. * @@ -5966,6 +5985,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 48c7d1a8615..13ecade8129 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -108,6 +108,21 @@ 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, and the relation does not use REPLICA IDENTITY FULL + * (which makes the protection below unnecessary). 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 f90cf1ef896..255c80b0dc5 100644 --- a/src/include/commands/publicationcmds.h +++ b/src/include/commands/publicationcmds.h @@ -33,6 +33,8 @@ 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); +extern bool pub_has_row_filter(Oid pubid, Relation relation, + List *ancestors, bool pubviaroot); 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 3561c6bef0b..7a126065546 100644 --- a/src/include/utils/relcache.h +++ b/src/include/utils/relcache.h @@ -89,6 +89,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 e2c83670053..5134a022984 100644 --- a/src/test/subscription/t/028_row_filter.pl +++ b/src/test/subscription/t/028_row_filter.pl @@ -799,6 +799,62 @@ 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 INSERT cannot carry an +# unchanged externally stored value of a column that is not part of the +# replica identity, so the update is rejected on the publisher rather than +# losing the value on the subscriber. + +$node_publisher->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_publisher->safe_psql('postgres', + "INSERT INTO tab_rowfilter_toast_loss VALUES (3, repeat('a', 5000))"); + +# The update changes the replica identity key and leaves the out-of-line +# column unchanged, so it could be transformed into an INSERT that cannot +# carry the value; it must fail. +my ($ret, $stdout, $stderr) = $node_publisher->psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET id = 7", on_error_die => 0); +is($ret, 3, + 'update with unchanged out-of-line value in a row-filtered publication fails' +); +like( + $stderr, + qr/cannot update table "tab_rowfilter_toast_loss" because it is published with a row filter and column "val" has an unchanged externally stored value/, + 'error message names the out-of-line column'); + +# Changing the out-of-line column together with the key is safe: the new +# value is WAL-logged. +$node_publisher->safe_psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET id = 7, val = repeat('b', 6000)"); +is( $node_publisher->safe_psql('postgres', + "SELECT length(val) FROM tab_rowfilter_toast_loss"), + qq(6000), + 'update changing the out-of-line value succeeds'); + +# With REPLICA IDENTITY FULL the old tuple image is complete, so the +# update is safe even when the out-of-line column is unchanged. +$node_publisher->safe_psql('postgres', + "ALTER TABLE tab_rowfilter_toast_loss REPLICA IDENTITY FULL"); +$node_publisher->safe_psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET id = 8"); +is( $node_publisher->safe_psql('postgres', + "SELECT length(val) FROM tab_rowfilter_toast_loss"), + qq(6000), + 'update succeeds with REPLICA IDENTITY FULL'); + +# 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