Logical replication between two Cloudberry clusters: what the server needs to provide

[my-ship-it]

The problem

PostgreSQL logical replication reassembles a transaction from one WAL stream. On
Cloudberry a transaction is not in one stream. A distributed transaction leaves a
piece of WAL on each participating segment and a commit record on the
coordinator, and nothing in the WAL ties those pieces together.

That is fine for the server itself — two-phase commit already knows who the
participants are, in memory, while the transaction is alive. It is not fine for
anything reading the WAL afterwards. The coordinator's record says which
transaction committed. It does not say where its data went.

A receiver that wants to apply distributed transactions to another cluster has to
answer two questions that a single-node receiver never asks:

1. Which segments does this transaction have parts on? Without an answer it
   must wait for every segment in the cluster — and a segment that was never
   dispatched to will never speak, so the wait never ends.
2. In what order did the transactions commit? Each segment's stream is
   ordered within itself, but there is no order between segments. Applying the
   three streams independently can run a later DELETE before the earlier
   INSERT of the same row.

Both answers exist on the coordinator and nowhere else.

What it is for: disaster recovery between clusters that are not the same size

The reason I care about this is not "replication" in the abstract. It is that
today a Cloudberry cluster can only be stood behind by a copy of itself.

Mirrors protect a segment, not a site. For a second site, the choices are a
backup and restore, or a second cluster kept in step by physical means — and
physical means require the two clusters to have the same shape: the same
number of primary segments, segment for segment. That is a hard thing to promise
over the life of a system. The DR site is often smaller than production, on
purpose; production gets expanded and the DR site does not, or not at the same
time; a cluster is moved to different hardware with a different segment count.
Any of those ends the arrangement.

Logical replication does not have that constraint, because it ships rows and
commit order, not blocks and segment positions. The receiver re-routes every
row through the target's own distribution policy, so:

• source and target segment counts may differ — verified both ways: 3 → 2 and
  2 → 3, as well as 3 → 3;
• the target is an ordinary cluster the whole time — queryable, not a replay-only
  standby waiting to be promoted;
• a subset works: replicate the tables that matter to the DR site rather than the
  whole instance;
• the two sides may be expanded independently, since neither side's segment count
  is part of the contract between them.

That is the capability I am after.

And the target does not have to be a cluster

Nothing in the server changes knows what the other end is. They add two things —
a list on a WAL record, and an optional output-plugin callback — and an output
plugin is the standard way a third party gets at a logical stream. What it does
with the stream afterwards is none of the server's business.

That matters because "another Cloudberry cluster" is only one of the things
people want to feed:

• A message bus. A CDC pipeline into Kafka or Pulsar wants the same two
  answers this proposal supplies, for the same reason. Reading the segment
  streams independently — which is all a tool can do today — produces rows with
  no transaction boundary across segments and no order between them, so a
  downstream consumer sees a later DELETE before the earlier INSERT of the
  same row and has no way to know the two belong to one transaction.
• Object storage / a lake table. Appending change files to S3-compatible
  storage, Iceberg, Hudi and so on needs to know when a distributed transaction
  is complete before it commits a snapshot. "Every segment the coordinator
  named has delivered its part" is exactly that condition, and it is not
  derivable from the segment streams alone.
• Another database entirely, or an in-house consumer — anything that already
  speaks the logical replication protocol.

None of those need my receiver, and none of them need a change to the server
beyond what is here. They need an output plugin of their own, which is a
supported extension point, plus the two answers the plugin can now be given.

This is also why commit 6 exists in the shape it does: a cluster-wide slot is
the same slot on every node, made all-or-nothing, and every such tool needs
one. The server exports the piece that creates a slot and keeps the
cluster-wide policy out of itself, so a third party can build that policy rather
than wait for the server to grow one opinion about it.

So I would rather the discussion treat these as an interface than as a hook
for one receiver. If the shape is right, a CDC vendor can use it next week
without talking to me; if it is wrong, it is wrong for all of them at once —
which is the argument for settling it now rather than after the fact.

What I am proposing

The coordinator's distributed commit record carries the list of participating
segments, and logical decoding hands that list to an output plugin through one
new optional callback. Nine commits, ~900 lines, of which only two change an
interface:

• XACT_XINFO_HAS_DISTRIB_SEGS — the list on the record, written only where
  wal_level already admits logical decoding.
• distributed_commit_cb — an optional output-plugin callback, called in WAL
  order for XLOG_XACT_DISTRIBUTED_COMMIT and XLOG_XACT_DISTRIBUTED_ONEPHASE,
  for the slot's own database only.

