From 97771b10e3f37c6b5072004929640a5a5a526bbf Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumarb@google.com>
Date: Fri, 28 Aug 2026 15:24:03 +0530
Subject: [PATCH v73 2/2] 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      |  32 +++
 src/backend/replication/logical/conflict.c | 235 +++++++++++++++++----
 src/backend/utils/adt/json.c               | 130 +++++++++++-
 src/include/utils/json.h                   |   2 +
 src/test/regress/expected/subscription.out |   3 +-
 src/test/subscription/t/035_conflicts.pl   |  77 ++++++-
 6 files changed, 439 insertions(+), 40 deletions(-)

diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 8af06a689aa..66a01e7ab4b 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2398,6 +2398,13 @@ DETAIL:  <replaceable class="parameter">detailed_explanation</replaceable>[: <re
          <literal>delete_missing</literal> conflicts).
         </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 an <literal>omitted</literal> marker
+        containing its length in bytes.</entry>
+       </row>
       </tbody>
      </tgroup>
     </table>
@@ -2408,6 +2415,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 16kB is replaced in the
+     JSON 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
+     JSON 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
+     JSON 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 0515ff3de9d..2c438aba550 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,40 @@ 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.
+ *
+ * conflict_values_to_json() enforces this as an actual output size limit via
+ * json_set_size_limit(), not an estimate from a value's storage size, so the
+ * bound below holds regardless of the column's type or content (see that
+ * function's header for why a storage-size-based estimate is not safe for
+ * every type). Each column can overshoot the limit by at most a small,
+ * bounded amount of in-flight structural JSON (a field name, closing
+ * brackets/braces as nested array/composite frames unwind) before the limit
+ * is noticed; call this overshoot O, comfortably under 1kB in practice. So
+ * even a table with the maximum supported number of columns
+ * (MaxHeapAttributeNumber = 1600) produces a serialized JSON datum of at
+ * most ~1600 * (16kB + O), well 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 stays comfortably within PostgreSQL's 1GB (MaxAllocSize)
+ * in-memory row/tuple size limit.
+ */
+#define CONFLICT_MAX_VALUE_SIZE (16 * 1024)
+
 /*
  * Schema for the elements within the 'local_conflicts' JSON array.
  */
@@ -138,13 +178,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 +222,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 +374,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 +388,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 +404,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.
@@ -478,7 +523,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.
  */
@@ -488,7 +533,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);
@@ -503,7 +548,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);
@@ -1009,27 +1054,132 @@ 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 produces the same JSON object row_to_json() would 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. The JSON form can be far larger
+ * than the tuple's own storage, and not just by a small constant factor: an
+ * array or composite value can render every one of its NULL elements or
+ * fields as several bytes of JSON while costing as little as one bit of
+ * storage each, so raw storage size is not a safe proxy for the size of the
+ * JSON it will produce. We therefore serialize each column under an actual
+ * output size limit (json_set_size_limit()) rather than estimating from its
+ * storage size beforehand; this is safe for any type, not just the ones we
+ * have thought to check, because the limit is enforced by the JSON
+ * serializer itself as it recurses through arrays and composites.
+ *
+ * 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. A truncated
+ * value would also not necessarily be well-formed JSON, since the cutoff
+ * point has no regard for token or nesting boundaries.
+ *
+ * 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)
+{
+	StringInfoData result;
+	bool		needsep = false;
+
+	initStringInfo(&result);
+	appendStringInfoChar(&result, '{');
+
+	for (int i = 0; i < tupdesc->natts; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+
+		if (att->attisdropped)
+			continue;
+
+		if (needsep)
+			appendStringInfoChar(&result, ',');
+		needsep = true;
+
+		escape_json(&result, NameStr(att->attname));
+		appendStringInfoChar(&result, ':');
+
+		if (isnull[i])
+		{
+			appendStringInfoString(&result, "null");
+			continue;
+		}
+
+		{
+			JsonTypeCategory tcategory;
+			Oid			outfuncoid;
+			Datum		attrjson;
+
+			json_categorize_type(att->atttypid, false, &tcategory, &outfuncoid);
+
+			json_set_size_limit(CONFLICT_MAX_VALUE_SIZE);
+			attrjson = datum_to_json(values[i], tcategory, outfuncoid);
+
+			if (json_size_limit_exceeded())
+			{
+				Size		rawsize = (att->attlen == -1) ?
+					toast_raw_datum_size(values[i]) - VARHDRSZ : 0;
+
+				appendStringInfo(&result,
+								 "{\"omitted\":true,\"length\":" UINT64_FORMAT "}",
+								 (uint64) rawsize);
+				*omitted = true;
+			}
+			else
+			{
+				text	   *t = DatumGetTextPP(attrjson);
+
+				appendBinaryStringInfo(&result, VARDATA_ANY(t),
+									   VARSIZE_ANY_EXHDR(t));
+			}
+
+			/* Don't leak the limit into unrelated json.c calls. */
+			json_set_size_limit(0);
+		}
+	}
+
+	appendStringInfoChar(&result, '}');
+
+	return PointerGetDatum(cstring_to_text_with_len(result.data, result.len));
+}
+
 /*
  * 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);
 }
 
 /*
@@ -1043,12 +1193,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;
 
@@ -1065,15 +1215,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;
@@ -1125,9 +1271,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;
@@ -1182,7 +1338,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;
 
@@ -1241,6 +1398,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);
@@ -1297,12 +1455,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
@@ -1312,7 +1472,7 @@ 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;
 
@@ -1321,10 +1481,13 @@ insert_conflict_log_tuple(EState *estate, Relation rel,
 	 * conflicting rows, so set local_conflicts to NULL.
 	 */
 	if (conflicttuples != NIL)
