Multigres: Bringing Horizontal Scaling and Enterprise Operations to PostgreSQL (Sugu Sougoumarane)

[music] >> I don't give a [ __ ] about my sequel. I don't give a [ __ ] about auto go. >> [music] >> I don't give a [ __ ] about CLICK HOUSE. I POST DRESS. >> [screaming] >> I DON'T GIVE A [ __ ] ABOUT ANYBODY BUT ME. This is it. We're super excited. We're having the great Sugo here with us. Uh I he doesn't need introduction, but like he's like the distributed database guy for years. Uh he built the test at YouTube, but one of the world's largest uh MySQL clusters at YouTube. Then he was a co-founder of uh PlanetScale. And in the last year or so, he's been now Supabase building out a sharded version of Postgres called MultiGrids. So, as always as while Sugo is giving his talk, if if you have any questions, feel free to unmute yourself and fire away anytime and that way he's not talking to himself for an hour on Zoom. And then and of course, Sugo, thank you coming back a second time. The last time you gave a talk with us is during the pandemic. It's now 5 years later to talk about MultiGrids. The floor is yours. Go for it. Cool. All right. Actually, as uh I forgot that uh I had to introduce myself. Uh so, I guess I'll I'll start with a story um about my oldest interaction with Postgres. My oldest interaction with Postgres was in 1996. Or 90 uh somewhere around that time, 96 or 97. This was when uh uh Stonebraker had built this company called Illustra. And I was at Informix at that time and Informix acquired Illustra. And I was so fascinated by uh his work that I actually transferred myself into the Illustra team and uh worked on building uh DataBlades, which is actually Illustra's uh invention. And believe it or not, that uh that experience actually inspired many of Vitess's designs and many of uh MultiGrids's designs even to the day. Uh it was such an awesome experience. Anyhow, and uh the the other part is the last time I gave a talk uh for Andy was about uh Vitess. And uh I ran out of time. So, I my hope is to uh not have that happen again. So, let's see if that happens. Uh okay. So, about uh MultiGrids. Um most of you think that MultiGrids is a sharding solution for Postgres and that is what it intends to be. But uh I'd like to see MultiGrids as much more than uh just that, uh which is um MultiGrids actually the goal is to redefine how we approach databases. So, when we started this project, we kind of took a look at, you know, where the Postgres ecosystem is, what is the tooling, what are the problems people face. And then we found quite a few issues with how things are in the state of the art with Postgres and we thought, you know, we should take a fresh look at things and these are the kind of some of the things that our thoughts, you know. One thing we realized is Postgres just doesn't run itself. If you start running Postgres, soon you realize that you need to do backups and restores, you need to worry about connection pooling, you need to worry about failovers. So, all these things come into play and there's a lot of tooling that is required for Postgres to run well. Uh the other thing is we have noticed is that people kind of, you know, have learned to accept some of the problems that exist in Postgres. And we said, let us see what what can we do to, you know, why should we have to accept these problems, you know? Can we really solve these problems and solve them better and rethink about how people why people have accepted. So, I'll I'll as I go into the talk, I'll talk I'll you'll see how we approach some of the problems that Postgres has in this area. And the other one is um uh the tooling uh ecosystem is kind of fragmented. Um There is like each tool solves a specific problem within Postgres. But a problem has a blast radius, right? And uh and then when two tools come into play, they need to cooperate and we didn't see much of that happening. And that is another problem that uh MultiGrids solves. So, essentially what our goal is is to be to have MultiGrids solve all your problems uh holistically. And the goal is that once you are uh once you use MultiGrids, you shouldn't want to go back and use anything else. That's kind of our goal. Whether we'll succeed or not, we don't know, but that's kind of what our objective is. So, if you look at MultiGrids, there are I think of it as three pillars. These are the three main areas of MultiGrids. So, the first area that MultiGrids solves is everything about cluster management. The second area is uh uh how to do sharding. And the third area is called materialization. So, these three features work together to actually um the these three these three put together is actually what MultiGrids is. And these three concepts actually come from Vitess. Vitess has all these three concepts also. So, what about cluster management? So, cluster management essentially goes back to uh the first statement I made, which is Postgres doesn't run itself. Everything that we do to make sure that Postgres runs itself comes under cluster management. Okay. Uh the first part of uh there are a few sub parts within the cluster management. The first one is the provisioner. What the provisioner does is make sure that um it uh it does resourcing and component starts. So, uh in MultiGrids, this is actually a Kubernetes operator. But the by design, the provisioner is pluggable. So, today, let's say you don't want to run MultiGrids in Kubernetes, you can, for example, provide a um you know, Teradata um the Terraform, sorry. Uh Terraform operator and uh that can then deploy MultiGrids on your own cluster without the need to run under Kubernetes. And uh what the provisioner does is make sure it gets all the resources, starts all the MultiGrids components, and then that's its job is done. After that, MultiGrids takes care of itself. The components know how to discover themselves and wire themselves up against each other. Okay. So, the next part is uh connection pooling. So, connection pooling is kind of a mess today in uh Postgres. You know, there are a lot of limitations and because there is nothing better, people have learned to live with it. So, what we have done is we have solved connection pooling the right way. First of all, in MultiGrids, connection pooling has no special modes. Uh if you like if you use PG Bouncer, for example, you have to connect and be in transaction mode. If you're in transaction mode, there are some things you can't you can't do. And if you're in the other there is the other one called session mode, I think. And in that, things you cannot do transactions. In MultiGrids, uh there is no need for you to worry about it. You connect to MultiGrids as if it's a Postgres server. You don't have to worry about transaction mode or session mode and just do what you like. But under the covers, MultiGrids will uh smartly still have a connection pool and keep it within the limit of the number of connections that Postgres can handle. And the other part is obviously multi uh it is multi-threaded, which means that you are not limited to one CPU. You can you can run a large connection pool server. Uh so, to explain, you may be wondering, how can you do connection pooling and be fully compatible? What if I go set a variable, uh then that connection gets tainted, uh how do you make sure that, you know, that tainted connection doesn't get reused by someone else? That comes down to how we have architected uh MultiGrids. So, uh just to catch up on uh on the left-hand side, you see this provisioner. This is basically a Kubernetes operator. And the way a user will start MultiGrids by saying that they'll push a YAML saying create a cluster for me. And then what happens is the provisioner allocates nodes uh for uh getting the cluster running. It'll bring up a few components. It'll bring up this multi-pooler. It'll bring up Postgres here and allocate the storage for it. And on the upper side, it'll bring up this multi-gateway. So, you'll wonder like why is this so complex? Why is this two-layer architecture? As the system expands, you will see that all these components are needed. Can you maybe share if you're going to talk about this ahead of time, feel free to shut me down. Like, can you share maybe why you picked gRPC over you know, maybe something that's more lightweight? Oh, yeah. I will I was not going to talk about it. So, let me answer that right away. Uh mainly because uh it is free. Uh it uh it has actually a large number of features that are very very useful, actually too useful for us. Uh but there there may come a time where where we can actually change it to something better or more efficient. At this point, it's the best of the worst. Let's put it that way. Or worst of the best uh right now. There's a question from Yeah, Andre, go for it. Unmute yourself if you can. Thank you so much. Quick question. Why not you you you you do my sockets? Why over the network? Oh, we do Unix domain sockets wherever possible. Like, for example, Multipuller is on the same node as Postgres and we will use Unix domain sockets there. So, here actually I can talk a little bit more about this. Uh the storage, right, is another one is another question. Uh should the storage be local or should the storage be mounted? And in today's cloud services, most of the storage is are mounted. The reason is because uh durability or HA is not a solved problem within the Postgres ecosystem. So, the only safe way to make sure that you don't lose your data is to rely on a replicated mounted storage like EBS. Uh or Google's GCP, I think. Uh yeah. So, the uh so, but then, once you have HA, I'll talk about it. We don't you don't need the mounted storage. Uh in a database, I think I'm probably speaking to the choir. There's a huge value in having the storage being local uh because of the complexity of queries. Uh we'll talk more about that later. So, in my logical view, the Multipuller the Postgres instance and the storage should ideally be within one node. I see them as one box. And uh multi-gateway, uh as a matter of fact, this is actually an interesting um uh way to look at things. People talk about um separating storage from compute. You know, you hear this so much that it's it like if you say that it's taken for granted. That's actually one of the worst decisions, in my opinion, for a database. Uh separating compute from storage because of the number of IOs {slash} round trips the database has to make against storage. So, the way I uh my retort to that is yes, uh Multigress also separates storage from compute, but the storage API is not a file system. The storage API is a database API and this is your storage. The Multipuller, Postgres, and storage is your storage layer and what you scale horizontally is the multi-gateway. Uh is another way to look at it. So, the test is the test doesn't have the Multipuller, right? That that's an artifact of like you know, just to be blunt, the crappy connection handling of the multi-process architecture of Postgres. Like, you need the Multipuller because Postgres doesn't scale with the number of network connections. Whereas, like in MySQL, because it's multi-threaded, it does scale better. It scales better, but it's still not as good. And so, therefore, we do have a equivalent for that. It's called the VT tablet. Yeah. So, this is the identical architecture to Vitess. Okay. >> So, like in terms of like orders of magnitude, uh a simple like uh one of our team members tried to build a demo. On his laptop, he was able to like make 20,000 connections to multi-gateway. Hm. Uh and like with, you know, 100 connections on on Postgres and he was able to do a demo. So, the the order of magnitude difference is huge in terms of what a go process can accept in terms of as far as connections are concerned. I mean, would that be because the multi-gateway is something that you wrote that Supabase wrote from scratch? It doesn't inherit any of the C code of of the existing Postgres infrastructure. Correct. This is this is uh essentially copied from Vitess. Got it. Okay. So, here uh you can see that there is a topo server. That's basically etcd. Uh this is what Multigress uses as discovery mechanism for components to find each other. Like, when a Multipuller comes up, it publishes itself into the topo. And the multi-gateways are watching them and then they discover and connect to them. So, this is how uh this mechanism works. So, in this case, uh I need to explain uh why Multigress can handle uh connections that are stateful. Uh like, you why you don't need special modes. The reason is because multi-gateway has a parser. The multi-gateway parses your statement and if you set a variable, it knows that you actually tried to set that variable. And it connect it transmits that metadata to Multipuller, at which point Multipuller actually tags that connection as having that taint. So, if a new connection comes in that has the same properties, then it will reuse that connection. And that is the reason why it is modeless. So, you don't have to differentiate between transaction mode versus session mode. It is all done for you transparently. And there is still uh some explanation needed as to why these are two different layers. They will become obvious as I move forward with more features. Because right now, in this architecture, there's no reason for multi-gateway to be different from Multipuller. They could both be the same. So, the next part is the consensus protocol. Uh So, uh if you go and look at consensus, um the problem What What problem does consensus solve? Right? What is the use case for consensus? Uh I I talk to a lot of people, say, "Can you define what consensus is?" They usually say something like, "Oh, it's a bunch of nodes agreeing on it agreeing with each other, make sure that things don't diverge." But the real use case for consensus is just durability. When a system accepts your transaction, it wants to guarantee that it will not lose that data. And in today's world, uh the the definition of durability is that the data has to be elsewhere. If it is just within the current node, then if you lose that node, then you've lost your data. So, consensus essentially, what it says is that you need to be able to make sure that the data is elsewhere and you can find it. You know, if this if you lose the current node, you can find it. So, so, consensus, in my opinion, like this from a use case perspective, is not uh something where uh bunch of nodes come to an agreement. Consensus is a way to solve the problem of durability. So, this is how essentially uh I redefined consensus. Uh this is kind of a very loose uh loose term, but uh is it like What does What is the problem that consensus solve? Just make sure that I don't don't lose my data. Make sure that I can continue, you know, my transactions if there are failures, right? So, but then, if you look at consensus from this perspective, uh it feels like uh the consensus protocols uh basically uh is not an exact match. Uh like uh here is an example, right? Today, there are uh Postgres high availability setups, two-node high availability setup, where there's a primary node, there is a standby, right? If the primary node fails, you fail over to the standby and the standby continues to serve traffic. And you rebuild the primary node and then fail over back to the primary and continue. Right? This is a high availability solution. The solution also solves durability. But you cannot make Raft work for this. You cannot make Paxos work for this. Right? But it is a consensus problem. Uh you all the guarantees of consensus have to be uh provided by this system also. There are many other more complex things that people want. People say, you know, "I have four zones. I have like I don't know, I have like I have two nodes per zone, right? For me, my durability requirement is that the data just goes to another zone. You know, that is enough for me, you know? I don't care that there are eight nodes. As long as the data is in a different zone, it's good enough, right? How do you express this with the with a majority quorum, right? So, those are those are the real industry needs uh that consensus, as of uh the state of state of the art today, doesn't solve, right? So, like the majority quorum is a major limitation when it comes to solving real-life problems. Uh but today, what happens is people know only that uh only Raft works and therefore, you know, they make compromises about saying that, you know, either you need a three-node quorum or a five-node quorum. If you have specific requirements, make sure you distribute the nodes such way that, you know, your requirement is met. And the second problem is actually deeper, which is uh these algorithms are all black boxes. They they don't really uh you can't really break them up. You cannot [clears throat] like uh you cannot make Raft work for Postgres, right? You use Raft as a black box and you cannot break it up into smaller parts. So, uh what we have done is actually studied consensus enough. Uh like the the simple question you ask yourself is when Raft does this begin term, what does it actually achieve, right? What rules is it trying to follow? Like, if I changed the begin term to do something slightly differently, what would I break, right? Those are questions that are not answered. So, what we have done is we have redefined consensus into a set of rules saying that if you follow these rules, uh then your system will actually satisfy the properties of a consensus protocol. And we have also validated that uh algorithms like multi Paxos, Paxos, and Raft actually follow these rules. And these are actually um very they are all implicitly there within the algorithms. They're just not explicit. So, all we did was make them explicit, and then what we did was we actually uh re-implemented consensus within multi grass uh on top of Postgres using those rules. And uh if you want to know what those rules are, uh you can go to multi grass.com. There's a blog series. It's 11-part blog series. Unfortunately, it's a long read. But, it actually builds the entire case for uh not having uh to uh where you can actually re-implement consensus using uh any way you like. Uh so, the beauty of this algorithm that we have implemented is your durability policy is actually not part of the core algorithm. You actually provide that as a plugin. You say, "This is my durability policy." It can be any arbitrarily complex rule. You can say cross zone, cross zone at least two zones, or at least two regions. Whatever you like, right? You specify that the algorithm is agnostic of the policy. But, it will make your policy work for you. Uh so, that's the property number one. The property number two is it works on top of Postgres' synchronous replication today. It is not perfect. Uh but, later we plan to uh uh prob- uh probably try an extension, but if not an extension, minor modifications to Postgres uh so, that there is a two-phase synced replication, which will make actually the perfect consensus implementation. Uh the other big change we made was in in a system like Raft, the nodes themselves do the coordination. Actually, the act of coordination need not be done by the nodes, especially when we are going to allow like 5, 10, 15 nodes in a cohort. It doesn't make sense for 15 of them to be doing the coordination. Uh you you'll have fewer coordinators. So, we separated that as a separate role. We call them the multi ox, multi orchestrators. That's what they stand for. Uh and the other beauty of this is uh that uh because of how it is implemented, you can freely add nodes to the cohort, remove them. Uh like, oh, you need more replicas, just add them. Uh and uh all the invariants of a consensus protocol are maintained while these nodes are added and removed. Like, if you look at Raft, there's a completely separate section just to add and remove nodes. In multi grass, it's actually part of the protocol itself. So, what does that give us? It gives us something awesome, which is we now have theoretical proof to say that we will never lose your data once it is committed. Uh and today there are a lot of systems that give you high availability, but they don't give you that guarantee. But, we can give the guarantee because of the fact that it is based on a um based on following the rules of consensus. And we'll also make sure that your cluster never diverges. That's a lot of things I've said. I hope you have no questions. Uh but, uh I'll keep going until you interrupt me. It's so so Yeah, go for it, Andy. No, Jim, you let him rip. Yeah, I Uh Sugu, nice to have you here. Hi. I was wondering if you could go a little bit deeper into that previous slide because it sounds very good at the high level, but you kind of see that as properties of what you might get in the consensus protocol. Uh so, I was wondering if you can maybe bring out like what specifically uh uh is happening over here because, you know, you could you could look at Raft and say it has no diversions and data loss guarantees the of different flavors. So, maybe just go click one level deeper into that. All right. Uh so, let me um uh this uh essentially will explode is likely to explode into a huge uh So, this is basically uh in in my blogs Can you see my screen? Yep. Okay. So, this is actually the last the recap of my blog series. It actually summarizes the So, this this basically is the set of rules that you must follow. If you follow these rules, these are rules that I have shown that algorithms like Raft and Paxos and multi Paxos follow. We verified them. And also that if you follow these rules, all the invariants of a consensus protocol are maintained is basically what we have concluded. And these rules uh do not depend on a on a majority quorum. They just uh they can be defined for any arbitrary set of uh constraints. So, that is one. Uh I know this is this won't be satisfactory, so let me scroll down and show you a slightly more practical Okay. So, if you uh when Raft actually performs a leader change, what does it actually do? Uh it basically performs the leader change in two round trips, something something in that order. But, those two round trips achieve actually six different objectives. And these six uh these objectives are actually described can be described functionally, uh which is one thing is obtain a turn number. Actually, I call it the obtain a turn number is essentially it obtains the exclusive right to perform work, which means that the one of the rules of consensus is that even though there are multiple agents acting, only one of them can act at any given point of time. So, that's an implicit constraint. Uh it's not explicitly stated in the protocol, but that's what those protocols essentially do. And obtaining turn number allows an agent to obtain the exclusive rights to perform its actions. So, in other words, the way it does it is one, it it revokes the ability for all previous agents to do any work. It also establishes the candidacy, which is the future leader. The it recruits the nodes that are needed to uh get the future leader. It discovers the most progressed timeline. Uh propagates it and establishes the leadership. So, if you did these five steps as different things, you will still maintain all the properties uh that a consensus system requires. You don't have to do exactly what Raft does essentially is what we have concluded. I don't know if that makes sense to you. Got it. Got it. So, I get it that you have a different uh interpretation of Raft in terms of what steps and what are the properties for it. And so, is the big picture that you are essentially making this work with the idiosyncrasies of Postgres and and uh having that be correct? Correct. Correct. So, the problem the thing with Postgres is that Postgres does its own replication, right? It has its full sync replication. Uh but, um just that itself is not sufficient for for us to build a consensus system. Uh so, that's why we applied uh these rules. Now, if you if you use Postgres synchronous replication and then follow these rules, then this system uh becomes compliant and follows uh and satisfies the properties of a consensus protocol. The only additional improvement we've made is that uh it is not a majority quorum. Uh it doesn't have to be. Uh so, the original uh paper was by Heidi uh in 2016. Yeah, 2016, Heidi Howard published the Flex Paxos. This is kind of one level above Flex Paxos. I think Flex Paxos only talked about intersecting quorums. Uh here it is purely durability rules. Uh there is no need even need to define intersecting quorums. But, your rules are essentially defining the invariants. In a high-level property to your point a higher level than Paxos, Raft, and using an application, you're defining the experience you want to happen, of which the invariants In order for those experiences to happen, the invariants have to be enforced. Or in other words, what I have done is uh re-stated the actions performed by Paxos and Raft as functional uh functional achievements and saying that this is essentially what Paxos and Raft do under the covers. Um And the the operations that they do are too simple because they only work for majority quorum is how I would put it. If it is not a majority quorum, quorum, then these operations becomes more complex. How do you break it up into smaller parts and more complex operations such that uh the properties of consensus are maintained. But, like like now we're in the weeds here. Um like like Paxos has the liveliness property that you essentially have two-phase commit. You're not requiring No, sorry. You're not requiring a majority? Or you do require You don't require a majority. You do not require a majority. But, like you're not requiring a majority for the you have like the multi ox, if you go back to your slides, and then there was like using like Paxos terminologies, it's like there's acceptors, followers, or acceptors, proposers, and then there's also listeners, who are non-voting observers of the system. Yeah, listeners are not really part of the core algorithm. They are more of a uh for people to discover what the decision is. The core algorithm is only with acceptors and follow-ups. Got it. Okay. Um Okay, so keep going. This is fascinating. But I I think I think what you're you're sort of relying on like the you're sort of the old-school multi-Paxos where like the you have the proposer has a lease that's not going to that just gets renewed. So that way you don't have conflicting proposals of updates to the state machine occurring. You have the one guy that's one or small number that's just generating these things. You're relying on that property. Uh no, not necessarily. The way the way I would define it is like if you the simpler way I would put it is if Paxos uh in Paxos when the when someone ends up recruiting a majority of the nodes, what did it achieve, right? What it has achieved is basically it has revoked the ability for all previous nodes from completing any action. Right? So that's the reason why it can safely move forward. And that uh objective need that recruiting three nodes is one way to revoke the previous agent's objective to make progress. You could, for example, go to that node directly and say, "Please stop doing what you're doing." is also a successful revocation. Mhm. And as long as you revoke the previous node's property, establish the new one, and make sure that it resumes from the last point uh safely, uh you have maintained all the properties that a consensus protocol requires. And that is essentially all that Paxos does. What do you Got it. So the one way to state this, now I think I'm getting it, but maybe this is not right. What you're saying is that in your setup, Paxos is requiring that everyone eventually learns about the decision in a passive fashion at some later point in time, but you're saying in the specific ecosystem that you have with the Postgres replication, you know who all the actors that are involved, and you can specifically go and ask them to do something as opposed to wait for it to be discovered later. So you're going to basically get people on the same consensus sooner because of that. Is that an accurate way of thinking about it, or is it something else? Um It is more It is more in terms of the the way I would put it is if you uh a majority quorum is basically like if you if you told multi-Gresil, "Use majority quorum." that's your durability policy, then it would be exactly like it its behavior would be exactly like Raft. Does that make sense? Yeah. Uh but if you say, "Oh, there are eight nodes, I need my data only for my durability requirement is only one other node." uh then it will not look like Raft, but when the when a failover happens, the failover mechanism will actually visit a lot more nodes, but will make sure that anything any transaction that was committed resumes and is not lost. I don't know if that makes sense. So the the the the basic idea of Raft is that it will not lose your data. Right. So that same property is maintained, except your tolerance to failure is different. In Raft, the tolerance to failure is by majority. In this case, your tolerance to failure is defined by the user. Oh, I see. Okay. That makes sense. Defined by the user. Okay. Okay. So, yeah. It's a I'm trying to you know this this literally is 10 years of thinking and trying to compress it all, so it will be it's hard to explain it in in one go. So I actually wrote a one-pager, and then people said I they didn't understand, so that is what expanded into this 11-part blog series. Yeah. >> All right. So there's a question in the chat that says, "Can we see it as moving consensus into another layer? I.e., the SCD controller that decides which node is the ultimate leader instead of internally within the replica group?" No, this actually that is actually what we tested. In Vitess, we cheated. We we leaned on etcd to do this consensus. In multi-Gres, it is pure it is pure bottom-up consensus protocol. All right, this is gold. Keep going. All right, so there you go. I'm going to run out of time again. Yay. >> [laughter] >> But but I'm happy this I'm I'm really happy that you know this the the last time I tried to present it, people like couldn't understand what I was talking about, so I'm happy here there are people to get this. All right, so if you let's go back to the slideshow. Uh So we'd already said that connection pooling has no special modes and is multi-threaded. So now that you've brought consensus, connection pooling acquires two more properties. Uh we give you automatic replica load balancing, which means that you connect to the multi-gateway, and it will and you say I want to talk to a replica, it'll distribute your it'll distribute your reads across multiple replicas. But the cooler feature is the follow-the-leader part, where multi-gateway always knows who the leader is and sends the traffic to the leader. If there is a failover, it'll follow that leader. It'll say, "Oh, there's a new leader now. I'll now send traffic to the new leader." And the even cooler feature is if there is a planned failover, it'll actually hold off traffic until the failover completes, and then sends it to the leader. So essentially the user will see So one of our policies in multi-Gres is no errors served on any planned operations. So if there's any kind of planned failover, the user will see no errors at all. And uh here is actually a diagram that shows how So by the way, our first release within Superbase at least we are going to deploy a traditional three-node system with majority quorum, which means that one other node should have it from the primary. Uh except that we have the mechanism to come up with more sophisticated topologies in the future. Uh there is a question. In the example you just mentioned, if the durability requirement is just two nodes, and there are a lot of nodes to which the changes are not made durable, but there are at least two to which it has been made durable, then how is there a sense of agreement between all the nodes that the two nodes we have been talking about should be considered the source of truth. This is a fantastic question. The answer to this is actually in the Flex Paxos paper that Heidi published. The way you do that is let's say there are five nodes, right? And uh and your requirement is durability is two nodes, and you performed a write, that means that the failover mechanism needs to recruit at least four nodes. If the failover mechanism recruits four nodes, then it is guaranteed to discover any transaction that has been made durable. And therefore uh any transaction that became durable will not be lost. I don't know if that answers your question. The failure case is is more complicated. And that is actually the wisdom behind this approach is uh a consensus protocol does not worry about performance, but in real life we do have to worry about performance. We typically a system does thousands of transactions per second, but how often do you do failovers, right? You do failovers once a day, once a week. Uh Once a week, that operation can afford to be expensive, right? Uh you want the fast path to be efficient. So that's the reason you want fewer nodes, but having a lot of nodes gives you more latitude. Cool. Oh, I accidentally I don't know how to go back, so I have to go back and click here. >> [laughter] >> All right. So here is actually an zoomed-out diagram. These smaller boxes are actually the multi-polar Postgres and storage together. So the way it works is uh these three nodes, for example, are part of a consensus cohort, and the multi-arc is an external agent that watches over them and performs failovers as needed. And uh the way we have designed it is you can have any number of multi-arcs as you desire. The multi-arcs do not know about do not need to know about each other, and the protocol still works correctly. No problem at all. And uh the arc would connect to a multi-gateway, and it will say, "I want to connect to a primary." in which case the traffic gets sent to a primary. Uh if it says, "I want to connect to a replica." then it'll load balance. And there's actually a special node we are going to introduce, which is consistent replica reads, which means that once you have written to a primary and you issue a read, it can wait for the replica to catch up to that point and then perform the read there for better distribution. Cool. Uh how much time do we have? Okay, I'll try to wrap everything up within 10 minutes. You got You got plenty of time. Keep going. Okay. Okay. So, we have integrated backups. I have a lot of content to cover, by the way. >> [laughter] >> We use PG back rest. So, the consensus protocol actually works with backups to make sure that you know, the system like for example, if you say add a node the node automatically restored itself from backup and joins the cohort consensus system registers it and then it becomes part of your your system. All right. And then finally Multi-Gres allows you to be has this concept of a cell. I'll actually it's better shown as in this diagram. You can actually define failure zones in Multi-Gres as cells and then have traffic routing that are cell aware. So, the there are two ideas here. One is when you have when you are in different zones in different regions, you don't want your replica reads to go you know, across across zones. You want your reads to be local. So, that is something Multi-Gres can do. The other thing is actually if there is a partition network partition, if a cell gets partitioned, it's designed in such a way that a single cell can survive by itself for as long as it can. And we use this a lot at YouTube because in YouTube, guess what? People are mostly watching videos. We could tolerate like 2 hours of you know, network partition. After that, we say okay, no things are getting stale. We cannot continue to do this. This is the olden days. I'm sure the not tolerances are much lower now. Um with the current YouTube. So, I don't I won't go into the details, but the beauty of all this is the provisioner now makes sure these things are up. You don't really need to maintain this complexity. These these are components that you know, maintain by themselves. The one thing you can see here is that there are six nodes in this in this cluster. There are team three multi-arks. The idea is that if there is a network partition, at least one other one multi-ark will have enough connectivity to perform a failover. So, that's the idea behind this. All right. Now, switching gears, we are going into sharding. Here's a hot take. I I think I don't have to tell you this. It is theoretically impossible to build a scalable distributed database. If anybody wants to challenge me bring me your database and I'll bring it to its knees. >> [laughter] >> What I have to say. But yes. Um Why would a guy who's trying to scale scalable databases say such a thing? Well, reality is reality and knowing these rules, we can make good decisions is my opinion. So, here is kind of a quiz of some sort. We have known the CAP theorem. We have known that databases required acid. But these two No, these two things never met because CAP was mostly used for key value replicated stores. But how does CAP come into play for distributed databases is the question, right? How does acid and CAP you know, interact with each other when it comes to databases? So, I I usually explain the difference between atomic and isolation because that is something that people get wrong all the time. I assume you're all educated, so I'm going to skip that part. So, and going to go through this slightly quickly. So, which part of CAP does atomic require? Atomicity is a consistency property. What about isolation? Isolation is also a consistency property. What about durability? Durability was actually not a CAP property at all because what is written to disk was considered durable. But now with consensus data becomes durable only when it is distributed, therefore it starts to be affected by the CAP theorem. So, durability is also a consistency property. Believe it or not consistent of acid is the only thing that has nothing to do with the CAP theorem because consistent of acid deals with you know, foreign key integrity and those kind of things, whereas the CAP theorem's consistency deals with read after write consistency. So, so if you look at this, right? The simple answer is the database wants all your data to be together. If your data is not together, you are going to lose on availability. That's essentially what this slide is trying to tell you. So if you're trying to distribute data you have to know this. You have to know that you are going to cause availability issues. Or if not availability, you're going to cause things to be slower. You know, there is not the CAP theorem, there's the PACELC theorem, right? I don't know how you spell it, but it's the one that says if things are distributed, things are going to be slow. Which is the Abadi's theorem. So how do you then do Oh yeah, and the link back. So, how do you make this work, right? How do you build a scalable system where these properties are maintained and it is also consistent and it is also available. The answer is kind of simple. Keep the things that have to be together together. Keep the things that can be separated separate. So, in reality, if you're like if you have petabytes of data in your database one thing I can tell you for sure everything is not related to everything else. You know, there are there will always be groups of data that are related there are smaller groups of data within your large petabytes of system that are related. If you can bring them together the correct way then you mostly get the properties that you want. You mostly get the best of both worlds. Basically, in other words, you get all the acid guarantees that you want and you get the consistency and availability part of properties that you also want. And that is the philosophy that we use in Vitess as well as in Multi-Gres. So the part One part of sharding is how do you distribute the data? And this is the part that most distributed databases have solved. And and it's not actually any hard problem to solve. You know, distributing the data is the hard problem to solve in the art of sharding is how to collocate it, right? And this is where we believe we've got the secret formula right. Vitess has a fantastic powerful sharding scheme and that sharding scheme allows you to in a formal way make data that is related to each other be collocated. And once you have that you then have a powerful system that can scale forever for you. It won't always work. There is always there will always be one particular workload which is not well suited for this mechanism, in which case you have to pay the price that you have to pay. There is no escape from that. So, the first part of the the first part of Multi-Gres is the sharding scheme is also pluggable. Now, you can see this pluggable thing, right? This this all comes from my experience at Illustra, you know. Make important things pluggable, right? So, in Illustra, the index was pluggable, so so here the sharding scheme can be pluggable, which means that if you say I have a JSON file that decides where the shards live, Multi-Gres can use that JSON file and can you know, make it part of our relational model for sharding. And you it allows you to collocate related rows which allows multi-gateway to push down joins. If there are full scans, you can do scatters and if you collocate the correct data, most of your transactions will be single shard. In other words, your system will scale to petabytes and will continue to feel like you know, a small database. You'll continue to see the performance of a small database. Materialization. What What is materialization, right? Like what Why is it so important? This actually is actually the third secret weapon of both Vitess and Multi-Gres. So, what are What is materialization? Everybody knows materialization, which is basically you take a source table and say materialize this select statement from that table, right? So, that's a simple way to describe materialization. But you don't realize how powerful it is. If you materialize table A into table B and let's say they are identical. What you can do is actually if you're writing to table A you can you know, at some point of time say, okay, you know what? Stop writing to table A, start writing to table B and reverse the replication from table B to table A. Right? And things continue to work fine and no one has noticed a difference. And if you are in a sharded system and you do this, uh then uh nothing is disrupted is the last feature of this materialization. Why is this so awesome and so powerful? Why? Is because of all the things we can do with this. This is this tooling is so versatile that we are even calling it multi-graph. Somebody wanted to call it multi-graph but because it's as versatile as graph. So, what can you do with this feature? We can migrate all your data from any anywhere into multi-graph. You can migrate your data from multi-graph to anywhere else. If you're going to upgrade your Postgres version, we can do it for you, you know. You're running version 15, you want to come to version 17, we'll materialize this table for you in version 17, switch over traffic, everything works fine, continue. Oh, things are broken, fail back, go back to the old one, right? Saves uh version upgrades. We do re-sharding which is basically a materialization where we filter by the sharding key. Uh there are use cases where uh if in a large in a large petabyte scale database, you'll have these small tables that are reference tables, you know, they are they have only you know handful of rows or you know tens of thousands of rows, but those tables have to be joined with the sharded tables. In that case, uh these become cross-shard joins because you know shards that are there you may have like 500 shards and this is an unsharded table. So, what we can do is we can materialize this reference table, spray it across all shards, in which case all your joins become local joins. Uh we can do schema deployments, right? If you say I want to add a column and that causes Postgres downtime, we can materialize a new table with the new column for you and materialize it and then failover and then reverse, right? So, the list is literally endless. You just keep on adding you have found so many use cases that um when materialization becomes a first-class citizen within multi-graph, all these things are like click of a button type of operation and safe because they all have uh the ability to undo what you have done. And all these put together is uh what multi-graph is. So, what is our progress so far? Uh we have finished cluster management. We are going to start work on sharding and materialization. We are behind uh but the thing that we like is we are solving problems in ways that people have not solved before and therefore we think we we bring awesome value to this ecosystem and that's why we are super excited about the future. And that concludes my talk and I can answer more questions. If you don't have more questions, we can go back and nerd out on more consensus. >> [laughter] >> I will applaud on behalf of everyone else. That was awesome as always. Again, we have time for questions so feel free to mute yourself and go for it. If you want to go first, go for it. Yeah, I think this is great. As you I think alluded to a lot of it depends upon what you're trying to get out of your database system. So, is there class of database workloads that you are targeting with multi-graph as you build out this vision? Maybe would love to hear your thoughts in terms of what is that target? And Yeah, that's a fantastic question. I actually >> agents are involved and things like that, so go for it. Yeah, I actually I I was meant to I'm meant to talk about it but I missed it. Uh so, if you look at uh high-scale OLTP systems, uh 90% of them are multi-tenant. And uh a sharded system is perfect for multi-tenant workloads. Uh because uh you can like you can basically like you can scale you can go really really far uh by sharding by tenant, right? Like um let's say you have um So, like you can go up to the point where you know a shard contains just one tenant, right? That's the lowest uh lowest level of granularity you can go with multi-tenant sharding. And uh so, in that case which means that typically like um uh organizations like Slack are the first ones where tenant-based sharding did not scale for them. And then they actually changed their sharding key when they used Vitess. And uh even there uh it is actually common sense, right? Like when a tenant becomes big, it means that they also have a sub-sharding key because otherwise uh if the tenant were themselves, you know, their own database, how would you shard the database is the question you would ask. Most likely those tenants themselves have sub-tenants. Which means that they have a second sharding key and therefore you should be able to shard by that key. In other words, um when a tenant becomes too big, you will most likely be able to find a way to break that tenant up into smaller parts also. Uh Slack actually went a different route when they actually uh sharded by tenant and then they decided that doesn't scale because some of their tenants are massive. They actually switched over to user-based sharding where uh each user lives in their own shard model. Uh there are some there is one model which is actually not uh conducive to sharding which is uh the place where uh P2P things like for example payment platforms because anybody randomly pays someone else and that becomes always a two-phase transaction. Uh the way Block solved this uh previously known as Cash App is uh they just used a memo system where you'd subtract money in one shard and write a journal saying that this money has to go to the other person and then uh make the replay that journal in the other shard and complete it uh is how they solved it. Uh but you could also solve it through 2PC. But what I've noticed is that even in that case they are a lot of the operations are single-shard operations is is what happens. Uh But I would say that that that workload is not easily adaptable to a uh sharded system. Uh a local disk can get you so far. Uh yes. >> [laughter] >> The question is uh is there any The question is are there any plans to do for utilization of an object store? Oh, uh I I am I am against even uh using mounted storage for databases. I I somehow find find it, you know, like database is so IOPS-hungry uh that I want to use uh local storage. The only time I can think of object storage is if you have data that is uh that may be archived on because it's not uh used as often is one is when I thought uh when I felt I I If you ask me, I would go like people are all go already going the wrong route by trying to use mounted storage, you know. We should use uh we should use local storage um as much as possible. Yeah, yeah. I saw great great uh talk there. Um Yeah, with a lot of modern systems nowadays, you see uh utilizations of object store storage and uh tiering, right? So, tiering let's say like a different uh forms of caching, moving the data across different tiers, right? Uh depending on like accessibility patterns. So, like that's why I brought that up because local storage are so constrained local storage, right? And uh but you see like >> Yeah, yeah. So, tiering can be yeah. Uh it can be it can definitely be uh used. I think Auriol actually has uh has the ability to do tiered storage. I don't know how far they are. Um Like uh I guess the I come from, you know, the GitHubs and the Slacks, they don't care about tiering. They they want all the data immediately available all the time everywhere. They they want everything everywhere all at once. That's what That's where I That's kind of I may be biased by the way. No, no, no. Yeah. I like I'm with the with the local storage and like embedded embedded uh uh solution. But uh yeah, that's interesting for sure. Great talk and yeah, awesome. Thank you for answering the question. All right, other questions? All right, so I'll ask the last question. Go for it. I should say it hi. So, I Yeah, yeah, sorry. >> [laughter] >> Um So, again, given your experience uh you know, building out scale version of of MySQL with Vitess and now scale version of Postgres with multi-graph, um can you share us what are the aspects that are giving you the most problems in making this work in Postgres? Like in terms of like the architecture, you know, you mentioned the replication piece. Um I see you're a layer above it. You're not really touching so much the internals of the system. You're you're utilizing what already Postgres exposes to you. But is there any aspect of Postgres that's caused you the most headaches in the last year that you wish Postgres didn't you wish Postgres didn't do it? Uh the wall is a big problem for me. Uh yes, it is uh the reason is because um the reason is because you have um uh have very little control over the replicas. There is like the because the wall is a physical replication method. It is You you want to filter what you're sending when you send the wall or what do you want to do? Um Let me try and see why wall is a nightmare. Okay, so it's like here is what I thought of right? Like uh I was asking like if physically replication was little bit more logical which is actually what OrientDB does, we can do cool things. Like for example, one problem I want to solve is the problem of vacuum, right? Vacuum runs at an unexpected time of day and disrupts you, right? If there was logical replication, what I would do is I would run vacuum on the replica. Right? And then fail over to the replica and then run vacuum on the other replica for example, right? With the wall, I just can't do it. Yeah. So there are there are cool things like that with that we cannot do if we had the wall replication. Got it. Okay. Okay. Makes sense. Okay, awesome. But I miss you girl. You did it again. Awesome. Thank you so much for for giving this talk. Great one as always. I had fun. I had fun. And thanks for the questions. They are awesome. I wish I could spend you know more time talking about consensus. Unfortunately, it's not enough time. So let's see this. I mean let's not wait 6 years for you to give another talk with us, right? Let's let's you know let's let's do one sooner, okay? Oh, definitely. Yes, I'll be I'll be happy to.