本文转载自微信公众号《阿里技术》，作者姚玲。转载本文请联系阿里科技公众号。本文将深入介绍PolarDBMySQL在并行查询这一企业级查询加速特性上的技术探索、形态演变及相关组件实现原理。涉及的功能将随PolarDBMySQL8.0.2推出。一、背景1PolarDB云的兴起，给陈旧顽固的数据库市场带来了新的发展机遇。根据Gartner的预测，到2022年，75%的数据库将部署或迁移到云平台。云原生数据库的诞生，为各数据库厂商和云服务商提供了绝佳的弯道超车机会。如果你看看AWS在Re:invent2020上发布的Babelfish，就能闻到它在数据库市场的野心有多大。AWS在2017年发表的关于Aurora的论文[1]，引领了云原生关系型数据库的发展趋势。作为国内最早做云计算的厂商，阿里云也在2018年推出了自己的云原生关系型数据库，数据库PolarDB与Aurora的理念是一致的。PolarDB深度集成了云上的基础设施。目标是为客户提供特定于云的可扩展性、弹性和高可用性，同时具有更低的响应延迟和更高的并发性。吞吐量，其基本架构如下：底??层分布式共享存储突破了单机存储容量的限制，可以随着用户数据量的增加自动弹性扩展。计算层是典型的一次写入多次读取的拓扑结构。远程访问能力抵消计算和存储分离带来的额外网络开销。2挑战从上图可以看出，存储层会允许数据容量远大于单机（目前128T），甚至会出现部分用户在线，单表容量会达到的水平ofxxT。这是基于MySQL主从复制在传统部署中难以想象的。同时，大量用户会要求对业务数据进行实时分析，如统计、报表等，但大家对MySQL的直观印象是：小事务处理快、并发性强、分析能力强弱。对于这些实时分析查询，如何处理呢？3解决方案首先要说的是，随着互联网的发展和数据量的爆炸式增长，一定的数据分析能力和异构数据处理能力开始成为事务型数据库的标配。强化了自身的查询处理能力，包括子查询转换、hashjoin、窗口函数支持等。同时，PolarDBMySQL优化器团队也做了大量工作来提升复杂查询的处理能力，比如统计信息增强，子查询更多样化的转换，查询缓存等。并行查询（ParallelQuery）是PolarDBMySQL推出之初就自带的查询加速功能。本质上解决了一个核心问题：MySQL查询执行是单线程的，无法充分利用现代多核大内存硬件资源。.通过多线程并行执行减少包括IO和CPU计算在内的处理时间，响应时间大大降低。毕竟对于用户来说，如果1分钟10核就可以完成一个查询，那比10分钟1核更有意义。另外，所有成熟的商业数据库也都具备并行查询的能力。2并行查询介绍1特点并行查询可以说是PolarDBMySQL在计算层最重要也是最复杂的功能组件。随着PolarDB的上线，已经在线稳定运行多年，并不断演进。它具有以下特点：完全基于MySQL代码库，100%兼容原生MySQL，包括语法兼容、类型兼容、行为兼容。0额外费用。产品发布自带的功能不需要额外的存储资源，也不需要额外的计算节点。0维护成本。不同之处在于响应速度更快。集群部署，开箱即用实时分析单个配置参数（并行），不侵入业务，PolarDB原生部分，受益于REDO物理复制的低延迟统一底层事务数据提交也就是说，最终的表现是可以看到的。随着PQ的不断完善，对解析运算符和复杂查询结构的支持能力不断提升。所有运算符都是并行且高效的。复杂SQL结构支持，稳定可靠。作为企业级的特性，这无疑扩展了MySQL测试。系统多年在线积累了完善的诊断体系。以上这些听起来像是广告语，但确实是并行查询的核心竞争力。2进化并行查询功能不断积累。从最初的PQ1.0到PQ2.0，已经进入跨节点并行研发阶段，即将上线。这里不介绍跨节点并行能力，只介绍Followwhat'sonline。PQ1.0首次发布的并行查询能力的基本思想是将计算下推，将尽可能多的计算分配给多个worker并并行完成，从而可以同时进行IO等繁重的操作时间，但又不同于一般的share-nothing分布式数据库。由于底层共享存储，PolarDB并行中的数据分片是逻辑上的而非物理上的。每个工作人员都可以看到全量的表数据。下面介绍逻辑分片背后的执行者部分。典型的并行拆分计划形式如下：可以看到有几个特点：执行方式是简单的scatter-gather，即只有一个planslice，多个worker执行相同的功能，他们是尽可能聚合到leader去计算。leader负责完成不能从child推给worker的计算。该方案可以解决很多在线慢查询问题，达到很好的加速效果，但是也有一定的局限性。计划形式单一，导致算子并行化。方式单一，比如groupby+聚合，只能通过两阶段聚合完成：worker先进行部分聚合，leader进行最终聚合。一旦在leader上完成了聚合操作，如果有distinct/windowfunction/orderby等就只能在leader上做，形成单点瓶颈。如果存在数据倾斜，一些工作人员将无事可做，导致并行可扩展性差。在实现上还有一些需要改进的地方，比如少数算子不支持并行，还有一些复杂的查询嵌套结构不支持并行。总的来说，PQ1.0的并行形式和PostgreSQL社区的方案差不多，还有改进的空间。毕竟，所有商业数据库的并行形式都必须更加灵活和复杂。PQ2.0PQ2.0弥补了上述的局限性，在执行模式上对齐了Oracle/SQLServer，实现了更强大的多级并行。该计划的典型形式如下：首先看到的变化是有多个工作组。PQ2.0的执行计划是多阶段的，计划会被拆分成若干块（planslices）。Workers并行完成，通过交换数据通道在切片之间传递中间结果，触发后续切片的流水线执行。其中一些强化点包括：全新的Cost-based并行优化器，它根据统计信息和成本来确定最优计划形式。对所有算子的并行支持，包括上面提到的复杂的多层嵌套结构，也可以做完全并行化，引入exchange算子，即支持shuffle/broadcast等数据分发操作，引入一定的自适应能力。即使并行优化完成，也可以根据资源负载情况进行动态调整，如回滚串行或降低并行度。这些变化意味着什么？我们来看一个简单实用的例子：SELECTt1.a,sum(t2.b)FROMt1JOINt2ONt1.a=t2.aJOINt3ONt2.c=t3.cGROUPBYt1.aORDERBYt1.aLIMIT10;对于上面的简单查询，经过优化后，PQ1.0会生成图中的执行计划。在jointableset中，寻找可用于逻辑分片的表。如果三张表都不够分足够多的分片，那就选一张最多的。比如这里选择了t2，可能12个分片被拆分了，但是还是达不到16的并行度要求，导致4个worker因为读不到数据而空转。聚合操作首先对worker进行部分聚合，然后对leader进行聚合。如果每个worker上的group聚合不好，leader还是会从下面收到大量的group，leader还是会有繁重的聚合计算。如果leader计算慢了，就来不及接收到worker的数据，从而反压worker的执行速度，导致整体查询变慢。PQ2.0的执行计划如下。虽然数据分片只能在t2上进行，但是12个worker只需要完成t1joint2的操作即可。join完成后，一般数据量会膨胀。Shuffle(Repartition)会将更新的更多中间结果分发给后续的分片，从而以更高的并行度完成与t3的join。每个worker完成本地聚合后，如果还有很多group，可以根据groupbykey做一次Shuffle，将数据打散到下一层slice，下一组worker并行完成繁重的聚合操作，并且后续排序通过本地排序，最后leader只需要做归并排序的汇总，解决了单点瓶颈和数据量不足带来的可扩展性问题，实现了线性加速。为什么线性扩展如此重要？从上图可以看出，随着并行度的增加，E2E的响应时间线性下降，这对客户来说有两个重要的作用：随着业务的增长，数据不断扩大，通过相应增加并行度来使用匹配高度的计算资源，以持续获得稳定和可预测的查询性能。始终快速的分析时间可以推动快速的业务决策，使企业能够在瞬息万变的市场环境中保持竞争力。完美的线性加速是，ParallelRT=SerialRT/CPUcores，当然这不太现实3Architecture并行查询组件的整体架构如下核心部分包含在3层中，从上到下：Cost-basedParallelOptimizer，内嵌在MySQL优化器框架中，完成并行优化部分ParallelPlanGenerator，根据抽象的并行计划描述生成可被worker执行的物理执行计划ParallelExecutor，并行执行器组件，包括实现算子内部的并行函数、数据分发函数等具体组件的实现我会在后面详细介绍4。由于是个人文章，这里隐藏了具体的执行时间（可以上网搜索），主要看PQ2.0的查询加速能力，这里的并行度是32（有同学会疑惑为什么Q6/加速比Q12的超过32个，后面会详细讲）总个数为：100%的SQL可以加速，总加速比为18.8倍。5使用方法从易用性的角度，用户只需要设置一个参数就可以开启并行查询：setmax_parallel_degree=xxx;如果想查看并行执行计划，只需要像普通查询一样执行EXPLAIN/EXPLAINFORMAT=TREE即可。Explain对显示并行相关信息做了必要的增强，包括开销、并行模式、分配方式等三并行查询实现以上是一些通用的内容，没有任何技术细节，后面的章节会依次深入到各个模块介绍一下。1Paralleloptimizer在PQ2.0中，由于计划形式会变得更加多样化，如果拆分计划仅仅依靠简单的规则和简单的统计是很难得到最优解的，所以我们重新实现了一套基于代价的并行优化器。基本流程是MySQL串行优化后进一步进行并行拆分。说到这里，可能有同学会疑惑为什么没有像Oracle或者Greenplum那样集成，即在优化过程中统一考虑serial/parallel执行策略。究其原因，在MySQL的优化过程中，子步骤之间没有明确的界限，而其中嵌入的深度递归连接排序算法和半连接优化策略选择都使得代码逻辑和结构更加复杂，使得很难在不大量侵入本机代码的情况下实现集成优化。一旦社区代码损坏严重，将无法跟随社区后续版本迭代，享受社区红利。因此采用两步优化的流程，这也是业界常用的方法。例如Spark、CockroachDB、SQLServerPDW、Oceanbase等都采用了类似的解决方案。由于代价模型的增强是基于代价优化的，因此需要能够得到流程中各个算子并行执行的代价信息。为此，PolarDB也做了很多统计信息的增强工作：自动更新统计信息，为并行执行加强串行优化过程，比如修改扫表方式等。这就是为什么Q6/Q12在上述性能数据会有超线性加速比的原因是全算子统计信息的推导+成本计算，补充了一系列成本公式和基数估计推导机制。这里我们只能展示统计信息增强的效果。好处不仅是并行查询，还有串行执行。会改善。自适应执行策略在早期版本中，串行优化和并行优化、并行优化和并行计划生成之间存在一定的耦合，导致启动并行优化后无法退化回串行的问题。并发查询很多，会同时占用很多工作线程，导致CPU爆掉。新的并行优化器解决了这个问题。串行优化与并行优化分离。并行优化将重建抽象运算符树，并以此为输入开始枚举。并行优化和并行计划生成是分离的。优化的结果是plan子片段的抽象描述，作为plan的输出这样generation使得执行策略的灵活性成为可能，要么回归串行化，要么降低并行度，或者在资源不足的情况下进入调度队列排队资源。基于成本的穷举是一个比较大的话题。一般来说，并行优化是一个基于动态规划的自下而上的穷举过程。实现思路参考了SQLServerPDW论文[2]，在这个过程中，针对每个算子，枚举可能的并行执行方式和数据分布方式，并根据其物理属性（分布+顺序）构造物理等价类输出数据，从而进行局部剪枝，得到局部子问题的最优解传递给上层，最终由根算子得到全局最优解。下图是对算子t1NLJt2的枚举过程的简单示例：整体枚举完成后，会在计划空间中生成一系列数据分布为ExchangeEnforcer的物理算子树，其中最优的算子树将根据成本进行选择。可以使用excellenttree，然后以Enforcer作为子计划的切分点，构建一系列执行计划的抽象描述输出给计划生成器。2并行计划生成从工程实现的角度来说，并行计划生成可以说是整个组件中复杂度最高、陷阱最多的部分。这里采用了物理计划克隆的机制，即根据优化器生成的并行计划描述，从原来的串行计划中克隆出每个计划分片的物理执行计划。为什么要使用这种方法？还是跟MySQL本身的机制有关。MySQL优化和执行是耦合在一起的，没有明确的界限，即在优化过程中构建相关的执行结构。因此，没有办法直接根据一个独立的计划描述来构造每一个物理执行结构，只能从串行计划中“克隆”出来，这可以说是一切复杂性的根源。MySQL的执行结构非常复杂。表达式（Item）和查询块（SELECT_LEX）的交叉引用，内部查询和外部查询（Item_ref）的关联等等，都让这个任务变得更加困难。在这个过程中，团队也深入了解了MySQL优化后的执行结构，在社区中发现了很多bug……以上图中的简单查询为例SELECTt1.a,sum(t2.b)sumbFROMt1joint2ONt1.c=t2.cGROUPBYt1.aORDERBYsumb;虽然社区基于Iterator模型对executor进行了重构，但本质上，物理执行计划仍然是一个由QEP_TAB组成的序列，其中groupby+aggr由一个tmptable1完成，orderby由tmptable2完成。在做plangeneration的时候，有两个核心操作：clone根据串行物理plan和sub-slice的描述，克隆相应的结构到每个工作线程，如上图右下部分，执行的t1join在worker上克隆了t2和下推的聚合操作。refix原来的serialplan需要转换成leaderplan，所以需要替换掉不需要的执行结构，调整一些引用关系，如上图右上部分所示，由于t1joint2和一些聚合操作已经下推，leader需要移除不必要的结构，取而代之的是从一个收集器表中读取worker传递过来的数据。同时需要将后续步骤中引用的t1/t2表的结构转换为引用collector表的对应结构。这里只是举个最简单的例子，不涉及子查询和多阶段计划，实际项目实施成本要高很多。3并行执行器PQ在算子内部实现了一系列的并行机制，如表的逻辑分区和并行扫描、并行hashjoin等，使并行执行成为可能或进一步提高性能，以及多样化的子查询处理机制等.,这里介绍几个比较有代表性的。parallelscanPolarDB是共享存储，所有数据对所有节点可见，不同于分片的分布式系统，无法预先确定不同worker处理的是哪部分数据，所以采用了逻辑分区方案：在btree层次，会有把数据分成很多小的分片，不同的worker负责不同的分片来触发并行执行。这里有一些优化点：尽量做细粒度的分片，让分片的数量>>worker的数量，然后worker之间通过roundrobin的方式来“抢”shards来执行，这样自然就实现了更多的work对于那些可以的，并避免了由倾斜的数据分布引起的负载不平衡。这是共享存储系统的天然优势。拆分时，不必潜入叶子节点，即以页为最小分区单位，加快初始分区速度。ParallelhashjoinHashjoin是Community8.0引入的加速分析查询的功能。随着版本的演进，支持semihash/antihash/lefthashjoin。PolarDB也引入了这些patch，实现了完整的hashjoin功能，实现了多种并行执行策略。Parallelhashjoin支持build/probe两个阶段的build阶段，多个worker将数据插入到同一个共享的无锁哈希表中。在探测阶段，多个工作人员并行搜索哈希表。两个阶段之间没有重叠，从而实现了整个阶段的并行，但是并行hashjoin也有自己的问题，比如共享哈希表过大导致spilltodisk问题，虽然并行insert是无锁的，仍然有“同步”原语带来缓存失效。Partitionhashjoinpartitionhashjoin可以避免上述问题，但代价是引入了datashuffle的开销：如图所示，查询的执行过程分为三个阶段。build/probe双方根据joinkey做shuffling，将数据分发到目标分区；在每个分区中，构建端构建一个小哈希表；在每个分区中，探测端搜索对应的哈希表；这样co-locatedjoin是在每个分区完成的，每个hash表都比较小，避免了磁盘放置，构建时不存在并发问题。以上两种方案哪个更好？并行优化器根据成本做出决定。子查询并行-pushdownexec这里的子查询是表达式的一部分，可以存在于selectlist/where/having等子句中。对于关联子查询，唯一的并行化方式是将外层依赖的数据（表）下推给worker，在每个worker中完整执行。但是因为外层是并行化的，每个worker中子查询的执行次数还是可以等比例减少的。例如，以下查询：SELECTc1FROMt1WHEREEXISTS(SELECTc1FROMt2WHEREt2.a=t1.a<=EXISTSsubquery)ORDERBYc1LIMIT10;EXISTS子查询完全克隆到每个工作人员，并重复执行WHERE条件触发器的评估。子查询并行性-下推共享此并行子查询可以是表达式或派生表的一部分。粗略地说，这种并行方式适用于不相关的子查询，因此可以提前并行物化，形成一个临时结果表。后面的外层是并行的，每个worker在引用子查询的时候可以直接并行的从表中读取。得到结果数据。例如，以下查询SELECTc1FROMt1WHEREt1.c2IN(SELECTc2FROMt2WHEREt2.c1<15<=INsubquery)ORDERBYc1LIMIT10;另外，在在线用户的报表查询中，一种非常常见的Query方式是派生表的多层嵌套，对于这种类型的SQL，下推共享策略可以提高并行执行的性能，例如下面的例子：上图中每个彩色方块代表一层查询块，构成了多层派生表的嵌套逻辑，有的层是通过UNIONALL进行汇总，有的层是多个表（包括派生表）的join。对于这样的查询，MySQL会对每一个派生表做必要的物化，在外层形成一个临时的结果表来参与后续的计算，而PQ2.0对这种常见的查询方式提供了更通用的支持，现在每一层的查询执行是并行完成的，力求达到线性加速的效果。要为Exchange生成高效灵活的执行计划，数据分发组件必不可少。目前PolarDB支持三种分发方式：Shuffle/Broadcast/Gather，采用无锁共享环形缓冲区，以管道方式实现高效的数据传输。.下图是Shuffle(Repartition)的基本形式。到这里，已经大致介绍了在线版并行查询的功能和实现。作为成熟的企业级功能特性，团队也实现了一套完整的辅助工具来提高产品的易用性，实现可监控、可干预、可反馈的功能，但这里的篇幅已经太大了，我先不介绍了。4.未来规划在这里，未来规划并不准确，因为团队在跨节点并行方面做了很多工作，进入了开发周期的末期。跨节点并行将海量数据的复杂查询能力提升到另一个层次。层次：打通节点间的计算资源，实现更高的计算并行度，突破单节点在IO/CPU上的瓶颈，充分利用分布式存储结合全局节点管理和资源视图的高吞吐能力，平衡调度全局计算资源实现负载均衡的同时保证查询性能结合全局一致性视图，保证事务数据的正确读取[1]https://www.allthingsdistributed.com/files/p1041-verbitski.pdf[2]https://www.scinapse.io/papers/2160963784#fullText
