StarRocks multi_distinct_count() 深度解析：语义、示例与从优化器改写到底层 Hash Set 的实现原理

StarRocks multi_distinct_count() 深度解析语义、示例与从优化器改写到底层 Hash Set 的实现原理【免费下载链接】starrocksThe worlds fastest open query engine for sub-second analytics both on and off the data lakehouse. With the flexibility to support nearly any scenario, StarRocks provides best-in-class performance for multi-dimensional analytics, real-time analytics, and ad-hoc queries. A Linux Foundation project.项目地址: https://gitcode.com/GitHub_Trending/st/starrocksmulti_distinct_count(expr)是 StarRocks 中的去重计数聚合函数返回expr的去重行数语义上等价于count(distinct expr)。本文将完整覆盖该函数的语法、参数、返回值与示例用法并结合当前仓库中 FE 优化器改写规则RewriteMultiDistinctRule与 BE 聚合状态实现DistinctAggregateState说明一条count(distinct ...)查询是如何被改写、序列化、分阶段合并并输出最终结果的。读完后你可以准确使用该函数、理解它与count(distinct)、multi_distinct_sum、CTE 改写路径之间的协作关系并掌握其底层哈希集状态在 MPP 多阶段聚合中的工作方式。一、函数定义语法、参数与返回值根据官方文档 multi_distinct_count.md该函数定义如下语法multi_distinct_count(expr)参数exprmulti_distinct_count()所依据的列或表达式。当expr是列名时该列可以是任意数据类型。返回值返回一个数值。如果没有找到任何行返回 0。该函数忽略 NULL 值。结合后端源码可以进一步确认两点实现事实返回类型固定为 BIGINT。在 BE 聚合函数注册器中multi_distinct_count对每一种输入类型都以TYPE_BIGINT作为结果类型注册resolver-add_aggregate_mappinglt, TYPE_BIGINT, DistinctState(multi_distinct_count, ...)见 aggregate_resolver_distinct.cpp。结果永不为 NULL。multi_distinct_count返回 0 而非 NULL因此其结果列被标记为非空BE 聚合函数基类中对该行为有明确注释“multi_distinct_count returns 0 (never NULL); multi_distinct_sum can be NULL, so restrict this to COUNT”见 distinct.h 中is_result_non_nullable()的实现。支持的输入类型范围register_distinct()通过type_dispatch_all在所有聚合可用类型上注册multi_distinct_count并额外补充了TYPE_VARBINARYvoid AggregateFuncResolver::register_distinct() { auto multi_distinct_types aggregate_types(); multi_distinct_types.push_back(TYPE_VARBINARY); for (auto type : multi_distinct_types) { type_dispatch_all(type, DistinctDispatcher(), this); } }见 aggregate_resolver_distinct.cpp。也就是说整数、浮点、DECIMAL、DATE/DATETIME、VARCHAR 以及 VARBINARY 等类型均可作为去重计数的输入与文档中“任意数据类型”的描述一致。BE 单元测试也验证了 SMALLINT、INT、BIGINT、LARGEINT、FLOAT、DOUBLE、VARCHAR、DECIMALV2、DATETIME、DATE 等输入类型的注册与可执行性见 aggregate_test.cpp。二、使用示例继承官方文档示例官方文档给出的示例场景假设存在一张名为test的表按id查询每个订单的类别与供应商select * from test order by id; -------------------------------------- | id | country | category | supplier | -------------------------------------- | 1001 | US | A | supplier_1 | | 1002 | Thailand | A | supplier_2 | | 1003 | Turkey | B | supplier_3 | | 1004 | US | A | supplier_2 | | 1005 | China | C | supplier_4 | | 1006 | Japan | D | supplier_3 | | 1007 | Japan | NULL | supplier_5 | --------------------------------------示例 1统计category列的去重值个数。注意第 7 行的category为 NULL被忽略实际去重值为 A、B、C、D 共 4 个select multi_distinct_count(category) from test; -------------------------------- | multi_distinct_count(category) | -------------------------------- | 4 | --------------------------------示例 2统计supplier列的去重值个数共 5 个供应商select multi_distinct_count(supplier) from test; -------------------------------- | multi_distinct_count(supplier) | -------------------------------- | 5 | --------------------------------这两个示例同时演示了文档中的两条返回语义NULL 被忽略示例 1 结果为 4 而不是 5以及结果类型为整数值BIGINT。三、与 count(distinct expr) 的关系优化器如何生成 multi_distinct_count在 StarRocks 中multi_distinct_count不仅是用户可以直接调用的内置函数更是优化器改写count(distinct ...)的产物。函数名常量定义在 FE 函数目录中public static final String MULTI_DISTINCT_COUNT multi_distinct_count;见 FunctionSet.java。改写入口RewriteMultiDistinctRule当逻辑聚合算子下只有一层简单输入时优化规则TF_REWRITE_MULTI_DISTINCT会被触发其check()逻辑见 RewriteMultiDistinctRule.java包含三个关键判断存在“复杂类型的常量 count(distinct)”时触发——两阶段聚合不支持该写法会被替换为等价的any_value形式存在多个 distinct 聚合函数、且它们的去重输入列不完全相同时触发——这是多 distinct 列场景的典型情况多个 distinct 函数虽然共用相同的去重列但 split 聚合规则无法处理例如表只有一个 tablet时触发。transform()阶段根据条件在两条路径之间选择public ListOptExpression transform(OptExpression input, OptimizerContext context) { if (isComplexConstantCountDistinct(input)) { return rewriteComplexConstantDistinct(input); } if (useCteToRewrite(input, context)) { MultiDistinctByCTERewriter rewriter new MultiDistinctByCTERewriter(); return rewriter.transformImpl(input, context); } else { MultiDistinctByMultiFuncRewriter rewriter new MultiDistinctByMultiFuncRewriter(); return rewriter.transformImpl(input, context); } }即多列 distinct如count(distinct a, b)或开启 CBO CTE 复用等场景走 CTE 改写单列 distinct 场景优先改写为multi_distinct_count/multi_distinct_sum这类“multi distinct”内置函数。此外当cbo_cte_reuse关闭且 distinct 输入是复杂类型、JSON、group_concat(distinct ...)或array_agg(distinct decimal列)时规划器会直接抛出%s is unsupported when cbo_cte_reuse is disabled错误见 RewriteMultiDistinctRule.java——这为使用该函数族划定了明确的使用前提。具体替换规则MultiDistinctByMultiFuncRewriterMultiDistinctByMultiFuncRewriter遍历聚合项对带DISTINCT标记的调用做如下替换见 MultiDistinctByMultiFuncRewriter.javacount(distinct expr)→multi_distinct_count(expr)sum(distinct expr)→multi_distinct_sum(expr)array_agg(distinct expr)非 DECIMAL 参数→array_agg_distinct(expr)avg(distinct expr)→ 拆分为multi_distinct_sum(expr) / multi_distinct_count(expr)并在上层LogicalProject中完成除法DECIMAL V3 场景下会把 BIGINT 的 count 列隐式转换为 DECIMAL(38,0)/DECIMAL(76,0)。改写完成后聚合算子类型被置为AggType.GLOBAL即单阶段全局聚合避免在数据交换后再做局部去重合并的额外开销。对本文主题而言核心构造方法为private CallOperator buildMultiCountDistinct(CallOperator oldFunctionCall) { Function searchDesc new Function(new FunctionName(FunctionSet.MULTI_DISTINCT_COUNT), oldFunctionCall.getFunction().getArgs(), InvalidType.INVALID, false); Function fn GlobalStateMgr.getCurrentState().getFunction(searchDesc, IS_NONSTRICT_SUPERTYPE_OF); return (CallOperator) scalarRewriter.rewrite( new CallOperator(FunctionSet.MULTI_DISTINCT_COUNT, fn.getReturnType(), oldFunctionCall.getChildren(), fn), DEFAULT_TYPE_CAST_RULE); }这意味着用户直接写multi_distinct_count(x)或写count(distinct x)最终落到 BE 的往往是同一个内置聚合实现。四、BE 实现聚合状态、去重哈希集与两阶段合并聚合状态的结构multi_distinct_count的每个分组对应一个聚合状态aggregate state核心是DistinctAggregateState见 distinct.h。按输入类型分为两类特化定长类型整数、浮点、DATE/DATETIME 等内部维护一个基于phmap::flat_hash_table的HashSetWithAggStateAllocatorTupdate()插入键值distinct_count()直接返回set.size()void update([[maybe_unused]] MemPool* mem_pool, T key) { set.insert(key); } int64_t distinct_count() const { return set.size(); }字符串/二进制类型VARCHAR、VARBINARY内部使用AdaptiveSliceHashSet。它先以单层SliceHashSet收集键当去重基数每增长 65536 且内存池累计分配量达到agg::two_level_memory_threshold()时自动升级为两级哈希集SliceTwoLevelHashSetWithAggStateAllocator以降低大基数下的探测开销void try_convert_to_two_level(MemPool* mem_pool) { if (distinct_size % 65536 0 mem_pool-total_allocated_bytes() agg::two_level_memory_threshold()) { two_level_set std::make_sharedSliceTwoLevelHashSetWithAggStateAllocator(); ... } }字符串键从输入列复制到内存池分配的连续内存中allocate_with_reservememcpy保证状态自包含后续序列化时不需要再回溯原始列。面向 cache 的批量更新优化TDistinctAggregateFunction为批处理路径专门实现了update_batch/update_batch_single_state见 distinct.h先对 chunk 内所有行预计算哈希值并缓存到CacheEntry数组再按“提前 16 行”的经验值执行prefetch_hash最后用update_with_hash以预计算哈希插入。源码注释指出这是针对phmap::flat_hash_table的有效模式可以显著提升哈希表性能。这类优化对用户透明但解释了为什么multi_distinct_count在高基数列上仍具备良好吞吐。两阶段聚合序列化与合并multi_distinct_count是支持两阶段聚合的函数本地阶段先对每个 tablet 分组内的数据去重再把序列化后的哈希集而非原始明细行发送到全局阶段合并大幅减少 shuffle 数据量。序列化定长类型先写size_t长度头再逐键memcpyserialize_size()保证不小于MIN_SIZE_OF_HASH_SET_SERIALIZED_DATA 24 字节的下限字符串类型以“长度前缀 字节串”逐键写出见 distinct.h。合并merge()接收二进制备选列中的序列化数据。对字符串类型走deserialize_and_merge逐个解包并emplace仅新键才 memcpy对定长类型有一个边界处理——若输入 slice 小于 24 字节则按“单值”直接插入哈希集否则按序列化哈希集解析合并见 distinct.h。输出finalize_to_column中DistinctType COUNT时向Int64Column写入distinct_count()与文档“返回数值、无数据返回 0”的语义一致空哈希集的size()为 0。V2 状态与窗口函数路径仓库中还存在DistinctAggregateStateV2对应 BE 注册的multi_distinct_count2其序列化格式为紧凑的“定长值数组 长度头”并支持在update阶段同步累加sum供sum(distinct)/avg(distinct)复用。此外fused_multi_distinct_count等一系列窗口映射add_window_mapping用于count(distinct ...) OVER (...)这类窗口去重计数场景由 FE 的DistinctAggregationOverWindowRule生成见 aggregate_resolver_distinct.cpp。这说明multi_distinct_count所在的 distinct 函数族是整个去重聚合体系的公共底座。五、使用建议与适用前提综合文档语义与源码证据使用multi_distinct_count时可以参考以下要点语义等价性multi_distinct_count(expr)与count(distinct expr)结果一致均忽略 NULL、无匹配行时返回 0返回类型均为 BIGINT。在多数情况下直接写count(distinct ...)即可优化器会自动改写为multi_distinct_count显式调用内置函数则能绕过改写路径直接执行同一实现。单列 distinct 的优势路径单列去重计数走MultiDistinctByMultiFuncRewriter聚合算子被标记为GLOBAL配合 BE 侧“序列化哈希集再合并”的两阶段机制避免明细行 shuffle。多列 distinct 的边界条件count(distinct a, b)等多列 distinct 依赖 CTE 改写路径需要开启cbo_cte_reuse默认由会话变量控制否则规划器会报USER_ERROR。结果确定性由于结果列标记为非空COUNT 类型 distinct 函数is_result_non_nullable()返回 true在结果集、物化视图聚合复用等场景中不需要额外处理 NULL。相关函数同族的multi_distinct_sum可返回 NULL、array_agg_distinct以及面向窗口函数的fused_multi_distinct_count*变体均注册于 aggregate_resolver_distinct.cpp可按需查阅对应文档与源码。小结multi_distinct_count在 StarRocks 中是去重计数的“用户接口 引擎内建实现”双重身份文档层面的函数语义任意类型输入、忽略 NULL、返回 BIGINT由 BE 注册器与聚合状态实现严格落地FE 优化器则通过RewriteMultiDistinctRule将count(distinct ...)统一改写为该函数族并按列数、统计信息与会话开关在 multi-func 路径与 CTE 路径之间做选择。理解这一整条链路既能正确编写去重计数查询也能在遇到多 distinct 列、复杂类型输入或 CTE 相关报错时快速定位原因。【免费下载链接】starrocksThe worlds fastest open query engine for sub-second analytics both on and off the data lakehouse. With the flexibility to support nearly any scenario, StarRocks provides best-in-class performance for multi-dimensional analytics, real-time analytics, and ad-hoc queries. A Linux Foundation project.项目地址: https://gitcode.com/GitHub_Trending/st/starrocks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考