GitHub user my-ship-it created a discussion: Logical replication between two Cloudberry clusters: what the server needs to provide
## 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](https://github.com/my-ship-it/cloudberry/commit/41624effde7dca146098d8b7d6ad20b6f8260c44)<br>`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](https://github.com/my-ship-it/cloudberry/commit/d9f3fb03060383d2fc7636a142c8fabe876cd403)<br>`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](https://github.com/my-ship-it/cloudberry/commit/b5ad3e346caef13815e15706721a23a581d280f5)**<br>`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](https://github.com/my-ship-it/cloudberry/commit/e211a48511740f3ff168add66254ed8a2309137e)**<br>`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](https://github.com/my-ship-it/cloudberry/commit/310d256685cccc8d91981e9235218f9df7754cd4)<br>`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](https://github.com/my-ship-it/cloudberry/commit/7e967c4ed770574c039d421b6d5031ef064bbbf4)<br>`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](https://github.com/my-ship-it/cloudberry/commit/5c9f122de391d9fff17e888c2917132cc5b22ec3)<br>`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](https://github.com/my-ship-it/cloudberry/commit/66f797c46b69659fed654e82a386f176d1a9ce32)<br>`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](https://github.com/my-ship-it/cloudberry/commit/2f00ef5ec337373b03585cdec4825cdac09949b3)<br>`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. GitHub link: https://github.com/apache/cloudberry/discussions/2210 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
