一、Halo Cepat原理
Halo数据库原生支持分布式数据库的特性，具备无限水平扩展能力，性能随容量扩容而线性提升，能够按需扩展集群，可以支撑 PB 级别的 HTAP 分布式数据库。该功能称作Halo Cepat。
Halo Cepat采用 Shared Nothing 架构，节点之间没有应用数据共享，仅在其中三个节点之间共享元数据信息。Halo Cepat需要至少三个节点构成分布式数据库集群，避免单点故障，相比单机 Halo数据库，Halo Cepat可以使用更多的 CPU 核心，更多的内存数量，保存更多的数据。通过向分布式数据库添加节点，可以轻松的水平扩展计算、存储等能力。
Halo Cepat使用分片技术在多个节点上横向扩展数据库。分片过程自动完成并自动维护，对应用透明，减少应用改造成本。Halo Cepat的查询引擎在这些节点上采用并行化查询，以便在大型数据集上实现实时的响应。
二、基本架构
1.协调器和 Worker 节点均通过流复制做了高可用
2.协调器将 tab 表进行分片，表的实际数据存储在各个Worker节点的tab_x中，tab 表不存储任何数据，只存放元数据信息
3.协调器接收到客户端的请求后，根据元数据映射，将请求分发给对应的Worker节点
三、基本概念
本地表
在协调器节点由 CREATE TABLE 创建的普通表。
分布式表
在协调器节点经 create_distributed_table('table', 'column') 处理过的本地表， table 需要转换成分布式的表， column 根据该列进行hash运算并进行分片。
分布式表的数据通过分布式列的 hash 进行计算，将数据存储到不同的分片表中。业务应使用此种表类型。
参考表
在协调器节点经 create_reference_table 处理过的本地表。参考表在所有 Worker 上拥有相同的数据。
四、功能特性
高性能
数据库并发连接最大可达近百万级
平滑扩容
无需关机即可不中断业务在线纵向/横向扩容，性能随节点扩展而线性增长，轻松应对超高并发请求带来的性能挑战，支持业务快速增长
高可用
当工作节点出现异常时，自动切换到高可用节点
数据一致性
采用两阶段提交处理分布式事务，确保数据正确性
支持 HTAP 混合模式
同时支持在线事务处理 (OLTP) 及在线分析处理 (OLAP) 场景，帮助用户基于一套系统同时承载在线交易及数据分析业务
五、应用场景
支持 HTAP 业务系统
在数据库系统应用中，OLTP 场景涉及数据量小，但对返回实时性要求高，OLAP 类场景涉及的数据量和计算量大，但是对实时性要求不高，Halo Cepat可同时支持在线事务处理 (OLTP) 及在线分析处理 (OLAP) 场景，帮助用户基于一套系统同时承载在线交易及数据分析业务，大幅节约存储成本，提高商业决策效率。
海量数据存储
随着数据的爆发式增长，企业经常会遇到非常大的数据集存储需求，在单机的 Halo 数据库中，查询性能无法满足需求。Halo Cepat具有无限水平扩展能力，性能随容量扩容而线性提升，能够按需扩充集群，保证集群可以支撑到 PB 级别存储。
实时高并发事务系统
随着金融行业的蓬勃发展，使客户量不断增长，业务核心系统需经受高并发访问的考验。
Halo Cepat的 Share Nothing 架构可以在线平滑地扩展实例规模，从容应对此类场景；
Halo Cepat采用2PC来保证事务一致性。
六、Halo Cepat的安装
软件安装
Halo的产品安装包里已带有Halo Cepat的完整功能。因此，Halo数据库正确安装后，意味着Halo Cepat也已经安装完成。Halo的具体安装过程请参见往期文章。
启用Halo Cepat
要使用Halo Cepat，只需要简单的进行两个步骤：
1. 配置参数shared_preload_libraries
在数据库参数文件中增加：
halo0root=# Sshared_preload_libraries = 'cepat'这里需要注意的是cepat必须为shared_preload_libraries的第一个参数。
halo0root=# SELECT * from cepat_add_node('worker', 1921);
halo0root=# SELECT * from cepat_add_node('worker', 1922);
halo0root=# SELECT * from cepat_get_active_worker_nodes();2. 使用CREATE EXTENSION创建扩展
halo0root=# create extension cepat;至此，Cepat功能便启用成功了。
七、 Halo Cepat的基本使用
设置协调器节点和Worker节点
设置协调器节点
在协调器节点上运行以下命令，注意“coordinator”需要能解析成IP
halo0root=# SELECT cepat_set_coordinator_host('coordinator',1921);创建Worker节点
在协调器节点上运行以下命令，注意“worker”需要能解析成IP
halo0root=# SELECT * from cepat_add_node('worker', 1921);
halo0root=# SELECT * from cepat_add_node('worker', 1922);
halo0root=# SELECT * FROM cepat_get_active_worker_nodes();
创建分布式表
使用create_distributed_table 在协调器节点上创建分布式表，分片表会均匀地创建在各个worker节点上
halo0root=# CREATE TABLE events (
halo0root(# device_id bigint,
halo0root(# event_id bigserial,
halo0root(# event_time timestamptz default now(),
halo0root(# data jsonb not null,
halo0root(# PRIMARY KEY (device_id, event_id)
halo0root(# );halo0root=# SELECT create_distributed_table('events', 'device_id');使用pg_dist_shard_placement 查看分片情况
halo0root=# SELECT * from pg_dist_shard_placement order by shardid, placementid;
shardid | shardstate | shardlength | nodename | nodeport | placementid
---------+------------+-------------+----------+----------+-------------
102075 | 1 | 0 | worker | 1921 | 69
102076 | 1 | 0 | worker | 1922 | 70
102077 | 1 | 0 | worker | 1921 | 71
102078 | 1 | 0 | worker | 1922 | 72
(4 行记录)插入数据后查看执行计划是否调用各个Worker节点的分片表。默认设置下，通过EXPLAIN查看只会列出1个Task的执行计划信息，若要查看所有Task的执行计划信息，需要设置cepat.explain_all_tasks参数为on。
halo0root=# INSERT INTO events (device_id, data)
halo0root-# SELECT s % 100, ('{"measurement":'||random()||'}')::jsonb FROM generate_series(1,1000000) s;
INSERT 0 1000000
halo0root=# set cepat.explain_all_tasks='on';
SET
halo0root=# EXPLAIN SELECT count(*) FROM events;
QUERY PLAN
-------------------------------------------------------------------------------------------------------------------------
Aggregate (cost=250.00..250.02 rows=1 width=8)
-> Custom Scan (Cepat Adaptive) (cost=0.00..0.00 rows=100000 width=8)
Task Count: 4
Tasks Shown: All
-> Task
Node: host=worker port=1921 dbname=halo0root
-> Finalize Aggregate (cost=5785.18..5785.19 rows=1 width=8)
-> Gather (cost=5785.07..5785.18 rows=1 width=8)
Workers Planned: 1
-> Partial Aggregate (cost=4785.07..4785.08 rows=1 width=8)
-> Parallel Seq Scan on events_102075 events (cost=0.00..4386.65 rows=159365 width=0)
-> Task
Node: host=worker port=1922 dbname=halo0root
-> Finalize Aggregate (cost=6184.36..6184.37 rows=1 width=8)
-> Gather (cost=6184.25..6184.36 rows=1 width=8)
Workers Planned: 1
-> Partial Aggregate (cost=5184.25..5184.26 rows=1 width=8)
-> Parallel Seq Scan on events_102076 events (cost=0.00..4752.60 rows=172660 width=0)
-> Task
Node: host=worker port=1921 dbname=halo0root
-> Finalize Aggregate (cost=5386.00..5386.01 rows=1 width=8)
-> Gather (cost=5385.88..5385.99 rows=1 width=8)
Workers Planned: 1
-> Partial Aggregate (cost=4385.88..4385.89 rows=1 width=8)
-> Parallel Seq Scan on events_102077 events (cost=0.00..4020.71 rows=146071 width=0)
-> Task
Node: host=worker port=1922 dbname=halo0root
-> Finalize Aggregate (cost=5904.18..5904.19 rows=1 width=8)
-> Gather (cost=5903.97..5904.18 rows=2 width=8)
Workers Planned: 2
-> Partial Aggregate (cost=4903.97..4903.98 rows=1 width=8)
-> Parallel Seq Scan on events_102078 events (cost=0.00..4574.77 rows=131678 width=0)
(32 行记录)创建参考表
使用create_reference_table 在协调器节点上创建参考表，参考表同时创建到各个worker节点上并保证数据一致
halo0root=# CREATE TABLE device_types (
halo0root(# device_type_id int primary key,
halo0root(# device_type_name text not null unique
halo0root(# );
CREATE TABLE
halo0root=# SELECT create_reference_table('device_types');
create_reference_table
------------------------
(1 行记录)在协调器节点上插入数据，查看各个Worker节点表的情况
halo0root=# INSERT INTO device_types (device_type_id, device_type_name) VALUES (55, 'laptop');
INSERT 0 1Worker 节点 1：
halo0root=# select * from device_types;
device_type_id | device_type_name
----------------+------------------
55 | laptop
(1 行记录)Worker 节点 2：
halo0root=# select * from device_types;
device_type_id | device_type_name
----------------+------------------
55 | laptop
(1 行记录)