This video is based on a talk I gave recently at the SIGMOD conference where we published a paper
on Citus and so it's all about what Citus is, how it works, and why we built it the way we built it.
So if you're interested in using Citus either using the open source extension or using the
Hyperscale (Citus) service on Azure Database for PostgreSQL, or you're just generally interested
in distributed databases or Postgres extensions, then this is going to be a useful talk to watch.
So this describes some of our experiences over the past 10 years building Citus, which is the
first distributed database that is developed entirely using the PostgreSQL extension APIs.
So it adds two features to PostgreSQL: distributed tables, reference tables. Newer versions also
add columnar storage. And the goal of Citus is to give you the simplicity and flexibility of
using PostgreSQL with its SQL support and indexes and transactions and all the capabilities it has
at any scale. And that means building a multi-purpose database that can scale both
transactional workloads by routing and delegating transactions across a cluster of PostgreSQL
servers as well as analytical operations by parallelizing them across the cluster. Now over
the years we talked to a very large number of companies that were looking to scale PostgreSQL
and we noticed that their workloads at a high level mostly fit into four workload patterns.
And the reason is that these are workload patterns for which PostgreSQL is commonly used but they are
also likely to run into scalability bottlenecks that can be resolved through scale out.
For example a common pattern we've seen very much is software-as-a-service. So there are
many companies building software-as-a-service and they can very easily add more customers
and therefore grow quickly, but that also creates scalability problems in the database
because the working set is often very large. Now of course multi-tenant applications in general
have the benefit that they can scale along the tenant dimension, I mean a traditional
approach has been to put every tenant into its own database or schema,
but this does give you the burden of building your own distributed data management plane and
also for PostgreSQL that approach doesn't actually scale very well. So with Citus we can address this
using distributed tables that are sharded by the tenant id, using a table sharing approach.
Another common workload pattern is realtime analytics and particularly
analytical dashboards with a large number of users, which means they issue maybe hundreds
of queries or thousands of queries per second on a very large data set that
is continuously changing, continuously has new data being added to it. And so this is
an area where traditional OLAP stores are not very suitable because maybe they don't really have the
the secondary index capabilities or it's difficult to do rollups within the database, so PostgreSQL
is quite suitable, but of course doesn't have the scalability characteristics needed to handle very
large amounts of data. So that's another common workload pattern for scaling out PostgreSQL.
Two other common workload patterns are the more traditional NoSQL use case, key value storage,
for which PostgreSQL is also used quite often in particular due to its sophisticated JSON support.
So there's many options for querying JSON and indexing JSON but if you start updating you
get a lot of write amplification which then becomes a scalability bottleneck.
And also the traditional OLAP workloads where you have a large amount of data
and a smaller number of analytical queries that scan a lot of that data at once.
Now what's interesting about these workload patterns is that each of them has
a very different notion of what it means to scale, so for a data warehousing application
it's very important that you can scan very large amounts of data very quickly, whereas for a
multi-tenant OLTP application that's not super important at all, but you do need to be able to
support very complex data models involving many foreign keys and constraints, many tables
being joined and transactions involving many different tables and those need to scale for
the multi-tenant applications. So we need to build different types of scalability features
into PostgreSQL, but also we need to be able to support the existing features in PostgreSQL,
and also we need to support the features that are implicitly provided by the ecosystem in
the form of extensions and tools. And so that is why we architected Citus as an extension
rather than forking the PostgreSQL code base, because PostgreSQL is a core project that has
a new release every year that's packed with new features and performance improvements,
but then there is also this vast ecosystem of tools and extensions that always converges to
the latest Postgres versions and provides a lot of the capabilities around Postgres, such as
the PostGIS extension which adds geospatial data types and operations.
So an extension is basically a set of SQL objects like functions and tables to store metadata,
and a shared library which implements the functions and can actually manipulate
Postgres's behavior. So to create a Citus cluster you just set up regular Postgres servers
and install the Citus extension and do CREATE EXTENSION citus to create the SQL objects. And
we typically assign one node as the coordinator which holds the authoritative metadata,
that's the node through which the application connects, but the metadata is relatively static
so we can also synchronize that to all the other nodes or set up multiple coordinators.
So Citus adds two new types of tables into PostgreSQL, so one is distributed tables which are
transparently sharded across the worker nodes, the shards are just regular PostgreSQL tables
which can have indexes, foreign keys, constraints, custom types. Distributed tables can be colocated
with each other to enable foreign keys and JOINs locally on the worker. And then you have
reference tables which are replicated across all the nodes for fast JOINs and and foreign keys. Now
once you've set up a distributed table, the goal of Citus is essentially to make
it look like a regular table and for that it uses a number of hooks or extension APIs to
hook into different parts of the Postgres code. We already saw user-defined functions like
create_distributed_table but one of the most powerful hooks is actually the planner hook,
which allows you to replace the Postgres query planner from the top, you can also do that with
the executor. And then there is a custom scan hook which allows you to put custom nodes into
a PostgreSQL execution plan. There are critical parts of the transaction lifecycle at which you
can run custom code, there's a utility hook for dealing with DDL and COPY and other commands,
and then you can have background workers which are extra background processes that
are connected to the database so they can read from the catalogs and do transactions.
So if a query comes to a coordinator node, what Citus would do is send queries
in parallel to the worker nodes, just regular SQL queries that are planned separately on the worker
nodes just by Postgres and then aggregate the results and return the response to the client.
And the way that integration works is via the planner hook, so after Postgres parses a query
it calls the planner function and the only thing that does is it checks if the planner hook is
set and then if so it calls the planner hook and otherwise it calls the standard Postgres planner.
And that means that an extension when it is loaded it can set the planner hook to one of its own
functions, and that means that every query that is parsed by Postgres goes through the extension. Now
that gives a lot of responsibility but the responsible thing to do is check if
the query contains any distributed tables or reference tables, if so call the Citus planner,
otherwise go back to the standard planner. So the result of the planner hook should be a query plan
that contains execution nodes and in this case we would have at the top probably a hash aggregate to
do the GROUP BY, and under that we could have a custom scan which is an execution node that
just calls C functions to get tuples, and so in this case it would call a Citus component
called the adaptive executor, which parallelizes the query across the worker nodes using multiple
connections per worker node and then those tuples flow back into the Citus executor,
the Postgres executor. Now because we needed to support different workload patterns one thing that
happened over time in Citus is that we actually ended up with different layers in the distributed
query planner, so for queries on a distributed table that have a distribution column filter
they typically just go to one shard, so we can recognize and process those queries with very
minimal overhead to handle high throughput, basically high performance CRUD operations.
Now for a multi-tenant application we often have much more complex data models
and so for that case what the planner does is it mostly looks for filters on distribution columns
and whether those filters transitively apply via JOINs, if so then it can without further analysis
take the whole query and send it to the right worker node because the worker node is just
a PostgreSQL server, so it knows how to do JOINs and subqueries and window functions etc.
Now if the query doesn't have a distribution column filter, it will need to be parallelized
across the shards, but there, there are also some layers, in particular if the query has,
if all the distributed tables in the query are joined by their distribution column,
big parts of the query can typically be pushed down to the workers without further processing,
of course if the JOINs are not on the distribution column then the planner needs to decide
what is the best JOIN order, how to reshuffle the data across the cluster, so that involves
more overhead but that's acceptable because the types of applications that issue those queries are
usually reporting applications where the execution time expectation is relatively high and the number
of concurrent queries is relatively low, so the different planners in Citus roughly reflect the
different workload patterns for which we designed it. Now apart from SQL and query planning another
thing we need to deal with is distributed transactions. Now for simple transactions we
basically delegate the decision to a worker node and that means for example for a multi-tenant
application which has, where all the transactions, all those statements go to the same worker node,
you basically get the exact same semantics as you do in PostgreSQL in terms of MVCC,
but then in a way that scales because we route the transactions across the worker nodes.
Now if your transaction spans across multiple worker nodes we need to do a two-phase commit
and so we use for that the transaction callbacks, so there's a pre-commit callback where we send
PREPARE TRANSACTION to all the worker nodes, this is infrastructure that PostgreSQL already provides
in order to to do two phase commits and then in the post commit, well actually in the pre-commit
we also write a record to a commit log and then in post commit we send the COMMIT PREPARED to
the worker nodes. Now if something fails in the first phase we can just roll back, if something
fails in the second phase or a rollback fails, then we have a pending prepared transaction
on that worker node, and the way we resolve that is by using the background worker infrastructure
so we have a background worker that continuously polls the workers for their prepared transactions
and then looks at the local table containing the commit records written by the coordinator. So if
those records become visible it means the coordinator decided to commit and therefore
the prepared transactions have to be rolled forward, if they're absent then they need to
be rolled back, and that's how the 2PC recovery works and we provide atomicity. In addition to
2PC recovery, the background worker also does distributed deadlock detection,
because the Postgres protocol is interactive, it's very trivial to construct a scenario where
you have two transactions waiting for each other and that can be on different worker nodes, and to
handle that the background worker continuously polls all the workers for their lock graphs and
then constructs a distributed transaction lock graph and if there's a cycle in the lock graph
it cancels one of the transactions by just sending a signal to the process on the coordinator node.
Now we don't support distributed snapshot isolation because (a) there's not really a
hook for that in Postgres, (b) in these workload patterns we've also not seen a strong need or
heard a strong need from customers because for example multi-tenant applications, well they're
complex transactional applications but they always scope their transactions to a single node
by using colocations or reference tables, and you know, something like a data warehouse application
doesn't really have many concurrent transactions going on so it's also less important. But we
may we may still implement this in the future but it will require some changes to Postgres.
Now we also did some benchmarking to kind of demonstrate the scalability
in each of these workload patterns, so for example for multi-tenant bench workloads we used HammerDB
TPROC-C which is based on TPC-C where every warehouse is essentially a separate tenant
but some transactions do span across tenants. So we don't get quite linear scalability when going
from four to eight worker nodes, but pretty close. And then we also compared each of these workloads
to a regular PostgreSQL server, so all the servers have 16 cores, and so if we have no workers,
so essentially we're just sharding locally on the same server we don't see a lot of benefit
for transactional workloads, there's just some overhead so typically throughput is lower,
but for analytical queries we do still get parallelism locally and that can be very
beneficial. And then with four worker nodes and eight worker nodes, the data fits into memory
and we can see like how that scales for example with YCSB we can see linear scalability,
and with larger analytical queries we can often see more than linear scalability
because of more data fitting in memory. So this was the presentation, if you have any questions,
feel free to send me an email, you can also find more about Citus on GitHub, from the Citus slack,
or the documentation. If you are interested in building a PostgreSQL extension yourself,
well Postgres itself has very good documentation, but also there's this GitHub repository
that documents all the hooks that Postgres offers, so that can be a good place to consult.
Thank you very much and I hope you enjoy the paper.