海山数据库（He3DB）原理剖析：浅析OLAP数据库计算引擎中的统计信息

背景：

统计信息在计算引擎的优化器模块中经常被提及，尤其是在基于成本成本优化（CBO）框架中统计信息发挥着至关重要的作用。CBO旨在通过评估执行查询的可能方法，并选择最有效的执行计划来提高查询性能。而统计信息则提供了关于数据分布、数据倾斜等方面的关键信息，帮助CBO做出最优的决策。无论是传统数据库MySQL、PostgreSQL，还是Hive、Spark计算引擎、Doris、StarRocks 等OLAP引擎，都针对CBO模块做了大量开发工作。而统计信息作为影响CBO决策的最重要因素，计算引擎需要对统计信息的搜集、加工、利用多个阶段进行打磨优化，最终帮助用户业务提供成本最低的执行计划。

本文将会从统计信息的常见来源以及计算引擎如何利用统计信息多个方面着手，综合多个计算引擎的CBO统计信息框架优化案例，浅析计算引擎的CBO统计信息。

图1 计算引擎的优化器模块

一、统计信息的种类与来源

综合多个计算引擎的统计信息集成情况，可以发现大家获取统计信息手段不尽相同，一种是精准搜集统计信息，一种是采用多种手段对数据进行评估获取统计信息。

常见的基本统计信息类型有min/max/ndv/totalSize/fileNum，各家计算引擎会根据引擎特点继续扩展不同的统计信息，如涉及到数据整体评估的统计信息，MVC（高频非NULL值）、HiSTOGRAM（直方图）等。

1、基本类型统计信息

基本统计类型是相对精准的信息，一般用户analyze语句获取的统计信息多数是这种min/max等基本信息。一般表级别、分区级别、字段级别都会有相应的统计信息。在不同的SQL业务中，计算引擎会根据SQL中的谓词语句、table scan的具体表来分别利用哪种统计信息。

如select count(*) from testtbl; 显然这条语句会用到表级别统计信息，select count(distinct id) from testtbl where date=2024; 这条语句一般就会用到分区date以及字段id的统计信息。

基本类型统计信息一般都是实时获取+定时任务获取。如大数据中Hive每次执行写入任务之后会有线程启动搜集写入数据的基本统计信息，一定程度上保证统计信息实时准确，StarRocks会有定时任务自动执行内表的统计信息搜集。

2、估算类型统计信息

估算类型统计信息一般泛指对数据集的一个整体评估，能够在牺牲一定的准确性的情况下给出数据集的一个分布情况，常见的如MVC。这种统计信息适合对于在大规模数据集下，基本统计信息搜集相对代价较高且基本统计信息缺乏对整体数据集的极端分布（如数据倾斜）的体现，而估算类型的统计信息能够告诉计算引擎数据的分布状况，能够使任务避免数据倾斜等情况。

一般评估数据分布的统计信息都会采用histogram直方图，多数计算引擎根据自身的业务特性实现了不同的直方图。直方图的基本原理是将数据排序后分成若干个bucket桶，并记录每个桶中数据的最大值、最小值、频次出现等信息。常见的直方图有Equal-width Histogram、Equi-height Histogram等。像Doris和StarRocks，均实现了Equi-height Histogram。Equi-width Histogram（等宽直方图）是将数据最大、小值之间的区间等分为N份，每个桶中最大、小值之差都为整体数据最大、小值之差/N，既所谓“等宽”。

Equi-height Histogram（等高直方图），它的桶宽度并不相等，取而代之的是，等高直方图会保证每个桶中数值的频次之和接近总行数的 1/N，就是落入每个桶里的值数量尽量相等。数据数据分布范围比较大时也可以很好的保证误差。各种计算引擎会根据其擅长的业务特性去改进直方图，以尽可能避免直方图落入局部最优的境地，这里不过多详细解释直方图实现原理，大家可以参考直方图的基本原理以及具体计算引擎实现去做细致研究。

一般情况下，估算类型统计信息能够相对准确、相对高效的应用在一些大数据量的表上，尤其是一些数据湖的大表（PB级别），但是一般的估算统计信息实现还是需要对数据做一遍整体扫描，所以多数情况下的实现会判断大小表来决定是否整体数据评估信息搜集还是采样搜集。判断是否整体还是采样搜集各家计算引擎都会有不同的规则，如StarRocks、Doris（[Enchancement](statistics) Support sampling collection of statistics by weizhengte · Pull Request #18880 · apache/doris · GitHub）搜集histogram直方图统计信息支持设置最大行数采样、根据比例进行采样；也有的利用统计学方式如伯努利采样PrestoDB（Add reservoir_sample aggregation function by ZacBlanco · Pull Request #21296 · prestodb/presto · GitHub）。

apache datasketches是另一种可用的非常高效的数据分布计算工具，目前基于apache datasketches算法库实现的直方图统计信息如Hive4.0，草图中实现了各种近似估算方法，很多方法都来源于数据库领域的论文算法。Sketch 结构即「数据草图」结构，主要是为了计算海量的流式数据的概率指标而设计的一种数据结构。

一般占用固定大小的内存，不随着数据量的增加而增大。这种结构通过巧妙地保存或丢弃一些数据的策略，将数据流的信息抽象存储起来，汇总成 Sketch 结构，最终能根据 Sketch 结构还原始数据的分布，实现基数统计、分位数计算等操作。

