From a7f67e486c4a47efffb7bc947c99040569ee7dc8 Mon Sep 17 00:00:00 2001 From: B1 Implementer Date: Fri, 14 Aug 2026 22:06:19 +0800 Subject: [PATCH v7] 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. --- src/backend/access/heap/heapam.c | 127 +++++++++++++++++++++- src/backend/commands/publicationcmds.c | 41 +++++++ src/backend/utils/cache/relcache.c | 13 +++ src/include/catalog/pg_publication.h | 13 +++ src/include/commands/publicationcmds.h | 2 + src/test/subscription/t/028_row_filter.pl | 56 ++++++++++ 6 files changed, 250 insertions(+), 2 deletions(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 8b488cfd8f6..1f721ecc75e 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" @@ -76,6 +77,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, const ItemPointerData *tid, LockTupleMode mode, LockWaitPolicy wait_policy, bool *have_tuple_lock); @@ -3308,6 +3313,9 @@ heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, bool checked_lockers; bool locker_remains; bool id_has_external = false; + bool id_changed; + bool check_unchanged_external; + PublicationDesc pubdesc; TransactionId xmax_new_tuple, xmax_old_tuple; uint16 infomask_old_tuple, @@ -3366,6 +3374,23 @@ 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); + /* + * Likewise, fetch (from the relcache, ordinarily) whether some + * publication with a row filter exists for this relation, for the same + * reason: this may require catalog access, which we don't want to do once + * the buffer is locked. + */ + RelationBuildPublicationDesc(relation, &pubdesc); + + /* + * 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. + */ + check_unchanged_external = walLogical && RelationIsLogicallyLogged(relation) && + relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL && + pubdesc.rf_exists_for_update; + block = ItemPointerGetBlockNumber(otid); INJECTION_POINT("heap_update-before-pin", NULL); buffer = ReadBuffer(relation, block); @@ -3453,6 +3478,34 @@ heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, id_attrs, &oldtup, newtup, &id_has_external); + 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. + */ + if (check_unchanged_external && id_changed) + { + 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."))); + } + } + /* * If we're not updating any "key" column, we can grab a weaker lock type. * This allows for more concurrency when we are running simultaneously @@ -4096,8 +4149,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); @@ -4627,6 +4679,77 @@ 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); + + /* cheap tests before deforming anything */ + if (!HeapTupleHasExternal(newtup) || !HeapTupleHasExternal(oldtup)) + return InvalidAttrNumber; + + 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 440adb356ad..8dec8a51704 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -346,6 +346,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 19c4ff6e75e..6374d3e3d91 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5926,6 +5926,19 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) pubdesc->rf_valid_for_delete = false; } + /* + * Remember if the publication has a row filter for the relation and + * publishes both 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.) + */ + if (!pubform->puballtables && + pubform->pubupdate && + 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. * 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..71e35d96c14 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/test/subscription/t/028_row_filter.pl b/src/test/subscription/t/028_row_filter.pl index c666ae00483..ca60c190c79 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