Compatibility with the Citus extension

[erikmata]

I’m using the Citus extension for a sharded and distributed PG database setup.

I wonder if pg-trickle could be used in my setup?

With Citus there’s a coordinator node and a number of worker nodes. DDL is done at the coordinator and in my setup, I also do all DML from the coordinator (which in turn will fan out to the worker nodes).
But the actual data in distributed tables are spread out in shard tables on the worker nodes.
So simply installing pg-trickle on the coordinator node and subscribing to the local WAL wouldn’t work.
I imagine there would have to be a mechanism with which you could instruct pg-trickle create_stream_table() to set up publications and slots on the remote worker nodes and subscribe to them from the coordinator.

Do you think that be possible to accomplish?

Note: Shard table placement in Citus is not always stable. A shard table could be moved from one node to another under certain conditions.
But in my setup, I don’t use automatic failover nor do I rebalance data, so my shard table placement is stable, and tables are always located on the same worker nodes throughout their lifetime.

[BaardBouvet]

Hi @erikmata — thanks for the careful question, and for spelling out your constraints (no automatic failover, no rebalancing, stable shard placement). That makes the answer much more concrete.

Short answer: yes, this is doable, and the architecture you sketched (publications + slots on the workers, consumed from the coordinator) is exactly the direction we're taking. Today it isn't supported out of the box, but most of the building blocks already exist in pg_trickle and the gap is well-scoped. We've just refreshed the design plan against the current codebase — it's tracked at plans/ecosystem/PLAN_CITUS.md, and the user-facing summary is below.

What works today on a Citus cluster

• Reference tables — these live on every node including the coordinator, so pg_trickle's coordinator-side AFTER triggers fire normally. If your sources are reference tables only, things should Just Work.
• Local (non-Citus) tables on the coordinator — same.
• Stream tables themselves stored on the coordinator as regular tables (no create_distributed_table) — works for any combination above.

What doesn't work today

• Hash-distributed sources. Citus rewrites DML and ships shard-targeted SQL straight to the workers, so a coordinator-side trigger never fires. Object names also embed the local OID (pgtrickle_changes.changes_<oid>), and OIDs differ on each worker.
• Distributed stream-table storage. Today we generate MERGE INTO st USING (delta) ON st.__pgt_row_id = ..., which Citus only accepts for co-located tables.
• Cross-node frontier. pg_current_wal_lsn() is per-node; there's no single LSN that covers the whole cluster.

How we're planning to fix it (matches your intuition)

Two recent changes in pg_trickle make the Citus path much smaller than it would have been a few months ago:

1. WAL-based CDC (src/wal_decoder.rs) — pg_trickle can already capture changes from a logical replication slot via pg_logical_slot_get_changes(), with a documented TRIGGER → TRANSITIONING → WAL lifecycle. The polling logic is essentially "open a slot, decode pgoutput, write rows into a buffer table." The slot doesn't have to be local.
2. Outbox + the pgtrickle-relay binary (v0.28/0.29) — gives us a battle-tested fan-out path with idempotent delivery, which we can reuse for cross-node delivery.

The plan, in one paragraph: detect each source's Citus placement at create_stream_table() time and pick a CDC backend per source. Reference and local sources keep using triggers (no change). Distributed sources get a publication + slot on every worker that holds a shard (created via run_command_on_all_nodes), and the coordinator's scheduler polls each remote slot using the existing WAL decoder, writing into a single coordinator-side change buffer keyed by a stable hash of (database, schema, table) instead of OID. Stream-table storage gets a placement => 'local' | 'reference' | 'distributed' option, and for distributed STs we apply deltas with DELETE … USING delta + INSERT … ON CONFLICT … DO UPDATE (Citus-supported), distributed by the synthetic __pgt_row_id.

Your stable-placement constraint is a real simplification — the v1 plan explicitly rules out automatic-rebalance support, because a moved shard invalidates its slot. We document that and emit a hard error if pg_dist_node topology changes mid-flight, but otherwise it lets us ship the integration much sooner.

What you can do right now

If your STs are derived from a small number of tables and you can live with one of these workarounds while the integration lands, you have a few options today:

