關於 B+tree (附 python 模擬程式碼)
在之前,我一直只把 b+tree 當成是 btree 的一種變形,或者說是在某種情況下的一種優化,另外一些情況可能還是 btree 好些。但是做完之後才發現,b+tree 在各種情況都可以完全取代 btree,並能夠讓索引效能得到比 btree 更好的優化。因為 b+tree 設計的核心要點,是為了彌補 btree 最大的缺陷。
btree 最大的缺陷是什麼?
首先,我們知道對於 btree 和 b+tree 這種多路搜尋樹來說,一個很重要的特點就是樹的度數非常大。因為只有這樣才能夠降低樹的深度,減少磁碟讀取的次數。而樹的度數越大,葉子節點在樹中的比例就越大。假設度數為 1000,那麼葉子節點比他上一層內部節點的數量至少要多 1000 倍,在上一層就更加可以忽略不計了。可以說樹種 99.9% 的節點都是葉子節點。 但是對於 btree 來說,所有節點都是一樣的結構,都含有一定數量的資料和指向節點的指標。這兩項資料佔據 btree 節點的幾乎全部的空間。一個節點內的資料的數量比硬碟指標的數量少一,可以說和指標的數量幾乎相等。對於 python 這種動態型別語言感覺不出來,但是對於 C 這種固定型別語言來說,即使這個 children list 陣列為空,這個陣列的空間也都是預留出去的。導致的結果就是佔絕大多數的葉子節點的 children list 指標陣列所佔的磁碟空間完全浪費。
一個數據的大小和硬碟指標的大小取決於 key-value 中 key 和 value 大小的比值。假如說這個比值是 2:1。那麼 btree 浪費了幾乎 1/3 的空間。
b+tree 針對這個問題的,把葉子節點和內節點的資料結構分開設計,讓葉子節點不存放指標。因此同樣大小的葉子節點,b+tree 所能包含資料數量要比 btree 大。按照上面的假設就是大 1/2。數的深度很可能比 btree 矮,大範圍搜尋或遍歷所需要的載入磁碟的次數也少。
另外,b+tree 還有一個特點是所有資料都存放在葉子節點,這些葉子節點也可以組成一個連結串列,並把這個連結串列的表頭拿出來,方便直訪問資料。有些文章認為這對於範圍搜尋來說是個巨大的優化。但是就我的看法,這個特性最大的作用僅僅是讓程式碼更容易一些,效能上,只會比樹的遍歷差,而不會比樹的遍歷好。因為不管是用指向葉子節點的指標搜,還是用樹的遍歷搜,所搜尋的節點的數量都是幾乎相同的。在相同大小的範圍搜尋的效能,只取決於訪問順序的連續性。從樹根向下遍歷,那麼一次可以取得大量的子節點的範圍,並針對這些節點做訪問排序,得到更好的訪問連續性。如果是沿著指向兄弟節點的指標搜尋,一是兄弟節點也許是後插入的,存放並不一定和自己是連續的,二是隻有每次從硬碟中將該節點載入到記憶體,才知道兄弟節點放在硬碟哪個位置,這又變成了對硬碟的一個隨機的同步操作,效能的下降可想而知。
說 b+tree 因為有指向兄弟節點的指標方便資料庫掃庫這種結論,是不正確的。
還是上程式碼吧,依舊只是在記憶體對資料結構插入刪除查詢的模擬
from collections import deque def bisect_right(a, x, lo=0, hi=None): """Return the index where to insert item x in list a, assuming a is sorted. The return value i is such that all e in a[:i] have e <= x, and all e in a[i:] have e > x. So if x already appears in the list, a.insert(x) will insert just after the rightmost x already there. Optional args lo (default 0) and hi (default len(a)) bound the slice of a to be searched. """ if lo < 0: raise ValueError('lo must be non-negative') if hi is None: hi = len(a) while lo < hi: mid = (lo + hi) // 2 if x < a[mid]: hi = mid else: lo = mid + 1 return lo def bisect_left(a, x, lo=0, hi=None): """Return the index where to insert item x in list a, assuming a is sorted. The return value i is such that all e in a[:i] have e < x, and all e in a[i:] have e >= x. So if x already appears in the list, a.insert(x) will insert just before the leftmost x already there. Optional args lo (default 0) and hi (default len(a)) bound the slice of a to be searched. """ if lo < 0: raise ValueError('lo must be non-negative') if hi is None: hi = len(a) while lo < hi: mid = (lo + hi) // 2 if a[mid] < x: lo = mid + 1 else: hi = mid return lo class InitError(Exception): pass class ParaError(Exception): pass class KeyValue(object): __slots__ = ('key', 'value') def __init__(self, key, value): self.key = key self.value = value def __str__(self): return str((self.key, self.value)) def __cmp__(self, key): if self.key > key: return 1 elif self.key == key: return 0 else: return -1 class Bptree_InterNode(object): def __init__(self, M): if not isinstance(M, int): raise InitError, 'M must be int' if M <= 3: raise InitError, 'M must be greater then 3' else: self.__M = M self.clist = [] # 如果是index節點,儲存 Bptree_InterNode 節點資訊 # leaf節點, 儲存 Bptree_Leaf的資訊 self.ilist = [] # 儲存 索引節點 self.par = None def isleaf(self): return False def isfull(self): return len(self.ilist) >= self.M - 1 def isempty(self): return len(self.ilist) <= (self.M + 1) / 2 - 1 @property def M(self): return self.__M class Bptree_Leaf(object): def __init__(self, L): if not isinstance(L, int): raise InitError, 'L must be int' else: self.__L = L self.vlist = [] self.bro = None self.par = None def isleaf(self): return True def isfull(self): return len(self.vlist) > self.L def isempty(self): return len(self.vlist) <= (self.L + 1) / 2 # 刪除的填充因子 @property def L(self): return self.__L class Bptree(object): def __init__(self, M, L): # M為度, L為填充因子 if L > M: raise InitError, 'L must be less or equal then M' else: self.__M = M self.__L = L self.__root = Bptree_Leaf(L) self.__leaf = self.__root @property def M(self): return self.__M @property def L(self): return self.__L def insert(self, key_value): node = self.__root def split_node(n1): mid = self.M / 2 newnode = Bptree_InterNode(self.M) newnode.ilist = n1.ilist[mid:] newnode.clist = n1.clist[mid:] newnode.par = n1.par for c in newnode.clist: c.par = newnode if n1.par is None: newroot = Bptree_InterNode(self.M) newroot.ilist = [n1.ilist[mid - 1]] newroot.clist = [n1, newnode] n1.par = newnode.par = newroot self.__root = newroot else: i = n1.par.clist.index(n1) n1.par.ilist.insert(i, n1.ilist[mid - 1]) n1.par.clist.insert(i + 1, newnode) n1.ilist = n1.ilist[:mid - 1] n1.clist = n1.clist[:mid] return n1.par def split_leaf(n2): mid = (self.L + 1) / 2 newleaf = Bptree_Leaf(self.L) newleaf.vlist = n2.vlist[mid:] if n2.par == None: newroot = Bptree_InterNode(self.M) newroot.ilist = [n2.vlist[mid].key] newroot.clist = [n2, newleaf] n2.par = newleaf.par = newroot self.__root = newroot else: i = n2.par.clist.index(n2) n2.par.ilist.insert(i, n2.vlist[mid].key) n2.par.clist.insert(i + 1, newleaf) newleaf.par = n2.par n2.vlist = n2.vlist[:mid] n2.bro = newleaf def insert_node(n): if not n.isleaf(): if n.isfull(): insert_node(split_node(n)) else: p = bisect_right(n.ilist, key_value) insert_node(n.clist[p]) else: p = bisect_right(n.vlist, key_value) n.vlist.insert(p, key_value) if n.isfull(): split_leaf(n) else: return insert_node(node) def search(self, mi=None, ma=None): result = [] node = self.__root leaf = self.__leaf if mi is None and ma is None: raise ParaError, 'you need to setup searching range' elif mi is not None and ma is not None and mi > ma: raise ParaError, 'upper bound must be greater or equal than lower bound' def search_key(n, k): if n.isleaf(): p = bisect_left(n.vlist, k) return (p, n) else: p = bisect_right(n.ilist, k) return search_key(n.clist[p], k) if mi is None: while True: for kv in leaf.vlist: if kv <= ma: result.append(kv) else: return result if leaf.bro == None: return result else: leaf = leaf.bro elif ma is None: index, leaf = search_key(node, mi) result.extend(leaf.vlist[index:]) while True: if leaf.bro == None: return result else: leaf = leaf.bro result.extend(leaf.vlist) else: if mi == ma: i, l = search_key(node, mi) try: if l.vlist[i] == mi: result.append(l.vlist[i]) return result else: return result except IndexError: return result else: i1, l1 = search_key(node, mi) i2, l2 = search_key(node, ma) if l1 is l2: if i1 == i2: return result else: result.extend(l.vlist[i1:i2]) return result else: result.extend(l1.vlist[i1:]) l = l1 while True: if l.bro == l2: result.extend(l2.vlist[:i2 + 1]) return result else: result.extend(l.bro.vlist) l = l.bro def traversal(self): result = [] l = self.__leaf while True: result.extend(l.vlist) if l.bro == None: return result else: l = l.bro def show(self): print 'this b+tree is:\n' q = deque() h = 0 q.append([self.__root, h]) while True: try: w, hei = q.popleft() except IndexError: return else: if not w.isleaf(): print w.ilist, 'the height is', hei if hei == h: h += 1 q.extend([[i, h] for i in w.clist]) else: print [v.key for v in w.vlist], 'the leaf is,', hei def delete(self, key_value): def merge(n, i): if n.clist[i].isleaf(): n.clist[i].vlist = n.clist[i].vlist + n.clist[i + 1].vlist n.clist[i].bro = n.clist[i + 1].bro else: n.clist[i].ilist = n.clist[i].ilist + [n.ilist[i]] + n.clist[i + 1].ilist n.clist[i].clist = n.clist[i].clist + n.clist[i + 1].clist n.clist.remove(n.clist[i + 1]) n.ilist.remove(n.ilist[i]) if n.ilist == []: n.clist[0].par = None self.__root = n.clist[0] del n return self.__root else: return n def tran_l2r(n, i): if not n.clist[i].isleaf(): # 將i的最後一個節點追加到i+1的第一個節點 n.clist[i + 1].clist.insert(0, n.clist[i].clist[-1]) n.clist[i].clist[-1].par = n.clist[i + 1] # 追加 i+1的索引值,以及更新n的i+1索引值 n.clist[i + 1].ilist.insert(0, n.clist[i].ilist[-1]) n.ilist[i + 1] = n.clist[i].ilist[-1] n.clist[i].clist.pop() n.clist[i].ilist.pop() else: # 如果 i不空,但是i+1節點為空 # 則將i中的最後一個追加到i+1的第一個中,並重新整理n在i+1的索引值 # 上面的邏輯類似 n.clist[i + 1].vlist.insert(0, n.clist[i].vlist[-1]) n.clist[i].vlist.pop() n.ilist[i] = n.clist[i + 1].vlist[0].key def tran_r2l(n, i): if not n.clist[i].isleaf(): n.clist[i].clist.append(n.clist[i + 1].clist[0]) n.clist[i + 1].clist[0].par = n.clist[i] n.clist[i].ilist.append(n.ilist[i]) n.ilist[i] = n.clist[i + 1].ilist[0] n.clist[i + 1].clist.remove(n.clist[i + 1].clist[0]) n.clist[i + 1].ilist.remove(n.clist[i + 1].ilist[0]) else: n.clist[i].vlist.append(n.clist[i + 1].vlist[0]) n.clist[i + 1].vlist.remove(n.clist[i + 1].vlist[0]) n.ilist[i] = n.clist[i + 1].vlist[0].key def del_node(n, kv): if not n.isleaf(): p = bisect_right(n.ilist, kv) if p == len(n.ilist): if not n.clist[p].isempty(): return del_node(n.clist[p], kv) elif not n.clist[p - 1].isempty(): tran_l2r(n, p - 1) return del_node(n.clist[p], kv) else: return del_node(merge(n, p), kv) else: if not n.clist[p].isempty(): return del_node(n.clist[p], kv) elif not n.clist[p + 1].isempty(): tran_r2l(n, p) return del_node(n.clist[p], kv) else: return del_node(merge(n, p), kv) else: p = bisect_left(n.vlist, kv) try: pp = n.vlist[p] except IndexError: return -1 else: if pp != kv: return -1 else: n.vlist.remove(kv) return 0 del_node(self.__root, key_value) def test(): mini = 2 maxi = 60 testlist = [] for i in range(1, 10): key = i value = i testlist.append(KeyValue(key, value)) mybptree = Bptree(4, 4) for kv in testlist: mybptree.insert(kv) mybptree.delete(testlist[0]) mybptree.show() print '\nkey of this b+tree is \n' print [kv.key for kv in mybptree.traversal()] print [kv.key for kv in mybptree.search(mini,maxi)] if __name__ == '__main__': test()
實現過程:
- 內節點不儲存 key-value,只存放 key
- 沿著內節點搜尋的時候,查到索引相等的數要向樹的右邊走。所以二分查詢要選擇 bisect_right
- 在葉子節點滿的時候,並不是先分裂再插入而是先插入再分裂。因為 b+tree 無法保證分裂的兩個節點的大小都是相等的。在奇數大小的資料分裂的時候右邊的子節點會比左邊的大。如果先分裂再插入無法保證插入的節點一定會插在數量更少的子節點上,滿足節點數量平衡的條件。
- 在刪除資料的時候,b+tree 的左右子節點借資料的方式比 btree 更加簡單有效,只把子節點的子樹直接剪下過來,再把索引變一下就行了,而且葉子節點的兄弟指標也不用動。
轉載自: