From 97771b10e3f37c6b5072004929640a5a5a526bbf Mon Sep 17 00:00:00 2001 From: Dilip Kumar 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: detailed_explanation[: delete_missing conflicts). + + has_omitted_values + boolean + Indicates whether any value in this row was too large to be + recorded and was replaced by an omitted marker + containing its length in bytes. + @@ -2408,6 +2415,31 @@ DETAIL: detailed_explanation[: json formats for flexible querying and analysis. + + 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: + +{"a" : 1, "b" : {"omitted":true,"length":190000000}} + + A cap is unavoidable, because these columns are of type + json 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, + has_omitted_values is true, which + allows incomplete rows to be found without searching the + JSON columns: + +SELECT * FROM pg_conflict.pg_conflict_log_16392 WHERE has_omitted_values; + + Note that a large value which is overwritten as a result of the + conflict cannot be recovered from the conflict log table. + + Note that virtual generated column values appear as null 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