StarRocks关于ConcurrentModificationException 问题的解决

背景

本文基于 StarRocks 3.1.7
目前在基于Starrocks做一些数据分析的操作(主要是做一些简单的查询)，同事遇到了一些并发的问题:

ontent:2024-11-27 07:04:34,048 WARN (starrocks-mysql-nio-pool-214933|3593819) [StmtExecutor.execute():643] execute Exception, sql SELECT distinct(id) FROM `db`.`table` WHERE col1='xxx' AND col2='xxx'
java.util.ConcurrentModificationException: nullat java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:719) ~[?:?]at java.util.LinkedHashMap$LinkedValueIterator.next(LinkedHashMap.java:746) ~[?:?]at com.starrocks.sql.optimizer.statistics.StatisticsCalcUtils.deltaRows(StatisticsCalcUtils.java:176) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.statistics.StatisticsCalcUtils.getTableRowCount(StatisticsCalcUtils.java:114) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.statistics.StatisticsCalculator.computeOlapScanNode(StatisticsCalculator.java:257) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.statistics.StatisticsCalculator.visitLogicalOlapScan(StatisticsCalculator.java:225) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.statistics.StatisticsCalculator.visitLogicalOlapScan(StatisticsCalculator.java:161) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.operator.logical.LogicalOlapScanOperator.accept(LogicalOlapScanOperator.java:149) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.statistics.StatisticsCalculator.estimatorStats(StatisticsCalculator.java:177) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.task.DeriveStatsTask.execute(DeriveStatsTask.java:57) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.task.SeriallyTaskScheduler.executeTasks(SeriallyTaskScheduler.java:69) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.Optimizer.memoOptimize(Optimizer.java:595) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.Optimizer.optimizeByCost(Optimizer.java:201) ~[starrocks-fe.jar:?]at com.starrocks.sql.optimizer.Optimizer.optimize(Optimizer.java:134) ~[starrocks-fe.jar:?]at com.starrocks.sql.StatementPlanner.createQueryPlan(StatementPlanner.java:146) ~[starrocks-fe.jar:?]at com.starrocks.sql.StatementPlanner.planQuery(StatementPlanner.java:121) ~[starrocks-fe.jar:?]at com.starrocks.sql.StatementPlanner.plan(StatementPlanner.java:92) ~[starrocks-fe.jar:?]at com.starrocks.sql.StatementPlanner.plan(StatementPlanner.java:61) ~[starrocks-fe.jar:?]at com.starrocks.qe.StmtExecutor.execute(StmtExecutor.java:456) ~[starrocks-fe.jar:?]at com.starrocks.qe.ConnectProcessor.handleQuery(ConnectProcessor.java:392) ~[starrocks-fe.jar:?]at com.starrocks.qe.ConnectProcessor.dispatch(ConnectProcessor.java:506) ~[starrocks-fe.jar:?]at com.starrocks.qe.ConnectProcessor.processOnce(ConnectProcessor.java:782) ~[starrocks-fe.jar:?]at com.starrocks.mysql.nio.ReadListener.lambda$handleEvent$0(ReadListener.java:69) ~[starrocks-fe.jar:?]at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]at java.lang.Thread.run(Thread.java:829) ~[?:?]

涉及到的表如下：

CREATE TABLE `table` (`id` bigint(20) NOT NULL ,`create_date` datetime NOT NULL ,`col1` varchar(64) NOT NULL ,`col2` varchar(20) NOT NULL
) ENGINE = OLAP PRIMARY KEY(`create_date`, `id`
) COMMENT "" 
PARTITION BY date_trunc('month', create_date) DISTRIBUTED BY HASH(`id`) PROPERTIES ("replication_num" = "3", "in_memory" = "false", "enable_persistent_index" = "true", "replicated_storage" = "true", "partition_live_number" = "18", "compression" = "LZ4");

结论