-		values[attno++] = build_local_conflicts_json_array(conflicttuples);
+		values[attno++] = build_local_conflicts_json_array(conflicttuples,
+														   &omitted);
 	else
 		nulls[attno++] = true;
 
+	values[attno] = BoolGetDatum(omitted);
+
 	Assert(attno + 1 == NUM_CONFLICT_ATTRS);
 
 	tuple = heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls);
diff --git a/src/backend/utils/adt/json.c b/src/backend/utils/adt/json.c
index 28e5f3cf9c0..babbaed6392 100644
--- a/src/backend/utils/adt/json.c
+++ b/src/backend/utils/adt/json.c
@@ -13,6 +13,7 @@
  */
 #include "postgres.h"
 
+#include "access/detoast.h"
 #include "access/htup_details.h"
 #include "catalog/pg_type.h"
 #include "common/hashfn.h"
@@ -99,6 +100,67 @@ static void add_json(Datum val, bool is_null, StringInfo result,
 					 Oid val_type, bool key_scalar);
 static text *catenate_stringinfo_string(StringInfo buffer, const char *addon);
 
+/*
+ * Optional output size limit for datum_to_json_internal() and the
+ * recursive calls it makes for array/composite values. Zero (the default)
+ * means unlimited, matching the historical behavior of every caller in
+ * this file. json_set_size_limit() is meant to be used around a single
+ * top-level call (e.g. a call to datum_to_json()); the caller should reset
+ * the limit to 0 afterward so it doesn't apply to unrelated json.c calls
+ * later in the same backend.
+ */
+static Size json_size_limit = 0;
+static bool json_size_limit_hit = false;
+
+/*
+ * Set the output size limit checked by datum_to_json_internal(). Pass 0 to
+ * disable the limit. Also clears the "limit exceeded" flag, so this is
+ * safe to call both before starting a bounded call and after finishing one
+ * (to reset for the next caller).
+ */
+void
+json_set_size_limit(Size maxlen)
+{
+	json_size_limit = maxlen;
+	json_size_limit_hit = false;
+}
+
+/*
+ * Whether the most recent call made under a size limit set by
+ * json_set_size_limit() stopped early because the limit was exceeded.
+ */
+bool
+json_size_limit_exceeded(void)
+{
+	return json_size_limit_hit;
+}
+
+/*
+ * Would appending 'addlen' more bytes on top of 'currentlen' already-
+ * written bytes exceed the limit set by json_set_size_limit()? If so,
+ * record that the limit was hit and return true.
+ *
+ * This is for the handful of rendering paths in datum_to_json_internal()
+ * that build their entire output in a single call (escaping a text value,
+ * calling a type's output or cast function): the check at the top of that
+ * function only ever sees the total so far each time it is re-entered, so
+ * it catches unbounded recursion (e.g. many array elements or composite
+ * fields) but not one leaf value whose own rendering is disproportionately
+ * large. Each such call site knows, cheaply, the exact or worst-case size
+ * of what it is about to append, before appending it, and can check here
+ * instead of finding out only after building it.
+ */
+static bool
+json_size_would_exceed(int currentlen, Size addlen)
+{
+	if (json_size_limit && (Size) currentlen + addlen > json_size_limit)
+	{
+		json_size_limit_hit = true;
+		return true;
+	}
+	return false;
+}
+
 /*
  * Input.
  */
