From 39108715fdbfa84ab2aa802cde0b461b81e016d4 Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Mon, 21 Sep 2026 09:00:43 +0530 Subject: [PATCH v75 2/2] Don't use row_to_json() or record oversized values in the conflict log row_to_json() honours a user CAST (t AS json), whose output bears no relation to the size of the value received, so a tiny key could render past the 1GB limit and break apply. Render each value with its type's output function instead, as the server log does; values are now json strings. Also, a large replica identity value could push the json datum past its 1GB limit and break apply. Values over 1kB are replaced by a marker recording their length, flagged by the new has_omitted_values column. --- doc/src/sgml/logical-replication.sgml | 69 ++++++++++- src/backend/replication/logical/conflict.c | 131 +++++++++++++++++---- src/test/regress/expected/subscription.out | 3 +- src/test/subscription/t/035_conflicts.pl | 125 +++++++++++++++++++- 4 files changed, 300 insertions(+), 28 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 39bc3f7f273..47a83173cee 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2376,7 +2376,11 @@ DETAIL: detailed_explanation[: replica_identity json - The JSON representation of the replica identity key values identifying the conflicting row. This is NULL when replica_identity_full is true or when replica identity is not applicable. + The replica identity key values identifying the conflicting row, + as a JSON object keyed by column name; see below for + how the values are represented. This is NULL when + replica_identity_full is true or + when replica identity is not applicable. local_conflicts @@ -2391,14 +2395,71 @@ DETAIL: detailed_explanation[: delete_missing conflicts). + + has_omitted_values + boolean + Indicates whether any value in + replica_identity was too large to be recorded and + was replaced by an omitted marker. + - The replica identity key values (replica_identity) - and the associated local conflict details (local_conflicts) are stored in - json formats for flexible querying and analysis. + Every value in replica_identity is written as a + JSON string containing the same text a + SELECT of that column would show. Values are quoted + whatever the column's type, so numbers, booleans and jsonb + documents all appear as strings: + +CREATE TABLE t (id int, tag text, doc jsonb, PRIMARY KEY (id, tag, doc)); + +SELECT replica_identity FROM pg_conflict.pg_conflict_log_16392; + replica_identity +------------------------------------------------- + {"id":"1","tag":"abc","doc":"{\"x\": 1}"} + + + + + Extracting a value with the ->> operator yields a + form that can be cast back to the column's type, so a recorded replica + identity can be used to locate the row it identifies: + +SELECT t.* FROM pg_conflict.pg_conflict_log_16392 c, t + WHERE t.id = (c.replica_identity->>'id')::int + AND t.tag = c.replica_identity->>'tag' + AND t.doc = (c.replica_identity->>'doc')::jsonb; + + Because container values such as jsonb, arrays and composite + types are recorded as strings rather than as nested + JSON, looking inside one requires an explicit cast: + +SELECT replica_identity->'doc'->>'x' FROM ...; -- NULL +SELECT (replica_identity->>'doc')::jsonb->>'x' FROM ...; -- 1 + + + + + A value is either recorded exactly or not at all; it is never truncated, + because a partial value in a queryable table would silently give wrong + answers to equality and join conditions. A value larger than 1kB is + replaced by an object recording that it was omitted together with its + length in bytes, and has_omitted_values is set for that + row: + + {"id":"1","tag":"abc","doc":{"omitted":true,"length":4096}} + + The flag allows incomplete rows to be found without searching the + json column: + +SELECT * FROM pg_conflict.pg_conflict_log_16392 WHERE has_omitted_values; + + A value omitted this way is not recoverable from the conflict log table. + The limit is necessary because replica_identity is of + type json, no value of which may exceed 1GB, while the + textual representation of a value can be far larger than the value itself. diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index c614b6e79b1..8c4e9a21357 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,7 @@ #include "storage/lmgr.h" #include "utils/array.h" #include "utils/builtins.h" +#include "utils/json.h" #include "utils/lsyscache.h" #include "utils/pg_lsn.h" @@ -85,11 +87,27 @@ static const ConflictLogColumnDef ConflictLogSchema[] = { {.attname = "remote_origin", .atttypid = TEXTOID}, {.attname = "replica_identity_full", .atttypid = BOOLOID}, {.attname = "replica_identity", .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 replica identity value recorded verbatim; anything larger is replaced + * by a marker recording its length. See tuple_table_slot_to_indextup_json(). + * + * The cap has to hold against the widest ratio of JSON output to storage a type + * can produce. That is a container holding numerics: numeric output is bounded + * near 131kB by NUMERIC_WEIGHT_MAX and NUMERIC_DSCALE_MAX, and the cheapest way + * to store one is about 12 bytes inside an array, so the ratio is around 11000. + * With at most INDEX_MAX_KEYS key columns the worst case is therefore + * 1kB * 11000 * 32 ~= 360MB, comfortably inside the 1GB limit on a json value; + * a 2kB cap would not be. Ordinary keys are orders of magnitude below the cap, + * so in practice nothing is ever omitted. + */ +#define CONFLICT_MAX_VALUE_SIZE 1024 + /* * Schema for the elements within the 'local_conflicts' JSON array. */ @@ -139,7 +157,8 @@ static char *build_index_value_desc(EState *estate, Relation localrel, 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 void insert_conflict_log_tuple(EState *estate, Relation rel, @@ -1018,40 +1037,106 @@ build_index_value_desc(EState *estate, Relation localrel, 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; + TupleDesc indexTupDesc; Datum values[INDEX_MAX_KEYS]; bool isnull[INDEX_MAX_KEYS]; - HeapTuple tuple; - TupleDesc tupdesc; - Datum datum; + StringInfoData result; + int indnkeyatts; Assert(slot != NULL); Assert(CheckRelationOidLockedByMe(indexid, RowExclusiveLock, true)); indexDesc = index_open(indexid, NoLock); + indexTupDesc = RelationGetDescr(indexDesc); build_index_datums_from_slot(estate, localrel, slot, indexDesc, values, isnull); - tupdesc = CreateTupleDescCopy(RelationGetDescr(indexDesc)); - /* Bless the tupdesc so it can be looked up by row_to_json. */ - BlessTupleDesc(tupdesc); + /* + * Build the JSON object here rather than handing the values to + * row_to_json(), and render each of them with its type's output function. + * + * row_to_json() would go through json_categorize_type(), which honours a + * user CREATE CAST (t AS json) for any type at or above + * FirstNormalObjectId. That cast is a function no other part of apply + * ever calls -- the publisher sends the value using the type's output + * function, the index compares it with the opclass, and neither asks for + * a json representation -- so its result is unrelated to the size of the + * value we received, and a tiny key could render a json value above the + * 1GB limit and error out the apply worker. Using the output function + * keeps us to the same representation the publisher already produced, as + * the server log does in BuildIndexValueDescription(). + * + * Only the key attributes are recorded. FormIndexDatum() above also + * fills in any non-key (INCLUDE) columns, but those are not part of the + * replica identity. + */ + indnkeyatts = IndexRelationGetNumberOfKeyAttributes(indexDesc); - /* Form the replica identity tuple. */ - tuple = heap_form_tuple(tupdesc, values, isnull); - datum = heap_copy_tuple_as_datum(tuple, tupdesc); + initStringInfo(&result); + appendStringInfoChar(&result, '{'); - heap_freetuple(tuple); - index_close(indexDesc, NoLock); + for (int i = 0; i < indnkeyatts; i++) + { + Form_pg_attribute att = TupleDescAttr(indexTupDesc, i); - /* Convert to a JSON datum. */ - datum = DirectFunctionCall1(row_to_json, datum); - FreeTupleDesc(tupdesc); + if (i > 0) + appendStringInfoChar(&result, ','); + + escape_json(&result, NameStr(att->attname)); + appendStringInfoChar(&result, ':'); - return datum; + if (isnull[i]) + appendStringInfoString(&result, "null"); + else + { + Oid outfuncoid; + bool typisvarlena; + char *outputstr; + Size rawsize = 0; + + /* + * Don't record a value larger than CONFLICT_MAX_VALUE_SIZE; put a + * marker recording its length in its place. Only a varlena has a + * size worth testing; a fixed-length type is small by definition. + * + * The raw datum size is tested rather than the length of the + * type's output, 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, but only by a + * small factor, which the cap already accounts for. + */ + if (att->attlen == -1) + rawsize = toast_raw_datum_size(values[i]) - VARHDRSZ; + + if (rawsize > CONFLICT_MAX_VALUE_SIZE) + { + /* A marker is an object, so it is appended unescaped. */ + appendStringInfo(&result, + "{\"omitted\":true,\"length\":" UINT64_FORMAT "}", + (uint64) rawsize); + *omitted = true; + continue; + } + + getTypeOutputInfo(att->atttypid, &outfuncoid, &typisvarlena); + outputstr = OidOutputFunctionCall(outfuncoid, values[i]); + escape_json(&result, outputstr); + pfree(outputstr); + } + } + + appendStringInfoChar(&result, '}'); + + index_close(indexDesc, NoLock); + + return PointerGetDatum(cstring_to_text_with_len(result.data, result.len)); } /* @@ -1209,6 +1294,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); @@ -1267,7 +1353,8 @@ 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 { @@ -1286,9 +1373,11 @@ 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); else - nulls[attno] = true; + nulls[attno++] = true; + + 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 49026eb2cb7..b678b7fe007 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -705,7 +705,8 @@ WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0 9 | replica_identity_full 10 | replica_identity 11 | local_conflicts -(11 rows) + 12 | has_omitted_values +(12 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 5b8eeeb83fc..3df954ebb2f 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -341,7 +341,7 @@ is($clt_check_ba, 1, 'delete_origin_differs logged into CLT on Node B'); my $clt_row_ba = $node_B->safe_psql('postgres', "SELECT replica_identity_full, replica_identity::text, (local_conflicts[1]->>'xid') IS NOT NULL FROM $clt_BA WHERE conflict_type = 'delete_origin_differs';"); -is($clt_row_ba, 'f|{"a":1}|t', 'delete_origin_differs records RI key columns and local conflict xid'); +is($clt_row_ba, 'f|{"a":"1"}|t', 'delete_origin_differs records RI key columns and local conflict xid'); $log_location = -s $node_A->logfile; @@ -365,7 +365,7 @@ is($clt_check_ab, 1, 'update_deleted logged into CLT on Node A'); my $clt_row_ab = $node_A->safe_psql('postgres', "SELECT replica_identity_full, replica_identity::text, (local_conflicts[1]->>'xid') IS NOT NULL FROM $clt_AB WHERE conflict_type = 'update_deleted';"); -is($clt_row_ab, 'f|{"a":1}|t', 'update_deleted records RI key columns and local conflict xid'); +is($clt_row_ab, 'f|{"a":"1"}|t', 'update_deleted records RI key columns and local conflict xid'); # Remember the next transaction ID to be assigned my $next_xid = $node_A->safe_psql('postgres', "SELECT txid_current() + 1;"); @@ -418,6 +418,127 @@ my $clt_row_ab_full = $node_A->safe_psql('postgres', "SELECT replica_identity_full, replica_identity IS NULL, (local_conflicts[1]->>'xid') IS NOT NULL FROM $clt_AB WHERE replica_identity_full = true;"); is($clt_row_ab_full, 't|t|t', 'update_deleted with REPLICA IDENTITY FULL sets replica_identity_full=true and replica_identity=NULL'); +############################################################################### +# Check that a user-defined CAST (t AS json) is not consulted when recording +# the replica identity. +# +# Each key value is rendered with its type's output function, as the server log +# does, so a cast cannot substitute its own representation. This matters because +# such a cast is a function no other part of apply calls, and its output bears no +# relation to the size of the value received, so a tiny key could otherwise +# render a json value past the 1GB limit and error out the apply worker. +############################################################################### + +my $cast_ddl = qq{ + CREATE TYPE ri_color AS ENUM ('red', 'green'); + CREATE FUNCTION ri_color_to_json(ri_color) RETURNS json + AS \$\$ SELECT '"cast-was-used"'::json \$\$ LANGUAGE sql IMMUTABLE; + CREATE CAST (ri_color AS json) WITH FUNCTION ri_color_to_json(ri_color); + CREATE TABLE tab_cast (a ri_color PRIMARY KEY, b int); +}; + +$node_A->safe_psql('postgres', $cast_ddl); +$node_B->safe_psql('postgres', $cast_ddl); + +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A ADD TABLE tab_cast"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_B->wait_for_subscription_sync($node_A, $subname_BA); + +# Replicate a row, remove it locally, then delete it on node_A so that the +# delete cannot find its target and the replica identity gets recorded. +$node_A->safe_psql('postgres', "INSERT INTO tab_cast VALUES ('red', 1)"); +$node_A->wait_for_catchup($subname_BA); +$node_B->safe_psql('postgres', "DELETE FROM tab_cast"); +$node_A->safe_psql('postgres', "DELETE FROM tab_cast WHERE a = 'red'"); +$node_A->wait_for_catchup($subname_BA); + +my $clt_check_cast = $node_B->poll_query_until('postgres', + "SELECT count(*) > 0 FROM $clt_BA WHERE relname = 'tab_cast';"); +is($clt_check_cast, 1, 'delete_missing on tab_cast logged into CLT on Node B'); + +my $clt_row_cast = $node_B->safe_psql('postgres', + "SELECT replica_identity::text FROM $clt_BA WHERE relname = 'tab_cast';"); +is($clt_row_cast, '{"a":"red"}', + 'replica identity is rendered by the type output function, not by a cast to json' +); + +# Restore tap_pub_A to publishing only 'tab'. Later tests subscribe to it from +# another database, which does not have tab_cast. +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A DROP TABLE tab_cast"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); + +my $cast_cleanup = q{ + DROP TABLE tab_cast; + DROP TYPE ri_color CASCADE; +}; + +$node_A->safe_psql('postgres', $cast_cleanup); +$node_B->safe_psql('postgres', $cast_cleanup); + +############################################################################### +# Check that a replica identity value too large to record is replaced by a +# marker rather than stored, and that has_omitted_values flags the row. +# +# A value is omitted rather than truncated, because a partial value in a +# queryable table would silently give wrong answers to equality and join +# conditions. +############################################################################### + +$node_A->safe_psql('postgres', + "CREATE TABLE tab_big (a text PRIMARY KEY, b int)"); +$node_B->safe_psql('postgres', + "CREATE TABLE tab_big (a text PRIMARY KEY, b int)"); + +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A ADD TABLE tab_big"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_B->wait_for_subscription_sync($node_A, $subname_BA); + +# One key just under the cap, recorded verbatim, and one over it, omitted. +$node_A->safe_psql( + 'postgres', qq{ + INSERT INTO tab_big VALUES (repeat('s', 1024), 1); + INSERT INTO tab_big VALUES (repeat('L', 2048), 2); +}); +$node_A->wait_for_catchup($subname_BA); +$node_B->safe_psql('postgres', "DELETE FROM tab_big"); +$node_A->safe_psql('postgres', "DELETE FROM tab_big"); +$node_A->wait_for_catchup($subname_BA); + +my $clt_check_big = $node_B->poll_query_until('postgres', + "SELECT count(*) = 2 FROM $clt_BA WHERE relname = 'tab_big';"); +is($clt_check_big, 1, 'both delete_missing conflicts on tab_big logged into CLT'); + +my $clt_row_small = $node_B->safe_psql('postgres', + "SELECT has_omitted_values, length(replica_identity->>'a') + FROM $clt_BA + WHERE relname = 'tab_big' AND NOT has_omitted_values;"); +is($clt_row_small, 'f|1024', + 'a replica identity value at the cap is recorded verbatim'); + +my $clt_row_big = $node_B->safe_psql('postgres', + "SELECT has_omitted_values, + replica_identity->'a'->>'omitted', + replica_identity->'a'->>'length' + FROM $clt_BA + WHERE relname = 'tab_big' AND has_omitted_values;"); +is($clt_row_big, 't|true|2048', + 'an oversized replica identity value is replaced by a marker recording its length' +); + +# Restore tap_pub_A to publishing only 'tab', as the later tests expect. +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A DROP TABLE tab_big"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_A->safe_psql('postgres', "DROP TABLE tab_big"); +$node_B->safe_psql('postgres', "DROP TABLE tab_big"); + ############################################################################### # Check that the xmin value of the conflict detection slot can be advanced when # the subscription has no tables. -- 2.54.0 (Apple Git-157)