From 412018dffcd9583b06c6ab7850c1b3d5c791b305 Mon Sep 17 00:00:00 2001 From: B1 Implementer Date: Fri, 14 Aug 2026 23:44:25 +0800 Subject: [PATCH v6] 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. CheckCmdReplicaIdentity() now rejects updates on such tables up front: when a publication with a row filter publishes updates of a table that has a TOASTable column outside the replica identity, any UPDATE of the table raises an error instead of risking an incomplete row on the subscriber. The result is cached in the relcache as a new toastcols_valid_for_update flag in PublicationDesc, alongside the existing rf_valid_for_update etc. checks. Co-Authored-By: Claude Fable 5 --- src/backend/commands/publicationcmds.c | 42 ++++++++ src/backend/executor/execReplication.c | 14 ++- src/backend/utils/cache/relcache.c | 117 ++++++++++++++++++++++ src/include/catalog/pg_publication.h | 10 ++ src/include/commands/publicationcmds.h | 2 + src/test/subscription/t/028_row_filter.pl | 87 ++++++++++++++++ 6 files changed, 271 insertions(+), 1 deletion(-) diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 96838730fe1..e21b5afedf1 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -346,6 +346,48 @@ 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; + + /* only the nullness of the row filter is of interest here */ + 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/executor/execReplication.c b/src/backend/executor/execReplication.c index b2ca5cbf117..926ccbed7c0 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -1057,6 +1057,12 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd) * 3. All generated columns in REPLICA IDENTITY of the relation, are valid * - i.e. when all these generated columns are published. * + * 4. If a publication with a row filter publishes updates of the + * relation, every column that can be stored out-of-line must be part of + * the REPLICA IDENTITY. Such a row filter can transform an UPDATE into + * an INSERT during decoding, and an INSERT cannot represent an unchanged + * externally stored value. + * * XXX We could optimize it by first checking whether any of the * publications have a row filter or column list for this relation, or if * the relation contains a generated column. If none of these exist and @@ -1083,6 +1089,13 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd) errmsg("cannot update table \"%s\"", RelationGetRelationName(rel)), errdetail("Replica identity must not contain unpublished generated columns."))); + else if (cmd == CMD_UPDATE && !pubdesc.toastcols_valid_for_update) + ereport(ERROR, + (errcode(ERRCODE_INVALID_COLUMN_REFERENCE), + errmsg("cannot update table \"%s\"", + RelationGetRelationName(rel)), + errdetail("Column that can be stored externally is not part of the replica identity, and a publication row filter can transform the update into an insert during logical replication, which cannot represent unchanged externally stored values, so the subscriber would store NULL instead."), + errhint("To enable updating the table, set REPLICA IDENTITY FULL or remove the row filter from the publication."))); else if (cmd == CMD_DELETE && !pubdesc.rf_valid_for_delete) ereport(ERROR, (errcode(ERRCODE_INVALID_COLUMN_REFERENCE), @@ -1129,7 +1142,6 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd) errhint("To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE."))); } - /* * Check if we support writing into specific relkind of local relation and check * if it aligns with the relkind of the relation on the publisher. diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 9abbaeab4a9..bf44786dddd 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5784,6 +5784,83 @@ RelationGetExclusionInfo(Relation indexRelation, MemoryContextSwitchTo(oldcxt); } +/* + * RelationHasNonReplIdentToastableColumn - does the relation have a column + * that can be stored out-of-line but is not part of the replica identity? + * + * Such a column is hazardous when the relation is published with a row + * filter: decoding can transform an UPDATE into an INSERT (when the row + * moves into the set of rows satisfying the filter), and an INSERT cannot + * represent an unchanged externally stored value: the new tuple in WAL + * then carries only the on-disk toast pointer, so the decoder cannot + * supply the value for the transformed INSERT and the subscriber stores + * NULL instead. Values of replica identity columns don't have this + * problem: they are logged with the old tuple image and the decoder + * restores them into the transformed tuple. + * + * Only the relation's definition is consulted here; whether a particular + * value is actually stored externally, or whether the row filter would + * actually transform a particular update, cannot be known at this point, + * so the result errs on the safe side. + * + * False is also returned when the relation has no replica identity index + * (REPLICA IDENTITY NOTHING or DEFAULT without a suitable index); updates + * are rejected for that reason separately. With REPLICA IDENTITY FULL the + * old tuple image is complete, so there is no hazard either. + */ +static bool +RelationHasNonReplIdentToastableColumn(Relation relation) +{ + TupleDesc desc = RelationGetDescr(relation); + Bitmapset *idattrs; + AttrNumber attno; + bool result = false; + + /* Without a TOAST table, no value can be stored externally. */ + if (!OidIsValid(relation->rd_rel->reltoastrelid)) + return false; + + /* + * Get the replica identity columns. A NULL result means there is no + * replica identity index, which the caller handles separately (see the + * function comment). + */ + idattrs = RelationGetIndexAttrBitmap(relation, + INDEX_ATTR_BITMAP_IDENTITY_KEY); + if (idattrs == NULL) + return false; + + for (attno = 1; attno <= desc->natts; attno++) + { + Form_pg_attribute attr = TupleDescAttr(desc, attno - 1); + + if (attr->attisdropped) + continue; + + /* + * Only varlena columns can be stored out-of-line. attstorage is + * not consulted: ALTER COLUMN ... SET STORAGE PLAIN doesn't rewrite + * the table, so values stored externally before the change can + * still be present, and updates preserve them (see + * toast_tuple_init). + */ + if (attr->attlen != -1) + continue; + + /* replica identity columns are logged with the old tuple image */ + if (bms_is_member(attno - FirstLowInvalidHeapAttributeNumber, + idattrs)) + continue; + + result = true; + break; + } + + bms_free(idattrs); + + return result; +} + /* * Get the publication information for the given relation. * @@ -5797,6 +5874,11 @@ RelationGetExclusionInfo(Relation indexRelation, * 3. The generated columns of the relation for such publications. We consider * any reference of an unpublished generated column in REPLICA IDENTITY as * invalid. + * 4. The columns that can be stored out-of-line, if a publication with a row + * filter publishes updates of the relation. We consider them invalid if + * any such column is not part of REPLICA IDENTITY, because the row filter + * can transform an UPDATE into an INSERT during decoding, and an INSERT + * cannot represent an unchanged externally stored value. * * To avoid fetching the publication information repeatedly, we cache the * publication actions, row filter validation information, column list @@ -5813,6 +5895,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) Oid schemaid; List *ancestors = NIL; Oid relid = RelationGetRelid(relation); + bool rf_exists_for_update = false; /* * If not publishable, it publishes no actions. (pgoutput_change() will @@ -5827,6 +5910,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) pubdesc->cols_valid_for_delete = true; pubdesc->gencols_valid_for_update = true; pubdesc->gencols_valid_for_delete = true; + pubdesc->toastcols_valid_for_update = true; return; } @@ -5843,6 +5927,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) pubdesc->cols_valid_for_delete = true; pubdesc->gencols_valid_for_update = true; pubdesc->gencols_valid_for_delete = true; + pubdesc->toastcols_valid_for_update = true; /* Fetch the publication membership info. */ puboids = GetRelationIncludedPublications(relid); @@ -5926,6 +6011,27 @@ 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; the out-of-line column validation below needs + * to know that. (FOR ALL TABLES publications cannot have row + * filters. Whether inserts are published doesn't matter here: + * pgoutput checks the publish actions against the original UPDATE + * action, before the row filter transformation, so a publication + * that doesn't publish inserts can still emit a transformed + * INSERT.) + * + * Note that this flag is not guaranteed to be set if the loop exits + * early, but that only happens when updates are rejected for another + * reason anyway. + */ + if (!rf_exists_for_update && + !pubform->puballtables && + pubform->pubupdate && + pub_has_row_filter(pubid, relation, ancestors, + pubform->pubviaroot)) + rf_exists_for_update = true; + /* * Check if all columns are part of the REPLICA IDENTITY index or not. * @@ -5986,6 +6092,17 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) break; } + /* + * If a publication with a row filter publishes updates of the relation, + * decoding can transform an UPDATE into an INSERT (when the row moves + * into the set of rows satisfying the filter), and an INSERT cannot + * represent an unchanged externally stored value of a column that is + * not part of the replica identity, so check for such columns. + */ + if (rf_exists_for_update && + RelationHasNonReplIdentToastableColumn(relation)) + pubdesc->toastcols_valid_for_update = false; + if (relation->rd_pubdesc) { pfree(relation->rd_pubdesc); diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 89b4bb14f62..0399758b23e 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -118,6 +118,16 @@ typedef struct PublicationDesc */ bool gencols_valid_for_update; bool gencols_valid_for_delete; + + /* + * true if all columns that can be stored out-of-line are part of the + * replica identity, or no publication with a row filter publishes + * updates of the relation. During decoding, such a row filter can + * transform an UPDATE into an INSERT, which cannot carry unchanged + * externally stored values of columns that are not part of the replica + * identity. + */ + bool toastcols_valid_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..9f3f33a9bde 100644 --- a/src/test/subscription/t/028_row_filter.pl +++ b/src/test/subscription/t/028_row_filter.pl @@ -799,6 +799,93 @@ is($result, qq(), 'check replicated rows to tab_rowfilter_viaroot_part_1'); # Testcase end: FOR TABLE with row filter publications # ====================================================== +# ==================================================================== +# Testcase start: UPDATE on a table published with a row filter that has +# an externally storable column outside the replica identity +# +# 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 updates on such a table are rejected on the +# publisher to prevent 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 check does not depend on which columns the update modifies, or on +# whether the row filter would actually transform the update, so any +# update of the table is rejected. +my ($ret, $stdout, $stderr) = $node_publisher->psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET id = 7", on_error_die => 0); +is($ret, 3, + 'update of table published with a row filter and an externally storable column outside the replica identity fails' +); +like( + $stderr, + qr/cannot update table "tab_rowfilter_toast_loss"[\s\S]*Column that can be stored externally is not part of the replica identity/, + 'error reports the externally storable column outside the replica identity' +); + +($ret, $stdout, $stderr) = $node_publisher->psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET val = repeat('b', 6000)", + on_error_die => 0); +is($ret, 3, + 'update changing only the externally storable column also fails'); + +# Values stored externally before a SET STORAGE PLAIN are not removed +# (the command does not rewrite the table) and are preserved by updates, +# so the column remains hazardous and the update still fails. +$node_publisher->safe_psql('postgres', + "ALTER TABLE tab_rowfilter_toast_loss ALTER COLUMN val SET STORAGE PLAIN"); +($ret, $stdout, $stderr) = $node_publisher->psql('postgres', + "UPDATE tab_rowfilter_toast_loss SET id = 7", on_error_die => 0); +is($ret, 3, 'update still fails after SET STORAGE PLAIN'); +$node_publisher->safe_psql('postgres', + "ALTER TABLE tab_rowfilter_toast_loss ALTER COLUMN val SET STORAGE EXTERNAL"); + +# Deletes cannot be transformed into inserts, so they are not affected. +$node_publisher->safe_psql('postgres', + "DELETE FROM tab_rowfilter_toast_loss"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_rowfilter_toast_loss VALUES (3, repeat('a', 5000))"); + +# With REPLICA IDENTITY FULL the old tuple image is complete, so the +# update is safe again. +$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 = 7"); +is( $node_publisher->safe_psql('postgres', + "SELECT id FROM tab_rowfilter_toast_loss"), + qq(7), + 'update succeeds with REPLICA IDENTITY FULL'); + +# Externally storable columns covered by the replica identity are fine: +# their values are logged with the old tuple image. +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab_rowfilter_toast_in_ri (id int, val text, PRIMARY KEY (id, val))"); +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION tap_pub_toast_in_ri FOR TABLE tab_rowfilter_toast_in_ri WHERE (id > 0)"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_rowfilter_toast_in_ri VALUES (1, 'a')"); +$node_publisher->safe_psql('postgres', + "UPDATE tab_rowfilter_toast_in_ri SET id = 2"); +is( $node_publisher->safe_psql('postgres', + "SELECT id FROM tab_rowfilter_toast_in_ri"), + qq(2), + 'update succeeds when the externally storable column is part of the replica identity' +); + +# Testcase end: UPDATE on a table published with a row filter that has +# an externally storable column outside the replica identity +# ==================================================================== + $node_subscriber->stop('fast'); $node_publisher->stop('fast'); -- 2.47.3