From 61a7b628a228ffad750ab760816bc96de37bccde Mon Sep 17 00:00:00 2001
From: Amit Kapila <akapila@postgresql.org>
Date: Mon, 24 Aug 2026 17:01:51 +0530
Subject: [PATCH v72 3/3] Don't record oversized values in the conflict log
 table.

Recording a conflict serialized the entire tuple to JSON with
row_to_json(). The JSON form can be several times larger than the value
itself, because escape_json_char() expands every byte below 0x20 to a
six-byte \uXXXX sequence, so a sufficiently large value exceeded the 1GB
limit on a json datum. The resulting error aborted the apply
transaction, and since the origin did not advance, the apply worker
retried the same change indefinitely and replication could not proceed.

Record a value only if it is within CONFLICT_MAX_VALUE_SIZE, and replace
anything larger with an object noting the omission and the original
length. The new has_omitted_values column flags such rows so that they
can be found without searching the json columns. Values are omitted
rather than truncated because a partial value in a queryable table would
silently give wrong answers to equality and join conditions.

Note that a value discarded this way is not recoverable from the
conflict log table; no representation of it would fit.
---
 doc/src/sgml/logical-replication.sgml      |  31 +++
 src/backend/replication/logical/conflict.c | 273 ++++++++++++++++++---
 src/test/regress/expected/subscription.out |   3 +-
 src/test/subscription/t/035_conflicts.pl   |  77 +++++-
 4 files changed, 345 insertions(+), 39 deletions(-)

diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 39407778d77..c8f945cf85a 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2395,6 +2395,12 @@ DETAIL:  <replaceable class="parameter">detailed_explanation</replaceable>[: <re
          and the full local row image (<literal>tuple</literal>).
         </entry>
        </row>
+       <row>
+        <entry><literal>has_omitted_values</literal></entry>
+        <entry><type>boolean</type></entry>
+        <entry>Indicates whether any value in this row was too large to be
+        recorded, and was replaced by a marker. See below.</entry>
+       </row>
       </tbody>
      </tgroup>
     </table>
@@ -2405,6 +2411,31 @@ DETAIL:  <replaceable class="parameter">detailed_explanation</replaceable>[: <re
      <type>JSON</type> formats for flexible querying and analysis.
     </para>
 
+    <para>
+     A column value is either recorded exactly or not at all; it is never
+     truncated. Any value larger than 64kB is replaced in the
+     <type>JSON</type> columns by an object recording that it was omitted
+     together with its length in bytes, for example:
+<programlisting>
+{"a" : 1, "b" : {"omitted":true,"length":190000000}}
+</programlisting>
+     A cap is unavoidable, because these columns are of type
+     <type>json</type> and no value of that type may exceed 1GB, while the
+     <type>JSON</type> representation of a row can be several times larger
+     than the row itself. Values are omitted rather than truncated so that a
+     value present in the conflict log table can always be compared safely;
+     a truncated value would give wrong answers to equality and join
+     conditions without any indication. When a row contains such a marker,
+     <literal>has_omitted_values</literal> is <literal>true</literal>, which
+     allows incomplete rows to be found without searching the
+     <type>JSON</type> columns:
+<programlisting>
+SELECT * FROM pg_conflict.pg_conflict_log_16392 WHERE has_omitted_values;
+</programlisting>
+     Note that a large value which is overwritten as a result of the
+     conflict cannot be recovered from the conflict log table.
+    </para>
+
     <para>
      Note that virtual generated column values appear as
      <literal>null</literal> in these JSON columns, because such columns have no
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 03a56b85a2a..e42fb06c5d6 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -15,6 +15,7 @@
 #include "postgres.h"
 
 #include "access/commit_ts.h"
+#include "access/detoast.h"
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/tableam.h"
@@ -30,6 +31,8 @@
 #include "storage/lmgr.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
+#include "utils/json.h"
+#include "utils/jsonfuncs.h"
 #include "utils/lsyscache.h"
 #include "utils/pg_lsn.h"
 
@@ -69,6 +72,14 @@ typedef struct ConflictLogColumnDef
  * scalar columns (relid, conflict_type, commit timestamp) while these json
  * columns are per-conflict payload to inspect, not search keys.
  *
+ * A column value is recorded verbatim, or not at all: values larger than
+ * CONFLICT_MAX_VALUE_SIZE are replaced by a marker object recording their
+ * length (see conflict_values_to_json()). Values are never truncated,
+ * because a partial value in a queryable table would silently give wrong
+ * answers to equality and join predicates. 'has_omitted_values' is true if
+ * any value in the row was replaced this way, so that incomplete rows can be
+ * filtered cheaply without searching the json columns.
+ *
  * 'local_conflicts' is typed as an array of JSON objects (json[]), not a
  * single json object, so that a future conflict type needing to record
  * multiple local rows for one remote operation doesn't require a
@@ -86,11 +97,34 @@ static const ConflictLogColumnDef ConflictLogSchema[] = {
 	{.attname = "replica_identity_full", .atttypid = BOOLOID},
 	{.attname = "replica_identity", .atttypid = JSONOID},
 	{.attname = "remote_tuple", .atttypid = JSONOID},
-	{.attname = "local_conflicts", .atttypid = JSONARRAYOID}
+	{.attname = "local_conflicts", .atttypid = JSONARRAYOID},
+	{.attname = "has_omitted_values", .atttypid = BOOLOID}
 };
 
 #define NUM_CONFLICT_ATTRS ((AttrNumber) lengthof(ConflictLogSchema))
 
+/*
+ * Largest value recorded verbatim in the json columns of the conflict log
+ * table; anything larger is replaced by a marker recording its length. See
+ * conflict_values_to_json().
+ *
+ * This is generous enough that an ordinary row is always recorded in full,
+ * while excluding the large TOASTed values that would otherwise bloat the
+ * conflict log table, and the WAL written for it, out of proportion to the
+ * conflict being recorded.
+ *
+ * With CONFLICT_MAX_VALUE_SIZE set to 64kB, even a table with the maximum
+ * supported number of columns (MaxHeapAttributeNumber = 1600) undergoing the
+ * worst-case 6x JSON escape expansion will produce a serialized JSON datum of
+ * at most ~600MB, safely below PostgreSQL's 1GB varlena limit. Furthermore,
+ * across all three JSON columns in ConflictLogSchema (replica_identity,
+ * remote_tuple, local_conflicts), the total in-memory formed row size is
+ * bounded by ~1.8GB, ensuring it stays within PostgreSQL's 2GB row size
+ * limit (assuming local_conflicts contains at most one local tuple, which
+ * holds for all resolvable conflicts today).
+ */
+#define CONFLICT_MAX_VALUE_SIZE (64 * 1024)
+
 /*
  * Schema for the elements within the 'local_conflicts' JSON array.
  */
@@ -138,13 +172,18 @@ static void build_index_datums_from_slot(EState *estate, Relation localrel,
 										 bool *isnull);
 static char *build_index_value_desc(EState *estate, Relation localrel,
 									TupleTableSlot *slot, Oid indexoid);
-static Datum tuple_table_slot_to_json_datum(TupleTableSlot *slot);
+static Datum conflict_values_to_json(TupleDesc tupdesc, const Datum *values,
+									 const bool *isnull, bool *omitted);
+static Datum tuple_table_slot_to_json_datum(TupleTableSlot *slot,
+											bool *omitted);
 static Datum tuple_table_slot_to_indextup_json(EState *estate,
 											   Relation localrel,
 											   Oid replica_index,
-											   TupleTableSlot *slot);
+											   TupleTableSlot *slot,
+											   bool *omitted);
 static TupleDesc build_local_conflicts_tupledesc(void);
-static Datum build_local_conflicts_json_array(List *conflicttuples);
+static Datum build_local_conflicts_json_array(List *conflicttuples,
+											  bool *omitted);
 static void insert_conflict_log_tuple(EState *estate, Relation rel,
 									  Relation conflictlogrel,
 									  ConflictType conflict_type,
@@ -177,7 +216,7 @@ create_conflict_log_table_tupdesc(void)
  * Create a structured conflict log table for a subscription.
  *
  * The table is created within the system-managed 'pg_conflict' namespace to
- * prevent users from manually dropping or altering it.  This also prevents
+ * prevent users from manually dropping or altering it. This also prevents
  * accidental name collisions with user-created tables with the same name.
  *
  * The table name is generated automatically using the subscription's OID
@@ -329,7 +368,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
 
 	/*
 	 * Only LOG-level conflicts (i.e. resolved conflicts where the transaction
-	 * continues) are recorded in the conflict log table.  ERROR-level
+	 * continues) are recorded in the conflict log table. ERROR-level
 	 * conflicts halt replication and abort the transaction, so they are
 	 * always reported exclusively to the server log.
 	 */
@@ -343,7 +382,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
 		 * If a conflict log table was requested but it has been dropped
 		 * concurrently (e.g. a concurrent ALTER SUBSCRIPTION changed
 		 * conflict_log_destination), get_conflictlog_dest_and_table()
-		 * returned NULL.  Fall back to logging to the server log so that the
+		 * returned NULL. Fall back to logging to the server log so that the
 		 * conflict is not lost.
 		 */
 		if (log_dest_table && conflictlogrel == NULL)
@@ -359,7 +398,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
 	}
 
 	/*
-	 * Report the conflict to the server log.  When the server log is one of
+	 * Report the conflict to the server log. When the server log is one of
 	 * the destinations (or for ERROR-level conflicts), emit the full details.
 	 * Otherwise (table-only for LOG-level conflicts), emit a shorter message
 	 * noting that the details are captured in the conflict log table.
@@ -465,7 +504,7 @@ InitConflictIndexes(ResultRelInfo *relInfo)
  *
  * The table is opened with try_table_open(), so NULL is returned if the
  * conflict log table has been dropped concurrently (e.g. by an ALTER
- * SUBSCRIPTION that changed conflict_log_destination).  Callers must treat a
+ * SUBSCRIPTION that changed conflict_log_destination). Callers must treat a
  * NULL result for a table destination as "table unavailable" and fall back to
  * server-log reporting rather than failing.
  */
@@ -475,7 +514,7 @@ get_conflictlog_dest_and_table(ConflictLogDest *log_dest)
 	Oid			conflictlogrelid;
 
 	/*
-	 * Convert the text log destination to the internal enum.  MySubscription
+	 * Convert the text log destination to the internal enum. MySubscription
 	 * already contains the data from pg_subscription.
 	 */
 	*log_dest = GetConflictLogDest(MySubscription->conflictlogdest);
@@ -490,7 +529,7 @@ get_conflictlog_dest_and_table(ConflictLogDest *log_dest)
 
 	/*
 	 * Use try_table_open(): the table may have been dropped concurrently by
-	 * an ALTER SUBSCRIPTION that changed conflict_log_destination.  Returning
+	 * an ALTER SUBSCRIPTION that changed conflict_log_destination. Returning
 	 * NULL lets the caller fall back to the server log instead of failing.
 	 */
 	return try_table_open(conflictlogrelid, RowExclusiveLock);
@@ -996,27 +1035,176 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
 	return index_value;
 }
 
