From 8111fd6bc551c2126d03fb3fda68ab21e80f913b Mon Sep 17 00:00:00 2001
From: B1 Implementer <dev@example.com>
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 <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 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

