相信很多同学看过 MySQL 各种优化的文章,里面 99% 会提到:单表数据量大了,需要进行分片(水平拆分 or 垂直拆分)。分片之后,业务上必然面临的场景:跨分片的数据合并。今天我们就一起来瞅瞅 MyCAT 是如何实现分片结果合并。
跨分片查询大体流程如下:
flow
和 《【单库单表】查询》 不同的两个过程:
下面,我们来逐条讲解这两个过程。
execute_sql
经过 SQL 解析后,计算出需要执行 SQL 的分片节点,遍历分片节点发送 SQL 进行执行。
核心代码:
SQL 解析 详细过程,我们另开文章,避免内容过多,影响大家对 分片结果合并 流程和逻辑的理解。
handle_response
和 《【单库单表】查询》 不同,多个分片节点都会分别响应 记录头(header) 和 记录行(row) 。在开始分析 MyCAT 是怎么合并多分片结果之前,我们先来回想下 SQL 的执行顺序。
FROM // [1] 选择表
WHERE // [2] 过滤表
GROUP BY // [3] 分组SELECT // [4] 普通字段,max / min / avg / sum / count 等函数,distinctHAVING // [5] 再过滤表ORDER BY // [6] 排序LIMIT // [7] 分页
多个分片节点响应时,会响应多次 记录头(header) 。MyCAT 在实际处理时,只处理第一个返回的 记录头(header) 。因此,在使用时要保证表的 Schema 相同。
分片节点响应的 记录头(header) 可以直接返回 MySQL Client 吗?答案是不可以。AVG
函数 是特殊情况,MyCAT 需要将 AVG
拆成 SUM
+ COUNT
进行计算。举个例子:
// [1] MySQL Client => MyCAT :
SELECT AVG(age) FROM student;
// [2] MyCAT => MySQL Server :
SELECT SUM(age) AS AVG0SUM, COUNT(age) AS AVG0COUNT FROM student;
// [3] 最终:AVG(age) = SUM(age) AS AVG0SUM / COUNT(age)
核心代码:
MyCAT 对分片结果合并通过 AbstractDataNodeMerge
子类来完成。
merge_service
AbstractDataNodeMerge
:
END_FLAG_PACK
表示队列已结束。packs
。END_FLAG_PACK
到 packs
。AbstractDataNodeMerge_run.png
通过 running
标记保证同一条 SQL 同时只有一个线程正在执行,并且不需要等到每个分片结果都返回就可以执行聚合逻辑。当然,排序逻辑需要等到所有分片结果都返回才可以执行。
核心代码:
AbstractDataNodeMerge
有两种子类实现:
DataMergeService
:基于堆内内存合并分片结果。DataNodeMergeManager
:基于堆外内存合并分片结果。目前官方默认配置使用 DataNodeMergeManager
。主要有如下优点:
如果对堆外内存不太了解,推荐阅读如下文章:
本文主要分析 DataNodeMergeManager
实现,DataMergeService
可以自己阅读或者等待后续文章(?欢迎订阅我的公众号噢)。
DataNodeMergeManager
有三个组件:
globalSorter
:UnsafeExternalRowSorter
=> 实现记录行(row)合并并排序逻辑。globalMergeResult
:UnsafeExternalRowSorter
=> 实现记录行(row)合并不排序逻辑。unsafeRowGrouper
: UnsafeRowGrouper
=> 实现记录行(row)聚合逻辑。DataNodeMergeManager#run(...)
逻辑如下:
UnsafeRow
。UnsafeRow
插入对应组件。UnsafeRow
插入完后,根据情况使用组件聚合、排序。是否排序 | 是否聚合 | 依赖组件 | [2] | [3] |
---|---|---|---|---|
否 | 否 | globalSorter | 插入 globalSorter | 使用 globalSorter 合并并排序 |
是 | 否 | globalMergeResult | 插入 globalMergeResult | 使用 globalMergeResult 合并不排序 |
否 | 是 | unsafeRowGrouper + globalSorter | 插入 unsafeRowGrouper 进行聚合 | 使用 globalSorter 合并并排序 |
是 | 是 | unsafeRowGrouper + globalMergeResult | 插入 unsafeRowGrouper 进行聚合 | 使用 globalMergeResult 合并不排序 |
核心代码:
?看到这里,可能很多同学都有点懵逼,问题不大,我们继续往下瞅。
unsafe_row
记录行(row)写到 UnsafeRow
的 baseObject
属性,结构如下:
unsafe_row_object
unsafe_row_2.png
baseObject
。bitSetWidthInBytes
) = 字段占用的格子数 * 64 bits。[3]区域
所在的位置和长度。[2]区域
的位置 = baseOffset
+ bitSetWidthInBytes
+ 8 Bytes * 字段顺序。[3]区域
的位置,后 32 bits 为字段对应的值长度。写入 UnsafeRow
,MyCAT 可以顺序访问每个字段,而不需要在记录行(row)进行遍历。
?日常开发使用位操作的机会比较少,可能较为难理解,需要反复理解下,相信会获得很大启发。恩,该部分代码引用自开源运算框架 Spark
,是不是更加有动力列?。
核心代码:
如果使用 Java 实现 SELECT * FROM student ORDER BY age desc, nickname asc
,不考虑算法优化的情况下,我们可以简单如下实现:
Collections.sort(students, new Comparator<Comparable>() { @Override
public int compare(Student o1, Student o2) { int cmp = compare(o2.age, o1.age); return cmp != 0 ? cmp : compare(o1.nickname, o2.nickname);
}
}
});
从功能上,UnsafeExternalRowSorter
是这么实现排序逻辑。当然肯定的是,不是这么“简单”的实现。
sorter_write
UnsafeRow
会写入到两个地方:
List<MemoryBlock>
:内存块数组。当前 MemoryBlock
无法容纳写入的 UnsafeRow
时,生成新的 MemoryBlock
提供写入。每条 UnsafeRow
存储在 MemoryBlock
由 长度 + 字节内容 组成。LongArray
:每条 UnsafeRow
存储在 LongArray
由两部分组成:address + prefix。address
:UnsafeRow
存储在 List<MemoryBlock>
的位置。前 13 bits 记录所在 MemoryBlock
的 index,后 51 bit 记录在 MemoryBlock
的 offset。prefix
:UnsafeRow
第一个排序字段值前 64 bits 计算的值。UnsafeExternalRowSorter
排序实现方式 :提供 TimSort 和 RadixSort 两种排序算法,前者为默认实现。TimSort 折半查找时,使用 LongArray
,先比较 prefix
,若相等,则顺序对比每个排序字段直到不等,提升计算效率。插入操作在 LongArray
操作,List<MemoryBlock>
只作为原始数据。
另外,当需要排序特别大的数据量时,会使用存储数据到文件进行排序。限于笔者暂时未阅读该处源码,后续会另开文章分析。?
核心源码:
如果使用 Java 实现 SELECT nickname, COUNT(*) FROM student group by nickname
,不考虑算法优化的情况下,我们可以简单如下实现:
Map<String, List<Object>> map = new HashMap<>();// 聚合for (student : students) { if (map.contains(student.nickname)) {
map.put(student.nickname, map.get(student.nickname).get(1) + 1);
} else {
List<Object> value = new Array<>();
value.add(nickname);
value.add(1);
map.put(student.nickname, value);
}
}// 输出for (value : map.values) {
System.out.println(value);
}
从功能上,UnsafeRowGrouper
是这么实现排序逻辑。当然肯定的是,也不是这么“简单”的实现。
?具体怎么实现的呢?我们在《MyCAT 源码解析 —— 分片结果合并(二)》继续分析。