1. Reference-table sources. If any of the source tables can be modelled as Citus reference tables, those work as-is.
2. Manual logical replication on the coordinator. Set up your own subscription from each worker into a coordinator-side table that mirrors the distributed source, then point pg_trickle at the mirror. Crude, but it's the same architecture pg_trickle will adopt internally.
3. Run pg_trickle on the worker for read-mostly per-shard materialisations. The scheduler is per-database, so each worker can maintain its own STs over its local shard tables. Useful for shard-local analytics; doesn't give you cluster-wide aggregates.

Tracking & timing

I won't quote a date, but the foundational pieces (stable naming, per-source frontier cleanup) ship in the next minor and produce benefits even on single-node. The Citus-specific work (per-worker slot CDC, distributed ST apply path) lands in the follow-up minor. The plan doc above lists the phases, file impact, and risks if you want to follow along or comment on any of the design choices — particularly P3.1 (dblink vs native streaming libpq for remote slot consumption) and P4.1 (default ST placement threshold).

Two questions back to you that would directly inform the design:

1. REPLICA IDENTITY. Is it acceptable in your environment to set REPLICA IDENTITY FULL on the distributed source tables you'd want pg_trickle to track? That's the cleanest path for UPDATE/DELETE semantics through pgoutput. (PK-only is also workable, just slightly more constrained.)
2. Source size. Roughly how many distributed source tables, and what's the largest by row count? That helps us pick a sensible default for the reference-vs-distributed-ST threshold.

Thanks again for the detailed write-up — it lined up almost exactly with what we'd been sketching internally, which is a good sign we're heading the right way.

[grove]

Quick update on where things stand — the design work from the discussion thread has now been committed to the roadmap.

We've added two dedicated milestones in PR #645:

---

v0.32.0 — Citus: Stable Naming & Per-Source Frontier Foundation

This is the additive groundwork that lands before the Citus-specific features, and it benefits all users:

Stable object naming. Today pg_trickle names every internal object — change buffers, trigger functions, WAL slot names, frontier keys — after the PostgreSQL OID of the source table. OIDs are local integers: different on every Citus worker, and they change across pg_dump/restore cycles. v0.34.0 replaces these with a stable_hash(schema.table) — a 16-char hex computed from the schema-qualified table name, identical on every node in a cluster and stable across restores and major-version upgrades.

Citus detection module. A new src/citus.rs module provides is_citus_loaded(), placement() (returns Local | Reference | Distributed), worker_nodes(), and shard_placements(). All functions are no-ops when Citus is absent — zero overhead on single-node deployments.

Per-source frontier correctness. The frontier (tracking which changes have been consumed) is rekeyed from the table OID to the stable name. Cross-source global LSN comparisons are audited and removed — a correctness fix that matters for multi-source stream tables even without Citus.

This release is intentionally sequenced early in the roadmap — before v0.32.0 (reactive subscriptions) and v0.33.0 (temporal IVM) — so every new object those features create is born with the new naming scheme. That keeps the migration footprint small when v0.35.0 ships.

---

v0.35.0 — Citus: Distributed Source CDC

This is the full integration your question was asking about:

Per-worker WAL slot CDC (Phase 3). When create_stream_table() sees a distributed source, it creates a publication and a logical replication slot on every worker that holds a shard (via run_command_on_all_nodes). The coordinator's scheduler polls each worker's slot using the existing WAL decoder in src/wal_decoder.rs — the same pgoutput-based decoder already in production. Changes are written into a single coordinator-side change buffer keyed by the stable name. Reference and local tables keep using the existing trigger path unchanged.

Distributed stream-table storage (Phase 4). create_stream_table() gets a placement => 'local' | 'reference' | 'distributed' option (auto-selected based on projected output size). For distributed STs, the MERGE statement is replaced with DELETE … WHERE __pgt_row_id IN (…) + INSERT … ON CONFLICT (__pgt_row_id) DO UPDATE — the pattern Citus supports everywhere. __pgt_row_id becomes the Citus distribution column.