The other seven are consequences, and three of those are plain bug fixes that
stand on their own.

The nine commits, rebased on current main:
https://github.com/my-ship-it/cloudberry/tree/mpp-logicalrep-kernel

The nine commits

#	Commit	What it does	Why it is needed	Touches
1	Fix the cmockery unit test harness for PG16 41624effde7d	fmgrtab_mock.c stops at fmgr_builtins/fmgr_nbuiltins, but fmgr_isbuiltin() has used fmgr_builtin_oid_index and fmgr_last_builtin_oid since PG13; gpopt_mock.c calls into declarations that live only in the ORCA C++ headers; mock.mk passes DLSUFFIX unquoted so it never reaches the compiler as one string	make check in src/backend/*/test has not built since the PG16 merge. Nothing below can add a unit test until it does	2 files, test harness only
2	Fix the description of a two-phase prepare record d9f3fb030603	ParsePrepareRecord() walked the record with the upstream layout, but Cloudberry's StartPrepare() writes two extra arrays (databases to delete on commit and on abort). Everything after the subxact ids was read at the wrong offset. Also folds TwoPhaseFileHeader into xl_xact_prepare instead of keeping a second, drifted copy of the same field list	pg_waldump and the rmgr description of a PREPARE printed whatever happened to be at those offsets. A bug on its own; it also happens to be the record this work reads most	xactdesc.c, twophase_xlog.h, xact.h, + an isolation2 case
3	Name the participating segments in a distributed commit record b5ad3e346cae	New XACT_XINFO_HAS_DISTRIB_SEGS: the coordinator's distributed commit record carries the segments the transaction was dispatched to. Written only where wal_level already admits logical decoding	This is the whole point. Without it a reader must wait for every segment in the cluster, and a segment never dispatched to never speaks, so the wait never ends	cdbtm.c, xact.c, xactdesc.c, xact.h, xlog_internal.h
4	Decode distributed commits and let an output plugin see them e211a4851174	One optional callback, distributed_commit_cb, called in WAL order for XLOG_XACT_DISTRIBUTED_COMMIT and XLOG_XACT_DISTRIBUTED_ONEPHASE, for the slot's own database only. A segment's own stream carries the matching gxid (from the gid for two-phase; remembered on the ReorderBufferTXN for one-phase)	The list from commit 3 has to reach a receiver. No existing callback fits: the coordinator's side of a distributed transaction usually has no xid of its own, so there is no ReorderBufferTXN to hang commit_cb on, and a one-phase marker is not a commit record at all	decode.c, logical.c, reorderbuffer.c, pgoutput.c, 4 headers, test_decoding
5	Ask for a snapshot on behalf of a participant that wrote nothing 310d256685cc	A base snapshot is requested for a participating transaction even when nothing it did was decoded, so ReorderBufferReplay() reaches a callback instead of returning early	A segment that took part but wrote nothing has to say so, or it cannot be told apart from one that is merely slow. "Nothing to decode, so nothing to say" is right on one node and wrong across a cluster	decode.c, reorderbuffer.c, pgoutput.c, reorderbuffer.h
6	Let an extension create a logical replication slot 7e967c4ed770	create_logical_replication_slot() stops being static. Same function, same callers, one fewer static	A cluster-wide slot is the same slot on every primary segment and on the coordinator, made all-or-nothing. That policy belongs in whatever implements cluster-wide replication, not in the server — so the server exports the piece and keeps the policy out	slotfuncs.c, slot.h
7	Make the apply machinery usable from outside the server 5c9f122de391	Three things: relation.c must not look up a per-relation sync state when no subscription is attached to the process; worker.c's tuple helpers stop being static; execReplication.c skips the row lock for append-optimized and PAX	An apply implementation on an MPP target runs on gangs, where the process reading the stream is not the process writing the rows and no subscription is attached. None of the three is specific to one such implementation	relation.c, worker.c, execReplication.c, worker_internal.h
8	Let a logical slot hold the coordinator's WAL 66f797c46b69	Coordinator WAL retention takes replication slots into account, not only connected walsenders	Retention followed walsenders because the coordinator's standby is fed by a bare walsender with no slot. But a logical slot is a promise written to disk precisely so that it outlives the connection, and its subscriber is expected to be away sometimes. During every one of those gaps the WAL was recycled and the slot invalidated at the next checkpoint	xlog.c
9	Execute the invalidations of a transaction wherever it ends 2f00ef5ec337	ReorderBufferForget/Invalidate/Replay/StreamTXN execute a transaction's invalidation messages whether or not it has a base snapshot	The base-snapshot requirement went away upstream in c6ff84b06a6 (2016) when invalidations stopped being executed under a historic snapshot; the guard and its companion Assert stayed. On Cloudberry ordinary DDL is two-phase, so the prepare path hits the assert — measured, 313 core files in three minutes	reorderbuffer.c

Commits 1, 2 and 9 are bug fixes that stand entirely on their own and have
nothing to do with the rest.

How I plan to submit this

One PR per commit, not one PR with nine. They are independent pieces of
review: a build fix for the test harness, a WAL description that read the wrong
offsets, a change to a WAL record, a new output-plugin callback. Bundling them
would mean the interesting ones wait on the dull ones, and that a reviewer who
only cares about reorderbuffer.c has to read cdbtm.c to get there.

Suggested order, which is also what the textual dependencies allow:

• Independent, can go in any order: 1, 2, 9, 8, 6, 7
• Then the core: 3 (the record), 4 (the callback), 5 (the silent participant)

Where two commits touch the same file I will rebase and re-push as earlier ones
land: 2 and 3 both touch xactdesc.c and xact.h; 4, 5 and 9 all touch
reorderbuffer.c.

If the project would rather see one series than nine PRs, say so and I will
post it that way instead — the split is for the reviewers' benefit, not mine.

What a receiver does with this

A receiver can be an ordinary extension — nothing here needs to be in the server.
Sketching one only so the server-side asks make sense:

• One worker per subscription on the target coordinator. It connects to the
  source coordinator for the commit order, and to each source segment for
  the rows.
• The rows never pass through the worker. Readers hold them on the target's
  segments; the worker ships only the list of transactions that are ready, and
  an Explicit Redistribute Motion puts each row on the target segment that owns
  it under the target's distribution policy. Source and target may have
  different segment counts.
• A transaction is applied when every segment the coordinator named has
  delivered its part — never on a timeout, never half of one.
• Everything a batch does — the rows, the position, the deduplication set — goes
  in one distributed transaction on the target, so the recorded position and the
  data it describes commit together.

Different segment counts on the two sides is the case this was built for, not a
curiosity: a row's target segment is computed from the target's cdbhash and
numsegments, which is exactly where a local INSERT would have put it. Tested
3 → 2, 2 → 3 and 3 → 3 end to end, including a mixed schema under concurrent
load with faults injected.

Also working: replicated (DISTRIBUTED REPLICATED) tables, REPLICA IDENTITY FULL, TOAST, streamed (in-progress) transactions, TRUNCATE, and initial sync
by segment-to-segment COPY under a cut point.

Where I would like opinions

1. Is the WAL record the right place for the participant list?

The alternative is a separate record, or a catalog the receiver reads. A separate
record means two things to keep in step; a catalog means the answer is not in the
stream the receiver is already reading. Putting it on the commit record means
the answer arrives exactly when it is needed and cannot disagree with the commit
it belongs to. But it does grow a WAL record that everyone pays for whenever
wal_level = logical.

2. Should writing it be switchable?

I think not, and I want to be argued with. A GUC here is a GUC that turns off a
correctness input: a cluster that stops writing the list does not fail, it goes
on looking healthy while its subscribers lose the commit order. wal_level is
already the knob for "this cluster can be a logical source".

3. distributed_commit_cb — right shape?

It does not fit the existing callbacks because the coordinator's side of a
distributed transaction usually has no xid of its own: it dispatches and writes
no rows, so there is no ReorderBufferTXN to hang commit_cb on, and the
one-phase marker is not a commit record at all. A separate callback was the least
invasive thing I could find. If there is a shape that fits the existing model
better I would rather use it.

4. One of the fixes changes the server's own behaviour.

ReorderBufferForget/Invalidate/Replay/StreamTXN all declined to execute a
transaction's invalidation messages unless it had a base snapshot. That
requirement disappeared upstream in c6ff84b06a6 (2016), when invalidations
stopped being executed under a historic snapshot — but the guard and its
companion Assert stayed. On Cloudberry ordinary DDL is two-phase, so the
prepare path hits the assert and the segment dies: measured, 313 core files in
three minutes. Upstream has not hit it because upstream reaches that path only on
commit, where SnapBuildCommitTxn() has already supplied a snapshot.

Should the fix go upstream first and come back with the next merge, or land here
and be offered upstream in parallel?

5. Anything else that would want these two answers?

If someone is already building a CDC or DR path off Cloudberry's WAL, I would
rather hear what shape they need now than find out after the interface is
settled. The two questions a distributed transaction raises are the same ones
for all of us; the answers should not have to be invented twice.