Support for the citus PG extension

[erikmata]

First of all, I'm really excited about your initiative and this project.

I'm using the citus PG extension for a distributed, horizontally scaled database. It significantly improves performance for data import and OLTP.
But, I'm now looking into adding OLAP capabilities and would be interested in using pg_duckpipe for a seamless HTAP solution.

My question is: would it be possible to adapt pg_duckpipe to support citus? Or, perhaps some parts would have to be adapted by pg_duckpipe and others by citus.

The way citus works, you fire up a number of PG nodes having the citus extension. One acts as the master node, while the rest act as worker nodes. For each table that is marked for distribution, citus takes care of creating shard tables on the worker nodes and distributing the initial data. Citus also takes care of seamless fan-out/fan-in of queries so they hit the necessary shard tables that contain the actual data.

First of all, a single source table would be mapped to a single target table in DuckLake, although the actual source data would be distributed in multiple shard tables in different databases (worker nodes). The pg_duckpipe fan-in capability would have to be adapted to handle this situation, where WAL from multiple source shard tables need to be collected from the worker nodes and mapped to a single DuckLake table. Should pg_duckpipe run on the master node or should each worker node have its own pg_duckpipe?

Second, when issuing a query to a source table, the pg_duckpipe transparent query routing would have to be adapted so that pg_duckpipe and citus co-operate, since both use hooks to route queries (citus to shards on worker nodes and pg_duckpipe to DuckLake).

Would be glad to hear your take on this.

[YuweiXiao]

Hi @erikmata! Thanks for your interest and feedback. This is a meaningful use case for large-scale HTAP.

From end Citus user's perspective, the best UX would work the same as a single-node PG: a single add_table UDF triggers the streaming pipeline, and then all analytics queries (fired on the Citus master node) run against the underlying DuckLake columnar tables.

And your proposal leads to the following design decisions:

• Single-node analytics vs. distributed DuckDB execution
  • The DuckDB ecosystem is all about single-node processing. Scalability is mainly achieved through serverless nodes (i.e., one dedicated read-only node per query). In my opinion (at least for now), it is architecturally hard to fuse DuckDB into a Citus distributed query plan. This would likely require significant effort in refactoring pg_duckdb to support mixed execution, for example, dispatching subqueries for DuckDB execution.
• One DuckLake table vs. sharded table
  • One table means we use a single catalog to host the DuckLake table. A fan-in approach should satisfy the need: a duckpipe worker runs on every Citus node, and one node (e.g., the master) acts as the DuckLake catalog to coordinate concurrency.
  • Sharded DuckLake would require one more level of abstraction so that sharded tables appear as a single table to the user. Although Citus might help here, a sharded table would also require distributed DuckDB execution.

Back to your use case: it really depends on your overall data size: whether your volume can be handled by a single scale-up node. IMO, the currently most viable approach builds on single-node analytics + one DuckLake table. Given DuckDB's capabilities, several TB can be easily handled.

Btw, I've only skimmed through Citus and am not very familiar with its internals. I believe there are still capability gaps to be bridged to make duckpipe work in Citus mode, e.g., the shard rebalancing feature seems can easily break the streaming.

[erikmata]

Thank you for your quick reply.

I think your suggested approach, with single-node analytics + one DuckLake table, would work perfectly for my use case.

Regarding the data pipeline, I agree that shard rebalancing would be a problem in general. In my specific environment shard placement is stable, I don't have automatic failover and never run shard rebalancing.

For the transparent query routing, I'm not sure if it would work without requiring any changes.
From what I understand, citus expects to be first-in-line when it comes to the planner hook. I am worried that duckpipe will not get the chance to route OLAP queries to the DuckLake query executor, because citus (having precedence) will route such queries to the distributed worker nodes. Duckpipe would have to be first-in-line to make this work I believe. But, this seems to be something that would have to be changed in citus, rather than in duckpipe.

[YuweiXiao]

Hi @erikmata, I’m planning to work on a PoC for DuckDB + Citus integration, specifically for single-node analytics with one DuckLake table. Overall, I think this should be doable without too much effort. Let’s see how it unfolds.

[erikmata]

That’s awesome! I’d very much like to try it out when there’s a PoC.