Multigres: One stop PostgreSQL Management and Scaling | Deepthi Sigireddi

Hello, uh As Diana already mentioned, today's topic is MultiGrids, and uh MultiGrids is something that we think of as uh an operating system for Postgres. So, that is the vision that MultiGrids becomes the operating system for everyone who's running Postgres. Um a little bit about me. Prior to uh working on MultiGrids, I had actually worked on Vitess, which was the sharding solution for MySQL that came out of YouTube and was open-sourced in 2015. So, uh more than 10 years ago. And Vitess has been running production workloads for sharded MySQL clusters for longer than that at this point. And I worked on Vitess for 7 years prior to uh prior to this attempt to uh use the same expertise of sharding MySQL uh with Vitess for the Postgres ecosystem. So, the idea of MultiGrids, or the motivation for MultiGrids, comes from a desire to redefine how we approach databases. Uh everyone who is running Postgres at scale, or at any scale in production, knows that Postgres doesn't run itself. Uh there are DBAs and database reliability engineering teams for a reason. Our philosophy in approaching how how Postgres should be run, how we run Postgres, is that we don't accept accepted problems. Uh there is a there is a long-standing tendency in the community to say this is just how things are, and this is uh the status quo, and we just have to work with it. But, we think that uh a better future is possible. There are many tools in the Postgres ecosystem, and that's actually the strength of Postgres, the community and the tooling that the community has built to uh to be able to run Postgres in production on uh workloads of any size, small to large, and to achieve the sort of reliability that you need in today's world, where uh everything is running all the time, uh downtime is not acceptable, databases are at the back end, the backbone of critical production services, and these are business critical. They are not life-threatening services, but they are business critical services. And every minute that a database is not available uh loses millions of dollars for for certain companies. There are all these tools in the ecosystem for solving the things that uh are not solved at the database layer, and most of them have to do with things like availability and scaling. Uh so, there are connection poolers, there are uh solutions for achieving high availability, uh because running a single node large Postgres in production is just not viable. Uh there are backup and restore solutions. There are uh very good solutions to each of these problems in the uh ecosystem, in the community, and a lot of them were developed by the community, and some of them were later uh adopted in core. Uh but the thing is that anyone who's running these things in production kind of has to uh figure out how to get them to work together. Many of the tools provide hooks for integration, but the integration is still being done uh as a one-off thing. Everybody is solving that problem separately versus everyone using the same solution, solving it once and reusing it across the community. So, that is another core guiding principle of how we are approaching this problem. Everything is connected. Connect them well. Our goal and vision is that once you use Multi Gress, you should not want to go back. In terms of what constitutes Multi Gress, there are actually three pillars. The first one is cluster management. This is your familiar high availability backup restore sort of problems that every Postgres deployment is solving. Then the next one is sharding. As Postgres deployments grow larger, this is becoming more and more urgent problem. The first attempt to solve sharding for Postgres actually started maybe about 10 years ago now. Some of you may be familiar with Citus. So, there have been multiple attempts to solve sharding for Postgres, and there are some more recent open source solutions that have come out for this as well. But, this is a core problem that has to be solved by anything that is attempting to to become the the operating manual or operating system for running Postgres because the scale just keeps increasing. Humans keep creating more data, and now we have agents creating more data as well. So, sharding is a core pillar for multi grass. The next one is materialization. So once one of the ways one of the things you have to do in order to shard is to be able to re-shard. So let's say you start with a single monolithic database, it gets to a certain size and you decide to shard. You don't want to shard it a hundred ways or two hundred ways right away because each shard requires hardware resources and that would be extremely expensive and wasteful if you don't need the scale. But at some point once you start sharding you have to re-shard. And in order to do that you have to materialize the data from your existing shards to the new shards. So this is a core um mechanism that sits underneath sharding. And once you have this it actually unlocks many other use cases as well besides just sharding. So let's talk about cluster management first. The requirements for any modern database are that it should be performant and scalable, it should be highly available and you don't want to lose your data and that is what cluster management management solves in the multi grass context. This is what a simple multi grass cluster looks like. I know that it doesn't look very simple but it only gets more complicated from here. And many of these diagrams are available on the multi grass website which I have a link to when we get to the last slide. >> [snorts] >> So in a simple multi grass cluster you have one post grass database and you have a connection pooler that mediates all access to that database. Everyone knows that connection pooling is a known problem for uh Postgres deployments, and there are many solutions to it. So, anything that uh claims to help you manage Postgres has to solve uh the connection pooling problem. Applications don't connect directly to the uh connection pooler, they connect to a gateway. And what the gateway does is that it routes uh it routes queries to the pooling layer. One of the things you get by separating out the gateway and the pooler is that the gateway doesn't worry about connection management. It only acts as a Postgres server. So, it implements the Postgres server protocol, accepts client connections, terminates TLS, uh handles uh authentication, and secure authentication. And it can forward all the queries to the pooling layer via a single GRPC connection. And all the queries can be multiplexed on that one connection. You don't need thousands of connections between the gateway and the pooler in order to support thousands or tens of thousands or uh or any even larger number of client connections to the gateway. The pooler does the uh actual connection pooling, and uh when it comes to connection pooling, our goals for this are ambitious, and we have actually realized many of them already in code. Uh we want to be able to support a very high number of connections at the gateway level with a small number of connections at the Postgres level. And uh we want the connections to be uh able to switch between session mode, transaction mode. Uh you should not have to choose between those two while um um giving a fair allocation of connections to clients. So, uh that is everything that the connection pool is supposed to do. Now, uh in order to run this whole infrastructure you need a a provisioner which can provision the resources you need the actual uh cloud resources or physical resources that you need. So, Multi Cluster comes with a Kubernetes operator which acts as a provisioner for running in Kubernetes and running in Kubernetes is the default mode for Multi Cluster. However, we also have a local provisioner that people can use to try this out, run it on their local machines and uh the provisioner pattern allows creation of new provisioners. So, for someone who wants to run Multi Cluster on their on-prem infrastructure, it is possible to do that by uh writing a new provisioner which hooks into the existing uh binaries and runs them just like the operator does for Kubernetes and the local provisioner does for local testing. And as far as storage goes, uh the storage that backs Multi Cluster could be cloud storage uh for the single instance database so that you still get uh data durability and you don't lose data if you lose a single disk. So, that's what a simple cluster looks like. Uh I kind of talked about everything that the provisioner does already. Uh storage, compute, networking, and it also uh makes sure to launch all of the components that are needed. The Kubernetes operator doesn't just launch the components, it manages the full life cycle of all of the components in the cluster. Um whereas the local provisioner pretty much just starts everything and then you can you kind of start and stop the cluster life cycle management is not that important for local testing. Um, connection pooling. So, the connection pooling is fully compatible. No special modes, no session mode, no transaction mode, and it is multi-threaded. So, this is a known limitation in some of the existing uh connection poolers that the architecture is single-threaded and uh that actually becomes a bottleneck in scaling the number of connections that your database can support or your database uh architecture can support. >> [snorts] >> The other thing you uh get even with a single node Multi GRASS is that uh backups are integrated. And uh for this we are just using PG BackRest, which is an existing backup and restore solution in the ecosystem. The uh next iteration or evolution from a single node cluster is to go to a high availability cluster. Now, the state of art in this uh area for Postgres deployments is that you have a primary uh which which takes rights and then you have a standby a hot standby that you can fail over to uh if your primary goes down. Multi GRASS goes beyond that uh in the default configuration or the recommended configuration of an HA Multi GRASS cluster, we actually run three uh nodes and not two. Um the reason for this is that we want rights to be durable. So, um Um, when you have two nodes, if you are running it with a synchronous commit and you lose the primary, you can fail over to the standby, but the standby cannot actually complete any rights because it has no other standby to get a synchronous commit acknowledgement from. And running without synchronous commit is not safe because you can have a the primary acknowledge that a transaction has been committed and die or crash or become unreachable for whatever reason before that transaction has actually flown to the standby, which means that failing over to the standby me has the possibility of going backwards in time. And in order to avoid that and provide high availability along with data durability, you do need three nodes and not just two nodes. So, that's what a high availability cluster looks like and this introduces new additional component into the multi-dress architecture, which is multi-arc. This stands for multi-dress orchestrator. And this is the component that is watching all of the post dress nodes, running health checks on them, detecting if the primary becomes unavailable, failing over to one of the standbys, and reconfiguring replication so that the remaining two nodes can be fully functional and available. In order to do this, in order for us to build this sort of a high availability, which comes with data durability guarantees and not just availability guarantees, we had to redefine what it means to do consensus. And and the only reason consensus is coming in is because as soon as you have a distributed system with multiple nodes, you are immediately in consensus territory and there are well-known algorithms for how you achieve consensus in a distributed system. But the idea here is that in order to do the same thing for a database, you have to redefine what it means because it doesn't mean exactly the same for uh Postgres or any specific database as it does for a for a a general system where you're trying to agree on one value. In a database, you're not trying to agree on one value. You're trying to make sure that the system is up and correct. Correctness is very important. So, uh this is how we have redefined consensus. A consensus system must ensure that every request is saved elsewhere before it's completed and acknowledged. So, this is the durability, which is uh I'm the primary. I receive uh a write. Somebody is trying to insert a row into the database. Before I acknowledge that this row has actually been written into the database, I should make sure that it is saved somewhere else so that if I as the primary crash, it is still there. It has not been lost. Uh the second part of this is that if there is a failure after the acknowledgement, the system must have the ability to find the saved requests, complete them, and resume operations from that point. So, this combines both durability and availability. So, resume operations from that point is where availability comes in. So, let's say there is a failure after the acknowledgement. We know that the write has been saved elsewhere. We have to find where that elsewhere is and make sure that that itself is durable. We lost one copy of the data. We have one copy. Before we restart operations, we have to make sure that it is copied in one more place so that the next failure doesn't end up losing data completely. And while while people may think that even I used to think this that two failures in quick succession should be rare as if each failure is an independent event. And the probability of each failure is the same. But actually experience has shown us that failures are not always independent events. So it is not as if if your failure rate is once in 6 months you can expect that the mean time between fail that the actual time between failures is actually 6 months. There can be cascading failures because the system is overloaded. You do one failover and your new primary c- gets overloaded right away. These things happen. So any consensus system for a database has to be designed for not just single failure modes but also for cascading failure modes. Now, why did we not adopt one of the existing consensus algorithms? There were two reasons. One is that every one of them requires majority quorums. What that means is that let's say for the sake of scaling reads, we want to add more replicas. Every decision has to be a majority decision. And these systems slow down once you go beyond five nodes because with five nodes, you have to get agreement from three. At seven nodes, you have to get agreement from four. And and the network round trips just end up taking too much time and things slow down. So, uh requiring a majority quorum, we see it as a limitation. But, the other thing is that uh they are actually not flexible and not adaptable to custom durability requirements. So, depending on the guarantees that are give that you have from underlying storage and compute, it is possible to design custom durability requirements. If you're running on commodity hardware and uh you have 10,000 uh nodes, all of them with local storage, you can expect that failures of one uh storage node at a time are fairly frequent and you have to design for that. Whereas, if you're running on replicated cloud storage, then uh maybe uh the failure rate is much lower and you want to design for that. So, uh it should be possible to design your database system to adapt to uh custom requirements driven by the characteristics of the underlying hardware. Uh so, how does what does MultiGrus consensus do and uh how does it address these things? So, first of all, the durability policy is pluggable. The default durability durability policy is going to be that you run three nodes and you require two nodes to have the data before anything is acknowledged. That will be the default. But, uh it is possible to run it single node and say, "I don't care. I don't need uh a replica." It's possible to run it with asynchronous replication. So, all of those things are possible as long as the tradeoffs are acceptable. So, this allows uh users, meaning administrators of the database system, to make tradeoffs. Uh Uh, by default, we will run with synchronous replication. Uh, Postgres synchronous replication has some limitations which we may need to uh, fix, but to start with uh, any failures caused by those are going to be extremely rare. So, uh, we will be using Postgres synchronous replication as is. Uh, the the consensus as implemented in Multigress also takes care of cohort registration. So, this is a problem that all consensus algorithms kind of have to solve. If you have uh, five nodes already running and talking to each other, then you can run the consensus algorithm to come to agreement on a value. But, how do you bootstrap the cluster? How do you deal with nodes which go down, which have to be deregistered, and then if you bring back uh, if you bring in a replacement node, you have to register it. Uh, Raft actually does address this to to a large extent uh, with with their joint consensus idea of changing cluster membership. But, cluster membership and changes to cluster membership uh, are a core part of any consensus algorithm, and the Multigress consensus addresses that as well. Uh, the other thing that we are doing in this is that if you look at uh, the existing consensus algorithms, coordination of um, leader coordination for the purposes of leader election uh, and cohort membership are handled by the same nodes as uh, the ones that are accepting requests and uh, trying to move the system forward. Uh, what we have chosen to do instead is that the coordinator role is actually separated from the nodes that participate in the actual consensus. And this allows us to have separation of concerns. There is the main group of database nodes which worry about keeping the system going and then the coordinators which watch over those. And one of the reasons for this also is that we want to run unmod unmodified Postgres. We don't want to have to change Postgres in order to accomplish this. So, Postgres can keep doing what it's good at which is actually servicing queries and replicating it to its replicas and the orchestrator nodes can watch over the cluster. And one of the guarantees or two of the guarantees that MultiPress provides is that there will be no data loss and there will be no divergence of data. This is the split-brain or split-timeline problem that you can get into. In this HA architecture connection pooling and query routing has to do a little more than just connection pooling that we saw for the simple cluster. It retains its previous characteristics but we add a few like read scalability. Now that we have replicas, we can actually use them for reads and load balance read queries across them. And we also want the the failovers to be as close to zero downtime as possible. And the multi-gateway query routing follows the leader and buffers requests during planned maintenance. Because one of the problems that any cluster configuration of a database has to deal with is that you not only have unplanned failures, you also have planned maintenance events and you don't want those planned maintenance events to actually cause any downtime to the client applications. Backups are integrated into the uh HA cluster configuration. Uh we continue to use PG backrest, but the backups will be used for uh the cluster bootstrap because in order to set up standby servers, you have to seed them with a backup taken from the primary. And uh that is all taken care of by the the provisioning and orchestration that's part of Multigress. Uh the other feature you get uh or you need when you have uh when you are running multi-node Postgres in an HA configuration is that you we may actually want to add new replicas for whatever reason. Either because we lost one and we need to add a new one or we decide that instead of having three total, we'll have four total. And all of this is enabled using the backups. Beyond the HA cluster, we uh progress to multi-cell clusters. So uh an HA cluster can run in uh a single cloud availability zone, a single data center, or it can be spread across cloud availability zones or even cloud regions in the future. And uh what that gives and and across data centers if if desired. So uh the reason to do that is so that you build resiliency to total regional failures or total zonal failures. Uh and in a multi-cell configuration uh of Multigress, you could lose one cell completely. You lose the gateway orchestrator post dress all of them and you can still operate. You can still converge on what the data state should be and then open it up again to queries without losing availability. The other thing this allows us to do is to add resiliency and high availability at higher levels of the architecture. So, we want to be highly available at the database level, but that is no good if the gateway is not highly available. So, we can distribute gateways across zones, cells and even if one of them is down or under maintenance for some reason, there are others which can take up the load. In this multi-cell architecture of multi-dress, connection pooling and query routing has to become more sophisticated as well. So, when you when all of your replicas are running in in one domain, whether it's a data center or a zone, there is no cost to choosing one versus the other. It's all about load balancing. But when you have them across multiple cells, then you want to do the routing in in a better more sophisticated way so that you avoid network latency and also the cost associated with transferring data, large amounts of data between domains, between zones, for instance. So, the query routing becomes more sophisticated for the multi-cell architecture. Okay, let's get to sharding and I hope I can get through everything in the time we have. All right. So, sharding there are different parts to what it means to shard your data, to shard your database. So, the first question we want to answer when we are sharding is how do you distribute the data? Uh and the goal is to spread the data across nodes so that we can get horizontal scalability, but we also balance load across the cluster. So, let's uh talk for a minute about what motivates sharding. There are uh different reasons why people end up having to shard. And the first one is the database resources are just not enough to handle the uh the traffic. So, your query rate is too high and you just can't keep up and you have got to shard. And you have already uh split the load by uh running reads on uh replicas, but your write volume is so much that your single primary is not able to keep up. So, that's the first motivation. But there's usually uh a second motivation as well, which is that data size as it gets bigger even adding uh more CPU, more RAM doesn't always uh get you better characteristics because with larger data size, you actually uh tend to run into bottlenecks in processing on the posters level. So, breaking it up into smaller databases can also help with performance uh and uh at least in theory and in practice with MySQL, you can get better than linear scaling by sharding because at the point when people decide to shard, they are actually not able to get the best possible performance from the database because of either their scale or size. So, uh distributing data in uh an optimal way to actually get that horizontal scalability and to actually balance the load is important. At the same time, it is also important to collocate data. Related rows in different tables should stay on the same node so that joins uh can be performant and so that as many transactions as possible stay within a shard. When transactions stay within a shard, we get all of the asset semantics that you get with Postgres basically for free. You don't have to do anything more. Whereas, if a transaction is spanning shards, then it is spanning more than one uh Postgres database, and then we have to do distributed transactions. The Multigres sharding scheme is flexible and pluggable so that there uh can be custom sharding strategies. Uh there will be some built-in sharding strategies, but it will allow for custom sharding strategies. Uh and uh sharding aims to collocate data so that performance can be optimized, joins can be pushed down, and uh transactions can stay single shard. Uh but, it also allows you to distribute the data, and when necessary, you can scatter the queries to all the shards if it's a full table scan for whatever reason. It should be possible to uh scatter the queries across all the shards. It should also be possible to uh do aggregations uh on the shard and then uh return those results to the gateway layer which will collate all the results and then return them back to the client. Let's talk about materialization next. The features that materialization gives you, first of all, it is basically select. So, materialization is saying, I want to be able to select data from my database based on some criteria and then I want to see it somewhere else. Not in the same database. And this the way this powers sharding is that in order to shard, what do we do? We say, let me select a subset of the data that matches this sharding criteria. And move it elsewhere because I want to break it up, right? So, select actually powers that. The The next thing you have to do once you actually use materialization to do the sharding is that you have to ensure correctness. When you are sharding, let's say you're going from monolith to two shards, uh every row has to be in one of those two shards and each row has to be in only one of those two shards. So, each row has has a placement and each row has a unique placement and that's what the correctness is all about. You want to make sure that you're not missing any data. You want to make sure that you're not putting the same row in two shards which would be really bad. >> [snorts] >> Um Sharding always implies migration and a cutover and during the cutover, you want to make sure that there is no glitch. There may be a small period of right unavailability while you start routing traffic to new shards. But that routing has to be dynamic and it cannot be manual. It has to be automated by the infrastructure, by multi-dress. The other feature that is very important is reverse replication. So, let's say you have done your sharding, you have you have your original monolith to new shards and you do a migration cutover and your new shards are the ones that are serving traffic. Something goes wrong, the caches are not warm, the buffer pool settings are not right, something or the other goes wrong and you want to quickly roll it back. Reverse replication allows you to do that because it keeps the source and destination synchronized after the cutover. For some user-defined period at whatever point the migration is complete, you issue a command saying the migration is complete and then you can drop, decommission the original database. What are the use cases for materialization? So, the first use case is actually migrations. So, let's say there is a monolithic database that is being managed using whatever tools and infrastructure and we want to migrate it into multi-dress, we will use materialization to do that. Resharding. Whether it is from one to two or two to four or two to 10, whatever number that is, that is enabled by materialization. Materialization also enables changing the sharding strategy. Maybe you choose a a sharding strategy and you decide that that is not right, you want to change it. You should be able to do it in an online way without taking down time. It enables schema changes. Schema changes can be a big problem in large post-class deployments because of locking and um not be essentially not being able to access that table while the schema change is in progress. And this really depends on the type of schema change and some are uh more dangerous than others. But we want to be able to do reversible schema changes in a zero downtime fashion and it uses the same principle that I already talked about which is to maintain a a stream of changes so that it's possible to uh roll back and then roll forward. Uh major version updates major version updates are a huge pain point for large Postgres deployments. Uh it is possible to do them in an almost zero downtime fashion but it requires a lot of human involvement and orchestration and we aim to automate those. Uh beyond that we can do reference tables, we can do sharded views and more. Where are we uh on this vision? So where we are is that we have cluster management. We have the Kubernetes operator, the provisioning, the HA, consensus, failovers, all of those uh exist today. We did uh an alpha release in early June about 3 weeks ago and that contains all of those. We are yet to uh get started on the sharding and materialization parts of this. >> [snorts] >> Um so that was everything. I have some references uh where you can find the source code. It's on GitHub uh Multigress uh is the org and uh Multigress itself and the Kubernetes operator live in different repos. We have documentation on the website and we have a blog series. Uh we have a blog and many blog series on the blog. Most recently along with the alpha release we published a series of short explainer videos on how things work. I highly recommend uh going and watching them. I talked almost entirely about Multigress. I haven't talked about what we are doing at Superbase to get this into people's hands. Uh Whether or not you already use Superbase, we will uh be having uh an alpha program for Multigress for people to try it out uh either just through deploying the open-source version of it or to try it out on Superbase when it becomes available. So, if you're interested, please uh express your interest in GitHub discussions on Multigress or uh reach out to me or to uh anyone you know at Superbase.