[GitHub] [doris] Kikyou1997 commented on issue #13628: statistics persistence

[GitBox <gi...@apache.org>]

Kikyou1997 commented on issue #13628:
URL: https://github.com/apache/doris/issues/13628#issuecomment-1290089018

   # 统计信息设计文档
   
   - [统计信息设计文档](#统计信息设计文档)
     - [持久化](#持久化)
       - [表更新](#表更新)
       - [统计信息粒度](#统计信息粒度)
     - [缓存](#缓存)
     - [收集](#收集)
       - [SQL](#sql)
     - [自动收集](#自动收集)
     - [TODO](#todo)
   
   ## 持久化
   
   目前主流的方案是将统计信息放在单独的表中，Postgres，MySQL，StarRocks等皆如是。我门也将遵循这个设计。
   
   Doris兼容了MySQL5.1, 在`information_schema`库中保有`statistics`表，但其字段与我们的需求有较大的gap，且Doris不对该表进行维护，该表永远为空。另外Doris对该库做了很多特殊的处理，我们不希望这些逻辑对统计信息的代码构成干扰。
   
   因此，将使用一个独立的库表用以保存统计数据，这样维护起来以较为方便也方便用户和开发者对已有的数据做进一步的聚合查询。
   
   库名：`internal_statistics`
   
   表类型：`OLAP`
   
   表副本数： 1，`fe.conf`增加配置项`statistics_replica_num`
   
   主要有如下几张表：
   
   表名：`column_statistics`
   
   bucket col：`id`
   
   unique key: `id`
   
   表字段及类型：
   
   |字段名|类型|Nullable|Comments|
   |--|--|---|--|
   |id|VARCHAR(4096)|false|由`catalog_id`  `db_id`  `tbl_id`  `col_id`  `part_id`拼接而成，分隔符`-`|
   |catalog_id|VARCHAR(1024)||
   |db_id|VARCHAR(1024)|n|
   |tbl_id|VARCHAR(1024)|n|
   |col_id|VARCHAR(1024)|n|名为id实则name|
   |part_id|VARCHAR(1024)||为空表示该row为表级别数据|
   |row_count|BIGINT||
   |ndv|BIGINT||
   |null_count|BIGINT|
   |min|VARCHAR(ScalarType.MAX_VARCHAR_LENGTH)||
   |max|VARCHAR(ScalarType.MAX_VARCHAR_LENGTH)||
   |data_size|BIGINT||
   |update_time|DATETIME||数据插入时间|
   
   注意，在这里，对于列上的最大最小值我们用字符串记录其原始值，在从表中load时转换为double类型。
   
   
   ### 表更新
   
   此外，由于存在列名被修改的情况，我们需要在相关的alter语句执行时，对`column_statistics`中的记录执行相应的modify操作。但这里如果在列的alter语句执行时，这个操作一并阻塞执行的话，那么存在潜在的性能问题，如果是异步的操作，则存在一致性问题。如果更新不及时，还存在拿到错误的统计信息的可能，比如A列被重名为B， C列被重命名为A的情况。
   
   结论：
   
   这种情况提交一个作业异步的删除旧的统计信息，由于不太可能对表schema的高频且持续不断的修改，并且统计信息本身的收集本身就必然存在滞后性，一定时间窗口内的部分的不一致是可以接受的，而通过周期性的收集任务我们可以保证表中列的统计信息和实际的列最终一致。
   
   ### 统计信息粒度
   
   
   就前者而言，当SQL中存在下推到OlapScan的等值条件`slot=constant`，且左侧为distributed column时，Doris会根据hash值对tablet进行裁剪，统计每个tablet的信息有助于我们在这种条件下更准确的估计scan出的rowCount，但如果将列的统计信息的收集拆分到`TABLET`的级别的话，如何合理的估计distinct是个问题，
   
   考虑到一些用户会设置bucket数量为一个非常大值，如果再乘以partition数量，那么每张表上将需要执行非常多的任务去收集统计信息。这样
   
   1. 一方面会对其他逻辑构成较大影响
   2. 另一方面也意味着`column_statistics`表的过分膨胀。
   3. FE任务调度的压力也更大
   4. 基于bucket粒度的统计信息 聚合估计出的ndv偏差更大
   
   class：`StatisticsStorageManager`
   
   method：
   
   * saveStatistics：在StatisticJob的状态被更新为true时，调用
   * removeExpiredStatistics：周期性任务。由于统计信息本身不会占用太多的资源，对一致性要求也不高，我们不必追求及时删除。这里有一个周期性的任务检查statistics表中记录的库表列和目前元数据中保有的之间的diff，然后起一个SQL把statistics中多余的记录删掉即可。
   * createStatisticsTableIfNotExists：FE启动时检查
   
   ## 缓存
   
   我们引入一个缓存层，这样的做的目的是
   
   1. 在统计信息收集和统计信息查询之间引入一个中间层，这样之后对前者进行重构时，不会对其他部分产生影响，同时屏蔽了收集过程中的各类状态，使得代码结构更为清晰易懂
   2. 另外，目前除非用户显式的执行删除操作，否则已经收集到的统计信息会永远驻留在FE内存中，在拥有持久化能力之后，可以将长期不用的统计信息淘汰掉减少FE内存占用
   
   各个操作基本设计如下：
   
   * 载入：按需从statistic表中查询，如果没有相应的统计数据，返回一个默认的值
   * 更新：每次执行`ANALYZE`语句，或`ANALYZE JOB`周期性执行时
   * 删除：
   
         1. 过期（超过指定时间未被access, ，可通过`information_schema_stats_expiry`配置），或相应的db/table/column被删除时删除对应的cache。
         2. cache条目达到上限,可通过参数`statistics_max_cache_size配置`
   
   * 刷新时间：一天，从statistics表重新load数据
   * 淘汰策略：LRU
   * 多FE同步：Doris常部署多个FE提供query服务，这里我们不通过额外的通信来保证各个FE间cache的统计信息的一致性，通过定时的刷新机制来保证cache中的数据不会是严重过时的
    
   
   class: `StatisticsCache`
   
   method:
   
   * getStatistics(dbid, tblid, colid)
   * tryToLoadStatitistics(dbid, tblid, colid)
   * eraseEntry(bid, tblid, colid)
   
   Doris中大量使用了Google Guava库，我们将基于该库提供的`LoadingCache`结构实现缓存能力，[Guava LoadingCache 相关文档](https://github.com/google/guava/wiki/CachesExplained)
   
   
   ## 收集
   
   目前的实现中，将每个列的统计信息的收集拆分为了多个单独的SQL去执行，同时为了避免统计信息收集占用过多资源，如果队列中的Job数量超过20个，用户执行`ANALYZE`语句时会抛出异常。这么做的问题有：
   
   1. 用户体验较差，用户执行ANALYZE语句不应因为这样粗糙的推测可能负载较高而直接报错
   2. 代码较为复杂，不便维护和扩展。比如，如果由于每个ColumnStat的统计项都有一个单独的SQL去收集，那么，我们要么在每次一项执行完后，执行一次insert，要么引入同步机制，等到所有的统计项收集完毕后再执行insert进行持久化。此外，为StatisticJob定义了，`FINISHED` `FAILED`等状态，如果Job中的任意一个task失败了就认为Job执行失败了，但是该Job中由其他task完成收集的统计项依然能够被查询到。
   
   重构计划：
   
   1. 删除`TABLET`这一级的收集操作
   2. 移除任务拆分逻辑，改为仅基于partition做收集
   3. 统一为使用 SQL查询各项统计信息
   4. 通过SQL粒度的会话变量限制SQL执行开销
   5. 不再对聚合表和明细表进行区分，统一用SQL收集行数
   6. 修改`TableStat` `ColumnStat`相关字段为`final`
   7. 在`fe.conf`中添加变量`statistics_max_parallism`控制同时可执行的SQL数量，默认为10。
   8. 统一命名：单个对象的统计信息使用`XXXStatistic `如`ColumnSatistic`，抽象表示整个统计信息框架，或聚合了多个对象的统计信息类的，用`Statistics`
   
   ### SQL
   
   partition统计信息SQL定义如下：
   
   ```sql
       private static final String ANALYZE_PARTITION_SQL_TEMPLATE = "INSERT INTO "
               + StatisticConstants.STATISTIC_DB_NAME + StatisticConstants.STATISTIC_TBL_NAME
               + " SELECT "
               + "CONCAT($catalogId, $dbId, $colId, $partId) AS id, "
               + "COUNT(1) AS row_count, "
               + "NDV($columnName) AS ndv, "
               + "SUM(CASE $colName IS NULL THEN 1 ELSE 0) AS null_count, "
               + "MAX($colName) AS max, "
               + "MIN($colName) AS min, "
               // SUM(LENGTH($colName)) as data_size
               + "$dataSizeFunction AS data_size, "
               + "MAX(LENGTH($colName)) AS max_len "
               + "FROM $tblName "
               + "PARTITION $partition_name";
   ```
   
   data_size的估计上
   
   * 字符类型，我们主要有两个函数可用：`CHAR_LENGTH` `LENGTH`，前者返回字符数，后者返回字节数，我们暂定采用后者，从BE实现上来看，后者更快。且后者返回真实字节数更贴合我们的需求
   * 数字类型则直接用类型长度乘以改列行数即可
   
   对于ColumnStat中的avgRowSize可通过 $data_size / row_count$ 得到，理想上ndv则通过对partition的统计信息的hll类型的ndv做`hll_union_agg`得到，但实际上，由于HLL类型目前在Doris上只有聚合表可以用，而且HLL类型对应的聚合类型必须是HLL_UNION，不能是REPLACE，那么更新该表数据时，我们就不得不先删除旧的记录，因此只能将其保存为BIGINT，故，我们只能通过一个单独的子句去完成表级别上的列的ndv的估计。SQL模板定义如下：
   
   ```sql
       private static final String ANALYZE_COLUMN_SQL_TEMPLATE = "INSERT INTO "
               + StatisticConstants.STATISTIC_DB_NAME + StatisticConstants.STATISTIC_TBL_NAME
               + "SELECT "
               + "CONCAT($catalogId, $dbId, $colId, $partId) AS id, "
               + "SUM(row_count) AS row_count, "
               + "NDV(col_ndv) AS t.ndv, SUM(null_count) AS null_count, "
               + "MAX(CAST max AS $type) AS max, "
               + "MIN(CAST min AS $type) AS min "
               + "FROM internal_statistics.column_statistics, "
               + "(SELECT NDV($colName) "
               + "FROM $tblName) AS t "
               + "WHERE internal_statistics.column_statistics.db_id = $dbId and "
               + "internal_statistics.column_statistics.tbl_id=$tblId";
   ```
   
   通过如下变量限制每条SQL的资源占用：
   
   * exec_mem_limit 默认2G
   * parallel_fragment_exec_instance_num 默认1
   
   SQL超时时间
   
   * query_timeout 默认5min
   
   ## 自动收集
   
   当前没有语法去专门的创建周期性的收集任务，也没有必要。
   
   `fe.conf`中增加如下可配置参数：
   
   * enable_full_statistics_collection：配置是否对自动所有表（除系统内部内部表外）进行analyze，默认为false，只对用户执行过`analyze`语句的表进行周期性的重新analyze。
   
   为此需要
   
   1. 将analyze语句中的信息封装为AnalyzeJob并保存在内部表`analyze_job_information`中。
   2. 允许用户删除指定的AnalyzeJob　语法：DROP ANALYZE
   
   class: AnalyzeJobManager
   
   method:
   
   * constructAnalyzeJob
   * replayCreateanalyzeJob
   * removeAnalyzeJob
   * replayCreateanalyzeJob
   * getAnalyzeJob
   * updateJobStatus
   
   库：`information_schema`
   
   表：`analyze_job_information`
   
   类型：unique
   
   |字段名|类型|Key|Nullable|Comments|
   |--|--|---|--|---|
   |job_id|BIGINT|Y|n||
   |db_id|BIGINT||n|
   |tbl_id|BIGINT||n||
   |col_name|VARCHAR(64)||n|
   |db_name|VARCHAR(64)||n|
   |tbl_name|VARCHAR(64)||n|
   |status|VARCHAR(32)||n|PENDING/RUNING/FAILED/FINISHED|
   |type|VARCHAR(32)||n|SAMPLE/FULL|
   
   ## TODO
   
   支持如下场景
   
   * ~~云原生~~ 暂不考虑
   * 外表
   * 复杂类型
   * 统计数据增量更新
   * 日后支持列的id后，存量表如何处理？
   * BE支持SQL优先级后，调低统计信息相关SQL的优先级
   * 为Doris内部database规定一个统一前缀，禁止用户创建/修改
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org