CMU 15-445/645-笔记-13-Query 执行-part2
## 课程目标
这节课主要讲的内容，是如何并行执行这些查询
## 为什么要关心并行执行?
因为现代 CPU 和 GPU 都有很多可用的 Core，利用多核可以有效提升性能(虽然并不绝对)
同时也有更高的吞吐量，而这意味着，每秒可以执行更多的查询，能处理的数据更多，同时延迟更小
最终，整个数据库系统能对请求进行更快的响应
TCO(Total Cost of Ownership)是用来考虑一个数据库系统成本方面的术语，它指的不仅仅是购买机器、购买软件许可证之类的成本，它指的是某一段时间内，运行这个数据库所要花的成本。如果能用更少的硬件来做更多的工作，那么 TCO 这个成本将会降一大截
为什么需要多核，就是能省机器，也就是省钱!
## 并行和分布式的区别
相同点: 都是将一个数据库分散到多个 **资源**(多台机器、CPU、磁盘等) 上，这能够改善数据库系统的各个方面，比如性能、成本、延迟等
不同点:
- 并行
1. 可用资源在物理意义上彼此相邻(比如多个 CPU 之间通过高速总线进行通信，这个距离很短)
2. 资源通信速度快、成本低，可靠性高(数据通信时不会丢失)
- 分布式
1. 资源间的距离可以很远
2. 资源间的通信，因为距离的原因，就得使用一个速度比较慢的通信信道(比如公共广域网络)
3. 又因为使用了网络协议做数据通信，所以可靠性不能保证，其他还有一些不能忽视的问题比如成本之类
## Process Model
Process Model 就是一种通过组织系统，调度多个 worker 来处理并发的请求方式
为什么需要调度多个 worker ?一个应用程序可能会发送一个很大的请求，或者同一时间发送了多个请求，给数据库系统，这个时候就可以把这个请求打散掉，然后分配给多个 worker 去处理
> 这里的 worker 就是一个任务的抽象，worker 可以是一个进程也可以是一个线程，也可以是一个查询优化器的执行任务
有 3 种 Process Model
### Process Per Worker
使用单个 OS 进程作为一个 worker 使用
上图就是这个 Model 的模型，最左边是数据库，最右边是应用发出的查询，OS 通过调度器，使用了一个进程作为 worker 连接数据库，然后用这个 worker 处理查询
但是这里有一个问题，如果是多个 worker 的情况，那么有可能，同一个 page 会有多个副本保存在多个 worker 代表的进程中，这样就会造成内存的浪费
那么如何解决?
可以使用共享内存，这种方式可以允许这些不同进程在内存中，拥有各自独立的地址空间，那么这些进程，就可以共享这些全局数据了
多进程好处是，一个进程挂了不影响其他进程工作
为什么上述老牌数据库系统都采用这种方式? 因为当时线程的 API 标准还没出来(尽管各个操作系统里面都有线程)，为了最大的兼容性以及最小的代码修改，就采用了这种方式，因为 fork 和 join 是当时各个 OS 的基本原语，也就是它们都支持进程并且 API 相对一致，也就是同一个数据库能跑在不同的 OS 中
### Process Pool
Process Pool 是 Process Per Worker 的改进版本，关键的改变在于，这里不会为进来的每个连接都去创建一个进程，而是里面会有一个进程池
进程池会分配进程去处理查询(同时进程可以复用)，而不是每一个查询进来就 fork 一个去处理，因为那样代价很高
好处就是能做并行查询，部分工作量可以分给其他进程
> work-stealing，是指某些 worker 在处理数据时花费了太多时间，还有一堆它需要处理的工作堆积在它要执行的队列中，那么调度器会从这个 worker 手中拿走一些工作，交给其他 worker 处理
### Thread Per Worker
这是大多数现代操作系统的选择，即一个线程就是一个 worker
用一个进程来运行数据库系统，在进程里面的线程，根据需要去对任务进行调度
上图是这个 Model 的模型
使用这种多线程模型才是正确的做法，这种方式非常省事，因为不用去管 OS 里面的共享内存或者进程管理之类的东西，并且上下文切换的开销更低，因为进程使用的是 OS 提供的共享内存空间，很难做到保护。然后就是执行任务会变得很快
## Scheduling
调度是决定将一个查询拆分成多少任务，决定使用哪个 CPU Core 来执行这些任务，哪条线程要等其他相关线程执行完之后再继续往前执行，的这么一个东西
并且，一旦某个线程执行的任务生成来结果，这个结果应该往哪里放，也是调度的内容
## 并行查询
分为 2 个
1. Inter-Query
可以在同一时间执行多个做不同事情的查询，能改善系统的吞吐量以及延迟
2. Intra-Query
将一个查询拆分成多个子任务或者片段，然后在不同的资源上同时并行执行这些任务
Inter-Query 的思想是，用多个 worker(线程) 同时处理这些请求
但是也会有一个问题，如果有多个线程同时对数据库进行更新操作，那么就要处理数据的冲突问题，要加锁
Intra-Query 适用于分析型查询，利于并行计算。因为有多个 worker，它里面有多个线程或者 CPU Core，就可以使用它们来执行同一个查询
可以用 生产者-消费者 模式来重新思考 Query Plan Tree 中的 operator 的并行算法
每个 operator 不仅仅是数据的生产者(调用 next 生产一些数据)，也是数据的消费者(消费在它下面的 operator 传上来的数据)
用多线程更新和构建 hash table，用多线程来对 hash table 进行检测
可以从当前 operator 下面的 operator 传上来的数据进行分区，用多线程来处理这些数据分区
好处就是不需要对这些同时工作中的不同的 worker 进行协调了
Intra-Query 也有 3 个方法
这 3 个方法并不是说独立使用，而是可以结合起来用
### 并行 Grace Hash Join
上图是之前提到过的 Grace Hash Join 算法，那么怎么做并行呢
让一个 worker 对其对应的一层 bucket 做处理，然后进行 Join 操作，并生成输出结果
麻烦的在于如何将这些输出结果组合在一起
### Intra-Operator(Horizontal)
如果要对一张表进行扫描，可以对其扫描生成结果后的多个实例，用不同的线程对不同的 fragment 进行相同的处理，每个线程都会去扫描表的不同部分，然后将数据收集到一起
这里数据库系统会使用一个叫做 exchange 的 operator 的玩意儿
这个 exchange operator 会放在查询计划中的某个位置，提示数据库系统，在这个位置，可以对这些 fragment 进行并行处理，然后将处理完后的 fragment 结合到一起
例子
如图所示，对 A 进行扫描，然后将结果传入 filter operator
为了做到并行执行查询，可以将这个查询计划拆分为 3 个不同的 fragment，每个 fragment 中都有一个 scan operator 以及 filter operator
然后对数据库进行拆分，将它们拆分成多个 page
然后在一个指定的 fragment 中，对不同的 page 进行操作
对于 exchange operator，它会调用 Next，而后在 scan operator，它也会调用 Next，然后从一个特定的 page 上拿 data
对于其他的 fragment 也做同样处理
exchange operator 的特征如下图
1. Gather(收集)
将不同 worker 线程执行任务所得到的结果，或者是这些 operator 生成的不同输出进行合并，合并成一个单一的输出流，再向上传给下一个 operator
2. Repartition
有些时候需要将数据按区间进行分块，这样就方便做并行扫描，然后将它们放进 repartition exchange operator 进行处理，根据看到的值进行拆分
3. Distribute
拿到 1 个输入流，将它们拆分变成 2 个不同的输出流
一个更复杂的例子
上图中的查询，经过了
1. 对 A 进行扫描
2. 分配 3 个不同的 worker 来执行工作
3. 在查询计划的 fragment 中，先对 fragment 进行 scan 操作，然后是 filter 操作，接着构建 hash table
上面的 hash table 是一个由这 3 个 fragment 共同维护的 hash table
然后这里会有一个 exchange operator，只有等到 hash table 全部都得到更新之后，这个 exchange operator 才会开始工作
然后对 B 做类似的操作，这里通过 partition 来拆分 B 的数据
然后再进行 Join，可以用单线程也可以用多线程(这里使用多线程)
此时可以在 Join 里面将任务进行拆分。同时也可以通过不同的线程来对 B 那边的 partition 进行检测
综合来讲，就是将数据进行拆分，将拆分后的数据交给多个线程，由它们来执行相同的逻辑，然后将这些线程执行所得到的结果进行合并。
最后，把这些过程整合为一个很大的树形结构，这样就可以做并行执行了
### Intra-Operator(Vertical)
对于每个 operator tree 来说，可以将每个 operator 作为一个单独的 worker 来执行
例子
对于此处的 worker 来说，可以使用一个 CPU Core 或者一个 worker 来执行这个 Join 操作，它只需要从它的子 operator 处获取数据即可
Join 操作完成之后，结果会发送给另一个 worker，上面这个 operator 会拿到 Join operator 发送给它的数据，然后执行条件判断或者 projection 操作，最后将生成的结果进一步发给查询计划树
所以执行这些 operator 的线程，都在自旋等待数据的传入，也就是形成了 生产者-消费者 模型
但是这里就会存在一个问题，假设 2 个线程在访问同一个数据，那么它就会用到锁，这样就会导致，一个线程在读时，另一个没法写，或者一个在写时另一个没法读。
所以如果生成的 tuple 越多，这里就会越卡(但是可以用环形队列 + 锁 方式改善)
当然还有一种方式是，在这个 Join Operator 上使用水平并行，然后通过垂直并行将它和这个 projection 操作整合在一起，即对于单个线程，高内聚它的任务；对于多个线程，低耦合它们的任务
### Bushy Parallelism
一个 inter-operator parallelism 的拓展版本
基本思路是，让不同的 worker 在同一时刻对一个查询计划的不同部分进行操作
这里依然使用 exchange operator 来在这些 operator 之间移动数据
比如上图这样的 4-way Join 查询
交给 worker 处理就是这样，它们会并行处理这些 fragment，将处理完的结果向上传递给 exchange operator，然后再通过其他 worker 对 exchange operator 后的结果进行处理
## I/O Parallelism
如果处理请求时，硬盘 I/O 瓶颈始终存在，那么就只能使用 I/O parallelism 了
基本思想是，对数据库系统的文件和数据进行拆分，将它们分散到储存设备上的不同位置处
有很多方法可以做到这一点，比如上图中列出来的
### Multi-Disk Parallelism
这有点像是 RAID
> RAID: Redundant Arrays of Independent Disks，即
独立磁盘构成的具有冗余能力的阵列
对系统进行配置，让多个存储设备以单个逻辑设备的形式来供数据库使用，可以用硬件也可以用软件来做这个事情
一个关于 RAID 的例子，也被叫做 Stripping(数据分条技术)
RAID 控制器会根据 Round-Robin 策略来决定该往哪个存储设备上写数据，在它内部会有一些元数据，这些数据保存了 page 在磁盘上的位置
另一种常见的做法就是做磁盘镜像(Mirroring)，就是每个存储设备上面都会保存一份该 page 的副本，可以通过某种纠错码或者其他方式，来确保在一个磁盘挂掉的情况下，还可以通过其他没挂掉的磁盘上保存的 page，进行重新创建该 page
### Database Partitioning
将数据库中的数据拆分为不相交的子集，然后将它们分配给离散的磁盘
同时可以设置软链接/硬链接，来让不同的目录(一个目录/文件代表一个数据库)指向不同的磁盘
log 文件保存的是对记录所做的所有的修改，如果有多个不同的存储设备，就需要对 log 文件进行分片
partition 的基本思想是，对于单个逻辑表，将该表的数据拆分成不相交的子集，然后将这些子集分别存放到不同的存储设备上，对它们进行管理
### Vertical Partitioning
这个垂直分区方式就是之前讲的列式存储方面的东西
假设有一个表，里面有 4 个属性
那么可以拿走其中该表中的一个属性，并将它保存在一个单独的分区中，然后这个分区，会以一个单独文件的形式，放在一个单独的磁盘或者存储设备上。虽然这种方式跟列存储很相似，但是，如果这些列中的值都是相同的，Vertical Partitioning 并不会对这列的数据进行压缩处理
### Horizontal Partitioning
假设有一个表，里面有 4 个属性
进行水平分区后，一个 tuple 中所有数据就会被放在一个分区中
## 结论