+/*
+ * conflict_values_to_json
+ *
+ * Serialize the given attribute values to a JSON object datum for the
+ * conflict log table.
+ *
+ * This behaves like row_to_json() on the equivalent tuple, except that a
+ * value larger than CONFLICT_MAX_VALUE_SIZE is not stored; in its place we
+ * emit an object recording that it was omitted along with its length, for
+ * example:
+ *
+ *	   {"a" : 1, "b" : {"omitted":true,"length":190000000}}
+ *
+ * A cap is required, not merely desirable. The result must become a json
+ * Datum, and a varlena cannot exceed 1GB, so a tuple whose JSON form is
+ * larger than that simply cannot be stored. Note that the JSON form can be
+ * far larger than the tuple: escape_json_char() expands every byte below
+ * 0x20 to a six-byte \uXXXX sequence, and bytea output is hex, so the
+ * expansion factor depends on the value's *content*, not just its length.
+ * Capping the stored size keeps the bound predictable regardless of content.
+ *
+ * We omit rather than truncate. A truncated value in a queryable table
+ * would silently return wrong answers to equality, join and diff predicates,
+ * with no way for the consumer to tell. The server log can truncate because
+ * a human reads it and ExecBuildSlotValueDescription() appends an ellipsis.
+ * Here the contract is that any value present is byte-exact.
+ *
+ * The marker is placed on the value rather than in a side list of column
+ * names because the same column may be oversized in one payload of a
+ * conflict log row and fine in another, and because a sibling metadata key
+ * could collide with a real column of that name.
+ *
+ * *omitted is set to true if any value was omitted; it is never set to
+ * false, so callers can accumulate the flag across several tuples.
+ */
+static Datum
+conflict_values_to_json(TupleDesc tupdesc, const Datum *values,
+						const bool *isnull, bool *omitted)
+{
+	Datum	   *args;
+	bool	   *nulls;
+	Oid		   *types;
+	int			nargs = 0;
+	bool		found_oversized = false;
+	Datum		result;
+
+	/*
+	 * First pass: look for oversized values. We test the raw datum size
+	 * rather than the length of the type's output representation so that a
+	 * toasted value is not detoasted merely to find out that we are going to
+	 * discard it; for a toasted datum this only reads the pointer header. The
+	 * two differ for types whose output is wider than their storage (bytea
+	 * renders as hex, for instance), but only by a small constant factor,
+	 * which the cap already accounts for.
+	 */
+	for (int i = 0; i < tupdesc->natts; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+
+		if (att->attisdropped || att->attlen != -1 || isnull[i])
+			continue;
+
+		if (toast_raw_datum_size(values[i]) - VARHDRSZ >
+			CONFLICT_MAX_VALUE_SIZE)
+		{
+			found_oversized = true;
+			break;
+		}
+	}
+
+	/*
+	 * Nothing oversized, which is the overwhelmingly common case: form the
+	 * tuple and hand it to row_to_json(), so that the stored representation
+	 * is exactly what it would have been without this check.
+	 */
+	if (!found_oversized)
+	{
+		HeapTuple	tuple;
+		Datum		json;
+
+		tuple = heap_form_tuple(tupdesc, unconstify(Datum *, values),
+								unconstify(bool *, isnull));
+		json = DirectFunctionCall1(row_to_json,
+								   heap_copy_tuple_as_datum(tuple, tupdesc));
+		heap_freetuple(tuple);
+
+		return json;
+	}
+
+	/*
+	 * Otherwise build the object ourselves, substituting a marker for each
+	 * oversized value. Note this renders with json_build_object()'s spacing
+	 * rather than row_to_json()'s; both are valid json and these columns are
+	 * payload to inspect, not compared as text.
+	 */
+	args = palloc_array(Datum, tupdesc->natts * 2);
+	nulls = palloc_array(bool, tupdesc->natts * 2);
+	types = palloc_array(Oid, tupdesc->natts * 2);
+
+	for (int i = 0; i < tupdesc->natts; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+		Size		rawsize = 0;
+
+		if (att->attisdropped)
+			continue;
+
+		/* Key. */
+		args[nargs] = CStringGetTextDatum(NameStr(att->attname));
+		nulls[nargs] = false;
+		types[nargs] = TEXTOID;
+		nargs++;
+
+		if (!isnull[i] && att->attlen == -1)
+			rawsize = toast_raw_datum_size(values[i]) - VARHDRSZ;
+
+		if (rawsize > CONFLICT_MAX_VALUE_SIZE)
+		{
+			StringInfoData marker;
+
+			initStringInfo(&marker);
+			appendStringInfo(&marker,
+							 "{\"omitted\":true,\"length\":" UINT64_FORMAT "}",
+							 (uint64) rawsize);
+
+			/*
+			 * Passed as json so that it is emitted verbatim rather than as a
+			 * quoted string.
+			 */
+			args[nargs] = CStringGetTextDatum(marker.data);
+			nulls[nargs] = false;
+			types[nargs] = JSONOID;
+			pfree(marker.data);
+
+			*omitted = true;
+		}
+		else
+		{
+			args[nargs] = values[i];
+			nulls[nargs] = isnull[i];
+			types[nargs] = att->atttypid;
+		}
+		nargs++;
+	}
+
+	result = json_build_object_worker(nargs, args, nulls, types, false, false);
+
+	pfree(args);
+	pfree(nulls);
+	pfree(types);
+
+	return result;
+}
+
 /*
  * tuple_table_slot_to_json_datum
  *
  * Helper function to convert a TupleTableSlot to JSON.
  */
 static Datum