Spark3.5之后也使用apache datasketches实现了一些估算函数如ndvApache Spark ❤️ Apache DataSketches: New Sketch-Based Approximate Distinct Counting | Databricks Blog ，基于sketch的估算函数能有效加速Spark对大数据集的一些数据分布估算，比如在此之前，统计数据集中的ndv，一般spark用户都会count(distinct)，这种计算十分耗时，而且计算结果不会有临时存储，每次都需要重新计算，而datasketch能够利用经过验证的统计学算法，快速的返回计算结果，结果还可以持久化到sketch的存储中，能够大大加速一些统计类型查询。

二、统计信息带来的常见收益

我们常常说统计信息能够加速查询，能够优化执行计划，那么从计算引擎角度来说，统计信息利用最多的地方有哪些呢？这里列举几个关键的点，我们可以从不同的计算引擎中了解统计信息是怎样给用户的业务带来加速。

1、join选择

join能力是考量OLAP引擎的关键指标。如何在复杂的SQL语句中找到优化的join方式是CBO优化要做的事情。分析中常见的hash join，涉及到大小表join，一个关键的因素是怎么判断表的大小，最直接的指标就是表的统计信息，优化器根据表大小，把小表作为build side来构造哈希表放入内存，大表作为probe side，这样可以有效避免数据的shuffle过程，主流的计算引擎都会支持这种高效的join方式。

还有join reorder优化，经常会根据计算过程中生成的临时统计信息对执行计划动态调整，修改join算法，简而言之，join优化的基本要素就是需要有相对准确的统计信息，最直接的统计信息如rowCount判断表大小。计算引擎一般利用这些基础的统计信息再结合一些reorder算法或者自定义的规则，完成join查询的最优执行路径选择。

2、自适应任务执行

Adaptive Query Execution即AQE也是计算引擎高阶优化经常谈到的一个点。AQE执行可以理解为动态CBO，可以根据运行期的一些临时数据的统计信息，动态调整CBO选择的执行路径。典型的一个是Spark AQE，其根据在运行时统计信息（runtime statistics）在查询执行的过程中进行动态（Dynamic）Spark的查询优化，AQE可以Spark运行query stage阶段准确获取统计信息，然后进行CBO优化剩余的stage，可以有效的动态合并Spark shuffle分区，避免join阶段的一些数据倾斜问题。

无独有偶，除了Spark AQE，其他计算引擎也都有很多类似AQE优化。（个人理解AQE优化一般针对中间数据有落盘的计算过程，如上面提到的Spark（shuffle阶段），所以可以推测其他有中间数据可以物化/落盘行为的计算引擎也可以去做这种优化。）。

TrinoDB近年来增强了自身的容错计算能力，即设计了中间shuffle数据落盘的一种计算模式（fte），可以在部分task运行失败时从磁盘中恢复中间执行数据然后重新执行，TrinoDB的这种fte模式很适合使用AQE优化，用于减少运行期启动不必要的task，如Adaptive planning framework in FTE by gaurav8297 · Pull Request #20276 · trinodb/trino · GitHub 就是利用运行期统计信息做一些自适应优化；

PrestoDB虽然没有fte执行模式，但是其曾经也做过一些中间数据物化以提高task容错的开发如[Design] Exchange Materialization · Issue #12387 · prestodb/presto · GitHub ，其思想会把中间数据物化成一个临时表供下游task消费，那么很显然的优化就是获取这个中间表的统计信息来对下游的CBO执行计划做动态自适应调整Initial Support of Adaptive Optimization with Presto Unlimited by pguofb · Pull Request #14675 · prestodb/presto · GitHub。

类似的，Hive3其实就有AQE优化，核心思想缓存中间运行期的统计信息，动态修正CBO执行计划、动态调整分区裁剪优化等[HIVE-17626] Query reoptimization using cached runtime statistics - ASF JIRA 。所以，一旦清楚了AQE思想，每一种计算引擎都可以根据自己运行期的统计信息特点做进一步动态优化，给与业务最好的加速体验。

3、聚合下推优化

计算引擎中的聚合算子如sum、count是相对比较消耗计算资源的操作，常规执行逻辑就是扫描数据的每一行来进行各种加减操作。但是如果已经搜集了存储表的统计信息如rowCount，那么像这种count算子就是一个O(1)的简单元数据操作，计算引擎不需要计算直接返回已经搜集的统计信息即可。

这种聚合下推的优化在各个计算引擎中基本都有实现，尤其是针对底层存储采用Parquet/ORC这种开发式列存的文件格式（如Iceberg的metadata文件就记录了详细的Parquet/ORC统计信息），如Spark利用Iceberg的统计信息，做一些下推的优化操作，如TrinoDB也做了类似的基于统计信息的聚合下推操作优化Add aggregation pushdown support for count using Iceberg Metrics by osscm · Pull Request #15832 · trinodb/trino · GitHub 。

当然，Hadoop之上经典的Hive计算引擎也早就有这种聚合下推优化，比如有些Hive优化参数会控制是否启动MR分布式任务，如参数hive.compute.query.using.stats，该参数开启的情况下，Hive计算引擎会去判断当前表的统计信息rowCount是否最新，如果统计信息最新，则在SQL语句中涉及到count的操作算子直接通过统计信息返回，避免了启动分布式任务去计算。