Draft distributed plan

[mdianjun]

功能列表

• 基于分布式架构重新设计clickhouse，分为meta-service，compute（计算层），store（存储层）三种角色。
  • meta-service负责存储集群元数据（基于clickhouse-keeper改造）：
    • 提供访问权限控制信息（/clickhouse/access）：原生clickhouse支持，配置参数是<user_directories>。
    • 数据模型信息（/clickhouse/metadata）：由Replicated database engine提供。
    • 集群节点拓扑信息（/clickhouse/clusters）：自主开发。
    • 当ck-server实例扩容或重启时，从meta-service上加载元数据。加载database是自主开发，加载table是由Replicated database engine提供。
    • 多节点的session共享（/clickhouse/sessions）：（自主开发）把session数据存储到meta-service上，以session id为标识。初始接收query的计算节点负责创建session并同步到meta-service，其他节点在处理该query时从meta-service上加载session数据。共享内容包括过期时间，settings配置等。对于同一个query来说，initial节点主动发给其他计算节点和存储节点。对于相同session的多个query来说，只有initial节点才会从meta-service上加载session。
  • compute层功能包括：接收用户query、生成分布式执行计划、承担大部分计算负载。
  • store层功能包括：读写数据、承担小部分计算负载（如读数据后的filter操作）、数据块的后台处理（如合并、清理等）。
• meta-service的优化：
  • 接口优化：递归watch。每一个计算层节点和存储层节点都需要watch。
  • 存储优化：将meta-service的数据存储由内存转移到磁盘上，使用rocksdb实现（TODO：性能测试）。
• 分布式DDL：在一个计算节点执行DDL语句，其他所有节点即可自动同步到结果。支持对库、表元数据和权限控制元数据的CREATE/DROP/ALTER等操作。
  • table级别的DDL：由Replicated database engine提供。
  • database级别的DDL：（自主开发）使用defalut数据库，派生其他database，实现方式类似于“在Replicated database里执行table DDL”的逻辑。
• 分布式admin query：（自主开发）将“SYSTEM”、“OPTIMIZE”等语句改造为分布式执行，实现方式类似于“Replicated database的DDL”。
• 分布式DML：在一个计算节点执行DML语句，所有存储节点即可自动同步到结果。支持对表数据的UPDATE/DELETE/INSERT操作（以无副本的MergeTree表为核心）：
  • UPDATA/DELETE：由Replicated database engine提供。
  • INSERT：（自主开发）参考分布式表的写操作，实现MergeTree表的分布式写。可以使用“随机shard”或“指定shard”方式，将数据写到任意存储节点。
• 分布式DQL：一个计算节点接收到SELECT查询语句，首先生成分布式执行计划，然后将构建执行计划片段（plan fragment）需要的参数信息发送给其他计算节点和存储节点，其他节点根据原始语句构建plan fragment并执行。目前已经支持的查询类型及算子包括：

select ... from
insert into ... select
create ... as select 
aggregate + group by
distinct
order by
limit
union
in
join: broadcast join
subquery：select+subquery, in+subquery, join+subquery
materialized view
with totals/rollup/cube
extremes
input function: insert into ... select ... from input()
cancel query

• 外部数据源：接入外部数据源通过特定的表引擎或表函数完成。对外部数据源的查询和写入是在计算节点执行，该功能将支持产品的数据导入/导出功能。已支持的外部数据源包括：kafka, mysql, hdfs, s3，input。

实现过程中的关键点或难点

• 模块耦合问题：
  • 计算逻辑和存储逻辑的耦合
  • 逻辑执行计划和物理执行计划的耦合
• 暂时无法实现最好的方案，只能采用折中方案：
  • 所有节点都是从头到尾构建执行计划；较好的方案是发送plan fragment。
  • 基于zookeeperclient接口实现某些功能，复杂度较高；较好的方案是在server侧作开发，把复杂度封装到server里，避免分布式场景下不一致的问题。
• 当开发出分布式执行计划的框架之后，无法以设计驱动的方式继续完善，而是以测试驱动的方式case by case地补漏洞。
• 分布式执行计划怎么切stage：
  • union与它的子节点之间
  • full join与它的两个节点之间：将左右表都汇总到一个节点
  • left join与它的右子节点之间：将右表广播
  • right join与它的左子节点之间：将左表广播
  • aggregate：partial aggregate与final aggregate之间
  • sort：partial sort与final sort之间
  • dinstinct：partial distinct与final distinct之间
  • limit：partial limit与final limit之间
• 子查询：包括三种形式：select from subquery, join subquery, in subquery。先计算子查询的结果集，把结果集发送到父查询对应的所有节点。
• 物化视图：由表找到它关联的物化视图，构建一个pipeline用于写该物化视图，这个pipeline读的就是insert时的那个Block数据。

分布式执行计划的待优化项

1. 物理执行计划的序列化。
2. 避免内存数据与网络数据之间相互转换时的序列化和反序列化开销。
3. 重构生成分布式执行计划的代码，以“transformation rule”形式提供可扩展接口。
4. 生成分布式执行计划时会添加某些逻辑算子，把这些代码放到生成单机执行计划的过程中。
5. shuffle算子。
6. 利用数据分布信息，对分布式执行计划进行更精细的调度。

核心代码流程图

1. 分布式执行计划

2. 分布式DDL