From b7d6cdafd37140ee1aa8dbefc1375fe06c7b0dba Mon Sep 17 00:00:00 2001 From: Mikhail Kharitonov Date: Sat, 1 Aug 2026 00:07:24 +0200 Subject: [PATCH v4 1/2] logical replication: make old_tuple's flag match its contents Under publish_via_partition_root, a leaf partition's change is published using the root partitioned table's OID and tuple descriptor, but the old-tuple payload is built from the leaf that actually stored the row. pgoutput, however, chose the old-tuple flag ('O', a full old tuple, versus 'K', replica-identity key columns only) from the root's replica identity. When a leaf's replica identity differs from the root's, the flag could disagree with the payload: for example, a root with REPLICA IDENTITY FULL over a leaf using a key-based identity emitted 'O' while sending only the key columns. That violates the logical replication protocol, in which an 'O' message promises a complete old tuple. Choose the flag from the leaf relation's replica identity instead. logicalrep_write_update() and logicalrep_write_delete() now take both the published relation, whose OID and tuples go on the wire, and the leaf relation, whose replica identity selects the flag; the single caller in pgoutput_change() passes both. (The two relations coincide unless publish_via_partition_root is in play.) This is a wire-protocol conformance fix for consumers that read the replication stream directly, such as external CDC tools. It does not affect the built-in apply worker, which inspects the flag only to learn whether an old tuple follows and does not branch on 'O' versus 'K' afterwards. It also does not change which columns are sent, so on its own it does not resolve subscriber-side row-matching divergence when a subscriber lacks a suitable replica-identity index; that is a separate concern. Add a TAP test that reads the raw pgoutput stream and asserts the exact old-tuple flag of each UPDATE/DELETE, one leaf per transaction, and that the subscriber stays consistent with the publisher. Author: Mikhail Kharitonov Discussion: https://postgr.es/m/CAKkoVatDVuH=AcnecrZSOQ0_Md6RfW4ExvCKJzJJ4JTrX1wbxQ@mail.gmail.com --- doc/src/sgml/logical-replication.sgml | 14 ++ src/backend/replication/logical/proto.c | 41 +++-- src/backend/replication/pgoutput/pgoutput.c | 4 +- src/include/replication/logicalproto.h | 11 +- .../t/039_partition_replica_identity.pl | 159 ++++++++++++++++++ 5 files changed, 207 insertions(+), 22 deletions(-) create mode 100644 src/test/subscription/t/039_partition_replica_identity.pl diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 36298cacb75..5b9e7d783ef 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -972,6 +972,20 @@ HINT: To initiate replication, you must manually create the replication slot, e row filter is used. + + When publish_via_partition_root is true, the + relation OID and the tuple layout in logical replication messages correspond + to the root partitioned table. For + UPDATE and DELETE, however, the old + tuple's flag (O, a full old tuple, versus + K, replica-identity key columns only) reflects the replica + identity of the leaf partition that actually stored the + old row, since that is the identity the old-tuple payload was built from. + This matters mainly to consumers that read the replication stream directly; + the built-in apply worker uses the flag only to tell whether an old tuple is + present. + + diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 86ad97cd937..b6d93a59bfe 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -445,38 +445,46 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup) /* * Write UPDATE to the output stream. + * + * 'pubrel' is the relation as published (the root partitioned table under + * publish_via_partition_root, otherwise the table itself); its OID and tuple + * descriptor go on the wire. 'leafrel' is the relation that actually stored + * the row, and its replica identity decides whether the old tuple is a full + * tuple ('O') or the replica-identity key ('K'). The two are the same + * relation except under publish_via_partition_root. */ void -logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, +logicalrep_write_update(StringInfo out, TransactionId xid, + Relation pubrel, Relation leafrel, TupleTableSlot *oldslot, TupleTableSlot *newslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type) { pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE); - Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT || - rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL || - rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX); + Assert(leafrel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT || + leafrel->rd_rel->relreplident == REPLICA_IDENTITY_FULL || + leafrel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX); /* transaction ID (if not valid, we're not streaming) */ if (TransactionIdIsValid(xid)) pq_sendint32(out, xid); /* use Oid as relation identifier */ - pq_sendint32(out, RelationGetRelid(rel)); + pq_sendint32(out, RelationGetRelid(pubrel)); if (oldslot != NULL) { - if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL) + if (leafrel->rd_rel->relreplident == REPLICA_IDENTITY_FULL) pq_sendbyte(out, 'O'); /* old tuple follows */ else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldslot, binary, columns, + logicalrep_write_tuple(out, pubrel, oldslot, binary, columns, include_gencols_type); } pq_sendbyte(out, 'N'); /* new tuple follows */ - logicalrep_write_tuple(out, rel, newslot, binary, columns, + logicalrep_write_tuple(out, pubrel, newslot, binary, columns, include_gencols_type); } @@ -523,16 +531,19 @@ logicalrep_read_update(StringInfo in, bool *has_oldtuple, /* * Write DELETE to the output stream. + * + * See logicalrep_write_update() for the meaning of 'pubrel' and 'leafrel'. */ void -logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel, +logicalrep_write_delete(StringInfo out, TransactionId xid, + Relation pubrel, Relation leafrel, TupleTableSlot *oldslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type) { - Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT || - rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL || - rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX); + Assert(leafrel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT || + leafrel->rd_rel->relreplident == REPLICA_IDENTITY_FULL || + leafrel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX); pq_sendbyte(out, LOGICAL_REP_MSG_DELETE); @@ -541,14 +552,14 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel, pq_sendint32(out, xid); /* use Oid as relation identifier */ - pq_sendint32(out, RelationGetRelid(rel)); + pq_sendint32(out, RelationGetRelid(pubrel)); - if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL) + if (leafrel->rd_rel->relreplident == REPLICA_IDENTITY_FULL) pq_sendbyte(out, 'O'); /* old tuple follows */ else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldslot, binary, columns, + logicalrep_write_tuple(out, pubrel, oldslot, binary, columns, include_gencols_type); } diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 4ecfcbff7ab..3a1e4b1bcd6 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -1616,12 +1616,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, relentry->include_gencols_type); break; case REORDER_BUFFER_CHANGE_UPDATE: - logicalrep_write_update(ctx->out, xid, targetrel, old_slot, + logicalrep_write_update(ctx->out, xid, targetrel, relation, old_slot, new_slot, data->binary, relentry->columns, relentry->include_gencols_type); break; case REORDER_BUFFER_CHANGE_DELETE: - logicalrep_write_delete(ctx->out, xid, targetrel, old_slot, + logicalrep_write_delete(ctx->out, xid, targetrel, relation, old_slot, data->binary, relentry->columns, relentry->include_gencols_type); break; diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 058a955e20c..787dd3bc8f5 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -228,16 +228,17 @@ extern void logicalrep_write_insert(StringInfo out, TransactionId xid, PublishGencolsType include_gencols_type); extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup); extern void logicalrep_write_update(StringInfo out, TransactionId xid, - Relation rel, TupleTableSlot *oldslot, - TupleTableSlot *newslot, bool binary, - Bitmapset *columns, + Relation pubrel, Relation leafrel, + TupleTableSlot *oldslot, TupleTableSlot *newslot, + bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type); extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup); extern void logicalrep_write_delete(StringInfo out, TransactionId xid, - Relation rel, TupleTableSlot *oldslot, - bool binary, Bitmapset *columns, + Relation pubrel, Relation leafrel, + TupleTableSlot *oldslot, bool binary, + Bitmapset *columns, PublishGencolsType include_gencols_type); extern LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup); diff --git a/src/test/subscription/t/039_partition_replica_identity.pl b/src/test/subscription/t/039_partition_replica_identity.pl new file mode 100644 index 00000000000..769fd20cdc1 --- /dev/null +++ b/src/test/subscription/t/039_partition_replica_identity.pl @@ -0,0 +1,159 @@ +# Copyright (c) 2025-2026, PostgreSQL Global Development Group + +# Test that, under publish_via_partition_root, pgoutput chooses the old-tuple +# flag ('O' full old tuple vs 'K' replica-identity key) from the *leaf* +# partition's replica identity rather than the root's. +# +# The change is published under the root's OID and tuple descriptor, but the +# old-tuple payload is built from the leaf that actually stored the row, so the +# flag has to match that leaf. A leaf with REPLICA IDENTITY FULL must emit 'O'; +# a leaf using its replica-identity key must emit 'K'. This is a wire-protocol +# conformance property: the built-in apply worker only checks the flag to learn +# whether an old tuple is present, so both leaves still replicate correctly. + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $pub = PostgreSQL::Test::Cluster->new('publisher'); +$pub->init(allows_streaming => 'logical'); +$pub->start; + +my $sub = PostgreSQL::Test::Cluster->new('subscriber'); +$sub->init; +$sub->start; + +# part_table_sect_1 stores 'first' and has REPLICA IDENTITY FULL. +# part_table_sect_2 stores 'second' and uses its default replica identity, the +# primary key (id, ts). +$pub->safe_psql( + 'postgres', q{ +create table part_table( + id int generated always as identity, + ts timestamp, + load text, + constraint part_table_pk primary key(id, ts) +) partition by range(ts); + +create table part_table_sect_1 partition of part_table + for values from ('2000-01-01') to ('2024-01-01'); +create table part_table_sect_2 partition of part_table + for values from ('2024-01-01') to (maxvalue); + +alter table part_table replica identity full; +alter table part_table_sect_1 replica identity full; + +create publication pub_part_table + for table part_table + with (publish_via_partition_root = true); +}); + +# A dedicated slot lets us read the raw pgoutput stream and inspect the +# old-tuple flag of each UPDATE/DELETE message. +$pub->safe_psql('postgres', + q{select pg_create_logical_replication_slot('slot_test', 'pgoutput');}); + +$sub->safe_psql( + 'postgres', q{ +create table part_table( + id int, + ts timestamp, + load text, + constraint part_table_pk primary key(id, ts) +) partition by range(ts); + +create table part_table_sect_1 partition of part_table + for values from ('2000-01-01') to ('2024-01-01'); +create table part_table_sect_2 partition of part_table + for values from ('2024-01-01') to (maxvalue); +}); + +my $connstr = $pub->connstr . ' dbname=postgres'; +$sub->safe_psql( + 'postgres', qq{ +create subscription sub_part + connection '$connstr application_name=sub_part' + publication pub_part_table;}); + +$sub->wait_for_subscription_sync($pub, 'sub_part'); + +# Return the ordered publisher-relevant contents of both nodes' tables, so we +# can assert the subscriber stays in step with the publisher. +sub dump_table +{ + my ($node) = @_; + return $node->safe_psql('postgres', + q{select id, to_char(ts, 'YYYY-MM-DD HH24:MI') as ts, load + from part_table order by id, ts}); +} + +# Consume the messages decoded since the previous call and return the list of +# old-tuple flags of the UPDATE/DELETE messages found, e.g. ('UO', 'DK'). With +# proto_version 1 and no streaming, an UPDATE/DELETE message is a one-byte type +# followed by the 4-byte relation OID and then the old-tuple flag byte. +sub old_tuple_flags +{ + my ($node) = @_; + my $hexdump = $node->safe_psql( + 'postgres', q{ + select encode(data, 'hex') + from pg_logical_slot_get_binary_changes('slot_test', null, null, + 'proto_version', '1', + 'publication_names', 'pub_part_table')}); + my @flags; + foreach my $hex (split /\n/, $hexdump) + { + next if $hex eq ''; + my $data = pack('H*', $hex); + my $type = substr($data, 0, 1); + next unless $type eq 'U' or $type eq 'D'; + push @flags, $type . substr($data, 5, 1); + } + return \@flags; +} + +$pub->safe_psql( + 'postgres', q{ +insert into part_table values (default, '2020-01-01 00:00', 'first'); +insert into part_table values (default, '2025-01-01 00:00', 'second'); +}); +$pub->wait_for_catchup('sub_part'); +is(dump_table($sub), dump_table($pub), 'subscriber matches publisher after insert'); + +# Drop the INSERT messages we are not interested in. +old_tuple_flags($pub); + +# UPDATE touching the replica-identity key column (ts) of each leaf, one leaf +# per transaction so we can attribute each message to a known partition. +$pub->safe_psql('postgres', + q{update part_table set ts = ts + interval '1 hour' where load = 'first';}); +is_deeply(old_tuple_flags($pub), ['UO'], + 'UPDATE on leaf with REPLICA IDENTITY FULL uses O (full old tuple)'); + +$pub->safe_psql('postgres', + q{update part_table set ts = ts + interval '1 hour' where load = 'second';}); +is_deeply(old_tuple_flags($pub), ['UK'], + 'UPDATE on leaf with key replica identity uses K (old key only)'); + +$pub->wait_for_catchup('sub_part'); +is(dump_table($sub), dump_table($pub), 'subscriber matches publisher after update'); + +# DELETE always carries an old tuple; again one leaf per transaction. +$pub->safe_psql('postgres', + q{delete from part_table where load = 'first';}); +is_deeply(old_tuple_flags($pub), ['DO'], + 'DELETE on leaf with REPLICA IDENTITY FULL uses O (full old tuple)'); + +$pub->safe_psql('postgres', + q{delete from part_table where load = 'second';}); +is_deeply(old_tuple_flags($pub), ['DK'], + 'DELETE on leaf with key replica identity uses K (old key only)'); + +$pub->wait_for_catchup('sub_part'); +is($sub->safe_psql('postgres', 'select count(*) from part_table'), + '0', 'subscriber has no rows left after delete'); +is(dump_table($sub), dump_table($pub), 'subscriber matches publisher after delete'); + +done_testing(); -- 2.50.1 (Apple Git-155)