Query Execution II

Last lecture, we started talking about how to compose operators together to execute a query plan. We previous talked about how to execute with a single worker, and now we want to talk about how to execute with multiple workers. 

This can result in better throughput, latency, and system responsiveness and availability. The system will feel more lively and respond more quickly to our request, in the domain of disk-oriented database systems. This would likely result in a lower Total Cost of Ownership (TCO). We want our database system to take advantage of a system that has more cores.

In both Parallel and Distributed Databases, we have the database spread accross multiple resources that improves different aspects of the Database System, but it appears as a single database instance to the application. SQL query for single database would generated the same result as a distributed database. 

The term parallel vs distributed are different, but the names are often mixed.

Parallel DBMS are resources that are physically close to each other and connect over high-speed interconnect. The communication is assumed to be cheap and reliable.

The Distributed DBMSs allow the resources to be farther away from each other, but resources communication happens slower and there is way more significant costs/problems in comparison to distributed DBMSs. Parallel sockets have a bunch of cores running at the same time. 

OLTP is a bunch of small request and OLAP is requests bigger but bigger chunk of them. The process model defines how the system is architected, and the worker is the DBMS component that is responsible for executing tasks on behalf of the client and returning results. The worker is either a process or thread, depending on what process you use. 

We can have a process per DBMS worker, process pool and Thread per DBMS worker as the 3 different approaches, respectively. 

In process per worker, we have a single worker represented by a single OS process. What happens is your application sends a request for a query to open connection to database system, wome dispatcher gets request, then worker responsible for handling the connection, and now the worker is responsible is responsible for executing whatever request the query wants. Also, a process crash doesn't end up taking a whole system because only one guy failed. 

In shared memory, malloc() is private address,  but with shared memory any person is also able to read and write to that particular space. Unless I have shared memory, every single worker is going to have the same buffer pool. You would use processes over threads because there was no standard thread APIs, especially in the 19980as before posix and P-threads, but now P-thread is the most useful. 

An extension is a process pool has a bunch of workers sitting around and the dispatcher tells a worker to execute a particular query. A worker uses any process free in the pool, and still relies on the OS scheduler and shared memory. However, this is bad for CPU cache locality. You don't want the CPU to burn cycles. 

Most modern systems have thread per worker where we have multiple worker threads (pthreads) and these threads in the DBMS manage their own scheduling. Because we know what the tasks are, we can do a better job on how everyone is doing and as a result schedule threads in how things are doing.

Using multi-threader architecture has less overhead per context switch and does not have to manage shared memory. The thread per worker model doesn't not mean the system supports automatic parallel query execution. It can't break queries up across threads and run all of them is parallel. 

Both processes and threads are independent sequences of execution. The typical difference is that threads (of the same process) run in a shared memory space, while processes run in separate memory spaces.

In scheduling, the dispatcher can understand what tasks should execute, what should pause into another thread, and where an output actually goes, in general, we need to know the tasks/CPU cores to use and store in output. However the Database System always know better than the OS, so we can decide these things better. 

Inter-Query Parallelism are multiple things that can be executed concurrently, and Intra-Query parallelism is executing operations of a single query in parallel. 

Inter-Query Parallelism Improves overall performance by allowing multiple queries to execute simultaneously. If queries are read-only this requires little coordination between queries. If multiple queries are updating to database, then this is hard to do correctly. 

Intra-Query Parallelism is useful for analytical queries with multiple resources and workers available for us. We will focus on compute parallelism here. A way to think of how to organize this is the query plan, and we think of the operators in the terms of producer/consumer paradigm. There are parallel algorithms for every relational operator. 

Parallel Grace Hash Join is an algorithm used in intra-query parallelism. We use separate worker to perform the join for each level of bucket for R and S after partitioning. 

We can sort work out in sequential scan, divide them out, and have them work out in parallel. 

There are 3 kinds of parallelism (not mutually exclusive): 

Intra-Operator (Horizontal).

Inter-Operator(Vertical) and Bushy. I can use these techniques to get the best performance from the workload. 

The exchange operation is location in a query plan where the database system injects artificially to combine. It was invented by the same guy who came up with the volcano iterator model. The exchange operator into the query plan helps to coalesce results from children operators. We feed up data for all the pages, and carry out the exchange operators. These are streams, and this keeps shoving up data until the system tells it to stop. 

We can allocate threads like "I want one thread to process the data where values are less than 1,000 then less than 2,000), etc. We won't want to do this parallel stuff if the order in which we process data matters. 

There are 3 exchange types.

One type is gather, combining the results from multiple workers into a single output stream.

Another type is Repartition is reorganizing multiple input streams across multiple output streams.

The final type is distribute, where we split a single input stream into multiple output streams. 

Inter-operator parallelism is when we have different operators run in separate threads at the same time. 

For example I can have one worker just do the join that emits it up to another worker doing the projection and sending things up in the query plan as a result.

(Inter operator parallelism diagram).

Here is the Intra operator Parallelism Diagram 

The final thing to talk about is bushy parallelism. The difference is that worker execute multiple operators from different segments of a query plan at the same time and we need to use exchange operators to combine intermediate results from segments. We have 2 operations running in parallel and subsequently shoving data up to the exchange operator.

One tuple can do the join, where the other one can do the projection. Try to do as much sequential I/O as much as possible and we do want to build the hash table in parallel, as this operation is super slow. 

Bushy parallelism is virtually the same thing as inter-operator parallelism, but with more parallel workers running at the same time, where workers execute multiple operators from different segments of a query plan at the same time. We want tuples to execute in the same partitions. Here is the diagram: 

If we're bottlenecked all these buffers will not help at all. We can solve this through I/O parallelism, which splits the DBMS installation across multiple storage devices. This is very similar to RAID (Redundant Array of Independent Disks). 

(RAID Diagram)

A simple example is a database that has 6 pages and what happens is there's some RAID approach that tells which advice to write particular data to. Another common approach is to do mirroring, which means that every single device has a complete copy of every single page. 

The idea of database partitioning is we can split data into particular subsets in distinct disks. These DBMS allow you to specify the disk location of each individual database. This is easy to do at the filesystem level if DBMS stores each database in a separate directory. 

Let's talk about partitioning a little bit. In Partitioning, we take a single logical table, and split it into disjoint segments that are stored and managed separately. Ideally, partitioning is transparent in an application. The application accesses logical tables and does not care about how things are stored.

There are 2 types of partitioning: vertical partitioning and horizontal partitioning. Vertical Partitioning is to store different columns in Separate storage device. 

Horizontal partitioning means that all the data in a single tuple will be located together in a single parittion, so there can be multiple workers working in parallel and operating in the partitions at the same time. The following diagram reduces from 4 to 1:

So, finishing up, parallel execution is important, and we can develop them right in inter, intra, and bushy parallelism. However, how can we coordinate multiple threads operating at the same time?