英文源地址
本章将使用Go语言实现一个不可变地B+树.这是一个最小实现, 因此很容易理解.

Node节点的格式

我们的B树最终将被持久化到磁盘上, 因此我们首先需要为b树节点设计数据格式.如果没有这种格式, 我们将无法知道节点的大小以及何时拆分节点.
一个节点包含:

• 一个固定大小的头部, 包含节点的类型(叶子节点或内部节点)和键的数量
• 指向子节点的指针列表(内部节点使用)
• 指向每个键值对的偏移量列表
• 封装的KV对
  
  以下是KV键值对的格式.长度后紧跟着数据.
  
  为了简单起见, 叶子节点和内部节点都使用相同的格式.

数据类型

既然我们最终要将b树转存至磁盘, 为什么不使用字节数组作为我们的内存数据结构呢?

type BNode struct {data []byte
}const (BNODE_NODE = 1BNODE_LEAF = 2
)

我们不能使用内存指针, 这些指针是64位整数, 引用磁盘页而不是内存节点.我们将添加一些回调函数来抽象这个方面, 以便我们的数据结构仍然是纯粹的数据结构代码.

type BTree struct {root uint64get func(uint64) BNodenew func(BNode) uint64del func(uint64)
}

页大小定义为4K字节.更大的页面大小(如8K或16K)也可以.
我们还对键和值的大小做了一些约束.所以只有一个kv键值对的节点总是可以放在一个页上.如果需要支持更大的键或更大的值, 就必须为它们分配额外的页, 这会增加复杂度.

const HEADER = 4const BTREE_PAGE_SIZE = 4096
const BTREE_MAX_KEY_SIZE = 1000
const BTREE_MAX_VAL_SIZE = 3000func init() {node1max := HEADER + 8 + 2 + 4 + BTREE_MAX_KEY_SIZE + BTREE_MAX_VAL_SIZEassert(node1max <= BTREE_PAGE_SIZE)
}

解码b树节点

由于节点只是一个字节数组, 因此我们将添加一些辅助函数来访问其内容.

func (node BNode) btype() uint16 {return binary.LittleEndian.Uint16(node.data)
}func (node BNode) nkeys() uint16 {return binary.LittleEndian.Uint16(node.data[2:4])
}func (node BNode) setHeader(btype uint16, nkeys uint16) {binary.LittleEndian.PutUint16(node.data[0:2], btype)binary.LittleEndian.PutUint16(node.data[2:4], nkeys)
}func (node BNode) getPtr(idx uint16) uint64 {assert(idx < node.nkeys())pos := HEADER + 8 * idxreturn binary.LittleEndian.Uint64(node.data[pos:])
}func (node BNode) setPtr(idx uint16, val uint64) {assert(idx < node.nkeys())pos := HEADER + 8 * idxbinary.LittleEndian.PutUint64(node.data[pos:], val)
}

关于偏移列表的一些细节

• 偏移量与第一个kv对的位置有关
• 第一个kv对的偏移量始终为零, 因此它不存储在列表里
• 我们将到最后一个kv对的偏移量存储在偏移列表, 这用于确定节点的大小

func offsetPos(node BNode, idx uint16) uint16 {assert(1 <= idx && idx <= node.nkeys())return HEADER + 8 * node.nkeys() + 2 * (idx - 1)
}func (node BNode) getOffset(idx uint16) uint16 {if idx == 0 {return 0}return binary.LittleEndian.Uint16(node.data[offsetPos(node, idx):])
}func (node BNode) setOffset(idx uint16, offset uint16) {binary.LittleEndian.PutUint16(node.data[offsetPos(node, idx):], offset)
}

偏移列表用于快速定位第n个KV键值对

func (node BNode) kvPos(idx uint16) uint16 {assert(idx <= node.nkeys())return HEADER + 8 * node.nkeys() + 2 * node.nkeys() + node.getOffset(idx)
}func (node BNode) getKey(idx uint16) []byte {assert(idx < node.keys())pos := node.kvPos(idx)klen := binary.LittleEndian.Uint16(node.data[pos:])return node.data[pos+4:][:klen]
}func (node BNode) getVal(idx uint16) []byte {assert(idx < node.nkeys())pos := node.kvPos(idx)klen := binary.LittleEndian.Uint16(node.data[pos+0:])vlen := binary.LittleEndian.Uint16(node.data[pos+2:])return node.data[pos+4+klen:][:vlen]
}

并确定节点的大小

func (node BNode) nbytes() uint16 {return node.kvPos(node.nkeys())
}

b树插入操作

代码被分解成了小步骤.

查找key

要将键插入叶节点, 我们需要查找它在kv列表中的位置

func nodeLookupLE(node BNode, key []byte) uint16 {nkeys := node.nkeys()found := uint16(0)for i := uint16(1); i < nkeys; i++ {cmp := bytes.Compare(node.getKey(i), key)if cmp <= 0 {found = i}if cmp >= 0	{break}	}return found
}

查找对叶子节点和内部节点都有效, 注意, 第一个键将被跳过以进行比较, 因为它已经从父节点进行了比较.