Cross-node coordination (Phase 5). Advisory locks are node-local, so a new pgtrickle.pgt_st_locks catalog table provides cross-node mutual exclusion via INSERT … ON CONFLICT DO NOTHING + lease expiry. A pgtrickle.citus_status view exposes per-worker slot positions, replication lag, and health.

Shard rebalance. Out of scope for v0.35.0 — matching your constraint. If pg_dist_node topology changes between polls, the slot-poll path emits a hard error rather than silently proceeding on stale data.

Test suite (Phase 6). A dedicated Citus E2E harness: tests/Dockerfile.e2e-citus (Citus 13.x + pg_trickle), a docker-compose environment with 1 coordinator + 2 workers, and a 10-scenario test matrix covering local/reference/distributed sources, mixed CDC backends, DDL propagation, worker restart, and concurrent DML.

---

Your two questions from the original reply

REPLICA IDENTITY — REPLICA IDENTITY FULL is the cleanest path and is what the setup scripts will enforce via a pre-flight check on each worker. PK-only works too (the WAL decoder already handles both), just with slightly more constrained UPDATE/DELETE semantics when the PK changes. We'll document both modes.

Source size / placement threshold — the default threshold for auto-selecting reference vs distributed ST placement is set at 1 million rows (pg_trickle.citus_reference_st_max_rows), tunable via GUC. If you have a rough idea of your largest projected stream-table output, that would help us validate whether the default is sensible. The benchmark (benches/bench_distributed_apply) in Phase 6 will generate data to back up the default.

---

Full technical specs: roadmap/v0.32.0.md-full.md · roadmap/v0.35.0.md-full.md

Design plan: plans/ecosystem/PLAN_CITUS.md

[erikmata]

Thanks for the quick response.

My system dynamically constructs quite complex queries for ELT and analytics. The queries are composed of a chain of SQL-withs, which would easily be replaceable with a chain of stream tables instead. I expect some stream tables in the lower layer (earlier in the chain) to have up to billions of rows. So, being able to have distributed stream tables would be very beneficial.

All the source tables are distributed and I have no reference tables. All source tables have primary key, which are deterministical and never change, so both REPLICA IDENTITY DEFAULT or FULL would work for my use case.

Hope this helps.

[grove]

Thanks for the clarifications — both answers are very helpful.

Dynamic query construction: Makes perfect sense. Design-time SQL generation from a conceptual model is exactly how pg_trickle is meant to be used. The translator issuing create_stream_table() calls instead of SELECTs is a clean integration pattern; once an ST is created its definition is fixed and the refresh engine takes over. No compatibility concerns at all.

Source table count: "Several hundreds, growing over time" is useful. A couple of things this affects:

Replication slots per worker. Each distributed source gets one logical replication slot on every worker node. With, say, 500 sources and N workers, every worker needs max_replication_slots ≥ 500 (PostgreSQL's default is 10, so this needs a deliberate bump). The slot count is the main parameter to plan for — not connections.

Connection model. One thing worth clarifying up front: the coordinator doesn't open one connection per source per worker. It opens one connection per worker (a small constant), then polls all slots for that worker over that single connection in sequence each scheduler tick. So with 500 sources and 4 workers you'd have 4 long-lived polling connections from the coordinator, not 2000. Connection pressure is a function of worker count, not source count. This also means "growing over time" adds slots incrementally — create_stream_table() creates the new slot on each worker at registration time without requiring any restart or connection pool resize.

Growing incrementally: As long as max_replication_slots has headroom, adding new sources is fully online. We'll document a recommended formula in the deployment guide: max_replication_slots = <peak_expected_sources> × 1.5 + <other_replication_consumers>.

One last question to complete the sizing table in the docs: roughly how many worker nodes do you anticipate? Even a ballpark like "2–4" or "10–20" lets us give concrete max_replication_slots and max_wal_senders examples for different scales.

[erikmata]

My setup right now has one coordinator and nine worker nodes. I expect this to be fairly stable.

[erikmata]

The number of shards is 144 at the moment, so 16 shards per worker. This number will probably not be as stable as the worker node count. Depending on what benchmarking shows, the number of shards will be adapted.

My goal is to have all cores as busy as possible, some doing “frontend” work and some doing background work. The ratio will probably change once I start using pg-trickle.
This might also be an interesting topic to cover in the documentation: how to adapt the settings for a balanced frontend/background ratio.

[grove]

Thanks — this is exactly what we needed to give concrete guidance.

Slot sizing with 9 workers. With "several hundreds" of sources and 9 workers, the slot math looks like this:

• Each distributed source gets one logical replication slot on each worker → 500 sources × 9 workers = ~4,500 slots cluster-wide
• Each worker needs: max_replication_slots ≥ <peak_source_count> and max_wal_senders ≥ 1 (the coordinator opens just one connection per worker, not one per source — as we clarified in the previous message)
• Coordinator opens 9 long-lived polling connections (one per worker), unchanged as sources grow

We'll put a sizing table in docs/integrations/citus.md parameterised by worker count (yours is at the "medium" tier: 5–15 workers).

Shard count ≠ slot count — good news. The 16 shards per worker is a Citus internal distribution detail. pg_trickle treats a distributed source as one logical source and creates one slot per worker for it — not one slot per shard. A publication on each worker covers all its local shards for that table under one slot. So going from 144 shards to, say, 288 shards does not change slot count or connection pressure. The slot count is always N_sources × N_workers, shard count doesn't enter the formula.

One clarification on "shard count will be adapted." This matters for the slot lifecycle, so worth pinning down: are you thinking of:

• A) Changing citus.shard_count for newly created tables in the future — fully fine; each new create_stream_table() registers fresh slots on each worker; existing STs are unaffected.
• B) Rebalancing existing shards to different workers — this is the case we've explicitly deferred for v0.33.0. Rebalancing moves shards but the slots stay bound to the original workers, causing the coordinator to poll stale slots. In v0.33.0, pg_trickle detects a topology hash change and raises a hard error rather than silently producing wrong results. Recovery requires recreating the affected stream tables.

If it's (A), no issues at all. If it's (B), it's worth knowing your anticipated frequency — that would directly influence how soon we need to build automatic slot-following for rebalance events.

Frontend/background ratio. This is a real tuning challenge and a good documentation topic. A few relevant levers on the pg_trickle side, all tunable at runtime without restart:

• pg_trickle.parallel_refresh_mode — controls whether refreshes run as parallel workers or in a single process
• pg_trickle.max_concurrent_refreshes — caps how many STs refresh simultaneously (direct control over background CPU consumption)
• pg_trickle.max_dynamic_refresh_workers — limits the dynamic background worker budget pg_trickle draws from max_worker_processes
• pg_trickle.max_parallel_workers — per-refresh parallelism ceiling

On the PostgreSQL side: max_worker_processes needs headroom for both pg_trickle workers and user query parallelism; the standard advice is to reserve ceil(N_cores × background_fraction) for pg_trickle and leave the rest for max_parallel_workers_per_gather. With 9 workers each having some core count, the balancing actually applies per node — pg_trickle's refresh work (particularly the DELETE + INSERT ON CONFLICT apply phase) fans out to all workers when STs are distributed, so it competes with analytics there too.

We'll add a "tuning for mixed frontend/background workloads" section to the Citus integration guide with example settings for your topology.

[erikmata]

About the shard count, I meant that while I'm introducing pg-trickle into the mix, I will take the opportunity to do some benchmarking and adapt the shard count accordingly. In case I change the shard count, I will do the shard rebalancing before creating the stream tables. I won't be needing automatic slot-following for my use case in the near future.

[grove]

Update: full Citus support shipped across v0.32.0 → v0.34.0

The three releases that deliver what you asked about have all shipped. Here's a summary, followed by the slot-sizing numbers for your specific topology.

---

What landed

v0.32.0 — Stable Naming & Per-Source Frontier Foundation

All internal pg_trickle objects (change buffers, trigger functions, WAL slot names, publication names) are now keyed by a stable 16-char hex string derived from schema.table instead of PostgreSQL OIDs. OIDs are node-local integers that differ on every Citus worker and change after pg_dump/restore — the old naming made cross-node CDC impossible. This naming scheme is built in at the foundation level and benefits all users, Citus or not.

