From 8111fd6bc551c2126d03fb3fda68ab21e80f913b Mon Sep 17 00:00:00 2001 From: B1 Implementer Date: Fri, 14 Aug 2026 22:06:19 +0800 Subject: [PATCH v2] 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. --- doc/src/sgml/logical-replication.sgml | 21 ++++ src/backend/access/heap/heapam.c | 123 +++++++++++++++++++++- 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 | 56 ++++++++++ 8 files changed, 258 insertions(+), 11 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 3a61e2d6889..c6484e12239 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -927,6 +927,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 d0ffad19a16..b819ac8218d 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,8 @@ 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 = false; TransactionId xmax_new_tuple, xmax_old_tuple; uint16 infomask_old_tuple, @@ -3366,6 +3373,19 @@ 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 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 (walLogical && 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); @@ -3747,6 +3767,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 */ /* @@ -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,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 96838730fe1..c4eef7f5d37 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..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