StarRocks 对分区带有TTL的表，会后台启动线程轮询的去删除分区，轮询的间隔受到 dynamic_partition_check_interval_seconds 控制，
然而在查询的时候, Starrocks会做语法解析，以及基于CBO的优化，在这期间会统计涉及到的表的分区信息统计，而此时恰好遇到了后台线程的分区删除，导致了ConcurrentModificationException 并发异常。
目前可以参考这个issue ConcurrentModificationException when query during drop partition,以及 增大 dynamic_partition_check_interval_seconds（目前是10分钟） 这个参数来降低这种冲突的概率。

分析

这里主要涉及到两个部分:
一个部分是查询部分(主要是StmtExecutor.execute)，一个是后台清理分区部分(DynamicPartitionScheduler)

查询部分

每个SQL查询都会经过 StmtExecutor.execute方法，进而生成物理执行计划，而在生成物理执行计划的阶段，会经过Optimizer 阶段，这个阶段由于默认情况下是基于CBO的优化，所以会统计涉及的表所扫描的数据量,最终会走到 StatisticsCalcUtils.deltaRows方法：

private static long deltaRows(Table table, long totalRowCount) {long tblRowCount = 0L;for (Partition partition : table.getPartitions()) {long partitionRowCount;TableStatistic tableStatistic = GlobalStateMgr.getCurrentStatisticStorage().getTableStatistic(table.getId(), partition.getId());if (tableStatistic.equals(TableStatistic.unknown())) {partitionRowCount = partition.getRowCount();} else {partitionRowCount = tableStatistic.getRowCount();}tblRowCount += partitionRowCount;}if (tblRowCount < totalRowCount) {return Math.max(1, (totalRowCount - tblRowCount) / table.getPartitions().size());} else {return 0;}}

这里会对 table.getPartitions 进行 迭代，也就是OlapTable的 idToPartition.valus 进行迭代, 注意 idToPartition 是HashMap类型的，
总体的流程如下:

StmtExecutor.execute||\/
StatementPlanner.plan||\/
StatementPlanner.planQuery||\/
StatementPlanner.createQueryPlan||\/
Optimizer.optimize||\/
Optimizer.optimizeByCost||\/
Optimizer.memoOptimize||\/
SeriallyTaskScheduler.executeTasks||\/
DeriveStatsTask.execute||\/
StatisticsCalculator.estimatorStats||\/
StatisticsCalculator.computeOlapScanNode||\/
StatisticsCalcUtils.getTableRowCount||\/
StatisticsCalcUtils.deltaRows

后台清理部分

对于这种带有TTL的分区表来说，会有 DynamicPartitionScheduler 这个后台线程进行分区的删除。具体代码见：

protected void runAfterCatalogReady() {if (!initialize) {// check Dynamic Partition tables only when FE startinitDynamicPartitionTable();}setInterval(Config.dynamic_partition_check_interval_seconds * 1000L);if (Config.dynamic_partition_enable) {executeDynamicPartition();}executePartitionTimeToLive();}

其中删除分区的频率就是由 Config.dynamic_partition_check_interval_seconds 也就是dynamic_partition_check_interval_seconds来决定的，
其中executeDynamicPartition方法就是执行分区删除,具体数据流如下 ：

executeDynamicPartition||\/
executeDynamicPartitionForTable||\/
getDropPartitionClause||\/
GlobalStateMgr.getCurrentState().dropPartition(db, olapTable, dropPartitionClause);||\/
olapTable.dropPartition(db.getId(), partitionName, clause.isForceDrop());||\/
idToPartition.remove(partition.getId());

其中 在 executeDynamicPartitionForTable 中 RangePartitionInfo rangePartitionInfo = (RangePartitionInfo) olapTable.getPartitionInfo(); 会根据PartitionInfo的信息来进行判断，只有 RangePartitionInfo类型支持partition TTL删除，也就是Expression partitioning (recommended) 和Dynamic partitioning支持.

在最后的idToPartition.remove(partition.getId())中就会删除正在进行查询迭代的idToPartition.values，就是导致并发问题