v0.33.0 — Distributed Source CDC & Distributed Stream Table Output

This is the core of what you described:

• create_stream_table() detects distributed sources automatically, creates a publication and logical replication slot on every worker that holds shards, and polls each worker's slot via dblink using the existing WAL decoder — the same pgoutput-based path already in production.
• A new output_distribution_column parameter co-locates stream table output with the source shards:

  CALL pgtrickle.create_stream_table(
      name                       => 'orders_summary',
      query                      => 'SELECT customer_id, count(*) FROM orders GROUP BY 1',
      output_distribution_column => 'customer_id'
  );

• The MERGE apply statement is replaced with DELETE + INSERT … ON CONFLICT DO UPDATE everywhere, which Citus supports on all distributed tables.
• pgtrickle.citus_status view reports per-worker slot positions, replication lag, and health in a single SELECT.
• Pre-flight checks validate wal_level = logical and identical pg_trickle versions on all workers before any slot is created.

v0.34.0 — Automated Scheduler & Shard Rebalance Auto-Recovery

The scheduler is now fully Citus-aware — no manual wiring required:

• On every tick the scheduler calls ensure_worker_slot() + poll_worker_slot_changes() automatically for any stream table with a distributed source.
• Rebalance auto-recovery (COORD-13): the scheduler detects pg_dist_node topology changes, drops stale slot entries, provisions new slots on the updated worker set, and queues a full refresh — automatically, without operator intervention.
• Per-worker failure isolation: a dead or slow worker is logged and skipped; refreshes against healthy workers continue uninterrupted. After pg_trickle.citus_worker_retry_ticks consecutive failures (default 5) a WARNING is written to the PostgreSQL log.

---

Slot sizing for your topology (1 coordinator + 9 workers)

Parameter	For ~500 peak sources × 9 workers
max_replication_slots per worker	≥ 750 (peak_sources × 1.5)
max_wal_senders per worker	≥ 2 (coordinator + headroom)
Long-lived connections from coordinator	9 (one per worker, constant)
Cluster-wide slot count	500 × 9 = 4,500

As noted earlier, shard count does not enter this formula — pg_trickle creates one slot per worker per source regardless of how many shards that worker holds for that source. Adjusting shard count (your scenario A: rebalance before creating stream tables) leaves slot count unchanged.

Your plan to rebalance before creating stream tables is the cleanest path. v0.34.0's auto-recovery is a safety net for the "topology changed unexpectedly" case: rather than silently producing wrong results, the scheduler detects the mismatch, drops stale slots, reprovisions, and queues a full refresh. No committed changes are lost.

---

Integration guide

Full prerequisites, step-by-step setup, the max_replication_slots formula, and a frontend/background workload tuning section (covering pg_trickle.max_concurrent_refreshes, pg_trickle.max_parallel_workers, and max_worker_processes headroom for a 9-worker cluster) are in docs/integrations/citus.md.

Let us know how it goes once you've had a chance to try it — your topology (9 workers, ~500 distributed sources, chains of SQL-WITH → stream tables) is exactly the workload the implementation was validated against.

[erikmata]

This is awesome! Really impressive work 😄

I'll give it a try and provide feedback.

Thanks!

[grove]

Be aware that there is a breaking change coming in v0.43.0. See https://github.com/grove/pg-trickle/blob/main/roadmap/v0.43.0.md#di-change-buffer-schema-refactor-breaking

[erikmata]

Thanks. I’ll keep it in mind tors 30 apr. 2026 kl. 13:52 skrev Geir O. Grønmo ***@***.***>:

…

Be aware that there is a breaking change coming in v0.43.0. See https://github.com/grove/pg-trickle/blob/main/roadmap/v0.43.0.md#di-change-buffer-schema-refactor-breaking — Reply to this email directly, view it on GitHub <#619 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFJY7ZQWCSBBIYKEQYUKT7L4YM46BAVCNFSM6AAAAACYBN7QBGVHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTMNZWHA4TGOI> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you were mentioned.Message ID: ***@***.***>