@@ -184,6 +246,19 @@ datum_to_json_internal(Datum val, bool is_null, StringInfo result,
 
 	check_stack_depth();
 
+	/*
+	 * Stop as soon as a size limit set by json_set_size_limit() is exceeded,
+	 * rather than continuing to append. Every recursive call for array
+	 * elements or composite fields comes back through here, so this one check
+	 * bounds the total output of the top-level call regardless of how deeply
+	 * arrays and composites are nested inside one another. (This alone is not
+	 * enough for a leaf value whose own single-shot rendering is
+	 * disproportionately large; see the individual checks below and
+	 * json_size_would_exceed().)
+	 */
+	if (json_size_would_exceed(result->len, 0))
+		return;
+
 	/* callers are expected to ensure that null keys are not passed in */
 	Assert(!(key_scalar && is_null));
 
@@ -273,14 +348,30 @@ datum_to_json_internal(Datum val, bool is_null, StringInfo result,
 			}
 			break;
 		case JSONTYPE_JSON:
-			/* JSON and JSONB output will already be escaped */
+
+			/*
+			 * JSON and JSONB output are already escaped, so we can call their
+			 * output functions directly without extra escaping. Check the
+			 * rendered length before appending to result.
+			 */
 			outputstr = OidOutputFunctionCall(outfuncoid, val);
+			if (json_size_would_exceed(result->len, strlen(outputstr)))
+			{
+				pfree(outputstr);
+				return;
+			}
+
 			appendStringInfoString(result, outputstr);
 			pfree(outputstr);
 			break;
 		case JSONTYPE_CAST:
 			/* outfuncoid refers to a cast function, not an output function */
 			jsontext = DatumGetTextPP(OidFunctionCall1(outfuncoid, val));
+			if (json_size_would_exceed(result->len, VARSIZE_ANY_EXHDR(jsontext)))
+			{
+				pfree(jsontext);
+				return;
+			}
 			appendBinaryStringInfo(result, VARDATA_ANY(jsontext),
 								   VARSIZE_ANY_EXHDR(jsontext));
 			pfree(jsontext);
@@ -289,10 +380,29 @@ datum_to_json_internal(Datum val, bool is_null, StringInfo result,
 			/* special-case text types to save useless palloc/memcpy cycles */
 			if (outfuncoid == F_TEXTOUT || outfuncoid == F_VARCHAROUT ||
 				outfuncoid == F_BPCHAROUT)
+			{
+				/*
+				 * escape_json_char() expands every byte below 0x20 to a
+				 * six-byte \uXXXX sequence, the worst case for this path;
+				 * check against that before escaping rather than after.
+				 * toast_raw_datum_size() gives the logical (uncompressed)
+				 * length without detoasting a value we may be about to
+				 * discard.
+				 */
+				Size		rawsize = toast_raw_datum_size(val) - VARHDRSZ;
+
+				if (json_size_would_exceed(result->len, 6 * rawsize))
+					return;
 				escape_json_text(result, (text *) DatumGetPointer(val));
+			}
 			else
 			{
 				outputstr = OidOutputFunctionCall(outfuncoid, val);
+				if (json_size_would_exceed(result->len, 6 * strlen(outputstr)))
+				{
+					pfree(outputstr);
+					return;
+				}
 				escape_json(result, outputstr);
 				pfree(outputstr);
 			}
@@ -461,6 +571,15 @@ array_dim_to_json(StringInfo result, int dim, int ndims, int *dims, const Datum
 			array_dim_to_json(result, dim + 1, ndims, dims, vals, nulls,
 							  valcount, tcategory, outfuncoid, false);
 		}
+
+		/*
+		 * Stop looping once a size limit set by json_set_size_limit() has
+		 * been hit; datum_to_json_internal() already stopped appending, so
+		 * further iterations would just add separators to no purpose, and for
+		 * a very large remaining element count that adds up.
+		 */
+		if (json_size_limit_hit)
+			break;
 	}
 
 	appendStringInfoChar(result, ']');
@@ -585,6 +704,15 @@ composite_to_json(Datum composite, StringInfo result, bool use_line_feeds)
 
 		datum_to_json_internal(val, isnull, result, tcategory, outfuncoid,
 							   false);
+
+		/*
+		 * Stop looping once a size limit set by json_set_size_limit() has
+		 * been hit; datum_to_json_internal() already stopped appending, so
+		 * further iterations would just add field names and separators to no
+		 * purpose, and for a very wide row that adds up.
+		 */
+		if (json_size_limit_hit)
+			break;
 	}
 
 	appendStringInfoChar(result, '}');
diff --git a/src/include/utils/json.h b/src/include/utils/json.h
index 2f4be40518d..8a2d236b095 100644
--- a/src/include/utils/json.h
+++ b/src/include/utils/json.h
@@ -31,5 +31,7 @@ extern Datum json_build_object_worker(int nargs, const Datum *args, const bool *
 extern Datum json_build_array_worker(int nargs, const Datum *args, const bool *nulls,
 									 const Oid *types, bool absent_on_null);
 extern bool json_validate(text *json, bool check_unique_keys, bool throw_error);
+extern void json_set_size_limit(Size maxlen);
+extern bool json_size_limit_exceeded(void);
 
 #endif							/* JSON_H */
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

