From fc935f1a648148987674ceafe022b9e2d958253d Mon Sep 17 00:00:00 2001
From: B1 Implementer <dev@example.com>
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 <command>UPDATE</command> is transformed into an <command>INSERT</command>.
    </para>
 
+   <note>
+    <para>
+     An <command>INSERT</command> resulting from this transformation cannot
+     carry the value of a column that is stored out-of-line (see
+     <xref linkend="storage-toast"/>) 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
+     <command>UPDATE</command> 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
+     <literal>FULL</literal> (see
+     <xref linkend="sql-altertable-replica-identity"/>), or remove the row
+     filter from the publication.
+    </para>
+   </note>
+
    <para>
     <xref linkend="logical-replication-row-filter-transformations-summary"/>
     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