-tuple_table_slot_to_json_datum(TupleTableSlot *slot)
+tuple_table_slot_to_json_datum(TupleTableSlot *slot, bool *omitted)
 {
-	HeapTuple	tuple;
-	Datum		datum;
-	Datum		json;
-
 	Assert(slot != NULL);
 
-	tuple = ExecCopySlotHeapTuple(slot);
-	datum = heap_copy_tuple_as_datum(tuple, slot->tts_tupleDescriptor);
-
-	json = DirectFunctionCall1(row_to_json, datum);
-	heap_freetuple(tuple);
+	/* Make sure the tuple is fully deconstructed */
+	slot_getallattrs(slot);
 
-	return json;
+	return conflict_values_to_json(slot->tts_tupleDescriptor,
+								   slot->tts_values, slot->tts_isnull,
+								   omitted);
 }
 
 /*
@@ -1030,12 +1218,12 @@ tuple_table_slot_to_json_datum(TupleTableSlot *slot)
  */
 static Datum
 tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
-								  Oid indexid, TupleTableSlot *slot)
+								  Oid indexid, TupleTableSlot *slot,
+								  bool *omitted)
 {
 	Relation	indexDesc;
 	Datum		values[INDEX_MAX_KEYS];
 	bool		isnull[INDEX_MAX_KEYS];
-	HeapTuple	tuple;
 	TupleDesc	tupdesc;
 	Datum		datum;
 
@@ -1052,15 +1240,11 @@ tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
 	/* Bless the tupdesc so it can be looked up by row_to_json. */
 	BlessTupleDesc(tupdesc);
 
-	/* Form the replica identity tuple. */
-	tuple = heap_form_tuple(tupdesc, values, isnull);
-	datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+	/* Convert the replica identity key values to a JSON datum. */
+	datum = conflict_values_to_json(tupdesc, values, isnull, omitted);
 
-	heap_freetuple(tuple);
 	index_close(indexDesc, NoLock);
 
-	/* Convert to a JSON datum. */
-	datum = DirectFunctionCall1(row_to_json, datum);
 	FreeTupleDesc(tupdesc);
 
 	return datum;
@@ -1112,9 +1296,19 @@ build_local_conflicts_tupledesc(void)
  *
  * Example output structure:
  * [ { "xid": "1001", "commit_ts": "...", "origin": "...", "tuple": {...} }, ... ]
+ *
+ * XXX Each element's tuple is bounded by conflict_values_to_json(), but
+ * the array as a whole is not, and it is itself a varlena limited to 1GB.
+ * That is unreachable today because the only conflict type that can produce
+ * more than one element, CT_MULTIPLE_UNIQUE_CONFLICTS, is reported at ERROR
+ * level and so is never written to the conflict log table (see
+ * ReportApplyConflict). If configurable resolution strategies later make
+ * such a conflict non-fatal, this will need a bound on the total size, and
+ * the caller should also avoid recording the same local row once per
+ * violated unique index.
  */
 static Datum
-build_local_conflicts_json_array(List *conflicttuples)
+build_local_conflicts_json_array(List *conflicttuples, bool *omitted)
 {
 	Datum	   *json_datum_array;
 	Datum		json_array_datum;
@@ -1167,7 +1361,8 @@ build_local_conflicts_json_array(List *conflicttuples)
 
 		/* Convert conflicting tuple to JSON datum. */
 		if (conflicttuple->slot)
-			values[attno++] = tuple_table_slot_to_json_datum(conflicttuple->slot);
+			values[attno++] =
+				tuple_table_slot_to_json_datum(conflicttuple->slot, omitted);
 		else
 			nulls[attno++] = true;
 
@@ -1226,6 +1421,7 @@ insert_conflict_log_tuple(EState *estate, Relation rel,
 	TransactionId remote_xid;
 	XLogRecPtr	remote_final_lsn;
 	TimestampTz remote_commit_ts;
+	bool		omitted = false;
 	HeapTuple	tuple;
 
 	Assert(conflictlogrel != NULL);
@@ -1282,12 +1478,14 @@ insert_conflict_log_tuple(EState *estate, Relation rel,
 			values[attno++] = BoolGetDatum(false);
 			values[attno++] = tuple_table_slot_to_indextup_json(estate, rel,
 																replica_index,
-																searchslot);
+																searchslot,
+																&omitted);
 		}
 		else
 		{
 			values[attno++] = BoolGetDatum(true);
-			values[attno++] = tuple_table_slot_to_json_datum(searchslot);
+			values[attno++] = tuple_table_slot_to_json_datum(searchslot,
+															 &omitted);
 		}
 	}
 	else
@@ -1297,11 +1495,14 @@ insert_conflict_log_tuple(EState *estate, Relation rel,
 	}
 
 	if (!TupIsNull(remoteslot))
-		values[attno++] = tuple_table_slot_to_json_datum(remoteslot);
+		values[attno++] = tuple_table_slot_to_json_datum(remoteslot, &omitted);
 	else
 		nulls[attno++] = true;
 
-	values[attno] = build_local_conflicts_json_array(conflicttuples);
+	values[attno++] = build_local_conflicts_json_array(conflicttuples,
+													   &omitted);
+
+	values[attno] = BoolGetDatum(omitted);
 
 	Assert(attno + 1 == NUM_CONFLICT_ATTRS);
 
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 681dc66dca6..bd712fda02d 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -706,7 +706,8 @@ WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
      10 | replica_identity
      11 | remote_tuple
      12 | local_conflicts
-(12 rows)
+     13 | has_omitted_values
+(13 rows)
 
 -- Changing the subscription owner should also update the owner
 -- of the associated conflict log table.
diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index 5efef6614ef..0d0b214216c 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -157,6 +157,79 @@ $node_subscriber->wait_for_log(
 
 pass('multiple_unique_conflicts detected on a leaf partition during insert');
 
+# Truncate table to get rid of the error
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab_2;");
+
+##################################################
+# Test that a value too large to record in the conflict log table is omitted
+# rather than breaking apply. A JSON representation can be several times
+# larger than the value itself (every byte below 0x20 escapes to a six-byte
+# \uXXXX sequence), so an oversized value used to error out the apply worker,
+# which then retried the same change forever.
+##################################################
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE conf_tab_big (a int PRIMARY KEY, b text);");
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE conf_tab_big (a int PRIMARY KEY, b text);");
+
+$node_publisher->safe_psql('postgres',
+	"ALTER PUBLICATION pub_tab ADD TABLE conf_tab_big");
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION sub_tab REFRESH PUBLICATION");
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab_big VALUES (1, 'small');");
+$node_publisher->wait_for_catchup($appname);
+
+# Remove the row locally so that the next remote update cannot find it.
+$node_subscriber->safe_psql('postgres',
+	"DELETE FROM conf_tab_big WHERE a = 1;");
+
+# Update it on the publisher with a value whose JSON form is far larger than
+# the value itself, generating an update_missing conflict on the subscriber.
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab_big SET b = repeat(chr(1), 100000) WHERE a = 1;");
+
+# Apply must not break; replication continues past the conflict.
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab_big VALUES (2, 'after');");
+$node_publisher->wait_for_catchup($appname);
+
+is( $node_subscriber->safe_psql(
+		'postgres', "SELECT b FROM conf_tab_big WHERE a = 2;"),
+	'after',
+	'apply continues when a conflicting value is too large to record');
+
+# The conflict was recorded, with the oversized value marked as omitted
+# rather than truncated or dropped silently.
+is( $node_subscriber->safe_psql(
+		'postgres',
+		"SELECT has_omitted_values FROM $clt WHERE relname = 'conf_tab_big';"),
+	't',
+	'oversized value is flagged by has_omitted_values');
+
+is( $node_subscriber->safe_psql(
+		'postgres',
+		"SELECT remote_tuple->'b'->>'omitted' FROM $clt
+		 WHERE relname = 'conf_tab_big';"),
+	'true',
+	'oversized value is marked omitted in place');
+
+is( $node_subscriber->safe_psql(
+		'postgres',
+		"SELECT remote_tuple->'b'->>'length' FROM $clt
+		 WHERE relname = 'conf_tab_big';"),
+	'100000',
+	'omitted value records its length');
+
+# Values below the cap in the same row are still recorded exactly.
+is( $node_subscriber->safe_psql(
+		'postgres',
+		"SELECT remote_tuple->>'a' FROM $clt WHERE relname = 'conf_tab_big';"),
+	'1',
+	'values below the cap are recorded verbatim');
+
 ###############################################################################
 # Setup a bidirectional logical replication between node_A & node_B
 ###############################################################################
@@ -713,7 +786,7 @@ like(
 
 ###############################################################################
 # ALTER TABLE ALL IN TABLESPACE must skip conflict log tables, the same way it
-# skips catalog and TOAST tables, instead of failing.  Use an isolated database
+# skips catalog and TOAST tables, instead of failing. Use an isolated database
 # so the bulk move only touches the objects created here.
 ###############################################################################
 $node_subscriber->safe_psql('postgres', "CREATE DATABASE clt_ts_test");
@@ -751,7 +824,7 @@ is( $node_subscriber->safe_psql(
 	'0',
 	"ALTER TABLE ALL IN TABLESPACE skips the conflict log table");
 
-# Cleanup.  The subscription has no real publisher connection, so detach its
+# Cleanup. The subscription has no real publisher connection, so detach its
 # slot before dropping it.
 $node_subscriber->safe_psql('clt_ts_test',
 	"ALTER SUBSCRIPTION sub_ts_test DISABLE");
-- 
2.49.0

