CMU Database Group · 1:24:53 · https://www.youtube.com/watch?v=3wNmpE6DsSs
0:05 official members bounce bust off the Chrome realize it's real hell what we call Home official members bounce bust off the Chrome realize it's real hell where we from the ins side one I Vision fat lcks and M Clips the Chrome will make the average imitator do back flips you like a [Music]
0:27 pistol we have we have three lectures left um so for you guys in the class the road map to finish up the semester we're almost there um we no class obviously that should be Wednesday not Thursday uh this Thursday's Thanksgiving Wednesday's there's no class on campus or university so don't show up here uh I still will hold my office hours I'll I'll post it on slack or not slack post it on Patza
0:51 uh we'll do it over Zoom so if you want to talk to me on Wednesday I'll still hold my regular office hours even though we're not having having a lecture the Wednesday's class next week at the end of the semester we typically do this poper session where uh in the past we've had students vote for what systems they want me to cover so it's like flash talks but from Andy like I'll try to cover as many D systems I can in like five minutes we can do real ones we can do fake ones um or not sorry we can do hobby ones or joke ones parody ones uh like mango DB instead of mongod DB right things like that or we could do um if there's bunch of topics you guys you know want to learn more about we can cover like you know try to go off the cuff in like 5 minutes try to cover
1:31 those uh so I'll post on Pia what we want to do this this semester I'm up for anything you guys can vote for what you want to do in previous lectures we've done call-in shows where people like from the outside team you can call in and ask crazy questions you guys can ask questions so we we can do anything uh so I'll post on again P what we want to do project four we do uh on the Sunday December 8th at midnight homework 6 I think was the gr scope I said it said it was due December 4th the website it was due December 8th I've changed this now
1:59 gr scope and website now I'm giving you an extra day it'll be due on December 9th okay yes can we use can you use late days for project 4 yeah why notes don't allow us to uh I mean like I don't think the the university doesn't want me to have doesn't want anything due the week of
2:22 finals but like I don't care if you don't care right like uh so yeah you you can use late days I mean final grades aren't due till like December 18th or something like that right there's some some date I have to put in final grades so long as you get it done before that I don't care right
2:41 um and then the final exam will be December 13th again so we want to get the the part of the reason why making this do earlier rather than later because we want to give you the solutions everything by the 13th so we'll just we'll put them out like maybe the the 11th um but the the homework six isn't isn't there won't be a lot of topics covered in Homer 6 in the final exam because you're not building that you know it's not you're not building it for the projects right we we want to emphasize the things we do in the projects so again December 13th uh Friday 8:30 a.m. uh we'll do the study uh the final final exam review
3:16 Wednesday's next class before the pop re session um but there won't be early exam and then don't don't disappear before the date okay um there's two more uh talks datas talks this semester today we have the guys in China doing grip time DB the time series database built on data Fusion that'll be at 4:30 again on zoom and then next week we have another dude uh for another system in China called datab Bend but they've sort of in the same way that like data Fusion was sort of the Standalone piece that they built to do query execution uh open doll
3:50 is a data access layer meaning like think of like the the the the dis manager in busb like you have your buffer pool talk to some disc manager that then goes out to some disc and get gets gets gets the data so they have a layer that replaces the dis manager that you can replace the thing in bus toop to go read data from S3 read data from EBS or what whatever remote URL you want so that guy's gonna talk uh that should be
4:13 not November 25th whatever whatever Monday is next week I think December 2nd or something okay all right so I wasn't here last class I was in Chicago then I had to go to Amsterdam for uh an award session and then my parole officers all pissed off off cuz like whatever so cuz it was out the country but even though it was it was an award session it wasn't my award but I was pres I was giving a talk at the award so like oh that doesn't count for work because it's not my award but like whatever all right so that was last week so I apologize for not being here but you know way did a good job so last class W talked about again introduction to distributed databases and he presented these sort of the basic three
4:54 system architectures that you can have in a in a in a database system shared everything is what we understand bus tub or something like post to be right where it's like a single system that's running on a single box that has control of the disc the memory and the CPU and everything's all local to it the shared dis architecture is where something is like running on S3 or there's this distributed file system layer where you have multiple compute nodes that essentially aren't holding the data but know how to access that shared storage to go pull the data out and do something with it and then Shar nothing with the classic distributed system where every node is is a is a sort of holds a piece of the data that in your database uh and anytime you want to run queries or transactions across them you're sending messages between those noes and they know how to talk to between them right so in the before the cloud before something like S3 and the prominence of that shared nothing was the most common distributive system approach and Shar disc was actually considered a bad idea because you know
5:52 uh Oracle tried it and ter data tried it it was always kind of expensive but like with S3 that that's a game changer right snowflake was one of the big the first systems that came along and said hey let's build on Shar disc architecture on top of S3 so we don't have to do storage ourselves uh and that was a huge win for them again so the war ceremony I was at last week was the was the co-founder of of of uh of snowflake um then we saw different ways of split up the database
6:16 right the the partition charting schemes hash partitioning range partitioning it's just a way to say I have a a database it's larger than I can fit on a single node how do I split it up and distribute it across my my different nodes and even if you're doing a shared disc architecture something like S3 has technically infinite storage you still need a way to assign what compute nodes are allowed to touch or look at what pieces of data and a partitioning scheme would give you that and then he talked a little about at the end the different ways you can coordinate transactions and that's one of the big things we're talk about today is do I have this this sort of global traffic cop that can that is in charge of deciding what transactions are all to commit and when or am I doing decentralized architecture where I have the different nodes uh coordinate amongst themselves of how to commit transaction
7:00 right so we're we're not going to go too we're not going to see real distinction in centralized decentralized today but like the protocols we'll talk today ites actually doesn't matter which one of these two you're doing so just as a reminder for everyone what what what do we mean when we say we're doing olp right we have this distinction early in the semester we talked about olp versus olap or online transaction processing versus online analytical processing right the transaction processing World we're talking about uh queries that touch a small amount of data but actually can modify the database they can update records insert and delete and so forth
7:34 and they're be basically doing the same operations over and over again but we don't care whe they're doing the same thing and over again the thing we care about is that now they're modifying the database or now they want to see the database in a consistent state with all the asset properties that we we learned about before and now how do we how do we manage that and coordinate that across multiple nodes and I realize I'm only giving you a single lecture if you count last or last class as two lectures like distributed databases transaction Rees are super hard we could do years on this
8:03 uh I'm just trying to give you a a quick brain dump up like here's the things you got to be aware of so when you got in the real world uh you can at least be aware of the different things you have to consider and then Monday's class next week we'll talk about how to do distri olap system right basically how do you do a join across multiple nodes that's actually really hard
8:23 okay all right so if we go back to the diagram we had before from last class this distri the centralized coordinator right say we we have a transaction it's going to run the application server here it wants to touch data at these three three nodes and assume we're just doing you know simple partitions we have P1 or P4 so we get a begin request somehow we know to land on this one node where p1's located and we're going to designate that as the primary node of this
8:49 transaction right what that means we'll see in in a second but then now assuming that the application server can talk to any other other node it can send query requests to all the other noes to update things get some data do whatever it wants and then now and this is all done in the context of a transaction so then now when my application server says I want to commit this primary node is responsible for communicating with the other nodes and say hey all right we were all involved in this transaction is it safe to commit yes or
9:20 no and then once they all agree which we'll talk about how we get that agreement which is not easy uh then we can send the acknowledgement back to application server and say yes your transaction has successfully committed and again ass running with all the same asset guarantees that we talked about before on a single node shared everything system no matter what happens to my database system if I crash uh if any of these nodes crash or the whole thing crashes come back I should still see all my data or now if I tell the outside world my transaction is committed if anybody then tries to read data Maybe P3 I should see the changes
9:53 that I saw that created by this other transaction assuming has been updated yet so how actually achieve this is not easy and very very hard and and people often get it wrong and really wrong and often times in the nosql world they just don't do it because it's hard um so our big idea we're trying to do is what we talked about before is that we're going to have our database split across multiple uh resources so multiple compute nodes multiple uh uh you know multiple discs it doesn't matter but we want them to appear to the outside world to the application server as a single logical database system or single logical database so we don't know we don't care in our SQL code what you know what node is going to run our query we just say here's the query we want to run and so whether our database is split across one machine or a thousand machines our SQL query should just work all the
10:44 same but things we haven't we didn't haven't discussed so far is what do we do to make sure everyone agrees that a transaction is allowed to commit and then when it does say go to commit how do we make sure that it actually gets committed and again we come back we saw our changes so it's easy when everything's up and run up running just as fine right if there's no failures no big deal right but it's when when things start to go bad and they will in the real world then then Things become more challenging so things we'll talk about today is what happens if a node fails while are we running a
11:18 transaction right what happens if a message gets delayed right think just because it's tcpip it doesn't mean it'll guarantee within the stream is going to come in the right order uh um but like there might be some weird NE Hiccup and your message to one node for one stream might get delayed and then show up like a second
11:38 later or what if the other node does actually doesn't physically crash but they it's running a Java database database system and then the garbage collector runs and it pauses for like 5 Seconds which can happen so now the DAT that that system looks unavailable but it's still alive you can ping it you can send you know you can ping the node it's still going to be responsive but the database over process itself is is not responsive what do you
12:02 do and then what happens if we don't want to wait for everyone to agree that we're going to commit a transaction and we send back responses early okay well now what happens if one of those notes again crashes comes back how do we how do we account for
12:16 that so this this this I'm just want to reemphasize that this is very very hard to do and this again this is why you don't want your random JavaScript PR building these kind of things this is what you want to use the data system to do this for you and when we see Vector clock gets even worse if you let JavaScript programers handle
12:32 that all right so for today's lecture we're going to make a very very important assumption and that is we're going to assume that all the nodes in our distributed datab system are friendly or we control them they're in the same administrative
12:47 domain right that means they'll be the software itself will be well behaved now there could be problems like I said the jvm could kick in start doing garbage collection and pause your pause your node but but again it's not like we're worried about someone a hacker coming in and running malicious code on our databas node right so that means if we tell a node if one node tells another node let's commit transaction and that node says yep let's commit it I agree let's do it it's not going to reverse that decision later
13:16 on on purpose right failures is a different thing we we'll handle that right so if you don't trust the other nodes then the protocol you're need is not what we're talking about today you're going to need what it's called a Byzantine fault tolerant or bft protocol and this is basically how blockchains work right blockchain is going to run transactions it's basically a database they're going to run commit but they're not going to trust the other nodes so they're going to have this other this this this bft protocol to be able to handle that where noes could start lying and doing things that that that you don't want them to
13:50 do so don't do this unless you're doing Bitcoin right we don't need we don't want this anywhere near any any of our lb datab system systems right because the performance I've had for this is like 5 to 10x over what the protocols we'll talk about today it's bad enough we got to send Network messages potentially over the white AA Network and now we're dealing with the speed of light but then you got some protocol here because you're worried about you know Bitcoin Miners and stuff like that we're we're not going to do this and most applications unless it's unless you're money laundering through Bitcoin you don't you don't need this
14:23 okay this is clear so I mean I it always comes up like okay why are we doing this you do uh how we handle this if we want to do can we do the blockchain instead of doing all the stuff we're doing no blockchain is going to be a generative case of what we're talking about today and
14:37 slower okay so assume again all our nodes are friendly we wrote the software we're running the software we control it uh we don't expect them to do malicious things but we still need to account for failures and anomalies all right so first thing we have to talk about is how we're going to replicate the database across multiple nodes and then that'll lead us to discussions okay how do we extend propagate changes between these nodes to keep things
15:01 replicated then now comes to the question of now we have a bunch of replicas we've told them to do a bunch of stuff can we get everyone to agree that that those changes are allowed to happen and in the right order and then we'll finish up talking about the the consistency issues we have to deal with under the cap theorem or the the pelic extension to cap theorem
15:22 okay all right so the first thing we're going to do is replicate our database so now maybe thinking this doesn't sound like a distributed database where I'm just saying okay I have a node a primary a bunch of replicas that's a distributed database right it's a simp simplistic form of it because we'll see in a second because you're just having all your updates going rights going to one location but you still need to propagate them and you still need everyone to agree that those are the changes you want to
15:47 propagate I would say for the the basic non-p partition uh replication scheme we'll talk about today like 99% of people out there this is what you'd want to use right because you want to use this for scalability be able to have you know run extra queries we also wanted this for durability or availability right because the primary could go down and you can fail over to a replica we'll see we'll see the uh the multi Primary in a
16:09 second so this is to be independent of of what we're talking about today of whether we're doing partition or nonpartitioned uh databases meaning does does every node have a complete copy of the database or do they have some partition of it and also whether we're doing shared nothing or shared
16:23 dis because shared dis systems like Aurora they're basically doing the same kind of replication stuff we're talking about here but they're going to do this underneath the covers for you and hide it from you so the question is going to be how do we configure our replicas like where do the rights go where can reads go when when do changes actually occur right uh how do we wait for this those changes be acknowledged from other nodes and then what happens when we have to do uh how do we actually want to update the database do we run queries or do to replay the log so again this is basically a brain dump of all these different design decisions and we just go through one by one and understanding the trade-offs between
17:02 them all right so the first issue is what the configuration of the system is going to look like when we have replicas and as I said the most common one is going to be the primary replica and the sometimes this called leader follower in the old days used to be called Master Slave uh so if you want to look up you know older papers on this they they will refer to it Master Slave there's a lot of documentation using that terminology um so you might have to just use that term to find what you need
17:29 so the idea here is that there' be some node in our system in our distributed system or distributed D system that we're going to anoint or elect as the primary and then all re oh sorry all rights have to go to that to that primary and then it's the primary responsibility to then propagate those changes to the other to to the replica nodes the
17:54 followers and so if you ever want a consistent view of the database you can always go directly to in the latest version of the database you always go directly to the primary because all your rights are go in there you know things get committed there you'll see them right
18:06 away but you can offload readon transactions or readon queries to the replicas and that takes some of the load off of of the primary so it can just try to do the wrs as fast as possible and then you can do the reads on the on the replicas now again you may not get the latest version of the database on the read replicas but for some applications that's probably
18:25 okay right now the key thing we're going have to deal with and we'll see how to handle this in a second when we talk about uh consensus protocols is at some point the primary will go down either by accident like someone trips over the power cord or there there's a software Hardware failure but often times too when there's a a reg version upgrade like like Amazon will force you to upgrade your versions of your database you know every so often
18:50 so at that point you have to restart the server so at some point the primary is going to not be you know it's going to go down and you have to elect a new primary from one from the one of the replicas so we'll see how to handle that in a
19:01 second the second configuration scheme is to do what's called multi- primary sometimes called multi home and this is where the the any transaction can update any data object on any possible node and now the replicas have to synchronize between them when these changes occur to make sure that to figure out who's allowed to actually commit who's actually should be the light the correct version of a of of the database of an object
19:29 so let's look at two visual examples so the first one is primary replica so again you have the primary all your right queries can have to go here but then you can also can send could send read queries as well read transactions
19:40 and then in the background the primary is going to be able to send uh messages to the replicas to apply the changes that occur on the primary to the replicas and we'll see in a second this either could be the basically the right actually the most common one is is the right hand log that we talked about before so remember we we talked about when like the change data capture where
19:59 uh you could have all your updates occur on the on one node and then you have this other uh component or tool that looks at the right head log you're writing out the dis and also can send those red hog messages out to another node that you would use that for this so now the replicas are basically in kind of this pseudo recovery mode where they're getting these red head log messages over over the wire over the network from the primary and just repls them and applies the changes as if it was doing recovery like normal normally
20:27 but this it's like it's an infinite size or it's a NeverEnding log file right because you not some point you're going to reach the end like an Aries and say okay I'm done now you start running transactions you're kind of constantly always applying the changes as the primary sends them
20:42 over so the primary will always have again the latest version of the database uh if you want if you want to read the latest version of the database then you have to send your read queries here uh but you could also send read queries to the replicas and again you would do this to offload the work that the the primary would have to do so then maybe the Prime only does wrs which again which are expensive because you have to update indexes you have to generate right log messages right you have to run whatever the transaction protocol what Curren control scheme you need right and then so if I want to have my uh my I want to avoid my primary do a bunch of work for for my my read queries I can send the read queries to the replicas most workloads are read heavy right maybe like 1% of the queries or 1% of the transactions are updating 99% of them are reads just think how you browse the web you know it's usually the
21:33 case the other scheme is to do multi- primary and this is where you can do reads and writes to any possible node and then now when a transaction goes goes and commits now I have to figure out who you know what should be the correct version of this right so again we can use that right ahead log scheme that we talked about before just we're propagating updates between the different nodes you could have you know multi- tiers of hierarchy like where you have you say these are two different data centers you would have one leader within a data center that gets all the updates and knows how to propagate them to a bunch of followers that that has locally right but in general again
22:10 there's has to be some some process to figure out who's allowed to commit what changes if you know if two rights occur to the same object on different nodes so most people again most systems are going to need this it's only the really really large guys need something like this like F this is what Facebook uses right actually Facebook started off with this right and the the problem is of course that when Facebook became This Global thing this you know over 10 years ago uh if now I'm down in Brazil their data center is up here up in California
22:45 if I do a right to my timeline or whatever it's called when you post like your stupid cat cat video right that post would have to go up from Brazil to to uh the California data center do the right there but now if I click Refresh on my
22:59 timeline I may not be able to you know the change may not be propagated down to the replica down in Brazil so I may see a stale version of course now people get pissed like where's my cat video I just posted so so Facebook will play this little trick where they would actually put a cookie in your browser that that recorded the your timeline post so that when you refresh your page it would read the local browser copy to see your post and show it back to you right even though it actually hadn't been propagated to the database yet so you can do that when everything's in in the browser uh for some for some applications but you know as you start doing more different kinds of services and different kind of application types
23:35 they had to switch over to something like this so this is what they do now Google spanner will be like multi- primary right but again most your dates only need to be the primary replica all right so the next thing we have to consider is how many copies of data are we should we should we maintain at all time to make sure our database is still available I'll cover what I mean by available when talk about the cap the in a second the basic idea here is that it's a threshold to determine we assume nodes are going to go down how many nodes are are we allowed to go down or how many copies of our data are allowed to go down before we say we don't want
24:11 to uh allow any more reads and writes because we don't want to you know take a change from a transaction modify an object on the last copy of the data we have then that node crashes and we come back and now all our replicas that that we had before don't have that copy so in in the database world this is called case safety in the tribut systems World it might be called something different it's basically just to say again at at
24:35 at what point or how many copies I must have before I start uh a halt operations and we'll see in the case of the when we talking about the cap theorem again the no seal guys would would allow this to go below below uh majority where is in the sort of relational distributed dat this world we don't want to lose data so we very very careful about setting the thresold to a number to say okay once we go below this number then we just halt the system and so it makes us unavailable but that's better than we some people would argue that for some applications not being available is better than losing
25:11 data all right so now the question is also going to be how do we actually propagate the changes from either from the from the primary to replicas or between the the multi-home systems or multi primary systems and this is where this consistency stuff when I talked about asset before I said oh it's kind of of hand wavy when it's on a single node now this actually should make more sense when you do this in a in a Distributive
25:35 system so the idea is then when a transaction says I want to commit at what point do I tell do I send an acknowledgement back to tell them that they've committed again assuming I need to replicate the the change across multiple nodes do I wait for all those replicas to acknowledg that they got the change
25:52 do I wait for none of them or do I wait for some of them so so the two extremes are you wait for none of them and you wait for some of them sorry wait for none of them and wait for all of them if you wait for all of them that's called strong consistency so that means no matter what if I'm told a transaction committed no matter what node I then go to I'll see the I'll see the changes of that transaction
26:15 reflected in eventual consistency the idea is I won't wait for all my notes my replicas to say I got the change I'll immediately come back and say yeah I got it but now if another if another query tries to go read data on another node that hasn't applied the change from the primary I may not see that change from that transaction just committed even though I was told it was
26:38 committed so vual consistency the idea is that you're not going to wait for the the noes to acknowledge it they'll eventually get updated so basically looks like this so same assuming we have a really simple setup we have a primary and a replica and then all the rights go to the primary we tell the primary we want to go ahead and commit and so then we propagate the change to our to a replica and we tell it go ahead and flush right
27:07 we'll talk about how you know how it's actually staging the updates in the right hand log in a second but assume that it's got all the changes at this point and now we say go ahead and flush as again as if it was a transaction being run directly on the replica by by an application it's like the the primary is telling it I want to run this transaction go ahead and commit it so we got to make sure we flush all the changes because again it's the all the buffer pull stuff we talk before right it's it's the no the it's the steal no Force where you have to flush all the change flush all the right hand log changes to say our transaction has
27:37 committed so that means the primary has to stall and wait we send the flush message and has to wait and then at some point it does the flush sends back the acknowledgement and then at that point we can then tell the the application our transaction has
27:53 committed a sous commit or eventual consistency basically says I go go ahead and tell commit I go tell the other the replica node go ahead and flush this as well but I don't wait for them to respond I don't need
28:08 to and I immediately said about send back an acknowledgement now again this this node May crash and then this one gets selected as a new primary so I may come back here and actually now my transaction is missing because maybe this message didn't actually make it over you typically send the flush message first then you send back to acknowledgement but if you did in the the order if you send back the acknowledgement EV on the flush the flush may never make it right and now that transaction doesn't that the replica doesn't know what was the fate of the
28:37 transaction you saw this last class when we talked about consistent hashing right we talking about talk about Quorum rights so that's somewhere in the middle where I do a right to uh to commit my transaction my data is replicated three times but I have to wait for two out of three notes to acknowledge that they they flushed the data then I send back the uh the acknowledgement
28:58 right the idea is that again if you assume that failures are rare what's the likelihood that this thing's going to you know one of these guys is going to crash you know it's pretty low or or maybe they're both immediately going to crash at the same time potentially pretty
29:13 low so rather than paying this this penalty to wait I'll immediately get back my response okay we we'll see how this fits into the cap theorem uh in a second so most systems the no SQL guys are going to be down here doing some kind of version of eventual consistency
29:32 the sort of classic relational distributive relational Data Systems like a db2 an oracle uh non-stop SQL those guys at the top they're going to be doing synchronous commit because you they don't want to lose any of your data all right the next question is when do we actually propagate the changes from the primary to the replica and again it doesn't matter whether you're doing shared dis or or share nothing doesn't matter where where they're doing uh multi-home versus uh or multi- primary versus a single
30:06 primary if you do continuous propagation that means as the primary is getting or one of the nodes is getting all the changes from a from a transaction it's immediately sending out the redhead log messages the log messages to the replicas as soon as those those queries get get executed right think of like it's like again red head log I'm putting my red my my my log entries into my buffer that eventually get flushed to disk but as soon as I generate a log record I'll also immediately send that out over the network I can still batch it up maybe with other other messages but I'm not waiting to to acknowledge or not waiting for a commit message I'm just sending those things over the
30:45 wire the downside of this approach is that I have to also then send a commit message uh which is fine but then if I send in a boort message then my replicas may have been apply much of changes only to later out find out that the transaction didn't commit successfully and I they have to roll back the changes if you do if you do the propagation on Commit then basically you stage all the log records that you have
31:12 on the on the primary and only when you know the transaction is going to successfully commit then you ship that over to to the to the replica and now this guarantees that it only sees uh transactions that has committed and you don't have to was waste time roll ping things
31:29 back or boards are typically rare in in most systems I think how often again you browsing the web or doing you know using your bank website or something how often do you abort things you know not that often so the top one is always going to be better because you don't have to wait to know whether transaction is committed because you assume it's going to commit and you send your log messages as soon as you get
31:55 them okay all right this last one could beit confusing so bear with me and please ask questions if you're confused so it been kind of handwavy in PowerPoint showing arrows from from one node to the another without really saying what what are we actually sending along right but for the most part you you've can assume that we've been sending right ahead log messages so that means that when we run a transaction the queries are running on One Note code they're making changes to the database and then we're sending the right hand log messages them to get applied to the
32:35 replicas this is known as active passive scheme and the idea is that the transaction is actively executing on one on one node and then the other replicas are passively applying the changes that the transaction made on the Node where it was actively
32:56 running so but doesn't have to to be this way I could actually run the transaction in multiple locations let them run independently make whatever changes that they want I don't have to send any right toal log messages between them right the only thing I need to do is at the very end when the transaction commits does everyone says did you commit yeah I committed
33:19 done right so this is called active active because the transaction is actively running in two different locations at the the same time and depending on what queries they're doing they can run independently without coordinating between each other as they're running so again the only message they need to send at the end is did did you complete successfully yes I did
33:41 too so this only works if you can guarantee that the transactions are deterministic all right so how do you do that well if I have a random call in a seal query then I just make sure I I they had the same seed on both nodes that covers that
33:58 what if they call now I try to get a time stamp with current time okay well I just make sure that they you set the trans the transaction time stamp to be the same when they both start all right so now how to keep the clocks coordinated that's kind of hard but so maybe that the transaction shows up at a centralized system you assign it you know a random seed you assign it a time stamp when it starts and then you send it to the duud notes to execute
34:23 simultaneously yes so you save on communication because you don't have to like uh do it only in one and pass the data but then don't you at the end still have to make sure that all know have the same data so like what are you really saving here
34:35 yes the statement is and he's correct is that since I still have the at the end when a transaction commits since I have to just make sure that um I still have to make sure at the end that everybody's committed successfully so there's still a network round trip for that uh what am I actually saving it's avoiding it's the intermediate data I'm sending along which could be lot right I could update a billion records in my transaction and I got to send a billion rad log messages over the network to to propagate those
35:05 changes uh I avoid all that yesac ends up with the same result at each rep do we mean doing some work up front in the design of the database such that any transaction will be deterministic check at the end of each transaction and actually send something between the right so he's he's for the second question statement here I'm saying need to check at the end whether a transaction end with the same result at each replica is this something I'm doing at the application Level to make sure my transac transactions are all deterministic or or should the database system provide this guarantee for me
35:48 it's the latter so you would have to have a system that can guarantee that you had the same ordering serial ordering if you will of of of trans transactions on both same nodes and that the the anytime there is non-determinism in the queries like a random call like a timestamp that you guarantee that both both all the copies of where that transaction is running we'll see the exact same
36:18 thing almost very very few systems actually can do this uh the only one least I can think of is volb which is based on system called h which I help build uh there might be a a few others
36:33 um but you typically would see this in a system where you execute transactions that store procedures which we haven't talked about but think of a store procedure as like an RPC call you can make to the database system that has all your application logic things you can't normally have in SQL like if Clauses for Loops exceptions right you could have procedural logic imparative code that you run in the DAT system as a function call inter miixed with a bunch of SQL queries and so again now if I instead of sending these back and forth between the queries because the timing can get off between the client and the server I just say run this transaction for me it's a
37:08 looks up the catalog gets this function just runs beginning to end again avoiding any non-determinism in those functions you're guaranteed to end up at the same state yes how is act passive different from primary replica as show on the slides because from what I understand active is mainly like the primary that activ execute and replica just receiving the updates from rad off question is
37:34 question is how I'm saying this this is not the same as primary replica versus multi- primary I just mean this SEC like you could have um you could still have a uh you still could do an active active scheme with with uh primary replica but instead of me shipping the right ahead log to you I'm just telling you here's the transaction I'm about to ex you execute it
38:00 too and as long as you come back and say yep I did it and got the same same result you did and result is the state of the database that we care about and the reason why volb can do this is because they run all transactions in serializable ordering they're actually the execution engine is actually single threaded so I can have a single threaded engine per per CPU core
38:22 but now all I need to do is order the order I specify the order of the transactions when when they arrive and think like it's you know it's a bunch of a batch of uh s procedure I need to execute as long as both nodes execute them in the same order I end up with the same
38:39 state so most systems are going to do active passive because it's going to build off the right head log infrastructure we talked about before that we already have in our system and it's just easier again to put the put the replicas into recovery mode or just have the the ability to replay a rad log message uh that's coming over the network instead of coming from a file yes so active active going always be like higher lows for for like each node compared to active passive so active active is always be higher lows what do I mean I mean like workload if you like you're basically sending doing like the same query um running same query over every Noe that you have to replica instead of like um say if I I have like two transactions uh two
39:23 requests I could have like um if it's Prim a multi primary I could have like comput one in one Noe and comput the other one in another Noe but if you do like active active they have to do two like in both yes so his statement is his statement is it it it may be the case that the amount of work you have to do an act active would be more than you'd have to do under active passive again it's a cop out but it depends right so I could have a query where it doesn't
39:51 update uh and sets some value you know AAL AAL a plus 1 but I don't have a wear Clause I have a billion tupal I got to scan if I if I do active active then every note has to do that billion record scan to find the record it needs to update and apply that one change but under active passive I just I do the scan on one node then I send the the log record for that one record I found I need to update that then gets
40:19 applied right but again assume you're doing that um synchronous uh commit or where I I can only send the right Ahad log message and I have to wait for get it applied I got to scan the the the Bon records then send the one log message to get applied and wait for that to get acknowledged and come back to me whereas if everybody's doing independently yes that may take a long time for that one particular query but when they're done
40:44 it's just like hey you did it y we're good we're done like it's it's not multile round trips back and forth um and then there me another case where the query is super simple uh but I do a billion updates and I got to send billion log records again there's you can come up with different examples for all these where one's better than another from an engineering perspective this is going to be better because this is just relying on uh the right headl infrastructure we already have and to be honest most applications can't be most applications don't run the store procedures and most applications will have non-determinism in them in them
41:17 once you start bringing in lower isolation levels too then the problem is if if you're doing active active if I can run a different isolation levels then depending what other transactions are running at the same time I'm running I may get different results on different nodes and it's super hard to do Nobody Does it because it's not worth
41:34 it all right cool I so I I think we get the the gist of this this is good again most almost every system is going to be doing active passive okay so now if only one node is has to decide whether transaction is allowed to commit
41:53 then making that decision is e even under you know this precludes primary replica setups right because under primary replica you have to you know the primary say I want to commit but you got to wait till everyone else to say yeah we we we can go ahead and Commit This Too did they apply the replica the replicas apply the changes so if you're running on a single box committing transaction is still hard because we have to do all the con stuff we talked about before but at least the node can reason about what's going on in its in its own purview in its own
42:21 system things are much much much much much harder I can't emphasize this enough when we have to bring in multiple notes to decide whether transaction allowed to commit because first of all how do we get everyone to agree that transaction allowed to commit and get notified about that agreement and then what happens if now we try to go commit the transaction get everyone to agree
42:41 and then one of the nodes goes down the primary goes down and then now you need to to elect a new primary right so how do you how do you handle that so the main takeway from this is the same decision it takes to say whether a transaction is allowed to commit is actually going to be the same decision about who's allowed to be the primary it's just another
43:02 transaction with slight slightly different semantics so this is what is called an atomic commit protocol in the in the database world you call this Atomic commit protocol in the distrib systems world you call this a cons uh a consensus protocol or Atomic consensus protocol but a high level they're basically doing the same thing and the idea is that we want to be a to decide the order in which transactions are allowed to commit across our different
43:30 nodes so again if you come if you take an attributive systems class they they talk about uh these state state machines like in paxos it's basically the same thing because the state of of the database is the is defined by the order in which transactions commit including transactions that are not allowed to
43:47 commit in the state machine replica I think I don't know what example they Sorry in distributive system I don't know what example they always always use like you like a simple counter like who's allowed to increment the count or whatever it's the same thing the commit order of our transactions is is is determined by the state of this uh the distributed
44:06 system and again we're going to have to do this no matter whether our database is is partitioned or replicated or not we want to everyone to agree that a transaction is allowed to commit before we acknowledge to the outside world that it has successfully
44:18 committed so there's a there's a there's a handful of protocols that can do this um we're going to focus on these two here two base commit and paxos um because these are the most common ones in distribut databases although raft is probably the probably more common than paxos at this point but P it's just a it's a simplified version of paxos and so I'll I'll discuss PA paxos in a simplified way that basically is the same thing as raft right there's a there's a minor difference so two phase commit uh was first written about by uh or was always sort of assigned or attributed to Jim Gray the guy that met to locking and I'll we talk about before
44:58 uh cuz he has a paper in the 1970s that mentions it um but then he Jim Gray actually refers to some Italian guy that designed the first two- phas commit protocol for some the Italian version of of retirement system and social security system in the 1970s right but that Italian guy says he didn't vent it either it's basically this the same thing we use in the old days of of contract law right so this is actually taking something we do in the real world and applying it to a distributed database system
45:28 so two-phase commit came first Stonebreaker came along and did three-phase commit because they're trying to overcome this property of if if a no goes down you have to pause the system so three three-phase commit was an attempt to try to try to deal with that problem but it didn't nobody does this because it's complicated and doesn't handle the corner cases there's a four-phase commit from Microsoft they do that because they're running on Harbor accelerator me or um
45:50 using RDMA with like confab band stuff we could ignore that view stamp replication came out in 1988 uh again this was done in the distributed systems World um and then the the one everybody knows about is paxos so view St replication by Barbara lisov she won the touring award a few years ago and then Leslie Lampard did paxos but he didn't know about view stamp replication so I don't think he he cites her um Zab is from zookeeper it's the Zookeeper Atomic
46:17 broadcast protocol or BR yeah broadcast protocol only zookeeper uses it but it does exist and then Raph came out of Stanford it's supposed to be a simplified version of of of paxos right so for view stamp replication I think that was the one that was first provely was proved to be correct uh under all possible failures um and then the only system that that I know that is view stamp replication is a thing called tiger beetle a distributed distributed lb system out out of um South Africa but again most systems are going to do use tax
46:52 W so at a high level all these homic Protocols are basically doing the same thing and there's this great paper we mentioned in see in a second from from Jim Gray and leg Lamport that basically assess the state state of the world for all these Atomic protocols and sort of try to distill down what they have in common and what's essentially what they're all trying to do so at a high level is the idea is that the these notion of these resource managers I've been using the term node uh but basically there's something that that is can control or has has uh can determine the fate of of data in in our database
47:27 system and so that now when a transaction uh wants to commit across these resource managers we have to go through these series of steps to determine is this thing allowed to commit yes or no right and so the idea is you the basic state machine of any protocol is that when we start off in the working phase we're applying changes to the database at any point if any of the resource managers any of the noes decide this thing is not gone the way I want to go I'm going to go ahead and and abort it then you immediately go into the abort phase as soon as one node or one research manager says this thing is not allowed to commit that's game over that thing has to go down the transaction has to fail then at some point the transactions work is done and you enter the prepared phase where you say you're trying to get everyone to agree is this thing allowed to commit or not if no you go to abort it if yes then you're successfully
48:16 committed so for any commit protocol we want for our distributed database system we want to have these two important properties first is that the the the protocol has to be stable meaning that once we decide these transactions allowed to commit and everyone agrees least at least well long as the majority agrees in paos but in two- phase commit everyone has to agree has to be uh unanimous once you decide this thing's allowed to commit you can't reverse that
48:43 change right that that's what the blockchain guys are trying to trying to avoid that problem where people rolling back chang are you changing their minds in our world we don't worry about that and then we need to guarantee that our our protocol is consistent meaning at some point at un some undetermined point that everyone's going to agree that this is the correct state of the database and again for our world the state is the order which transactions are
49:11 committed we're also want to assume that we we that our protocol will be will be provid some guarantee of liveness and that just means that no matter what we there's some way to keep the system moving forward the state machine moving forward and try to make some progress
49:28 doesn't mean it's going to be efficient but at least we're going to at least not be stuck in a state where we can't get out of and can't keep you know can't commit new transactions right and the the the guarantee to achieve this basically says at some point we need to be alive the nodes need to be alive and and
49:47 and uh communicating with other nodes to to make forward progress on this so the first protocol discuss this two- phase commit we'll see what the issues with this are and some optimizations we can do and then we'll see how paxos uh can handle them and so the main takeaway is going to be again two-phase commit came first Pax came later but two-phase commit is actually a subset of paxos or degener case of
50:16 paxos and again this paper from lese Lamport and Jim Gray basically proves that shows that they're actually the same thing but the the PX has a different different requirement for deciding when when their transaction is allowed to commit so we have application server we have a bunch of nodes right and without determining how we decide that this one is the node that's going to be the the the one we want to talk to to commit a transaction we're going to go to this node one first so under the
50:41 the the par Lance for two phase commit the node that's in charge of deciding whether transaction is allowed to commit is called the coordinator and then all the other nodes that are going to be involved in this this committing this transaction would be called the participant Pax also will use different terms
50:57 right but they're basically the same thing so now the first phase of two phase commit is to send a prepare message to all the nodes and that's basically where they do whatever checks that they they need to do to say okay we're about to commit this transaction is this allowed to happen yes or
51:13 no and then if they if they if each node looking at its own uh context about what the transaction did on its local node because can't doesn't know what it did on the other ones potentially it can then decide yes this this is allowed to happen I'm gonna go ahead and commit or I I send that acknowledge like okay that we want to we're okay with preparing this thing for committing now I'm not showing this in the PowerPoint slide but
51:35 for all these messages I'm sending out uh sorry all the messages I'm I'm receiving and all the messages I'm sending out those are going to be another rad log messages I'm going to maintain and then when I want to go ahead and commit I have to flush all my network messages that I for involved in two phase commit out the dis so that if I crash and come back I can know where I was in the two phase Comm protocol
51:57 when when I was running a transaction you could just put in the same rad log you put for your your regular queries so now once the coordinator gets all the acknowledgements from every node that's a participant because it has to know which ones it needs to talk to all right it may not be all the nodes in my distribut database but it's the nodes that that are involved in this transaction once they all agree I get all the okay messages then I enter the second phase where I said okay to out all the other nodes go ahead and commit this transaction same thing the each node goes ahead implies the changes if I care about synchronous commits then I'll I'll flush the changes and and and wait
52:35 wait till the flush finishes before I send back acknowledgement but I don't have to and then now they come back Al send okay and at this point here the node can then respond to the application and say now your transaction has successfully
52:52 committed right so again the key thing in commit is you have to wait the coordinator has to wait for everyone to come back and to agree to say this transaction is allowed to commit yes how how particip okay to
53:11 S this question is what the question is how would participant decide on its own whether a transaction is allowed to go ahead and set the commit it's just all the stuff we talk about when we running on S anyway so say you're doing uh say you're doing OCC right and you then enter the valid phase because you got the commit message I may if I'm doing forward validation I may uh my right set May intersect with a transaction that's already running and therefore I can't commit right if it's two phase locking if you hold the locks and you completed all the updates it's fine but it depends on when you check for integrity
53:43 constraint violations so in some systems you can say go ahead and apply all my changes but only check the indexes to see whether I have a duplicate key when I go ahead and commit then you would fail this you would fail that
53:57 so it depends but it wouldn't be because like oh because because you did something on node two and then node three did something different because you're not talking to node two you're just looking at your your own system your own
54:12 node right so the abort case is basically what he's asking about so node one the the the coordinator sends a prep message say for whatever reason node three decides this transaction cannot cannot commit no no no no can't we can't happen can't let this happen so they send back the abort message then no matter what node two thinks is allowed to happen as soon as we get one abort message the game's over this transaction has to abort so we immediately enter the second phase where we tell everyone to go ahead and abort
54:39 we don't have to wait to we don't have to wait for the acknowledgement for the second phase from all the participants before we tell the application your transaction has has aborted right as soon as we get one we know this transaction is never going to commit so we immediately can return back to to the to the client
54:59 okay so this is repeating what what I just said right each node is going to record the inbound and outbound messages into some kind of right ahead log or nonv storage um and then now when you crash come back depending depending on whether you're the coordinator or participant you can do different things again depending on how far you got into the two base commit protocol right so if I was a participant and I crashed after sending back the okay on the prepare message when I come back online I look at my log I say oh I sent out an okay for this prepare message let me go figure out what actually was the final outcome of this transaction so you would go back to the coordinator and say hey remember transaction 123 I said it was going to be okay to commit did it actually commit and depending on the uh on that what the coordinate comes back with you could either reverse the transaction which you would have all its changes because it's in your rad log or you could could uh uh you could apply them and say the transactions actually successively
56:02 committed if you never got through the prepared phase uh then you just abort the transaction like if I was getting much of updates and then my no goes down that's essentially the same thing as as boarding transaction so no matter what the transaction will board so if everyone was the was saying let's go ahead and commit this and you're you're the coordinator and you go down then when you when you when you come back online if you do come back online
56:26 then you would send a message to the uh to all the nodes and say yes we go ahead and committed this even though you may have not may have not told the outside world your transaction has committed and under under the the transaction protocol the transaction guarantees of a DAT system if the if if the if we successfully save all the changes to disk but don't send the the commit or acknowledgement back to the to the client before we crash the transaction still considered as committed
56:57 so now it's up for the application server to say go back and do a look up and say did the thing I just sent you actually get committed or not all right so I've said all this a lot of this already but like what happens if the coordinat crashes well the participants can decide amongst themselves what to do after some some am of timeout right so typically there'll be like a heartbeat uh messages you're standing around between the different nodes to say are you awake yes I am and at some point the heartbeat heartbeat messages don't come back So eventually you'll time out and just say okay well this thing is uh we have to decide whether you know did you tell this did you tell the coordinator we're going to go to commit yes or
57:35 no uh and and then if everyone says yes then you can decide again whether you want to commit the transaction but the key thing about this is that the system will not be available while you're trying to figure this out because you can't commit any other
57:47 transactions while you're trying to figure out whether the current transaction you're at is is allowed to commit or not right because what happens if that that other transaction read something from this first transaction but then the first transaction ends up being aborted but we've already told the second transaction is allowed to commit and that would violate asset
58:09 guarantees all right if a participant crashes then the coordinator just going to assume that it it sent an abort message right if you don't get that first prepare okay back from the participant you assume that it's going to fail no matter what so you go ahead and and tell it tell every the transaction has aborted
58:31 so there are two optimizations you can do for two base commit one is very common one is not so common the first one is called early prepared voting the basic idea is that if you know that you're you're you know if you know that when you send a query request to a to a node in the system that this is the last query you're ever going to execute for this transaction on this node and you're never going to come back to it with another query then you just piggy back in the message and say hey run this query for me give me back the result and oh by the way also send back me uh your your vote for the prepare phase so you get back the query result and you get back either an okay or an abort message
59:12 so then now when you run two phase commit you don't have to wait for that round trip again between the coordinator to the participant because you already know whether it's going to it's going to commit or not from that last
59:21 query most systems don't do this because you don't have a way in the in like you know sequel to to know where the quer is actually going to run the first place so how would you know that this is going to be the last query you made S I think jdbc I think you may be able to send a flag say this is the last query for this transaction um I know in other systems Vol Tob you can do this but most of the times you can't do
59:46 this what is more common is to do early acknowledgement after prepare where instead of waiting once I get all once you get past the prepare phase to get okays from all the participants instead of waiting for them to say okay did you commit yes or no
1:00:01 which at that point because once you pass the prepare phase that transaction is committing no matter whether the notes go down or not so instead of waiting for them to all you know acknowledge that yes we got you got we we went ahead and committed you immediately shove the the acknowledgement back to the client because you assume the window between the time when the client the the time the time the the flush and you s make the acknowledgement it will be
1:00:23 small and you have a recovery prot anyway to deal with if one of the notes crashes during during the commit process right you assume it's going to go ahead and commit so you can send the commit message back right away and this is very common so that basically I get the application server they sement a commit request node one sends the prepare phase messages to all the different notes or the participant sorry the coordinator sends the prepare message to all the participants they all say yep good yes we can go ahead and commit this and then immediately once I get back all the okays from the participants then you can shove back the commit response the the success message to the to the client but then you still have to run the the the rest of the two-c commit
1:01:04 protocol right so you're just cutting off one round trip from uh uh you know you're cutting off one round trip waiting for one round trip on the commit message you still have to do it but just you don't have to wait for
1:01:22 it yes what if the cres his question is what if we're here we get prepare okay from all the participants and then we send back success now we crash which node crashes the so say what happens the coordinator crashes okay well as I said before the participants will have a heartbeat message that it eventually will time out you're not hearing back from the coordinator you totally you want to commit this transaction you can't commit any other transactions
1:01:57 until you figure out whether this transaction is committed right so some point some point you time out so then you have you need a way to say okay well who else is in my cluster oh node three knows about node two so they start talking to each other and say Hey you remember transaction two3 did you say it was allowed to commit yeah I did did you did you say it was allowed to commit
1:02:17 yes and then and then then then you go ahead and and apply make record that they successfully committed or you time out the coordinator comes back online right set saw that you sent saw that you got the okay messages out saw you sent the success messag out and said okay let me just make sure I tell these guys they hav they haven't figured it out yet make sure they know the
1:02:40 commit this is hard right there's L of these Corner cases you got to deal with yes could you say we need message for question is why do I still need this phase here no respon like like sorry this like why do I need this okay from everyone yeah
1:03:03 um question is why do I have to wait if I if I if I send a commit message to everyone why do I have to wait for the response back and say yes I went ahead and committed so it's basically for for that you know that everyone has has successfully applied the change and therefore when I go ahead and now commit the next transaction I know that everyone's in sync
1:03:27 uh so let's in this case I send the success to the application and there is another request from the application but at the same time I doesn't receive all the from the participant does that mean I can't start the next transaction so you could start it you just can't commit it so so they have another commit message shows up right so you send back success you haven't gone through the commit phase yet another commit request shows up what do you do well you would know that you're still still trying to finish off the first transaction you can't start committing the second transaction until you get back for everyone to acknowledge that they successfully applied the
1:04:06 changes because you don't want the case of like if I don't do the second phase right these guys don't know where this thing is actually committed yet because they don't know that everyone voted okay right so they're waiting to find out whether the transaction they just ran is actually going to commit or not right so now I do this now I say okay okay guys we all agreed to commit guess what they've committed great so at this point here all the the the NO2 and NO3 they can actually begin the process of committing another
1:04:36 transaction but this guy still can't because he doesn't know that everyone has has acknowledged that they have successfully applied the changes from this transaction because I asked him I I told I asked him to commit I don't know whether that message actually showed up yet it's only when I get back to the acknowledgement then then I I know that they applied the changes I told them to apply
1:04:56 so now when I commit the next transaction I know that transaction if it read if it read anything from the first transaction it will see the correct version of of those of those records okay uh we have 13 minutes left let's go through the hardest protocol and distributed systems in that time right but paxos again it's it's uh again it's going to be a degenerative case two commit Pax so if you understand two phase commit again the I'm going to simplify it so you'll see that how it just Maps into the same ideas and all the things we care about under two commit we still care about in in pexels
1:05:39 right so again this is the per provely resilient uh uh protocol Atomic protocol in the face of um a any Sous networks so it's basically gon to be the same thing under two base commit where we have a coordinator telling the world I want to commit this transaction but under two base commit you had to get had to get everyone to agree that this thing was allowed to commit under paxos you just have to have a majority of the nodes
1:06:07 agreed that something's allowed to commit right okay now you see why Two Face commit is a is ative case the original P paxos paper actually who here who here has read it what do you think of it we sh said no it's some C it's crazy I'm glad we have simp there actually P made simpol is not not good either right
1:06:31 it's not all right go read this paper just go spend five minutes looking at it it's crazy it's the best paper ever but in terms of like it's it's Pros it's not a traditional computer science paper it's written as if he's an archaeologist
1:06:44 finding this uh this ancient voting system from this tribe on the island of paxos in ancient Greece right has nothing to do with computers this is how they vote by throwing tablets into a hole and you come back and read later on what was thrown in the hole
1:06:58 right uh it's it's brilliant it's amazing but like if you try to take this paper and say Let Me Go implement this not going to happen right so the The Story Goes he wrote it in '89 submitted it to distributed systems conference and it got rejected right and if you go to Leslie lamport's website for like he has like this this chronological history of all the papers like what he was doing where he was living who he was dating what he was eating like for every single paper so he talks about this paxos paper
1:07:26 the the part-time parment about how it was a genius paper it was it was a work of art and you know those idiots on the reviewing committee they didn't know they didn't know how to appreciate it right so when I was in grad school I was assigned to read to present this paper in my distribut systems class and I basically said the same Lamport said like this paper's you know it's a genius of work of art and those reviewers they were stupid they didn't know they were doing the professor was actually was more Le hurle who used to be here at CMU
1:07:51 and now he's at Brown he was actually one of the reviewers of this paper and he's said that uh The Story Goes is like they were originally okay with this Greek archaeology metaphor all they asked him was put in the appendix a proof to show that like how you know what this thing was actually doing of the algorithm and Leslie Lamport basically said this is a work of art I'm not diing any change so he he got rejected so then he puts it The Story Goes he puts it in his in his in his uh filing cabinet forgot about it saw all these papers in the 90s where people were sort of dancing around the corners and kind of getting close what paxos had already implemented but wasn't solving the complete problem so then he pulled it out dusted it off uh then resubmit it and then this little there's a progue here that talks about how like it was uh this mission was recently discovered behind a filing cabinet of of the editorial office despite its age editor Chief felt that it was worth publishing so like they dug it out and and and uh
1:08:45 and and got it accepted or got it published then he pointed out there's a follow paper from Leslie lamber called paxos Made Simple that doesn't help either the best paper to read and understand in paxos is the uh Pax made live from Google that one describes how that's my opinion that's the best description of it at least in the modern papers um that that came out 2004 2003 right so again and then there's this other paper from leely Lamport and Jim Gray uh this came out in what is this
1:09:13 2006 right before Jim Gray disappeared at C does everyone know that story Jim Gray won the 21 datas in 1990s he was sanding his boat in the San Francisco Bay in like 2006 to dump his mother's ashes and he went to a part you normally don't sail to by yourself and then his boat disappeared all right and they never found it they didn't find any any sign of them um he got declared dead I think a few years later um but like
1:09:37 give you an idea how important Jim Gray was and how much he meant to a lot of people this is early days of like crowdsourcing they actually moved satellites in space to go over San Francisco Bay and take pictures of it and it was an early crowdsourcing effort where people like look at pictures of the ocean or of the bay and try to find his boat right and and they never found it so any this is one of the last papers that Jim grave wrote with uh before before he
1:10:00 disappeared um it's kind of heavy way to start this but let's go to it all right so paes what is it okay you have a commit request from the application server inste in calling these uh part coordinated participants paxos is can call these proposer and acceptors there's a third node type in paxos called Learners we can ignore that basically someone who doesn't participate in voting but can see the changes and update things think of like a a a non-voting
1:10:26 replica right so we send this proposed message out to everyone right say for whatever reason acceptor node 3 dies during this process just like two F commit they all agree that this thing wants to go ahead and commit and long as we get a majority back not all of them just majority from all our acceptors then the transaction is considered committed then we do the second phase and tell them to commit right and then they they they all send back
1:10:53 acknowledgements at some later point if the acceptor wakes up uh then they they can learn from the other nodes what what what happened with this transaction to get up to speed right so in raap they're going to call these uh I think they call it still propos an acceptors uh I think the Learners they call followers it's again they they're all basically doing the same thing but another way to think about this to guarantee the liveness property we talked about before we can always still make forward progress under
1:11:24 uh under two-phase commit if if we crash if a no goes down in the middle of a commit process they got to figure out you know you got to figure out what was the state of that of that transaction
1:11:36 and then uh then decide what to do again the system is unavailable during this time because you you trying to figure out whether transactions allowed to commit or not in paxis you don't have that problem because proposers can keep proposing new stuff and you you still you know there still things are moving forward so say along this timeline here we have a proposer over here he's got three acceptors and there's another proposer on the other side so he says all right I want to commit this transaction so let me goad propose that this is commit n right how they came out with this time stamp or this number it doesn't
1:12:05 matter all the exceptions say great fantastic yes we know about this transaction we agree to go ahead and commit transaction at at at uh in order n but then now say this other proposer comes along and is unaware that there's a this is the first proposer and they propos to commit n plus one soon as you see a new proposal to commit a transaction any previous uncommitted transactions have to abort because you always have to be accept new proposals and always keep marching forward at time
1:12:33 right in two- phase commit you have to wait until the the the first transaction commits before that you can propose to commit another transaction so now the all the acceptor is c n plus1 they haven't acknowledged that they're going to commit it soon as they see it that's enough to say that anything came before that is it's aborted so now when the first proposer comes back and says I want to go go ahead and commit this transaction you just got told me it was a good idea or not then they come back and say I want to reject this and tell it you know you you're telling us to commit transaction n we just saw n plus1 you're no good to us come back again right so then they all agree to commit n plus one they commit uh the the first propos says to go ahead and commit it they send back the acknowledgement and we're good to go and then this guy can then propose to commit the transaction again or abort it
1:13:20 entirely so obviously this would be very inefficient if you have a bunch of nodes in your system all proposing and commit transactions just claing each other like it'll be a the system will be live be you know you can accept new things but you never going to commit anything because you know contraction are going to keep getting aborted and over again and so this is what Google put together in this paper I mentioned about uh paxos made live um or and they put idea called
1:13:46 multipos and basically the idea is that you only have one node get anointed as the leader in a sort of paxos group and then they're the only ones that's allowed to commit transactions and if you have this you can just skip the proposed phase because you know no other node is going to come along and try to propose to commit a new
1:14:09 transaction if now though if one of the nodes goes down you fall back to uh to the regular PES protocol so to handle the case where the leader may go down and the system obviously would grind to a halt you set a lease or a timeout for the for how long they can be the leader and then when that timeout uh uh gets gets finishes off you run a leader election which is just another round of paos to decide who's allowed to be the new
1:14:39 leader and then now you basically introduce like you know incremental back off where if I try to say I want to be the new leader and I can't get it then I'll sleep for a random amount of time and then maybe somebody else will be able to sneak in and get it before I do again the nodes aren't malicious if I can't be the leader it's not the end of the world for me right if you're the leader that's fantastic we we can commit transactions together it's beautiful
1:15:01 right so how to set the the timeout depends on the system right the the longer it is the less time you have to spend doing these leader elections but now the the longer you may have to wait uh to time out on something because the leader has failed again the heartbeats can maybe pick that up earlier uh but it could depend on you know how often you're sending those so in Google spanner these leases are 10s I think they say in the paper yugabyte is 2 seconds GB I think the default is 5
1:15:31 minutes tidb is 10 seconds because they just follow Google right it depends on you know depends on the environment you're running in and the expectations of the nodes all right so quickly two-phase commit versus paxus versus raft again two-phase commit is going to be all transactions at the Block if the coordinator fails after the prare message is sent until the coordinator comes back or then you can then figure out uh uh you know amongst the NOS themselves who's allowed to commit paxis is non-blocking long as a majority of the participants are alive
1:16:05 and provided that you can keep going without continued failures raft is the same thing with this paxos but there's fewer no types there's basically just participant and and acceptor which I think they call follower and then when they do a leader election rather than letting anybody be the leader you have to have this requirement the basically says only the the the nodes that have the most upto-date log is allowed to become the leader so if someone is falling behind you don't want them to become the leader because obviously they're slow and going make things worse for you so if you're you have the freshest data or the freshest commit record of all the transactions
1:16:40 then you're allowed to go ahead and commit all right let me Blaze through cap theorem uh again uh I think this should be again if you take a attribut class this should be pretty pretty be aware aware of this already but we we can we can um touch this up next class
1:17:00 for refresh people all right so cap theum this proposed in the late 1990s by Eric Brewer at Google or he was at um uh some web startup basically they recognize that oh if you have a distributed system well again that's what a distributed database is that it's impossible to have three properties you can't guaranteed to be always consistent always available and still handle anytime there's a network partition
1:17:25 so the real way to think about the cap theorem is basically are you going to be a consistent system or an available system anytime there's a network failure right it doesn't make sense to be a consistent available system and when there's network network partitions or network failures communication failures because that that doesn't really make sense it really has to do with what do you do when there's a failure so consistency just means what we talked about before right the idea is that if I make a change on one node say I've copy uh of of uh my database I apply the change here I should then
1:17:58 apply the change also my replica once I do that then I could tell the outside world this transaction sucessfully committed now if anybody comes along and tries to read that value on the on another node they'll see they'll see the correct
1:18:13 version right availability just means that if a node goes down again assuming it's a simple primary replica then I can have any transaction run on uh on this node get back any changes and the other node on the other side of the network assuming we still can communicate can go ahead and R dat over here right that's fine so the datavis are still available even though some of them are gone
1:18:36 down the hard is Network partitions right say my network gets severed now the two nodes can't communicate with each other so what's going to happen so I again seeing I'm primary replica this primary can't talk talk to replica anymore it says okay no big deal I'm the primary this replica says I can't talk to the primary anymore well guess what I'm now the primary right I'm only showing two nodes you would do basically you know leader election between all the replicas on maybe on each side but this guy says yep I'm the primary too because he doesn't know that this guy's still alive because he can't talk to
1:19:11 them so now the problem's going to be trans you know both these guys think they're the primary you don't know about the other one because you can't talk to them because the Network's down so the first transaction modifies a the second transaction over there modifies a they both get acknowledgement say that your your transaction has
1:19:29 committed now the network comes back up and you got to reconcile these changes they both thought they were the primary they both thought these changes were allowed to happen but now you have conflicting values so this issue here when you have this this this the network fair you can't communicate this is called split brain right both sides of your brain can't communicate each other but they they both can operate
1:19:53 independently so again the question is what do you do when this happens if you care about consistency then you basically would halt the system You' halt the system where you don't have a the majority of the nodes right and then this means that when the network comes back up back up now the the the the node that the the portion of the network that had the majority of the nodes it maybe got a bunch of updates you you now propagate those updates to the replicas but at no point do do you deal with conflicting
1:20:22 values so again most relational distributed distributed relational Data Systems will will choose the first one the second one is you allow the split to happen you allow the rights to occur on both sides or sides you have then when the network comes back you got to do a little magic to figure out what should be the correct version of those values that got
1:20:42 modified so you can either let the server do this or they let the the client do this if the server does this then it's just last update wins like whatever had the the oldest time the newest time stamp that's the latest version good enough enough right so Cassandra does what most systems do most of the no signals do the other approach to do what's called Vector clocks this is actually invented by Leslie Lamport it's basically multiversion as you did before you keep track of here's the the version of a of a value or version of an object at this Tim stamp and it's like you have a list so now when I go read a value go get a instead of getting one value I get back a list of values with different time stamps and now my application code I got to figure out which one should I care about which one is the correct one and maybe go ahead and apply the changes
1:21:25 I only know one system actually did this it's called Rea uh and it was a nightmare because again your your average JavaScript programmer isn't know what the hell a vector clock is isn't going to know how to reason about if I update two things and now they both have different time stamps because one got modified one did get Modified by another side which is the correct version you're basically repeating the logic to figure out what is the correct version of multiversion cury control in
1:21:51 JavaScript that's not going to go well don't do this okay so I'll just say there's an extension to paxos or sorry cap theorem called pesic that came out uh in 2010 it's basically if there's a failure again do the same thing are you partition tolerant with predition are you always available or consistent but on a normal operations you got to make this tra off between latency and consistency and basically says if I have a primary I do my update in one data center I got a proper change to the other Data Center the question is how long should I wait for the to acknowledge it because I can wait an infinite amount of time and I'll guarantee to be
1:22:32 consistent right eventually I'll I'll get responses back but then now the system is not available right or maybe I only wait for one of the updates come back and uh or you know and then I can send back to acknowledgement so there's this trade up between how long do I want to wait versus uh how how quickly I can respond to things all right so hopefully again you recognize how hard this is and I sorry for going over uh Google is one of the first systems that said hey we don't want to do transactions because that you know we we we want to be web scale they you know they did this in big table and a bunch of systems copied what Google did but while everyone was recreating what Google did in the open source world
1:23:09 Google said wait a minute this looks like transactions are a good idea they hadn't published it yet but they've been working on a system called spanner for a long time so everyone else was trying to writing the the nosql wagon or nosql wave Google was actually building a real system spanner they didn't support SQL first that came later but they had this nice little uh snippet in here and it's basically say what I said before we believe it's better to have applications
1:23:32 programmers AKA JavaScript programmers deal with performance problems due to overuse and transactions rather than always coding around the lack of transactions mean deal with Vector clocks and inconsistent data so it basically says you you have some really smart people like you guys spend time making transactions run fast on your distributed database and let everyone else deal with that and not and they're make them better programmers
1:23:57 okay all right sorry for going over have a good break Monday next class distributed ol systems see I need something refreshing when I get fin manifesting a like SMI and WT and my thoughts hip hop related write a rhyme in my pants intoxicated lyrics are quicker with a simple Mo lior since I'm City Slicker play waves are picker rhes I create rotate at a way too quick to duplicate feel a breeze at escap Mike's at Fahrenheit when I hold it real tight
1:24:28 then I'm in flight then we ignite but starts to boil I heat up the party for you let my girl R me my mic down with oil Rec still turn with third de burn for one man I heat up your brain give it a suntan so just cool let the temperature rise to cool it off with s eyes
1:24:49 [Music]