CMU Database Group · 1:02:49 · https://www.youtube.com/watch?v=Vy2t_wZx4Is
1. <Untitled Chapter 1>
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
0:24 pistol is all right so I screwed up the recording for Wednesday's class uh the audio didn't get captured um at least for the court part of the lecture so it's I had to call my PE officer let me come here on a Saturday in the lab uh to record it so this will be re-recording up what was discussed on Wednesday so before we get into the material uh again everyone should be painfully aware of all these things uh homework uh sorry project two is due tomorrow on Sunday um and then today Saturday we're having the special office hours but again we've already announc this on PIAA and then homework for came out this week and that'll be due the following week on November 3rd and then we'll be releasing
1:06 project 3 uh probably this Sunday as well all right so last class we were talking about how we're actually going to execute queries it was basically discussion of how data moved between the operators in our query plan remember we talked about uh do you send a single Tuple do you send uh all the tupal that operator is going to admit or you send a batch of them and then we also discuss the the the way sort of the the control flow of the of the query plan works do you start at the top and go down and pull data up or you start at the bottom and and push data up so today we're going talk a little bit further about how we actually now run these these queries uh in our system now actually in in parallel are using using multiple
1:47 workers and so the all the discussions we had pretty much so far this en entire class except when we talk about concurrency tool and indexes has really assumed that the datab Sy has a single worker single thread or process that's running queries and it's not getting interfered uh it's not interfering with other queries running at the same time so now we got to worry about okay how do we actually uh how do we build a system to allow it to run multiple queries at the same time and then furthermore how do we allow it to take a single query
2:13 and run that in parallel across multiple workers throughout today's lecture I I'm going to try to use the word uh worker uh and not thread or process um because you see worker is a sort of higher level computational component of the system which could be a process could be a thread could be a com of them but I'll try to use worker but when I say thread or process to think think worker unless I'm explicit about it all right so the reason why you want to run queries in parallel is is kind of obvious right in a modern modern systems modern uh Hardware uh you have a lot of parallel
2:47 resources um and obvious things could be you have a lot of cores on a single CPU that pretty much everyone's laptop has at least four cores at this point um you can have multiple multiple CPU sockets you have multiple entire processors running together on the same box um but you also could have multiple uh storage devices non volatile storage devices and we'll see how to run those and handle those in a parallel system as well so why you want to do this is pretty obvious so some cases the the database system or the the database itself can't fit on on a single node and therefore you have to scale out horizontally or scale out vertically by adding more more
3:22 resources and and some cases we also get higher performance meaning our query is going to run faster we'll be to run more queries at the same time and then we won't talk about this so much today but when we talk about distributed systems we'll see how we can get better fault tolerance and redundancy by duplicating or replicating the database across multiple machines so now if any any one not goes down we can rely on another one to to pick up the work behind us the key thing that understand about this discussion is that even though underneath the covers we may be making copies of the data or taking a query and and running it in parallel across multiple uh multiple computational units
3:57 workers when an application invokes a SQL query that same SQL query should produce the same result whether or not it's running on a single you know single laptop with one CPU core or some giant uh distributed system with you know thousands of cores again this is the separation between the logical and physical storage of a representation of of data right you just clear your secret query this is the query this is the result I want and then the datab system is then decides how it wants to
4:24 distribute it and and execute it across you know one or more resources so again if it runs on a single node in theory it should run also across multiple nodes across multiple workers not always the case but uh for today's class we don't have to worry about
4:37 that so in this class also too this par this lecture today uh we're explicitly talking about what I would uh what I would call parallel databases and there is a distinction at least in my mind between what a parallel database is and a distributed database conceptually that the same thing you're using more resources ex your queries but the the key thing I want to differentiate is in a in a parallel data system the there's the resources whether it's storage or compute are going to be physically close to each other like if they think of like on a on a single box a single single a single laptop you have one CPU and you have multiple cores and therefore the cores are running basically right next to each other on the same cicon but they could also be potentially in the same same rack um right but but the when you when one thread wants to communicate a worker wants to communicate with another worker uh we're going to assume that that connection or that that communication is going to be fast
5:30 and reliable and potentially cheap uh meaning it's sort of you're not paying a high cost in terms of money or you know wait time latencies to make that do that communication so again think of like two threads running on the same CPU I can use you know they're running in the same process then they can easily just write into memory and the other other thread can read it right and that that's really fast if you if that communication starts going arai meaning like you send a message and the other thread doesn't get it then you're having like severe harb problems and the whole machine's crash anyway right now there's ordering issues we got to worry about and we we'll see how we handle that uh after or starting next week
6:08 but uh for now we just assume like we write a message the other guy going to get right away in a distributed database the the assumption is that the the the nodes the resources are be far from each other right either not in the same rack but least in the same data center in the same state the region of the country or in some cases worst case scenario on the other side of of the planet actually worst case scenario would be out in space where the latencies are are massive and so in this in this environment we we can assume that the communication cost is going to be fast
6:39 and we can't assume it's going to be reliable so that means in our algorithms we're have to uh accommodate uh message loss or node failure and so forth in in our protocols because we might send a message to a to a resource to get some data or do some computation and the other side never gets it so we have to account for that in our our implementation so again we're not going to worry about that so much in in this lecture here but we'll see how we have to handle this um a little bit when we talk about coner control but definitely when we talk about distributed databases
7:07 uh later in the semester at the end of the semester so for today's lecture I'm going talk s talk about process models uh is the way the we're going to organize the workers in our system then we'll talk about how the different variations of of parallel execution we have for queries then we'll talk about how to do uh variations of IO parallelism and then as I said we have a click house talk that I'll just cut over to from from class uh because it was actually really really good and covers a lot basically covers all the things we're talking about this lecture uh he the the Robert from click house basically says the exact same thing it talks about how click house does those things later on so we'll we'll cut rid of that at the end of
7:42 this all right so the dstm process model is is the way it's going to organize the the the the the the computation resources of the system to execute queries or requests in in in parallel tasks in parallel and as I said the the the the component we're going to use to schedule things or actually execute things uh and we schedule things for them is going to be what we'll call a worker and this is going to be some uh you know some computational piece of the system that can execute tasks have the client and also uh internal background jobs or internal maintenance jobs so we'll see later on when we talk about garbage collection and multiversion Cent control there's a background process you may have to run something like postgress that has to per periodically go through and and
8:29 uh do garbage collection or when we talked about large structure merg trees uh we had to do compaction in the background you you would assign that task to a worker but that the worker could have been used for executing queries as
8:41 well so there's G be there's basically two approaches to how to organize workers in a in in a d system and we'll go through the process per datas worker and the thread per worker and then there's a third category that doesn't actually line up exactly with this dichotomy between processes and threads called embedded systems um um but I think it's worth discussing because it's uh it is an important aspect of database systems how to how to build these things and we'll talk about what that looks like for the you know in the organization of the system so I would say the the most common approach of being used today if you're building a system is going to be the second one the thread per D worker um and this is because in the modern era writing threads uh is not a uh it's not as owners as it used to be back in like the 80s and 90s right so we'll talk about process for workers and see why people had to not use threads back in the day in the old days and then why they switched to threads later
9:34 on so the oldest uh process model is is a process per worker if you want to support pel execution and this is where every worker is its own entire OS process like you're calling Fork to get a separate process that has its own address space uh and is scheduled separately from all the other worker processes in in the system so the basic set would be something like this where you would have your application it would send a connection request to a dispatcher
10:00 process running on you know wherever the data system is actually running um and this thing dispatcher would actually be part of the data system so in postgress this is called the postmaster and then the dispatcher says okay well you want uh I can't actually the dispatcher can't actually execute queries so it's going to pick one of its worker processes to assign to this connection or or Fork a new process if one doesn't exist and then it tells the application hey here's the port number of the of the the worker you want to talk to and then the application then makes a second connection request to connect to the worker and then that's where it sends SQL commands for that worker that then execute the query and then return back results as
10:37 needed um so you could have the dispatcher just be like a like a coordinator uh where queries show up to the dispatcher and the disater just hands them off to other other work processes or you can do it like this where you have a sort of uh the dispatcher is really assigning uh workers to to to the to the application
10:57 so one nice advantage of this approach although you know does have additional overhead actually use this because now if you want to communicate with one worker wants to communicate with other workers you have to use some kind of IPC mechanism or in the case of postgress they use shared memory as as a way to communicate things right so there's a bit more work you have to do and of course now you're also relying on the OS to provide a bunch of the stuff for you right it's in charge of scheduling and so forth you can play around with uh priorities and the schedule but you really can't do more complex things that you may want to do if you have complete control of the of all the workers so the reason why old systems like postgress like postgress is old postgress was first started the first implementation
11:32 version of postgress was 1984 um but Oracle does this db2 does this the reason why these older systems use a process per per worker is because back in the 80s and early 90s uh the threading packages for the different versions of Unix out there were inconsistent right now we have P threads we have posix and all that but back in the day uh you know the threading package for HP would be different than the threading package for a and so it would be really you had to basically have all the reimplement all the the threading infrastructure if you want your datab support a bunch of different
12:05 uh architectures and systems and operating systems uh whereas like if you had posx fork and join were were the basic building blocks are a lot of things and all operating systems had those so you could just do do Fork processes and everything would just work across different systems without making major
12:24 changes the the modern implementation the modern approaches use uh a thread worker um and now you get your spawning P threads or whatever operating system you're running on and you're having your your system coordinate uh and schedule across those things so in this case the data system will actually be able to manage his own scheduling now uh because it has complete control of what thread's going going to run and when um the everything's all in the same address space so communication costs are pretty cheap between the different workers um now unlike in the process Pro worker model if a thread crashes uh
12:59 model if a process crashes it it's isolated to just that process if a thread crashes in this world in in this this model then that takes down the whole system so just like before the application will connect to the dispatcher uh and then the dispatcher canci say okay this is the thread you're going to run on and again we could either have the dispatcher be responsible for forwarding query requests to the the worker thread to execute or we just give it back say here's the port number you want to connect to uh and send your request to this all right so pretty much every data system implemented in the last 20 20 years uh implements supports this even systems like db2 and Oracle have gone back and added support for the thread thread uh thread per process thread per worker model even though they origin started off with process per worker like I think in db2 you can switch to say you want to use threads or processes and depending on what environment you on uh you choose one versus another um the only systems that really don't do this is that if you do if they're forked from
13:55 postgress uh you know they're based on postgress they just inherit that process per worker model that proo has so um again and then the other category would be the embedded database systems in uh so embedded systems don't really have their own uh typically don't have aren't in charge of their own threads uh and because they're embedded inside of a another application right so the idea is that the application makes calls into this uh embedded databas system library and then whatever thread made that call into it is then used by the datab system library to then evoke whatever the query they needs to do duck DB can uh you can tell it hey spin up additional threads that it can use to ex queries in parallel um but the and sort of the classic embedded system model especially like
14:50 sqlite uh and Rox CB it's whatever thread that's being entered into request to an operation that's the thing that's it's going to execute it and there so there isn't a dispatcher right so in the case of the application code um the application has has its own threads and then it can make calls into dat system to do whatever needs so as far as I know that one of the first embedded datab systems uh that supported this model uh was berly DB um that came out of obviously uhal University of California Berkeley in their early early 1990s um late 1980s that was bought by Oracle in in 2016 when a query shows up the day is is has decide how and where it wants to execute it uh and when should execute it
15:34 so most dat systems are going to do a really simple uh first in first out uh priority uh cue so whatever query shows up that gets the first execution gets to execute first um the but it doesn't have to be that right there's other policies and things you can actually implement but the DAT is also decide okay for this given query on this I know what data is going to access and therefore I can decide how many tasks should I break up the query into how many how many cores or workers should I use to exq things uh
15:59 when when a worker execute and produces some some output where should it go should it go in the same you know the same worker should it go to another worker right all these things the system can schedule for you and this is again why we want to do everything ourselves even though the operating system has a has a has their own scheduler it doesn't know what the queries are actually want to do and what's in the pipeline to execute so because it just sees blind processes or threads so again the data syst is always going to be better positioned than the operating system to figure out what to do and then the higher end systems like the Enterprise systems like Oracle db2 and single server have again they rewrite basically everything and don't rely on the operating system to do
16:36 anything so the so the different process models are going to allow us to again take advantage of different uh multiple workers in our system um and potentially run things in in in parallel but the key thing to understand is that just because
16:52 a a system supports multiple workers in uh you know running inside of it it doesn't mean that they're always going to be able to do what we'll call intro query parallelism meaning like just because they can have M workers run if a query shows up it doesn't mean they can divide that query up across those multiple workers like myo is probably the most famous of this like myo is a uses the thread per worker model uh so it's it's a multi-threading uh database system but one query shows up only one thread can execute that query into its entirety right uh they can't split up across multiple cores and so we'll we'll see that in a second and as I said before pretty much every system that is based on postgress or Reedus is probably the other most famous one that's the single process single worker system
17:34 unless you're forked of these things uh you're going to be support uh you know you support multiple workers now that we defined our process models we want to talk about how we're going to architect the system to support uh parallel execution of multiple operator instances multiple operators either running in part of the same query or multiple queries at at the same time um and the the high level approaches that we're talking about here today aren't going to really don't really matter whether you're using multi-threading or multiprocess or multi noode and distributive system right because it's just about like how do I say here's what you're going to execute and then where do you put your data and then that communication cost between the workers
18:10 you the cost of that will depend on what environment and and what process model we're actually using so there's two high L approaches to do par query parallelism so the first be inquery parallelism and the then we'll talk more detail about inquery
18:25 parallelism so inquery parallelism is is what I saying before where you can have multiple queries uh from you know different connections or different application requests running simultaneously in the system um and as I said most systems are just going to do first come first serve so whatever query shows up first uh it'll get scheduled execute before any queries that come after it in some systems like db2 you can specify transaction or or queuing priorities based on like the username so like a a the CEO might have their queries run faster than you know somebody in
18:55 sales so if all the queries are read only they're not actually modifying the database then this is pretty easy to do because you don't need to coordinate between the different queries running at the same time because you know no one's going to update anything now you still have to use those internal protection mechanisms that we talked about in the buffer manager and the indexes as I said before because they may be trying to access the internal data structures at the same time but the The Logical database isn't going to get changed so we'll see how we handle that uh how to hand updating queries uh in in in two classes for now um but for now now it's it's uh we're now just assume that's not going to happen that makes our lives a lot easier and the challenge is again if
19:34 if they do update things then we have to worry about who's reading what and when but again we'll cover this later uh in in lecture 16 so for INT quism this is typically most people think about say I run I'm running queries in parallel right the idea is that we have a single query that we can break up into uh multiple tasks that we then assign to multiple workers to run run at the same time and we can cut down the the execution time query potentially so you sort of think of the same way that we talked about operators in our query plan when it was single threat execution of you know sending data one oper to the next it's sort of this high level producer consumer model
20:09 we're going to do the same thing now when we run things in parallel but now we're going to have explicit control mechanisms to say I know I have these tasks running in parallel and I can wait until they all produce the results I need before I move on to the the next pipeline next stage of my query
20:23 plan so there's going to be two uh approaches to do intro parallelism the ver will be horizontal parallelism sort of scaling out uh multiple multiple instances of the same operator uh running at the same time and then vertical parallelism or interoperator Paralis that'll be different pipelines or operators running simultaneously and again and then there's explicit dependency graph between them so these techniques are not not mutually exclusive um you you can mix and match them again the higher end systems can do that and the way we're actually going to implement all the various algorithms we talked about so far uh there'll be
20:57 basically some minor changes to them but there there's almost parallel versions for all the algorithms that we talked about before uh so for example if we go to our our hash Jo before the grace hash join remember how we talked about how the in the sort the first phase we went through the the build side and the probe side the RNs tables and we generated buckets uh uh for each of them so that when we when we so we divide things up so that now we can just bring in uh things sequentially into memory and do our joins for just the the the buckets we know that things to match you can now sort of think of like the buckets now can can after I partition them up the bucket can then now be processed independently by different different workers at the same time so like the worker one can do the the first level of the hash table and it doesn't need to coordinate with the other workers
21:43 processing other other buckets or other levels because we've already divided things up before so basically where all the divide and conquer stuff we talked about before we can apply in in the same world as well and just but now we just run things in in
21:56 parallel so as I sort of said there there the horizontal parallelism there's the vertical parallelism and then I think the textbook mentions that there's this third category of bushy parallelism which in my mind is basically just the combination of the first one and the second one but we can talk about it and the way to think about how common these different approaches are the first one is is be obviously the most common one but having multi operators uh instance of the same operator running in parallel
22:19 uh an operator parallelism is slightly less common but you'll see this more in especially in streaming systems which we'll talk about a little bit uh later in the semester um and then the buer par parallelism that's just again the higher end systems can can do all these
22:33 things all right so in intralism the idea is that we're going to take our query plan uh and the operator the operator invocations are going to be then now duplicate or replicated across multiple workers that they run and and parallel at the same time but the key thing we're going to do is we're going to assign each operator instance each task for their operator to process a different subset of the input data so again think I have a table of a th tupal
23:00 and I have four worker threads I'll give the first uh 25% of of the data to the first worker the next 25% to the next worker and so forth and then now they can do the scan apply any filters do whatever operations they want in parallel and they don't need to coordinate to each other and we're not duplicating work because each one's uh processing a disjoint subset of of the database we we're sort of partitioning things and and handing them out to the different
23:22 workers but now we got to some cases we have to put the data back together so we sort of split things up allow them to to operate on physical subsets but now in some cases in order to do some uh some additional operators in our quer plan we got to put things back together because remember we talked about I can't do a hash join if I don't completely fill up my hash table because now I start probing it I may make it false negatives
23:45 so there's going to be this we're ince this new mechanism called an exchange operator uh it doesn't map to anything in relation to algebra but it's something we need to again do this coalescing we're going to introduce that almost as pipeline Breakers again to as a way to say I can't proceed up in the query plan until all the uh operators below me that I'm that are my children produce the results that I need and again it doesn't matter whether I'm doing a push versus a pool model doesn't matter whether I'm doing the iterator
24:11 the the materialization or the the vectorize uh processing model all right the exchange operator again is that barrier to make sure that we don't proceed in the query plan until we know we have all the the physical data we need postgress is going to call this uh this uh gather uh if you're familiar with like the map reduce world it's it's sort of the same same way it's the barrier between you start before you KN the next phase but the original paper that described this approach uh came
24:36 from this guy G graphy the same guy who wrote that b+g book that I was talking about before and the same guy who's gonna talk about query did a bunch of stuff we talk about query optimization uh in the next class right he wrote this paper uh in the early 90s or late 80s uh for this project called Volcano and then he defined out exactly this uh you know this exchange operator there all right so as they we're doing a query now joining A and B and where a idals b ID and then we have additional filter on A.V value and b. value so the first thing we're going to do is execute the
25:08 uh the scans on a so say now again we're divide the table a up into three three three partitions or three shards and we're going to assign each of those their own worker but now we actually can sort of start building up the pipeline
25:22 in our in each of these for each of these workers because again we know we don't need to coordinate between any of them yet because I each one can scan a but now they can also do the filter on a value here and then just keep going up the query plan pushing things up and then say we also want to then uh bring down the the projection to throw away a bunch of the data that we don't need
25:42 because again we know what we going need up in the query plan we're need Aid and uh to do the join so we could actually filter out anything else that is an aid after we do our filtering here and again we can do this All In Parallel we don't need to coordinate between the different workers to do to do projections and then now we're going to build our hash table and then this is where we introduced our exchange operator so for Simplicity just assume that there's a single Global hash table here in the uh in the uh you know in sort of the high higher end systems you can have multiple hash tables uh but we can IGN that for now butum Simplicity everyone's writing to the same hash table so they're building the hash table and then we don't uh we tell the exchange operator up above we don't sort of invoke that until we know that all the data we we weever going to process
26:30 is uh is done so the exchange operator again is the barrier waiting for the for all the the the children workers below to say I got everything I process everything I need to process I'm done you never never have to come back to me so you can start of think again these are the pipelines that are going run the these tasks one after another uh these operators one after another and then they're feeding the exchange operator saying like I got more I got got more I got more and then it keeps going until it says it's
26:53 done and then now above the exchange operator that's where we get the join operator because we know we built everything we needed in our hash table uh on the build side of the join so when that opor gets the notification every Everybody below then it says oh yes now now I'm good on my side we can do the join then now say they do the same thing on B again divide it up into three shards or three partitions they'll also do the the the filter and the projection just like before and then now they can also probe the the uh the hash table in parallel but we don't need exchange operator above above the probe right because we're doing the join to the hash table see we have a match and then two PS are coming out out of that operator
27:31 and again we don't need to coordinate anything at this point between the three different workers but we do need to know whether we've finished processing all of the the data in these these pipelines or B so we'll put exchange operator above that that says okay I know I have three workers below when they're all done uh let me know and it's producing a single output stream uh for these these pipelines down
27:52 below so this is the most common version of The Exchange operator this is The Gather approach again taking multiple input streams in producing a single output stream but there's other approaches you could do um and so this is a this is a sort of taxonomy that's defined or that's used by SQL Server to describe the different variations that they have of of exchange operator so again the The Gather one is taking multiple outputs and putting into a single out uh multiple out input streams and producing a single output stream the distribut operator is taking a single input stream and then recharting
28:22 repartitioning to spread across multiple output streams um and then the repartition one is is basically the combination of the two so I have multiple input streams uh and I'm producing multiple output streams but it doesn't need to be a one toone match between the inputs and outputs so I could take in my example here I'm taking three input streams from three sort of pipelines below me and I'm producing two output streams so the distribut operator the partition operator this is used often in distributed systems because you
28:49 may want to rebalance Things based on what the data looks like as it comes in so my example here is for dral big query which are the same systems Dremel is the internal name of what Google calls their bigquery system but the they have an explicit uh fre partition step between every single pipeline U this sometimes again called Shuffle the idea is that
29:11 when you first plan the query you you think the data is going to look a certain way so you say I'm going to run on say 100 workers but then as the data comes in you realize it's it's uh your filters are more selective so you have less data so you can scale down after the after repartition say I I thought I need 100 but I only need 80 you go the opposite I I thought I need 100 but I need 200 or I thought my distribution was going to look like this uh but
29:35 uniform distribution of the data now looks really skewed so rebalance things differently so it's a way to sort of like reorganize the query plan while it's running uh in a dynamic fashion or adaptive fashion most systems don't do this uh part that adaptivity but the repartition one is is can can can be common in the olap systems data
29:57 warehouses all right so now we going talk about okay the how do that was intra operator parallelism So within sort of one operator I have multiple versions of it running the same time now we talk interoperator parallelism where you have distinct operators that are running actually the same time even though there may be dependency of data between them but in some cases like the way the query plan is sort of set up I want to run them at the same time and have not have one an operator wait for all the the data come out of chot operator for it starts
30:24 running so the this is sometimes called pipeline parallelism we're still going to need the exchange operator that we talked about before because you need to know uh you may need to coess things from multiple shards from from down below and produce one or more output streams based on the input streams that are coming in right so really simp that would be something like this so they have that same join before but let's now say I do my join in one worker as one sort of operator instance and then this things is scanning through the the table uh the two tables and producing the join and then emitting them out as as the stream
30:59 but then now let's say I have this projection uh this projection operator instead of inlining or or fusing it in that we saw before in the pipeline I'm going to have it run as it own separate worker as well say for whatever reason this the projection here is very expensive to compute like we're Computing a hash of a hash of a hash or something and therefore it's very CPU intensive and therefore I don't want to burden the join down below with doing that that hashing I want to put that on on a separate worker right so now as the
31:26 the the the the first worker is doing the joint producing output it puts it in a queue that then gets sent up to the separate worker and then think this thing just waiting for new inputs to come in and soon soon as it get something it crunches it and produces the output again this is more common in in streaming systems like spark SQL Kafka and other things because in that world you sort of have this this continuous
31:48 query operation where there isn't actually an end to the data that you're inputting like it's an input stream of like stock ticks or temperature readings there's data just always coming in as a stream and therefore it's not like you know scan on a thick siiz table you scan it and you're done this thing is just always
32:03 running right and then the last one again as I said is is bushy parallelism uh and this is just intermixing or a combination of the horizontal inter interoperator sorry the horizontal parallelism with int operator and the vertical parallelism with interoperator just putting all together using them at at the same time and again we still going to need exchange operators because we need to know in some cases did I get all the data I expected from from my query down below me
32:28 so this is a contrived example but I'm doing a cartisian product between tables a b and c and d and so now I can do my my query plan I'll do my join on uh on A and B simultaneously as the join in C and D and then there's the the the cross joint up above that I'm running on worker three and four that just intermixing all these tupal uh and producing all possible combinations of them I buffers to keep track of what I've seen before and as new new new tubes come in I can match them right this is obviously be a very expensive operation uh and yes the the size of the in results would explode uh but you can again you can at least computationally run these in parallel and rather than just waiting for everything to get queued up and and
33:11 finished all right so everything we talked about so far it's about computational parallelism right how do we get multiple workers to run simultaneously and and process things in parallel of course though if now your bottleneck on the disc even though it's gotten a lot faster than uh than in the past in modern ssds with mvme you know that still could be a bandwith just you have to read a lot of data in your query um and then also too if you're if you're using a page table um and managing managing that with latches that could become a bottom neck but the the getting things off the disc and bring to memory
33:45 that can be a bottom neck depending on the speed of the the drive so in some cases it actually doesn't matter that if you have a bunch of parallel workers that can run the queries in parallel or run a query in parallel if they all have to go to dis and you're blocked on that then you're sort of the parallelism isn't going to help as as much now we'll see later on how to do tricks of like if I know one guy's got to go out the disc and get something and this other guy can can run because everything they need is in memory then I can play tricks of scheduling scheduling that guy first and then letting that disc iio Happ happen a thinker to see in the background so that way I'm always doing try to do useful work even though I'm blocked on dis but again in general I think if this is just slow then the parallelism can only go so far to help
34:24 you so this is why we're going to need IO parallelism uh and so the idea here is that if we can take our our our database systems data in our database and we can spread it across multiple storage devices then this could potentially produce our our disc bandwidth because I could have some part of the system read data from one dis and another dis and now ignoring whether you know my my pcie channel is the bottleneck I can get data I can have more current requests and have a higher throughput uh for my
34:53 data there's a bunch of different ways to do this uh like the most simple thing is like have a single database uh uh one one per database or sorry a single disc per database you could have uh put multiple disc for for a single database so so logically it seems like a single database but physically it's across multiple dorage devices you can put a single table single relation per disc you could split a disc across multiple discs or split a a relation or table across multiple discs in some systems you could actually say for the right intensive portion of of the the system like the right ahead log the log keeping track of all the updates I do you put that on one disc and then put the data from the regular tabl put that on another dis so you're always writing real fast at this thing and then you're reading your reads to that the data doesn't interfere uh with the rights
35:37 there's a whole different combinations ways to approach this in the higher end systems they they will they will have explicit control of how to handle IO parallelism like in Oracle you can say again put the right head log on this drive and then split my tables uh my data tables across these other drives and that the data system will manage all that for you uh in systems like you know my SQL for example the the disc can just be managed by taking a
36:05 um uh just in the actual data directory itself you just put SIM links to different drives uh and the data system doesn't know you've done that but just you know now you're getting that parallelism because the the multile disc can run simultaneously not interfere with each other postard having has table spaces you can say these tables belong in this table space and then tells you what what physical location it goes on dis but again the the the highend systems have complete control of all these
36:31 things all right so what does this look like so say now we have a a database and it has six files or sorry six pages right really simple so I'm going to show two extremes of of the different variations of how to support you know a multi-is parallelism um in the first case here say we have three disc drives
36:51 and we'll do the first approach be called striping where if you're familiar with raid it's raid zero and basically every single page uh in a round roben fashion we're going to write them out to one disc drive at a time uh so so page one goes to dis one page two goes to disc two and Page Three goes to page three goes to dis three and we just keep wrapp around looping around this round so now this is where that um remember that page director we talked about the beginning of the semester this is where we keep track of this information so now if I do a look on page one I'm going to know oh it's on dis drive one and therefore and this location Let me Go fetch it right because again the data system could could could manage this itself or you can rely on the hardware to do this for you where just transparently looks like a single logical disc even though physically it's broken up multiple disc devices and again the data system has control of this it can be responsible for deciding how to split things
37:38 up the other end of the spectrum of striping is called mirring and this is raid one and this is where for every single page we have in in our in our database we're going to write that out uh multiple copies out to every single uh unique disc drive we have so page one is going written on dis one dis two dis three page two is written on dis one dis two to three and so forth right and so the advantage of this now is uh if I have to do a bunch of reads that's going go faster because now I go fetch the I can go to actually any of the disc drives and I get exactly the copy that I need course now this makes the rights more expensive because any one right you do to a page has to be written in triplicate across these three drives whereas in striping it was always a one toone mapping between a page and and a
38:20 uh page and a in a disc drive so the there's Hardware actually do this all for you uh in there's you I'm saying is raid zero ra one there's different levels of raid that can inter mix and match all these things and there's different trade loss to all of them so in some cases you can get harder to do this for you so now it appears like a single logical device uh but physically it's it's all split across and this would be transfer to the database system D system wouldn't know that it has multiple disc drives it just thinks it has a faster disc or uh higher capacity disc based on what what you're
38:54 doing the uh the software based approach when the D can manage the themselves and now you can do BU of other tricks to not just have to do either striping or mirroring you can use a ratier coding to do some get some sort of blend of the two of them uh and this is typically be faster and more flexible than doing Hardware support uh but again not all systems actually support this so the way to sort of think about this uh the trade-offs for Io parallelism or disc parallelism is a tradeoff between
39:21 performance uh durability and capacity so in my examples before were I was doing um say I was doing uh the the mirroring approach I'm I'm basically wasting if you will multiple dis drives but I'm storing the data in in duplicate so one now if one dis drive crashes uh I can rely on the other two to still read my data so I'm getting better durability
39:43 and my reads are going to go faster because now I can have any one read from any one page be handled by any any of the the the three disc drives but my capacity is basically cut uh you know by a third because I'm or down to one/ third what it what it the total size could be because now every every every disc has the entire copy of it right where striping is I'm getting great capacity uh but my durability and better performance my durability isn't as good so there these trade-offs that you have to consider when you when you sort of deploy a real dat system of how you want to handle this S3 CH of changes all this because S3 is now managed by by Amazon or whatever the object store in your cloud provider you're using whatever the equivalent is and they're storing data I think in five times um but again it's all transparent to datab system you don't know if you do a read request or write request you know where it's actually being physically go going
40:34 um and S3 is actually not that fast that's like hundreds of milliseconds potentially per per read and write um but it's you know the the DST doesn't manage any of that Amazon manages that for you so again there's trade-offs of how much you want to do yourself versus how much you rely on external Services either through Hardware or through a cloud
40:51 service the other thing to also mention we'll talk a little B about we'll go more detail we talk about distributed databases is how we're actually doing this this splitting things up so I sort of said before when when when when we're talking about uh the horizontal parallelism the inquery uh interoperator intraoperator parallelism I would have my different operator instances crunch different portions of of a of a of a table or input data right that data is going be partitioned or sharded based on some some attribute um and so it's if the databas system is just stored in separate directories then that's really easy to do because you say this operator processes this directory this operator processes other directory um but we what we really want to be able to do is to do explicit partitioning within our database system where the database system responsible for deciding how to divide things up where to store it and then how to bring back things in into memory and process it in in in parallel
41:44 so again this is the advantage of of SQL again because the same query that would work on a database that's not partitioned that's not split up should still work in theory on if a database is partitioned is split up um because it's only processing these logical tables doesn't more actually the physical layout of things um doesn't always happen uh in some cases but that's that's the goal of of in you know doing this this partitioning scheme so again we're not going to talk about it any further right now uh we'll go into more detail of this when we talk about distributed databases uh around
42:17 Thanksgiving all right so finish up parel execution is super important uh and pretty much every single uh major data set you can think of is doing some variation of it even SQL light right SQL light's like this this it's a very sophisticated system but it's a you know it's an embedded datab system that is doesn't have all the features of like a full-fledged Oracle or or db2 or seal server um but it supports parel execution uh it does inter query inter query pism so you can have multile readers and a single writer um and that all works just fine but all the things we talk about that we we to get right
42:51 like the coordinate coordinating the different workers scheduling them uh if they have to start doing rights how do you handle that what if every's trying to read from the disc at the same time how do you handle that all these different topics and things like that you have to you have to account for when you start building things out and it's ideally you don't want to rely operating system to manage dis iio and other things for you you want to do everything
43:10 yourself all right so next class we're going to talk about query optimization uh so this is the idea of again taking our SQL career and then actually now generating the physical plan that we're going to execute on our workers uh and so this semester we're only going to do one lecture on this uh and it was so s to be a quick overview of the topic it's very very hard it's the thing I know the least about in databases but then uh in the in the Spring Class spring 2025
43:34 we'll be teaching of course entirely on creation so if you're interested in this kind of stuff uh we can go to this in more detail at click house Inc and I
2. Jump to
43:43 want to give you um a high level intro to the things that make click house equal database you probably think click house is a funny name I guess it is Click house was originally an abbreviation for clickstream data warehouse so that should give you an idea of what the system is about but you know people like the abbreviation and we simply kept
44:03 it uh here we go all right so what is Click house at a high level click house is an open source column oriented distributed olab database development started 15 years ago at Yandex eight years ago the code was open sourced in these eight years click house became super popular now with more than 2,000 contributors on GitHub the code is written in C++ and it compiles into a single statically linked binary which runs on pretty much anything from raspberry pies to the most powerful
44:31 servers being a column store people typically use click housee for filter and aggregation queries over billions and trillions of rows the storage layer under the hood is optimized for append only workloads um and it provides very fast inserts I get to that in a second
44:47 you can also build a cluster of Click housee nodes and um replicate and shart your data for higher availability and scalability if you do that the database will be by default only eventually consistent in the interest of high performance typical use case of Click house include business intelligence lock and and event analysis and realtime
45:09 dashboards let's dive a bit deeper into the architecture of the system there's a lot of stuff on the slide but the important parts are in red the query processing layer and in blue the storage layer the query execution in Click house follows the traditional approach of parsing a SQL query building and optimizing lot iCal Plan Building and optimizing a physical plan and then finally executing the the physical execution
45:33 plan the storage layer has a concept of plugable table engines which each represent the location and the format of the table data there's a similar abstraction in MySQL so we didn't come up with that idea by ourselves in the interest of time I will only talk about the native storage format of Click house which goes by the name merg Tre table engine it's actually a family of table engines therefore on the slide it's called merch tree star Family Table
45:59 engines this slide shows the Mery table engine in more detail what happens if you insert a bunch of rows into a table is that the new rows are sorted by the primary key Columns of the table and then stored as a so-called part on disk
46:16 once written a part cannot be changed anymore it is immutable we encourage our users to insert the data in batches for instance 20,000 rows at once to avoid that too many to little parts are created to prevent the parts from accumulating they are also continuously combined into bigger Parts by a merge chop which runs in the background now remember the parts are sorted this means that the merch chop can use a kway merge sort algorithm to combine the parts and that is um relatively cheap click house also provides an asynchronous insert mode
46:49 which basically um buffers the rows from multiple inserts before they are flushed to disk so if you really want to do lots of small smaller inserts with very few rows each you can use asynchronous inserts now if all of this reminds you of lock structured merge trees or LSM trees in short you're right the difference to classical LSM trees is that all parts in Click house are created equal they are equal they're not organized in a hierarchy this has advantages this has disadvantages on the one hand the merge can freely pick which parts are combined it is not bound to certain LSM tree levels on the other hand updates and delet become a lot more tricky but I'll not talk about these today for the majority of all use cases updates and deletes are the exceptions
47:33 so that's usually fine I talked about Parts on the previous slide this slide zooms into a single part the stuff shown here isn't terribly interesting I'll just say that after a few rounds of merging Parts typically contain a few dozen million rows logically each part is further divided into so-called granules Each of which contain contains about 8,000 rows by default all the dis exes are made at the granularity of granules not rows in the example on the left you see two granules
48:07 g0 in blue and G1 in red to reduce the data volume we also compress consecutive granules users can specify the compression algorithm to use there are generic codex like CSD and logical codex like Delta Okay now click house has a reputation for being fast one of the tricks we use to speed dat queries is aggressive data ping the idea is that you want your select queries to skip as much data as possible because every bite you don't read and don't process helps
48:37 performance click house uses three data printing techniques the first one are primary key indexes I mentioned earlier that each part is sorted by the primary key Columns of the table click house will additionally create an index facture which maps from the primary key column values to granules the mapping is quite small compared to the total table size and um kept in memory for example with just 1,000 entries you can index already 8 million rows now if you have a query that filters on a prefix of the primary key key columns click house will use the primary key index to find the right grenes directly instead of scanning the whole
49:19 thing the second pruning Technique we use are table projections think of a table projection SN or alternative version of a table a projection contains the same rows as the original table but it's sorted by different by a different primary key if you have lots of queries which filter on other columns than the original primary key columns you want to use a projection
49:42 the cool thing is this looks at part chity that means a part can have no or one or multiple projections and the query engine will choose for every part the optimal data representation to read from at query time
49:57 the third printing technique in Click house are skipping indexes the idea here is to annotate the granis with metadata that metadata enables to the scan to check if no rows or or some rows or all rows in the Grano will match the predicate there are now different types of skipping indexes in Click house for instance you could store the minimum and maximum value for every granal or the unique values or you could create a bloom filter for every granal each time of course has different trade-offs but generally speaking skipping indexes are more lightweight than projections but they're also a lot more data dependent
50:34 and hard to tune say again exactly click house also has a really cool feature that we call merge time data transformation I mentioned earlier that that parts are continuously getting merged in the background the merge job is able to apply additional transformations to the data for example
51:05 there are so-called replacing merges a replacing merge will check if there are rows with the same primary key values in the input Parts these rows are basically considered as duplicates if it finds such rows it only keeps the one from the most recently inserted part you can think of this conceptually as a kind of merge time update mechanism
51:27 another example are TTL or time to live merges they look at the age of the input parts and possibly compress or move or delete tles based on their age and finally there are aggregating merges they are really interesting but also a bit more complicated the problem we tried to solve here is that aggregation grou by over billions and trillions of rows becomes super expensive if you think about it a normal aggregation in a normal select query needs to scan the entire table build a hash table and then for each row find the Aggregates and update the Aggregates now aggregating
52:04 merges are supposed to help that what happens is that the aggregation is done at merge time a select quy will then simply use the precalculated Aggregates instead of scanning everything um you see an example on the right there are two parts which are combined by an aggregating merge into a new part um and then there are two aggregate columns the maximum and the average latency both of them are of a special data type which allows to update them incrementally for example the average latency column is internally
52:37 stored as a sum and a count in the top left part you find region Apec which has an average latency stored as 80 and uh stored as 80 for the sum and one for the count in the top right part you also find ape with a sum of 180 and a count of three in the merge part apeg is computed as a sum of 260 which is 80 + 180 and the count of four which is 1+
53:07 three that's tricky um but the point I'm trying to make on the slide is that merge time data transformation is a really powerful tool the calculations are decoupled from the normal queries and they don't affect them the only downside is that you cannot control when the transformation happens and on which parts they happen if that is important for your user case click house also provides means to run the Transformations at query time at least some of the
53:36 Transformations okay uh a few words about quer execution in Click house like every modern database click house will try to utilize all server resources and also all cluster resources if you happen to have more than one server to use all CPU Calles we unfold the physical execution plan into n Lanes with n equal 3 in the examp on the right each lane processes a disjoint range of the source data so we basically split the source table into three equally large ranges in the example on the right of course in practice things don't always go that smooth so you could have for example the situation that the lanes become imbalanced for example when the selectivities of the data ranges differ a lot from each other if you let that happen your course become underutilized
54:23 so what we do is we exchange we insert Exchange operat at s in certain places in the plan The Operators themselves pass multiple Tes at a time instead of single Tes or to say a bit more fancy click house uses the vectorized volcano model this has three three performance benefits first it has pretty good cach locality second it amortizes the cost of calling virtual functions and third it enables simply operations across multiple
54:52 tbes that's my last slide about simp simp instructions and and we use them a lot to speed up hot loops and make it really quick hot Loops in Click house are often implemented in multiple versions that we call Computer kernels
55:07 there's always a generic version and one or more vectorized versions um either using compiler Auto vectorization or handwritten intrinsics at runtime the database will check the system capabilities and execute the fastest available compute kernel the benefit of this approach is that the system utilizes um Modern Hardware there but it will also remain compatible with Legacy systems which our users happen to have a
55:36 lot that's it from my side today if you like to know more you could have a look at the overview paper that we just published this year at BLB so that's it thank you any questions sorry my my problem not your problem all right sorry any questions for Robert he basically described everything we talked about today right actually Robert can you go back like to the slides of the execution
56:01 engine which one this one perfect yes look so look exchange operators you were asking why would you ever repartition or distribute it's exactly what Robert said right if the data shows up unbalanced you have to you have to you know move things around uh and repartition and then distribute puts it puts it back together right the vectorized volcano model so the simd piece we didn't really talk about but uh volcano model we said was the iterator model but again the terms are used they're not they're used Loosely it's they're using the vectorized batch model we talked before so they're sending up batches of tups
56:32 between the operators to process now uh you know at the same time so now on the next slide can you go to the syy one so I I got all excited while you were talking you can't see me um so we didn't talk about simd I'll I'll send the lectures from last class but basically
56:47 it's sort of what I was saying before now you have a batch of tupos going up from one operator to the next now within instead of processing them one you know one Tu at a time in that for Loop you can say all right let me take eight tups
56:58 because again I have column let me take eight eight values uh across across eight different attributes eight same column eight values across eight different tuples and then instead of me again looping through one by one there specialized instructions on CPUs that do like you know 1 plus one or a number plus a number or something equals something you can do that in parallel so single instruction multi multiple data we don't talk about simd in this class but that's that's why they're getting amazing speeds not only are they running the the the queries and parallel doing all the things that we talked about today then now on a single worker
57:30 they're basically running uh operations on that worker in parallel at like the harbor level so that's why like again they get get amazing numbers sorry I got excited I I if anybody has any questions sorry any questions from
57:45 Robert all right uh so let me show you guys you can't see this here Robert uh I'm going to show them the click bench website so as I say say before the um click house has this amazing website where they keep track of the the the benchmarking for all these different database systems for this workload there so it's a a bunch of Select queries with no joins but you can look at the ranking here so this is from the Germans uh and then below that is Click house tuned
58:14 right and then after that is all is all Doris right so the Germans are all what's that I believe I believe umra is number one right now Umbra is number one yes yeah but you how far off you guys
58:26 uh yeah a little a little bit I mean like actually so do you so I mean do you know what's what's holding you back from beating Umbra in in your implementation that's a really good question um their code is unfortunately um close Force so we couldn't have a like a closer look correct um they use Query compilation a lot um everything's based around query compilation in umra um now we use quy compilation to some degree um but yeah it's it's really hard to answer at this
59:04 point question yes yeah are you guys using precompiled Primitives at all like for operators the question is are you use pre pre-compiled Primitives yes yes they are yeah right if so if you go back to the Z shell sharing yeah so this slide here right so these these multi-target function like that that gets pre-compiled and then at runtime they're figuring out what the harbra is and then based on that they like they know what they know what perim you got to run like does something does 32bit integer equals 32bit integer and then they'll have the the simd versions of that and then they can always fall back to the to the scaler version of
59:39 that click house is a freak system I mean Robert if you don't me saying this like like you have 20 versions of hash tables right you got you guys have all these amazing things where like most systems like they'll have one hash table but they'll have all these specialized based on the different data types and the size of them and that's just to do hash tables right and they have all these different storage engines so like
59:58 click house gets amazing performance because it just has so many different specialized components for different data types and and different sort of not workloads because still still doing analytics but like for different workload patterns right yeah absolutely that's and again the Germans are number one because they're like freaks uh or this one Thomas snowman he's the freak uh but click house again
1:00:22 it's I say this click house gets great performance it does has all these specializations for the different data types that we talked about Umbra it's just cuz he like what they do is a query shows up and then instead of actually running interpreting it like you guys doing busb uh and they're also not doing pre compile Primitives the way click house does he generates x86 assembly on the Fly runs that he runs that through the assembl and runs that uh you know the query starts running but then in the background he runs lvm then compile that assembly into like O2 and then then slides that in when it's ready so there's a bunch of other tricks the Germans are doing uh at T minut for Umbra but click house is like click house is sort of throwing everything at it rather than being sort of specialized
1:01:07 on one sort of one technique yeah right so so you can learn a lot from looking at the click Out Source Code because like how do they do hashes there's different 20 20 different versions of how they do it right all right any last
1:01:21 question and Robert are you in Amsterdam are you you're in Germany where are you I'm in chimy yeah got it okay the good part the no the other one okay that was a trick question yes all right guys let's get Robert a round of
1:01:40 applause again click house is amazing it's uh like how do you say this duct Tobe is nice for those small things running your laptop if you want to go hard you you do something like click house okay right all right Robert thank you so much thank you for being here perfect thank you see you guys
1:01:59 byebye I need something refreshing when I get fin manifesting cold a whole B like Smith and Wesson one court and my thoughts hip hop related write a rhyme and my pants intoxicated lyrics are quicker with a simple Mo lier since on City Slicker play waves are picker rhes I create rotate at a r too quick to duplicate feel a breeze at a skate mik at Fen when I hold it real tight then I'm in flight and we ignite what starts to boil I heat up the party for you let my girl R me my mic down with oil Rec still turn with thir Dee Burns for one man I heat up your brain give it a suntan so just cool let the temperature
1:02:40 rise to cool it all with eyes [Music]