更新叶子节点

在查找到要插入的位置后, 我们要创建一个包含新键的节点副本.

func leafInsert(new BNode, old BNode, idx uint16, key []byte, val []byte) {new.setHeader(BNODE_LEAF, old.nkeys()+1)nodeAppendRange(new, old, 0, 0, idx)nodeAppendKV(new, idx, 0, key, val)nodeAppendRange(new, old, idx+1, idx, old.nkeys()-idx)
}

nodeAppendRange函数从旧节点拷贝键至新节点

func nodeAppendRange(new BNode, old BNode, dstNew uint16, srcOld uint16, n uint16) {assert(srcOld+n <= old.nkeys())assert(dstNew+n <= new.nkeys())if n == 0 {return}for i := uint16(0); i < n; i++ {new.setPtr(dstNew+i, old.getPtr(srcOld+i))}dstBegin := new.getOffset(dstNew)srcBegin := old.getOffset(srcOld)for i := uint16(1); i < n; i++ {offset := dstBegin + old.getOffset(srcOld+i) - srcBeginnew.setOffset(dstNew+i, offset)}begin := new.kvPos(srcOld)end := new.kvPos(srcOld+n)copy(new.data[new.kvPos(dstNew):], old.data[begin:end])
}

nodeAppendKV函数复制一个kv对到新节点中

func nodeAppendKV(new BNode, idx uint16, ptr uint64, key []byte, val []byte) {new.setPtr(idx, ptr)pos := new.kvPos(idx)binary.LittleEndian.PutUint16(new.data[pos+0:], uint16(len(key)))binary.LittleEndian.PutUint16(new.data[pos+2:], uint16(len(val)))copy(new.data[pos+4:], key)copy(new.data[pos+4+uint16(len(key)):], val)new.setOffset(idx+1, new.getOffset(idx)+4+uint16(len(key)+len(val)))
}

递归插入

插入key的主要功能

func treeInsert(tree *BTree, node BNode, key []byte. val []byte) BNode {new := BNode(data: make([]byte, 2 * BTREE_PAGE_SIZE))idx := nodeLookupLE(node, key)switch node.btype() {case: BNODE_LEAF:if bytes.Equal(key, node.getKey(idx)) {leafUpdate(new, node, idx, key, val)} else {leafInsert(tree, new, node, idx, key, val)}case BNODE_NODE:nodeInsert(tree, new, node, idx, key, val)default:panic("bad node!")}return new
}

leafUpdate函数与leafInsert函数类似

处理内部节点

现在来处理内部节点.

func nodeInsert(tree *BTree, new BNode, node BNode, idx uint16, key []byte, val []byte) {kptr := node.getPtr(idx)knode := tree.get(kptr)tree.del(kptr)knode = treeInsert(tree, knode, key, val)nsplit, splited := nodeSplit3(knode)nodeReplaceKidN(tree, new, node, idx, splited[:nsplit]...)
}

分裂大节点

在节点中插入键会使其增加大小, 导致其超过页大小. 在这种情况下, 节点被分割成多个更小的节点.
允许的最大键大小和值的大小只能保证单个kv键值对始终适合于一个页.在最坏的情况下, 胖节点会分裂成3个节点(中间有一个大的kv键值对)

func nodeSplit2(left BNode, right BNode, old BNode) {}func nodeSplit3(old BNode) (uint16, [3]BNode) {if old.nbytes() <= BTREE_PAGE_SIZE {old.data = old.data[:BTREE_PAGE_SIZE]return 1, [3]BNode{old}}left := BNode{make([] byte, 2 * BTREE_PAGE_SIZE)}right := BNode{make([]byte, BTREE_PAGE_SIZE)}nodeSplit2(left, right, old)if left.nbytes() <= BTREE_PAGE_SIZE {left.data = left.data[:BTREE_PAGE_SIZE]return 2, [3]BNode{left, right}}leftleft := BNode{make([]byte, BTREE_PAGE_SIZE)}middle := BNode{make([]byte, BTREE_PAGE_SIZE)}nodesplit2(leftleft, middle, left)assert(leftleft.nbytes() <= BTREE_PAGE_SIZE)return 3, [3]BNode{leftleft, middle, right}
}

更新内部节点

在节点中插入一个键可以产生1,2或3个节点. 父节点必须相应地更新自身. 更新内部节点的代码类似于更新叶子节点的代码.

func nodeReplaceKidN(tree *BTree, new BNode, old BNode, idx uint16, kids ...BNode) {inc := uint16(len(kids))new.setHeader(BNODE_NODE, old.nkeys()+inc-1)nodeAppendRange(new, old, 0, 0, idx)for i, node := range kids {nodeAppendKV(new, idx+uint16(i), tree.new(node), node.getKey(0), nil)nodeAppendRange(new, old, idx+inc, idx+1, old.nkeys()-(idx+1))}nodeAppendRange(new, ol, idx+inc, idx+1, old.nkeys()-(idx+1))
}

我们已经完成了B树的插入操作. 删除操作和剩下代码将在下一篇进行介绍.