B+Tree go

en.wikipedia.org

package b_plus_tree

import (
        "errors"
        "slices"
)

var ErrKeyNotFound = errors.New("key not found")

type Entry[K any, V any] struct {
        key   K
        value V
}

func NewEntry[K any, V any](key K, value V) *Entry[K, V] {
        return &Entry[K, V]{key: key, value: value}
}

func (e *Entry[K, V]) GetKey() K {
        return e.key
}

func (e *Entry[K, V]) GetValue() V {
        return e.value
}

type Node[K any, V any] interface {
        isLeaf() bool
        keyLen() int
        getParent() *InternalNode[K, V]
        setParent(parent *InternalNode[K, V])
}

type InternalNode[K any, V any] struct {
        keys     []K
        children []Node[K, V]
        parent   *InternalNode[K, V]
}

func (i *InternalNode[K, V]) isLeaf() bool {
        return false
}

func (i *InternalNode[K, V]) keyLen() int {
        return len(i.keys)
}

func (i *InternalNode[K, V]) getParent() *InternalNode[K, V] {
        return i.parent
}

func (i *InternalNode[K, V]) setParent(parent *InternalNode[K, V]) {
        i.parent = parent
}

type LeafNode[K any, V any] struct {
        keys     []K
        values   []V
        next     *LeafNode[K, V]
        previous *LeafNode[K, V]
        parent   *InternalNode[K, V]
}

func (l *LeafNode[K, V]) isLeaf() bool {
        return true
}

func (l *LeafNode[K, V]) keyLen() int {
        return len(l.keys)
}

func (l *LeafNode[K, V]) getParent() *InternalNode[K, V] {
        return l.parent
}

func (l *LeafNode[K, V]) setParent(parent *InternalNode[K, V]) {
        l.parent = parent
}


func setChildrenParent[K any, V any](children []Node[K, V], parent *InternalNode[K, V]) {
        for _, child := range children {
                child.setParent(parent)
        }
}

type BPlusTree[K any, V any] struct {
        order int
        cmp   func(a, b K) int
        root  Node[K, V]
}


func NewBPlusTree[K any, V any](order int, cmp func(a, b K) int) *BPlusTree[K, V] {
        if order < 3 {
                panic("order must be >= 3")
        }

        if cmp == nil {
                panic("cmp must not be nil")
        }

        return &BPlusTree[K, V]{
                order: order,
                cmp:   cmp,
                root:  &LeafNode[K, V]{},
        }
}

func (t *BPlusTree[K, V]) Get(key K) (V, error) {
        leaf := t.findLeaf(key)

        index := t.findKeyIndex(leaf.keys, key)

        if index < len(leaf.keys) && t.cmp(leaf.keys[index], key) == 0 {
                return leaf.values[index], nil
        }

        var zero V
        return zero, ErrKeyNotFound
}

func (t *BPlusTree[K, V]) ContainsKey(key K) bool {
        _, err := t.Get(key)
        return err == nil
}

func (t *BPlusTree[K, V]) Put(key K, value V) error {
        leaf := t.findLeaf(key)
        index := t.findKeyIndex(leaf.keys, key)

        if index < len(leaf.keys) && t.cmp(leaf.keys[index], key) == 0 {
                leaf.values[index] = value
                return nil
        }

        leaf.keys = slices.Insert(leaf.keys, index, key)
        leaf.values = slices.Insert(leaf.values, index, value)

        if len(leaf.keys) > t.maxKeys() {
                t.splitLeaf(leaf)
        }

        return nil
}

func (t *BPlusTree[K, V]) Remove(key K) bool {
        leaf := t.findLeaf(key)
        index := t.findKeyIndex(leaf.keys, key)

        if index >= len(leaf.keys) || t.cmp(leaf.keys[index], key) != 0 {
                return false
        }

        minKeyRemoved := index == 0

        leaf.keys = slices.Delete(leaf.keys, index, index+1)
        leaf.values = slices.Delete(leaf.values, index, index+1)

        if leaf == t.root {
                return true
        }


        if minKeyRemoved && len(leaf.keys) > 0 {
                t.updateAncestorKeys(leaf)
        }

        if len(leaf.keys) < t.minLeafKeys() {
                t.rebalanceLeaf(leaf)
        }

        return true
}

func (t *BPlusTree[K, V]) RangeSearch(fromKey, toKey K) []*Entry[K, V] {
        if t.cmp(fromKey, toKey) > 0 {
                panic("fromKey must be <= toKey")
        }

        result := []*Entry[K, V]{}

        leaf := t.findLeaf(fromKey)

        for leaf != nil {
                for i := 0; i < len(leaf.keys); i++ {
                        key := leaf.keys[i]
                        if t.cmp(key, fromKey) < 0 {
                                continue
                        }

                        if t.cmp(key, toKey) > 0 {
                                return result
                        }

                        result = append(result, NewEntry(
                                key,
                                leaf.values[i],
                        ))
                }
                leaf = leaf.next
        }

        return result
}

func (t *BPlusTree[K, V]) findLeaf(key K) *LeafNode[K, V] {

        current := t.root

        for !current.isLeaf() {
                internal := current.(*InternalNode[K, V])

                index := t.upperBound(internal.keys, key)

                current = internal.children[index]
        }

        return current.(*LeafNode[K, V])
}


func (t *BPlusTree[K, V]) upperBound(keys []K, key K) int {
        low, high := 0, len(keys)

        for low < high {
                middle := (low + high) >> 1

                if t.cmp(keys[middle], key) <= 0 {
                        low = middle + 1
                } else {
                        high = middle
                }
        }

        return low
}

func (t *BPlusTree[K, V]) splitLeaf(leaf *LeafNode[K, V]) {
        splitIndex := (len(leaf.keys) + 1) / 2
        right := &LeafNode[K, V]{
                keys:   append([]K{}, leaf.keys[splitIndex:]...),
                values: append([]V{}, leaf.values[splitIndex:]...),
        }

        leaf.keys = leaf.keys[:splitIndex]
        leaf.values = leaf.values[:splitIndex]

        right.next = leaf.next

        if leaf.next != nil {
                leaf.next.previous = right
        }

        leaf.next = right
        right.previous = leaf

        separator := right.keys[0]
        t.insertIntoParent(
                leaf,
                separator,
                right,
        )
}

func (t *BPlusTree[K, V]) insertIntoParent(left Node[K, V], separator K, right Node[K, V]) {
        if left == t.root {
                newRoot := &InternalNode[K, V]{}
                newRoot.keys = append(newRoot.keys, separator)
                newRoot.children = append(newRoot.children, left)
                newRoot.children = append(newRoot.children, right)
                left.setParent(newRoot)
                right.setParent(newRoot)
                t.root = newRoot
                return
        }

        parent := left.getParent()

        childIndex := slices.Index(parent.children, left)
        parent.keys = slices.Insert(parent.keys, childIndex, separator)
        parent.children = slices.Insert(parent.children, childIndex+1, right)
        right.setParent(parent)

        if len(parent.children) > t.order {
                t.splitInternal(parent)
        }
}

func (t *BPlusTree[K, V]) splitInternal(node *InternalNode[K, V]) {
        middleIndex := len(node.keys) / 2
        separator := node.keys[middleIndex]
        right := &InternalNode[K, V]{}

        for i := middleIndex + 1; i < len(node.keys); i++ {
                right.keys = append(right.keys, node.keys[i])
        }

        for i := middleIndex + 1; i < len(node.children); i++ {
                right.children = append(right.children, node.children[i])
        }
        setChildrenParent(right.children, right)

        node.keys = node.keys[:middleIndex]
        node.children = node.children[:middleIndex+1]

        t.insertIntoParent(
                node,
                separator,
                right,
        )
}

func (t *BPlusTree[K, V]) rebalanceLeaf(leaf *LeafNode[K, V]) {
        parent := leaf.getParent()

        index := slices.IndexFunc(parent.children, func(child Node[K, V]) bool {
                return child == Node[K, V](leaf)
        })

        var leftSibling *LeafNode[K, V]
        var rightSibling *LeafNode[K, V]

        if index > 0 {
                node := parent.children[index-1]

                if node.isLeaf() {
                        leftSibling = node.(*LeafNode[K, V])
                }
        }

        if index+1 < len(parent.children) {
                node := parent.children[index+1]

                if node.isLeaf() {
                        rightSibling = node.(*LeafNode[K, V])
                }
        }

        // 左兄弟から借りる
        if leftSibling != nil && len(leftSibling.keys) > t.minLeafKeys() {
                last := len(leftSibling.keys) - 1

                borrowedKey := leftSibling.keys[last]
                borrowedValue := leftSibling.values[last]

                leftSibling.keys = leftSibling.keys[:last]
                leftSibling.values = leftSibling.values[:last]

                leaf.keys = append([]K{borrowedKey}, leaf.keys...)
                leaf.values = append([]V{borrowedValue}, leaf.values...)

                parent.keys[index-1] = leaf.keys[0]

                return
        }

        // 右兄弟から借りる
        if rightSibling != nil && len(rightSibling.keys) > t.minLeafKeys() {
                borrowedKey := rightSibling.keys[0]
                borrowedValue := rightSibling.values[0]

                rightSibling.keys = rightSibling.keys[1:]
                rightSibling.values = rightSibling.values[1:]

                leaf.keys = append(leaf.keys, borrowedKey)
                leaf.values = append(leaf.values, borrowedValue)

                parent.keys[index] = rightSibling.keys[0]

                return
        }

        // 借りられなければmerge
        if leftSibling != nil {
                t.mergeLeaves(
                        parent,
                        index-1,
                        leftSibling,
                        leaf,
                )
        } else if rightSibling != nil {
                t.mergeLeaves(
                        parent,
                        index,
                        leaf,
                        rightSibling,
                )
        }
}

func (t *BPlusTree[K, V]) mergeLeaves(parent *InternalNode[K, V], separatorIndex int, left, right *LeafNode[K, V]) {
        left.keys = append(left.keys, right.keys...)
        left.values = append(left.values, right.values...)

        left.next = right.next

        if right.next != nil {
                right.next.previous = left
        }

        // 親からseparatorを削除
        parent.keys = slices.Delete(parent.keys, separatorIndex, separatorIndex+1)
        parent.children = slices.Delete(parent.children, separatorIndex+1, separatorIndex+2)

        if parent == t.root && len(parent.keys) == 0 {
                t.root = left
                left.previous = nil
                left.setParent(nil)
                return
        }

        if parent != t.root && len(parent.children) < t.minInternalChildren() {
                t.rebalanceInternal(parent)
        }

        t.updateAncestorKeys(left)
}

func (t *BPlusTree[K, V]) rebalanceInternal(node *InternalNode[K, V]) {
        if node == t.root {
                if len(node.keys) == 0 && len(node.children) == 1 {
                        t.root = node.children[0]
                        node.children[0].setParent(nil)
                }
                return
        }

        parent := node.getParent()

        index := slices.IndexFunc(parent.children, func(child Node[K, V]) bool {
                return child == Node[K, V](node)
        })

        var leftSibling *InternalNode[K, V] = nil
        var rightSibling *InternalNode[K, V] = nil

        if index > 0 {
                sibling := parent.children[index-1]

                if !sibling.isLeaf() {
                        leftSibling = sibling.(*InternalNode[K, V])
                }
        }

        if index+1 < len(parent.children) {
                sibling := parent.children[index+1]

                if !sibling.isLeaf() {
                        rightSibling = sibling.(*InternalNode[K, V])
                }
        }

        // 左兄弟から借りる
        if leftSibling != nil && len(leftSibling.children) > t.minInternalChildren() {
                parentSeparator := parent.keys[index-1]
                lastChild := len(leftSibling.children) - 1
                movedChild := leftSibling.children[lastChild]
                leftSibling.children = leftSibling.children[:lastChild]

                lastKey := len(leftSibling.keys) - 1
                newSeparator := leftSibling.keys[lastKey]
                leftSibling.keys = leftSibling.keys[:lastKey]

                node.keys = append([]K{parentSeparator}, node.keys...)
                node.children = append([]Node[K, V]{movedChild}, node.children...)
                movedChild.setParent(node)

                parent.keys[index-1] = newSeparator

                return
        }

        // 右兄弟から借りる
        if rightSibling != nil && len(rightSibling.children) > t.minInternalChildren() {
                parentSeparator := parent.keys[index]

                movedChild := rightSibling.children[0]
                rightSibling.children = rightSibling.children[1:]

                newSeparator := rightSibling.keys[0]
                rightSibling.keys = rightSibling.keys[1:]

                node.keys = append(node.keys, parentSeparator)
                node.children = append(node.children, movedChild)
                movedChild.setParent(node)

                parent.keys[index] = newSeparator

                return
        }

        //借りられなければmerge
        if leftSibling != nil {
                t.mergeInternal(
                        parent,
                        index-1,
                        leftSibling,
                        node,
                )
        } else if rightSibling != nil {
                t.mergeInternal(
                        parent,
                        index,
                        node,
                        rightSibling,
                )
        }
}

// 内部ノードを結合
func (t *BPlusTree[K, V]) mergeInternal(parent *InternalNode[K, V], separatorIndex int, left, right *InternalNode[K, V]) {
        separator := parent.keys[separatorIndex]
        parent.keys = slices.Delete(parent.keys, separatorIndex, separatorIndex+1)

        left.keys = append(left.keys, separator)
        left.keys = append(left.keys, right.keys...)
        left.children = append(left.children, right.children...)
        setChildrenParent(right.children, left)

        parent.children = slices.Delete(parent.children, separatorIndex+1, separatorIndex+2)

        if parent == t.root && len(parent.keys) == 0 {
                t.root = left
                left.setParent(nil)
                return
        }

        if parent != t.root && len(parent.children) < t.minInternalChildren() {
                t.rebalanceInternal(parent)
        }
}

func (t *BPlusTree[K, V]) updateAncestorKeys(node Node[K, V]) {
        if node == t.root {
                return
        }

        parent := node.getParent()

        if parent == nil {
                return
        }

        index := slices.IndexFunc(parent.children, func(child Node[K, V]) bool {
                return child == node
        })

        if index > 0 && node.keyLen() != 0 {
                parent.keys[index-1] = t.firstKey(node)
                return
        }

        if index == 0 {
                t.updateAncestorKeys(parent)
        }
}

func (t *BPlusTree[K, V]) firstKey(node Node[K, V]) K {
        current := node

        for !current.isLeaf() {
                current = current.(*InternalNode[K, V]).children[0]
        }
        leaf := current.(*LeafNode[K, V])

        return leaf.keys[0]
}

func (t *BPlusTree[K, V]) findKeyIndex(keys []K, key K) int {
        var low int = 0
        var high int = len(keys)

        for low < high {
                middle := (low + high) >> 1

                if t.cmp(keys[middle], key) < 0 {
                        low = middle + 1
                } else {
                        high = middle
                }
        }

        return low
}

func (t *BPlusTree[K, V]) maxKeys() int {
        return t.order - 1
}

func (t *BPlusTree[K, V]) minLeafKeys() int {
        return t.order / 2
}

func (t *BPlusTree[K, V]) minInternalChildren() int {
        return (t.order + 1) / 2
}

func (t *BPlusTree[K, V]) entries() []*Entry[K, V] {
        result := []*Entry[K, V]{}
        leaf := t.firstLeaf()

        for leaf != nil {
                for i := 0; i < len(leaf.keys); i++ {
                        result = append(result, NewEntry(leaf.keys[i], leaf.values[i]))
                }
                leaf = leaf.next
        }

        return result
}

func (t *BPlusTree[K, V]) firstLeaf() *LeafNode[K, V] {
        current := t.root

        for !current.isLeaf() {
                current = current.(*InternalNode[K, V]).children[0]
        }

        return current.(*LeafNode[K, V])

}

追記

AIにベンチを取ってもらった

B+木 (B+Tree) 実装解説

b_plus_tree.go に実装されたジェネリックな B+木 (BPlusTree[K, V]) について、アルゴリズムの解説と、テスト・ベンチマークの実行結果の解説をまとめる。

1. アルゴリズム解説

1.1 データ構造

• InternalNode[K, V]: keys []K と children []Node[K, V] を持つ内部ノード。len(children) == len(keys) + 1 が常に成り立つ（B木の一般形と同じ）。
• LeafNode[K, V]: keys []K と values []V を持つ葉ノード。さらに next / previous で兄弟の葉ノード同士を連結リストとして繋いでいる。これが B+木特有の構造で、範囲検索を葉のリストの走査だけで済ませるためのもの。
• 各ノードは parent *InternalNode[K, V] を保持し、下から上へ辿れる（分割・再構成・別ノードの先頭キー更新に使う）。
• order（次数）によって「1ノードあたりの最大子数」が決まる。
  • 最大キー数（葉）: order - 1
  • 最小キー数（葉、root除く）: order / 2
  • 最小子数（内部ノード、root除く）: (order + 1) / 2

B木との違いは、すべてのデータ（value）が葉ノードにのみ格納される点。内部ノードのキーは「どちらの部分木を辿るか」を決めるための道しるべ（セパレータ）に過ぎず、実際のデータを持たない。これにより

1. 内部ノードが軽量になり、1ノードに収まるセパレータ数（=ファンアウト）を増やせる → 木の高さが低くなる → ディスクI/Oやキャッシュミスが減る（実務でDB/ファイルシステムのインデックスに多用される理由）。
2. 葉が連結リストになっているため、範囲検索（RangeSearch）が「開始位置を1回だけ二分探索し、あとは next を辿るだけ」で実現できる。

1.2 検索 (Get / findLeaf)

findLeaf はルートから葉まで、各内部ノードで upperBound（二分探索でキーより真に大きい最初の位置を探す = 何番目の子に降りるか）を使って一直線に降りていく。計算量は O(log n)。葉に着いたら findKeyIndex（下限二分探索）でキー位置を特定する。

findLeaf: ルート → 内部ノード upperBound で子を選択 → ... → 葉
Get:      findLeaf → findKeyIndex で二分探索 → 一致すれば値を返す

1.3 挿入 (Put)

1. findLeaf で挿入すべき葉を特定。
2. findKeyIndex で挿入位置を二分探索。既存キーなら値を上書き（更新）。
3. 新規キーならソート順を保ったまま slices.Insert で挿入。
4. 葉のキー数が maxKeys()（= order-1）を超えたら splitLeaf で葉を分割。
   • 右半分を新しい葉として切り出し、next/previous の連結を繋ぎ直す。
   • 新しい葉の先頭キーを「セパレータ」として親に挿入 (insertIntoParent)。
5. 親（内部ノード）の子数が order を超えたら splitInternal で内部ノードも同様に分割し、これを再帰的に繰り返す（伝播）。ルートが分割されると新しいルートが1つ生成され、木の高さが1増える。

挿入は常に葉から根に向かって伝播するボトムアップ方式で、最悪計算量は O(log n)。

1.4 削除 (Remove)

1. findLeaf + findKeyIndex でキーを特定して削除。
2. 削除したキーが葉の先頭キーだった場合、親（および必要なら祖先）が持つセパレータが古くなるので updateAncestorKeys で修正する。これは borrow（後述）によるセパレータ更新が「自分の右側」しか直さないため、削除の場合は先に明示的に直しておく必要がある、という設計上の工夫。
3. 葉のキー数が minLeafKeys() を下回ったら rebalanceLeaf で再調整:
   • 左兄弟に余裕があれば1キー借りる（borrow）。
   • なければ右兄弟から借りる。
   • どちらも余裕がなければ、兄弟とマージ (mergeLeaves)する。
4. マージで親の子が減り minInternalChildren() を下回ったら rebalanceInternal で内部ノードも同様に borrow/merge を再帰的に行う。
5. ルートの子が1つだけになったらルートを1段下げる（木が低くなる）。

削除も挿入と対称的に O(log n)。borrow/merge によって「各ノードは半分以上埋まっている」という B木の不変条件を常に維持している（これによって最悪ケースでも木の高さが O(log n) に保たれる）。

1.5 範囲検索 (RangeSearch)

fromKey を含む葉を findLeaf で1回だけ特定し、そこから next ポインタを辿って toKey を超えるまで線形に集める。内部ノードを再度辿る必要がないのが B+木の最大の利点で、計算量は O(log n + m)（m は結果件数）。

2. テスト

b_plus_tree_test.go に以下を実装・実行し、全件 PASS、カバレッジ 98.1% を確認した。

$ go test ./... -v
$ go test -cover ./...
ok      example.com/b_plus_tree 0.853s  coverage: 98.1% of statements

テスト内容

テスト	内容
TestNewBPlusTreePanicsOnSmallOrder / ...NilCmp	不正なコンストラクタ引数で panic することの確認
TestPutAndGet / TestGetMissingKeyReturnsError	基本の Put/Get
TestPutUpdatesExistingKey	既存キーへの Put が上書きになり、エントリ数が増えないこと
TestPutAllowsZeroValue	ゼロ値でも正しく挿入されること
TestPutMaintainsSortedOrderAcrossSplits	order=3,4,5,7,16 で 500件をランダム順に挿入し、挿入の都度 validateInvariants で構造不変条件（子数=キー数+1、最小占有率、親ポインタ整合性）を検証しつつ、最終的に全エントリがソート順であることを確認
TestRemoveExistingAndMissingKeys	削除の基本動作、二重削除・存在しないキーの削除が false を返すこと
TestDrainToEmptyPreservesInvariants	order×seed の組み合わせで挿入→シャッフル削除→空になるまでの全過程で不変条件を検証（最も厳しいテスト）
TestRangeSearch*	範囲検索の基本動作・空区間・不正区間でのpanic・マップとの突合検証（ランダム300件×50トライアル）
TestRandomOperationsAgainstReferenceMap	Put/Removeをランダムに3000回混在実行し、都度 map[int]string という信頼できる参照実装と突き合わせ、かつ構造不変条件も検証する統合テスト
TestLeafNextPreviousLinksAfterMutation	挿入・削除を経ても葉の双方向連結リスト (next/previous) が壊れていないことを確認

validateInvariants はテストコード内のヘルパーで、木全体を再帰的に辿り「内部ノードの子数がキー数+1」「rootを除き最小占有率を満たす」「親子ポインタの整合性」を検証する。単なる入出力テストだけでなく内部構造の正しさを継続的に検証しているのが特徴。

3. ベンチマーク

b_plus_tree_bench_test.go を以下のコマンドで実行した（AMD Ryzen 5 5500U、Go標準の testing.B によるデフォルト時間ベース実行、-benchmem でメモリ計測も有効化）。

$ go test -run=^$ -bench=. -benchmem ./...

対象パラメータ: - order（次数）: 4, 16, 64, 128 - size（事前投入件数）: 1,000 / 5,000 / 20,000

3.1 Put（挿入）

ベンチマーク	order=4	order=16	order=64	order=128
PutSequential (ns/op)	439.4	201.3	154.5	142.9
PutSequential (allocs/op)	4	0	0	0
PutRandom (ns/op)	1270	674.9	581.0	562.4
PutRandom (allocs/op)	2	0	0	0

解説: - order を上げるほど速くなる傾向が明確。order が大きいほど1ノードに多くのキーを詰め込めるため、同じ件数を挿入したときの木の高さが低くなり、findLeaf で辿る内部ノードの段数（ポインタ参照とキャッシュミスの回数）が減る。一方でノード分割 (splitLeaf/splitInternal) の頻度自体は減るので、slices.Insert によるスライス移動コストの総量も相対的に減る。 - Sequential（昇順キー）が Random（乱数キー）より速い理由は、逐次挿入では毎回「末尾付近」への挿入になりやすく、Go の append/スライス操作のメモリアクセスパターンが局所的でキャッシュに乗りやすいため。乱数挿入は木全体にまんべんなくアクセスするためキャッシュ効率が落ち、また分割位置も分散するので分割コストが平均化されにくい。 - allocs/op がほぼ0であるのは、b.N 回のうち大半が「スライスの再拡張(append)が不要な単純上書き・単純insert」で収まっており、Goのスライス事前確保（capacity）が効いているため。order=4だけ数allocsが残るのは、ノード分割自体の頻度が高く（maxKeys()=3ごとに分割が起きる）、新規ノード (&LeafNode{}, &InternalNode{}) や append([]K{}, ...) によるコピー割り当てが相対的に多く発生するため。

3.2 Get（検索）

order＼size	1,000	5,000	20,000
order=4	118.8 ns	155.1 ns	207.3 ns
order=16	90.53 ns	118.7 ns	144.3 ns
order=64	85.45 ns	107.0 ns	122.4 ns
order=128	86.26 ns	102.1 ns	121.9 ns

解説: - 全ケースで 0 B/op, 0 allocs/op。Get は読み取り専用でメモリ確保が一切発生しないことを裏付けている。 - サイズが大きくなるほど遅くなるのは O(log n) の理論通り。ただし増加は緩やかで、20倍のデータ量（1,000→20,000）でも所要時間は2倍にも満たない（対数的な伸び）。 - order が大きいほど速いのも Put と同じ理由（木の高さが低い）。ただし order=64 と order=128 でほぼ差がない、あるいは128がわずかに遅い場合があるのは、1ノードあたりのキー数が増えすぎるとノード内の二分探索コスト（upperBound）自体が無視できなくなり、「段数削減の恩恵」と「ノード内探索コストの増加」が釣り合ってくるため。B+木の次数を実運用でチューニングする際に「大きければ大きいほど良い」わけではない典型例。

3.3 Remove（削除）

order＼size	1,000	5,000	20,000
order=4	276.8 ns (19 B)	341.9 ns (19 B)	431.1 ns (19 B)
order=16	208.0 ns (66 B)	217.7 ns (69 B)	249.2 ns (69 B)
order=64	207.1 ns (155 B)	283.1 ns (195 B)	313.8 ns (198 B)
order=128	222.8 ns (257 B)	320.9 ns (363 B)	434.7 ns (383 B)

解説: - 削除は Get より重い。理由は削除後に rebalanceLeaf/rebalanceInternal（borrow/merge）や updateAncestorKeys が発生しうるため、単純な二分探索より処理経路が長い。 - 興味深いのは allocs/op はほぼ0なのに B/op（確保バイト数）はorderが大きいほど増える点。これは slices.Delete や mergeInternal/mergeLeaves 内の append がスライスの再アロケーションを毎回起こしているわけではないが、Goベンチマークの B/op はGC割り当て総量の推定値であり、slices.Insert/Delete/append がキャパシティ不足時にコピーする一時領域や、slices.IndexFunc のクロージャなどが積算されている。特に order=128 ではノード内のキー・子スライスが大きいため、borrow/merge時に動かすデータ量（append(left.keys, right.keys...) 等）自体が大きく、結果的に確保サイズも増える。 - サイズが大きくなるほど遅くなるのはGetと同様に O(log n) の探索コスト増加によるもの。

3.4 RangeSearch（範囲検索）

20,000件のツリーに対し、span（範囲の広さ）を 10 / 100 / 1000 で計測。

order＼span	10	100	1000
order=4	1436 ns (16 allocs)	9046 ns (109 allocs)	78904 ns (1012 allocs)
order=16	1080 ns (16 allocs)	6978 ns (109 allocs)	60939 ns (1012 allocs)
order=64	1041 ns (16 allocs)	6724 ns (109 allocs)	59749 ns (1012 allocs)
order=128	1035 ns (16 allocs)	6466 ns (109 allocs)	56697 ns (1012 allocs)

解説: - span（結果件数）が10倍になると所要時間もほぼ10倍になっており、O(log n + m) の m 項（結果件数に比例するコスト）が支配的であることがはっきり出ている。これは「最初の葉を見つけたあとは next を辿るだけ」という B+木の設計が効いている証拠で、内部ノードを再訪する必要がないため件数に対してほぼ線形。 - allocs/op が span にほぼ比例（span=10→16 allocs、span=1000→1012 allocs）しているのは、RangeSearch が result スライスに append する際、Goのスライス拡張戦略（capacity 2倍拡張）によって定期的に再アロケーションが起きているため。事前に make([]*Entry[K,V], 0, estimatedSize) のように容量を見積もって確保すれば allocs は減らせる余地がある（現状は result := []*Entry[K, V]{} で容量0スタート）。 - order による差は Get ほど大きくないが、それでも order=4 が最も遅い。これは最初の findLeaf 呼び出し部分（O(log n)）のコストが低次数だと大きいため。span が広がるほど「葉走査コスト」が支配的になり、order間の差（＝木の高さの差）は相対的に薄まっていく（span=10 では order=4 が order=128 の約1.4倍だが、span=1000ではほぼ差がない）。

4. まとめ

• 実装は挿入・削除ともに B+木の不変条件（キー数/子数のバランス、葉の連結リストの整合性）を正しく維持しており、3000回のランダムPut/Remove操作とGoの map を突き合わせる統合テストでも一致することを確認した。
• ベンチマークからは以下の設計上の示唆が得られる。
  • order（次数）を上げると Put/Get は速くなるが、上げすぎるとノード内探索コストが無視できなくなり頭打ちになる（order=64 と 128 でほぼ差がない、または逆転するケースがある）。
  • RangeSearch は結果件数に比例するコストが支配的で、B+木の葉連結リスト構造の恩恵がベンチマーク上も明確に確認できた。
  • Remove は Put/Get より重く、特に高次数ではノード間のデータ移動量（borrow/merge時の append）がメモリ確保量に反映される。

B Tree go

en.wikipedia.org

package b_tree

type BTree struct {
        t    int // minimum degree
        root *Node
}

type Node struct {
        keys     []int
        children []*Node
        keyCount int
        leaf     bool
}

func NewBTree(minimumDegree int) *BTree {
        if minimumDegree < 2 {
                panic("minimumDegree must be >= 2")
        }
        b := &BTree{t: minimumDegree}
        b.root = b.newNode(true)
        return b
}

func (bt *BTree) newNode(leaf bool) *Node {
        return &Node{
                leaf:     leaf,
                keys:     make([]int, 2*bt.t-1), // 最大キー数
                children: make([]*Node, 2*bt.t), // 最大子ノード数
        }
}

func (bt *BTree) Contains(value int) bool {
        return bt.search(bt.root, value)
}

func (bt *BTree) search(node *Node, value int) bool {
        var index int = 0

        // value以上のキーを探す
        for index < node.keyCount && value > node.keys[index] {
                index++
        }

        // 一致
        if index < node.keyCount && value == node.keys[index] {
                return true
        }

        // 葉ならこれ以上探せない
        if node.leaf {
                return false
        }

        // 子ノードへ降りる
        return bt.search(
                node.children[index],
                value,
        )
}

func (bt *BTree) Insert(value int) bool {
        if bt.Contains(value) {
                return false
        }

        currentRoot := bt.root

        if currentRoot.keyCount == 2*bt.t-1 {
                newRoot := bt.newNode(false)
                newRoot.children[0] = currentRoot

                // 古いルートを分割
                bt.splitChild(
                        newRoot,
                        0,
                )

                bt.root = newRoot

                bt.insertNonFull(
                        newRoot,
                        value,
                )
        } else {
                bt.insertNonFull(
                        currentRoot,
                        value,
                )
        }

        return true
}

func (bt *BTree) insertNonFull(node *Node, value int) {
        index := node.keyCount - 1

        if node.leaf {
                for index >= 0 && value < node.keys[index] {
                        node.keys[index+1] = node.keys[index]
                        index--
                }

                node.keys[index+1] = value
                node.keyCount++
                return
        }

        // どの子ノードへ降りるか決める
        for index >= 0 && value < node.keys[index] {
                index--
        }

        index++

        if node.children[index].keyCount == 2*bt.t-1 {
                bt.splitChild(
                        node,
                        index,
                )

                if value > node.keys[index] {
                        index++
                }
        }

        bt.insertNonFull(
                node.children[index],
                value,
        )
}

// 満杯の子ノードを分割する
func (bt *BTree) splitChild(parent *Node, childIndex int) {
        fullChild := parent.children[childIndex]
        rightNode := bt.newNode(fullChild.leaf)

        rightNode.keyCount = bt.t - 1

        for i := 0; i < bt.t-1; i++ {
                rightNode.keys[i] = fullChild.keys[i+bt.t]
        }

        if !fullChild.leaf {
                for i := 0; i < bt.t; i++ {
                        rightNode.children[i] = fullChild.children[i+bt.t]
                }
        }

        fullChild.keyCount = bt.t - 1

        for i := parent.keyCount; i >= childIndex+1; i-- {
                parent.children[i+1] = parent.children[i]
        }

        parent.children[childIndex+1] = rightNode

        for i := parent.keyCount - 1; i >= childIndex; i-- {
                parent.keys[i+1] = parent.keys[i]
        }

        parent.keys[childIndex] = fullChild.keys[bt.t-1]
        parent.keyCount++
}

// 値を削除する
func (bt *BTree) Delete(value int) bool {
        if !bt.Contains(value) {
                return false
        }

        bt.deleteValue(bt.root, value)

        if bt.root.keyCount == 0 && !bt.root.leaf {
                bt.root = bt.root.children[0]
        }

        return true
}

func (bt *BTree) deleteValue(node *Node, value int) {
        index := bt.findKeyIndex(
                node,
                value,
        )

        if index < node.keyCount && node.keys[index] == value {
                if node.leaf {
                        bt.removeFromLeaf(
                                node,
                                index,
                        )
                } else {
                        bt.removeFromInternalNode(
                                node,
                                index,
                        )
                }

                return
        }

        // 葉まで来ても見つからない
        if node.leaf {
                return
        }

        lastChild := index == node.keyCount

        if node.children[index].keyCount < bt.t {
                bt.fill(node, index)
        }

        if lastChild && index > node.keyCount {
                bt.deleteValue(
                        node.children[index-1],
                        value,
                )
        } else {
                bt.deleteValue(
                        node.children[index],
                        value,
                )
        }
}

func (bt *BTree) findKeyIndex(node *Node, value int) int {
        var index int = 0

        for index < node.keyCount && node.keys[index] < value {
                index++
        }

        return index
}

func (bt *BTree) removeFromLeaf(node *Node, index int) {
        for i := index + 1; i < node.keyCount; i++ {
                node.keys[i-1] = node.keys[i]
        }
        node.keyCount--
}

func (bt *BTree) removeFromInternalNode(node *Node, index int) {
        value := node.keys[index]

        if node.children[index].keyCount >= bt.t {
                predecessor := bt.getPredecessor(
                        node,
                        index,
                )

                node.keys[index] = predecessor
                bt.deleteValue(
                        node.children[index],
                        predecessor,
                )

                return
        }

        if node.children[index+1].keyCount >= bt.t {
                successor := bt.getSuccessor(
                        node,
                        index,
                )

                node.keys[index] = successor
                bt.deleteValue(
                        node.children[index+1],
                        successor,
                )

                return
        }

        bt.merge(node, index)

        bt.deleteValue(
                node.children[index],
                value,
        )
}

func (bt *BTree) getPredecessor(node *Node, index int) int {
        current := node.children[index]

        for !current.leaf {
                current = current.children[current.keyCount]
        }

        return current.keys[current.keyCount-1]
}

func (bt *BTree) getSuccessor(node *Node, index int) int {
        current := node.children[index+1]

        for !current.leaf {
                current = current.children[0]
        }

        return current.keys[0]
}

func (bt *BTree) fill(node *Node, index int) {
        // 左の兄弟から借りられる
        if index != 0 && node.children[index-1].keyCount >= bt.t {
                bt.borrowFromPrevious(node, index)
                return
        }

        // 右の兄弟から借りられる
        if index != node.keyCount && node.children[index+1].keyCount >= bt.t {
                bt.borrowFromNext(node, index)
                return
        }

        // 借りられない場合は結合
        if index != node.keyCount {
                bt.merge(node, index)
        } else {
                bt.merge(node, index-1)
        }
}

func (bt *BTree) borrowFromPrevious(node *Node, index int) {
        child := node.children[index]
        sibling := node.children[index-1]

        for i := child.keyCount - 1; i >= 0; i-- {
                child.keys[i+1] = child.keys[i]
        }

        if !child.leaf {
                for i := child.keyCount; i >= 0; i-- {
                        child.children[i+1] = child.children[i]
                }
        }

        child.keys[0] = node.keys[index-1]

        if !child.leaf {
                child.children[0] = sibling.children[sibling.keyCount]
        }

        node.keys[index-1] = sibling.keys[sibling.keyCount-1]

        child.keyCount++
        sibling.keyCount--
}

func (bt *BTree) borrowFromNext(node *Node, index int) {
        child := node.children[index]
        sibling := node.children[index+1]

        child.keys[child.keyCount] = node.keys[index]

        if !child.leaf {
                child.children[child.keyCount+1] = sibling.children[0]
        }

        node.keys[index] = sibling.keys[0]

        for i := 1; i < sibling.keyCount; i++ {
                sibling.keys[i-1] = sibling.keys[i]
        }

        if !sibling.leaf {
                for i := 1; i <= sibling.keyCount; i++ {
                        sibling.children[i-1] = sibling.children[i]
                }
        }

        child.keyCount++
        sibling.keyCount--
}

func (bt *BTree) merge(node *Node, index int) {
        leftChild := node.children[index]
        rightChild := node.children[index+1]

        leftChild.keys[bt.t-1] = node.keys[index]

        for i := 0; i < rightChild.keyCount; i++ {
                leftChild.keys[i+bt.t] = rightChild.keys[i]
        }

        if !leftChild.leaf {
                for i := 0; i <= rightChild.keyCount; i++ {
                        leftChild.children[i+bt.t] = rightChild.children[i]
                }
        }

        for i := index + 1; i < node.keyCount; i++ {
                node.keys[i-1] = node.keys[i]
        }

        for i := index + 2; i <= node.keyCount; i++ {
                node.children[i-1] = node.children[i]
        }

        leftChild.keyCount += rightChild.keyCount + 1

        node.keyCount--
}

func (bt *BTree) Inorder() []int {
        return bt.inorder(bt.root, []int{})
}

func (bt *BTree) inorder(node *Node, result []int) []int {
        var i int

        for i = 0; i < node.keyCount; i++ {
                if !node.leaf {
                        result = bt.inorder(node.children[i], result)
                }

                result = append(result, node.keys[i])
        }

        if !node.leaf {
                result = bt.inorder(node.children[i], result)
        }

        return result
}

追記

AIにベンチ取ってもらった

B-Tree 実装ノート — アルゴリズム解説とテスト/ベンチマーク結果

対象実装: b_tree.go テスト: b_tree_test.go ベンチマーク: b_tree_bench_test.go

1. B-Tree とは

B-Tree は二分探索木を一般化し、1ノードに複数のキーと複数の子を持たせることで 木の高さを抑えた平衡探索木。ディスク/ページ単位のI/Oを最小化する目的で 考案されたが、キャッシュライン単位でのメモリアクセスを減らす目的でも インメモリのインデックス構造として使われる（RDBMSのインデックス、ファイルシステムなど）。

最小次数 t と不変条件

この実装では BTree.t（b_tree.go:4）が最小次数（minimum degree）を表す。 NewBTree は t >= 2 を要求する（b_tree.go:16）。

各ノード（ルートを除く）は以下を満たす:

項目	下限	上限
キー数	t - 1	2t - 1
子の数（内部ノード）	t	2t

• ルートのみキー数が 1 個まで許容される（それ以外のノードは最低 t-1 個必要）。
• 全ての葉は同じ深さに位置する（=完全に高さ平衡）。
• ノード内のキーは昇順に並び、子ポインタ children[i] は keys[i-1] より小さく keys[i] より大きい値の部分木を指す。

Node 構造体（b_tree.go:8-13）はこれをそのまま表現しており、 keys/children スライスは newNode（b_tree.go:24-30）で 最大サイズ（2t-1, 2t）ぶん確保され、実際に使われている要素数は keyCount で管理する（スライス長そのものは使わない、固定長バッファ的な使い方）。

高さの上界

n 個のキーを持つ B-Tree の高さは O(log_t n) で抑えられる。 これが探索・挿入・削除が全て O(log n) になる理由であり、 二分探索木と違って「常に平衡している」ため最悪計算量が保証される （AVL木や赤黒木と同様の性質を、より少ない木の高さで実現する）。

2. 各操作の実装詳細

探索 Contains / search（b_tree.go:32-59）

ノード内を線形（または二分）探索し、 - 一致すれば true - 見つからず葉ならば false - 見つからず内部ノードなら該当区間の子へ再帰

計算量: ノードあたり O(t)（線形探索）、木の高さが O(log_t n) なので 探索全体は O(t log_t n)。

挿入 Insert / insertNonFull / splitChild（b_tree.go:61-163）

この実装は preemptive split（先読み分割）方式を採用している。 これはルートから葉へ降りる過程で、満杯（keyCount == 2t-1）のノードに 出会うたびにその場で分割してから先へ進む方式で、 「分割してからバックトラックする」必要がなく1パスで完結する。

1. ルートが満杯なら、新しいルートを作りルートを分割してから木の高さが1増える （b_tree.go:68-83）。
2. insertNonFull は葉に着くまで、通る子ノードが満杯なら splitChild で事前に分割する（b_tree.go:115-124）。
3. splitChild は満杯ノード（2t-1 個のキー）を中央値で2つに割り、 中央値を親に押し上げる（b_tree.go:133-163）。
4. 重複キーは Contains で事前チェックして拒否する（b_tree.go:62-64） ので Insert は多重集合ではなく集合として振る舞う。

計算量: O(t log_t n)（探索と同様、各レベルでの分割・シフトが O(t)）。

削除 Delete / deleteValue（b_tree.go:166-404）

削除は挿入よりケースが多い。標準的な B-Tree 削除アルゴリズム （CLRS 18.3）に沿っている。

• 削除対象のキーが葉にあれば単純に詰める（removeFromLeaf）。
• 削除対象のキーが内部ノードにあれば、
  • 左の子部分木が t 個以上キーを持てば「先行順の最大キー」で置換 （getPredecessor）、
  • なければ右の子部分木が t 個以上キーを持てば「後続順の最小キー」で置換 （getSuccessor）、
  • どちらも t-1 個しかなければ左右の子を結合してから再帰的に削除 （merge）。
• 削除対象のキーがそのノードになければ、該当する子へ降りる前に、 その子のキー数が t-1（最小）なら先に補充する（fill）:
  • 隣接兄弟が t 個以上キーを持っていれば1つ借りる （borrowFromPrevious / borrowFromNext）、
  • 借りられなければ兄弟と結合する（merge）。
• ルートが空になり葉でなくなった場合、唯一の子を新しいルートにする （b_tree.go:173-175）。木の高さが1減る唯一のケース。

この「先に補充してから降りる」方式により、挿入と同様に バックトラックなしの1パスで削除が完結する。

計算量: O(t log_t n)。

中順走査 Inorder（b_tree.go:406-431）

各ノードについて「子0 → キー0 → 子1 → キー1 → … → 子k」の順に再帰し、 ソート済みの全キー列を復元する。計算量は O(n)。

3. テスト（b_tree_test.go）

テスト	内容
TestNewBTreePanicsOnInvalidDegree	t < 2 で NewBTree が panic することを確認
TestBTreeBasic	挿入→存在確認→削除→非存在確認の基本フロー
TestInsertDuplicateReturnsFalse	重複キー挿入が false を返し、集合として振る舞うことを確認
TestDeleteNonExistentReturnsFalse	存在しないキーの削除が false を返すことを確認
TestContainsOnEmptyTree	空の木に対する Contains/Inorder の境界値確認
TestDeleteAllShrinksToEmpty	全キー削除後に木が空へ縮退することを確認（ルート縮小パスを通す）
TestPropertyInsertDeleteMatchesReference	t = 2..6 それぞれで、ランダムな挿入500件→ランダム削除を行い、Inorder() の結果が Go の map + sort.Ints で作った正解集合と完全一致することを検証するプロパティテスト

$ go test ./... -v
...
PASS
ok      example.com/b_tree      0.004s

全ケースPASS。特にプロパティテストは分割（splitChild）・借用 （borrowFromPrevious/borrowFromNext）・結合（merge）・ルート伸縮の 全経路を、複数の t で確率的に踏むように設計している。

4. ベンチマーク（b_tree_bench_test.go）

最小次数 t = 4（benchDegree）固定、キーサイズ n = 1,000 / 10,000 / 100,000 の3水準で計測。実行環境: AMD Ryzen 5 5500U（12論理コア）、go test -bench=. -benchmem。

goos: linux
goarch: amd64
pkg: example.com/b_tree
cpu: AMD Ryzen 5 5500U with Radeon Graphics
BenchmarkInsert/1k-12      9051    118097 ns/op    41088 B/op     642 allocs/op
BenchmarkInsert/10k-12      697   1740611 ns/op   413186 B/op    6456 allocs/op
BenchmarkInsert/100k-12      45  25926781 ns/op  4157559 B/op   64956 allocs/op
BenchmarkContains/1k-12  21228068     56.30 ns/op       0 B/op       0 allocs/op
BenchmarkContains/10k-12 13175751     92.00 ns/op       0 B/op       0 allocs/op
BenchmarkContains/100k-12 5439412    224.1 ns/op       0 B/op       0 allocs/op
BenchmarkDelete/1k-12       9604    127112 ns/op        0 B/op       0 allocs/op
BenchmarkDelete/10k-12       676   1776225 ns/op        0 B/op       0 allocs/op
BenchmarkDelete/100k-12       36  32390628 ns/op        0 B/op       0 allocs/op
BenchmarkInorder/1k-12     137306      8707 ns/op    25208 B/op      12 allocs/op
BenchmarkInorder/10k-12      8131    130995 ns/op   357626 B/op      19 allocs/op
BenchmarkInorder/100k-12      399   3107848 ns/op  4101378 B/op      28 allocs/op

BenchmarkInsert/BenchmarkDelete は「1回の b.N 試行 = n件を丸ごと 挿入/削除する」計測なので、ns/op を n で割ると1件あたりのコストになる。

操作	n=1,000	n=10,000	n=100,000
Insert（1件あたり）	118 ns	174 ns	259 ns
Contains（1回あたり）	56.3 ns	92.0 ns	224.1 ns
Delete（1件あたり）	127 ns	178 ns	324 ns
Inorder（1キーあたり）	8.7 ns	13.1 ns	31.1 ns

考察

• O(log n) の傾向が見える: n が10倍になるごとにコストは 一定倍（おおよそ1.5〜2.9倍）にしか増えておらず、n に比例していない。 t = 4 のとき理論上の木の高さは log_4 n で、 log_4(10,000) / log_4(1,000) ≈ 1.33、log_4(100,000) / log_4(10,000) ≈ 1.20 程度の伸びなので、定数倍のオーバーヘッド（キャッシュミス増加、 各ノード内の線形探索コスト増）を考慮すると木の高さの対数的な伸びと 整合的な結果と言える。仮に O(n) なら10倍で10倍のコスト増になるはずで、 実測はそれよりずっと緩やかな増加に留まっている。
• Contains はアロケーションゼロ: 探索は既存ノードを辿るだけで 新規メモリ確保が発生しないため 0 allocs/op。B-Treeが読み取り主体の ワークロードに強い特性を裏付けている。
• Insert/Delete の非対称性: Insert はノード分割のたびに newNode（make によるスライス確保）が発生するため allocs/op が 大きい一方、Delete は既存ノードの中身をシフト・上書きするだけで 新規ノードを作らない（merge は右ノードを捨てるだけでGC対象にするのみ） ため 0 allocs/op。ただし要素シフトのコストがある分、 1件あたりの実行時間は Delete の方が Insert よりわずかに大きい。
• Inorder はほぼ純粋な O(n): 1キーあたりのコストも増加しているが、 これは全キーをスライスに書き出す際の append 再確保コストと 再帰呼び出しのオーバーヘッドが、木が深くなるほど（=経由する内部ノードの 再帰呼び出し回数が増えるほど）積み重なるため。とはいえ増加幅は Contains などと同程度に緩やかで、実用上は線形走査として妥当な速度。
• メモリ確保数の内訳: 2t-1=7 キー・2t=8 子ポインタ分を ノード作成のたびに make しているため（b_tree.go:24-30）、 ノード1つあたり2回のヒープ確保が発生する。分割が起きるたびに 新規ノードが1つ増えるので、Insert の allocs/op はおおよそ 「分割が起きた回数 × 2」に相当する。

Red Black Tree go

en.wikipedia.org

package red_black_tree

type Node struct {
        value  int
        color  bool
        left   *Node
        right  *Node
        parent *Node
}

func NewNode(value int, color bool) *Node {
        return &Node{
                value: value,
                color: color,
        }
}

const (
        RED   bool = true
        BLACK bool = false
)

type RedBlackTree struct {
        null *Node
        root *Node
        size int
}

func NewRedBlackTree() *RedBlackTree {
        null := NewNode(0, BLACK)
        null.left = null
        null.right = null
        null.parent = null
        return &RedBlackTree{null: null, root: null}
}

func (r *RedBlackTree) Insert(value int) bool {
        var parent *Node = r.null
        var current *Node = r.root

        for current != r.null {
                parent = current

                if value < current.value {
                        current = current.left
                } else if value > current.value {
                        current = current.right
                } else {
                        return false
                }
        }

        // 新しいノードは赤で追加する
        node := NewNode(value, RED)
        node.parent = parent
        node.left = r.null
        node.right = r.null

        if parent == r.null {
                r.root = node
        } else if value < parent.value {
                parent.left = node
        } else {
                parent.right = node
        }

        r.size++
        r.fixAfterInsert(node)

        return true
}

func (r *RedBlackTree) fixAfterInsert(node *Node) {
        // 親が赤い場合、赤ノードが連続している

        for node.parent.color == RED {
                if node.parent == node.parent.parent.left {
                        uncle := node.parent.parent.right

                        if uncle.color == RED {
                                node.parent.color = BLACK
                                uncle.color = BLACK
                                node.parent.parent.color = RED
                                node = node.parent.parent
                        } else {
                                // LR型
                                if node == node.parent.right {
                                        node = node.parent
                                        r.rotateLeft(node)
                                }

                                // LL型
                                node.parent.color = BLACK
                                node.parent.parent.color = RED
                                r.rotateRight(node.parent.parent)
                        }
                } else {
                        uncle := node.parent.parent.left

                        if uncle.color == RED {
                                node.parent.color = BLACK
                                uncle.color = BLACK
                                node.parent.parent.color = RED
                                node = node.parent.parent
                        } else {
                                // RL型
                                if node == node.parent.left {
                                        node = node.parent
                                        r.rotateRight(node)
                                }

                                // RR型
                                node.parent.color = BLACK
                                node.parent.parent.color = RED
                                r.rotateLeft(node.parent.parent)
                        }
                }
        }

        // 根は必ず黒
        r.root.color = BLACK
        r.root.parent = r.null
}

func (r *RedBlackTree) Contains(value int) bool {
        return r.findNode(value) != r.null
}

func (r *RedBlackTree) findNode(value int) *Node {
        current := r.root

        for current != r.null {
                if value < current.value {
                        current = current.left
                } else if value > current.value {
                        current = current.right
                } else {
                        return current
                }
        }

        return r.null
}

func (r *RedBlackTree) Delete(value int) bool {
        target := r.findNode(value)

        if target == r.null {
                return false
        }

        r.deleteNode(target)
        r.size--

        return true
}

func (r *RedBlackTree) deleteNode(target *Node) {
        removedNode := target
        removedOriginalColor := removedNode.color
        var replacement *Node

        if target.left == r.null {
                replacement = target.right
                r.transplant(target, target.right)
        } else if target.right == r.null {
                replacement = target.left
                r.transplant(target, target.left)
        } else {
                removedNode = r.minimumNode(target.right)
                removedOriginalColor = removedNode.color
                replacement = removedNode.right

                if removedNode.parent == target {
                        replacement.parent = removedNode
                } else {
                        r.transplant(removedNode, removedNode.right)
                        removedNode.right = target.right
                        removedNode.right.parent = removedNode
                }

                r.transplant(target, removedNode)
                removedNode.left = target.left
                removedNode.left.parent = removedNode

                // 削除対象の色を引き継ぐ
                removedNode.color = target.color
        }

        // 黒ノードを削除した場合だけ修復
        if removedOriginalColor == BLACK {
                r.fixAfterDelete(replacement)
        }

        if r.root != r.null {
                r.root.parent = r.null
        }

        r.null.parent = r.null
}

func (r *RedBlackTree) fixAfterDelete(node *Node) {
        for node != r.root && node.color == BLACK {
                if node == node.parent.left {
                        sibling := node.parent.right

                        // 兄弟が赤
                        if sibling.color == RED {
                                sibling.color = BLACK
                                node.parent.color = RED

                                r.rotateLeft(node.parent)
                                sibling = node.parent.right
                        }

                        // 兄弟と兄弟の子がすべて黒
                        if sibling.left.color == BLACK && sibling.right.color == BLACK {
                                sibling.color = RED
                                node = node.parent
                        } else {
                                // 兄弟の遠い子が黒、近い子が赤
                                if sibling.right.color == BLACK {
                                        sibling.left.color = BLACK
                                        sibling.color = RED
                                        r.rotateRight(sibling)
                                        sibling = node.parent.right
                                }

                                // 兄弟の遠い子が赤
                                sibling.color = node.parent.color
                                node.parent.color = BLACK
                                sibling.right.color = BLACK
                                r.rotateLeft(node.parent)
                                node = r.root
                        }
                } else {
                        // nodeが親の右の子の場合
                        sibling := node.parent.left

                        if sibling.color == RED {
                                sibling.color = BLACK
                                node.parent.color = RED
                                r.rotateRight(node.parent)
                                sibling = node.parent.left
                        }

                        if sibling.right.color == BLACK && sibling.left.color == BLACK {
                                sibling.color = RED
                                node = node.parent
                        } else {
                                if sibling.left.color == BLACK {
                                        sibling.right.color = BLACK
                                        sibling.color = RED
                                        r.rotateLeft(sibling)
                                        sibling = node.parent.left
                                }
                                sibling.color = node.parent.color
                                node.parent.color = BLACK
                                sibling.left.color = BLACK
                                r.rotateRight(node.parent)
                                node = r.root
                        }
                }
        }
        node.color = BLACK
}

func (r *RedBlackTree) minimumNode(node *Node) *Node {
        for node.left != r.null {
                node = node.left
        }
        return node
}

func (r *RedBlackTree) transplant(oldNode, newNode *Node) {
        if oldNode.parent == r.null {
                r.root = newNode
        } else if oldNode == oldNode.parent.left {
                oldNode.parent.left = newNode
        } else {
                oldNode.parent.right = newNode
        }
        newNode.parent = oldNode.parent
}

// 左回転
func (r *RedBlackTree) rotateLeft(x *Node) {
        y := x.right
        x.right = y.left

        if y.left != r.null {
                y.left.parent = x
        }

        y.parent = x.parent

        if x.parent == r.null {
                r.root = y
        } else if x == x.parent.left {
                x.parent.left = y
        } else {
                x.parent.right = y
        }

        y.left = x
        x.parent = y
}

// 右回転
func (r *RedBlackTree) rotateRight(y *Node) {
        x := y.left
        y.left = x.right

        if x.right != r.null {
                x.right.parent = y
        }

        x.parent = y.parent

        if y.parent == r.null {
                r.root = x
        } else if y == y.parent.left {
                y.parent.left = x
        } else {
                y.parent.right = x
        }

        x.right = y
        y.parent = x
}

追記

AIでベンチ取った

赤黒木 (Red-Black Tree)

red_black_tree.go は自己平衡二分探索木の一種である 赤黒木 の実装。 各ノードに「赤/黒」の色を持たせ、5つの色に関する制約を常に満たすことで、 AVL木ほど厳密ではないが十分に高さを O(log n) に抑える。 AVL木より回転の回数は少なく済む(挿入は最大2回)一方、 若干高さが緩い(最大で 2*log2(n+1))という特徴を持つ。

1. アルゴリズムの解説

1.1 赤黒木の5つの性質

1. 各ノードは赤か黒のどちらか
2. 根は黒
3. 葉(番兵の nil)は黒とみなす
4. 赤ノードの子は必ず黒(赤が2連続しない)
5. あるノードから、その子孫であるnilまでのどの経路をたどっても、 通過する黒ノードの数(黒高さ, black-height)は同じ

この5つを常に保つことで、最も長い経路(赤黒交互)でも最も短い経路(黒だけ)の 2倍を超えることができなくなり、木の高さは O(log n) (具体的には 2*log2(n+1) 以下)に収まる。

1.2 番兵ノード null という設計

type RedBlackTree struct {
    null *Node // すべての葉が指す共有の番兵ノード(常に黒)
    root *Node
    size int
}

素朴なBSTのようにGoの nil を葉として使うと、node.left.color のような アクセスのたびに nil チェックが必要になり、コードが煩雑になる。 この実装ではすべての「存在しない子」を 同じ1個の黒ノード r.null に 向けることで、current.color や sibling.left.color を nilチェックなしに素通しで書ける。

削除の修復処理(fixAfterDelete)では、「二重に黒」を意味する仮想的な ノードとして r.null 自身が一時的に木に組み込まれ、 r.null.parent が一時的に書き換えられる(transplant 経由)。 修復が終わったタイミングで deleteNode の最後に r.null.parent = r.null へ必ず戻す後始末が入っている。

実装上のハマりどころ: NewNode はGoのゼロ値のまま left/right を返す(＝素の nil)。Insert で新規ノードを 作った直後に node.left = r.null / node.right = r.null を 明示的にセットしていないと、木が2〜3段深くなった時点で uncle.color や sibling.left.color の参照が素の nil に対する フィールドアクセスとなり panic: nil pointer dereference で 落ちる。番兵パターンを使うライブラリでは 「新規ノードの子は必ず番兵で初期化する」を徹底する必要がある。

1.3 挿入 (Insert / fixAfterInsert)

通常のBSTと同じ探索で挿入位置を決め、新しいノードは赤として追加する (黒として追加すると性質5の黒高さが即座に崩れてしまうため)。

赤ノードを追加した結果、親も赤だと性質4(赤の連続禁止)が破れる。 fixAfterInsert はこれを次の3パターンに分類して直す (親が祖父の左の子である場合。右の子の場合は左右対称):

ケース	叔父(uncle)の色	対処
1	赤	親と叔父を黒、祖父を赤に塗り替えて祖父から再チェック(色を上に伝播するだけで回転しない)
2 (LR型)	黒、かつ「く」の字	親を軸に回転して一直線(LL型)に矯正してからケース3へ
3 (LL型)	黒、かつ一直線	親を黒・祖父を赤に塗り替えて祖父を軸に回転

for node.parent.color == RED {
    if node.parent == node.parent.parent.left {
        uncle := node.parent.parent.right
        if uncle.color == RED {
            // ケース1: 色を上に伝播するだけ
            node.parent.color = BLACK
            uncle.color = BLACK
            node.parent.parent.color = RED
            node = node.parent.parent
        } else {
            if node == node.parent.right {
                node = node.parent
                r.rotateLeft(node) // ケース2 → ケース3に矯正
            }
            node.parent.color = BLACK
            node.parent.parent.color = RED
            r.rotateRight(node.parent.parent) // ケース3: 解消
        }
    } else {
        /* 左右対称の処理 */
    }
}
r.root.color = BLACK // 根は必ず黒に戻す

ケース1は回転を伴わず「赤が1段上に上がる」だけなので、最悪でも根まで O(log n) 回繰り返される。ケース2・3は回転して即座にループを抜けるため、 1回の挿入で必要な回転は高々2回(ケース2→3の連続適用のみ)。 これがAVL木(挿入は回転1回で済むこともあるが最大でも1回)と比べて 赤黒木の挿入が実務でよく使われる理由の一つ(判定・回転コストが軽い)。

1.4 回転 (rotateLeft / rotateRight)

回転はBSTとしてのinorder順序を保ったまま形だけを変える操作で、 AVL木と機構は同じ。ただし赤黒木のノードはheightを持たないため、 回転後の高さ再計算は不要で、ポインタの付け替えだけで完結する。

左回転 rotateLeft(x):        右回転 rotateRight(y):
    x                              y
   / \                            / \
  T1  y      ---->               x   T3
     / \                        / \
    T2  T3                    T1  T2
                 (xとyの位置が入れ替わり、T2が反対側の子に移る)

func (r *RedBlackTree) rotateLeft(x *Node) {
    y := x.right
    x.right = y.left      // 宙に浮くT2をxの右に付け替え
    if y.left != r.null {
        y.left.parent = x
    }
    y.parent = x.parent
    if x.parent == r.null {
        r.root = y
    } else if x == x.parent.left {
        x.parent.left = y
    } else {
        x.parent.right = y
    }
    y.left = x
    x.parent = y
}

rotateRight はこの鏡像。色の変更は回転そのものには含まれず、 呼び出し側(fixAfterInsert / fixAfterDelete)が回転の前後で 明示的に塗り替える点に注意。

1.5 削除 (Delete / deleteNode / fixAfterDelete)

BSTと同じ3ケース分岐で削除するが、transplant(部分木の丸ごと差し替え) を使う点がAVL木の実装と異なる:

1. 左の子がない: 右の子で置き換える
2. 右の子がない: 左の子で置き換える
3. 両方に子がある: 右部分木の最小値(中順後続, minimumNode)を探し、 その後続ノードを木から外して削除対象の位置に差し込む

removedNode := target
removedOriginalColor := removedNode.color
// ... transplantで実際に木から取り除く ...

// 黒ノードを削除した場合だけ、黒高さ(性質5)が崩れるので修復が要る
if removedOriginalColor == BLACK {
    r.fixAfterDelete(replacement)
}

削除したノードが赤だった場合は性質4・5に影響しないため何もしなくてよい。 黒だった場合、削除された経路の黒高さが1減ってしまうので、 「削除位置に来た子(replacement)は本来より1つ多く黒を背負っている (二重に黒, double black)」とみなして fixAfterDelete で解消する。 この二重黒の解消は次の4パターンに分類される (nodeが親の左の子である場合。右の子の場合は左右対称):

ケース	兄弟(sibling)の色	兄弟の子の色	対処
1	赤	―	親を軸に回転して兄弟を黒にすり替え、ケース2〜4に帰着
2	黒	両方黒	兄弟を赤に塗り、二重黒を親に押し上げてループ継続
3	黒	近い方(遠い方じゃない側)だけ赤	兄弟を軸に回転してケース4の形に矯正
4	黒	遠い方が赤	兄弟を軸に回転し、色を塗り替えて即座に解消(ループ終了)

for node != r.root && node.color == BLACK {
    if node == node.parent.left {
        sibling := node.parent.right
        if sibling.color == RED { // ケース1
            sibling.color = BLACK
            node.parent.color = RED
            r.rotateLeft(node.parent)
            sibling = node.parent.right
        }
        if sibling.left.color == BLACK && sibling.right.color == BLACK { // ケース2
            sibling.color = RED
            node = node.parent // 二重黒を1段上に押し上げる
        } else {
            if sibling.right.color == BLACK { // ケース3 → ケース4に矯正
                sibling.left.color = BLACK
                sibling.color = RED
                r.rotateRight(sibling)
                sibling = node.parent.right
            }
            sibling.color = node.parent.color // ケース4: 解消
            node.parent.color = BLACK
            sibling.right.color = BLACK
            r.rotateLeft(node.parent)
            node = r.root // ループを抜ける
        }
    } else {
        /* 左右対称の処理 */
    }
}
node.color = BLACK

ケース2だけがループを継続させ(二重黒が根に到達するまで伝播しうる)、 それ以外は1〜2回の回転で即座に解消する。よって削除1回あたりの 回転回数も定数回(最大3回)に収まる。

1.6 計算量

操作	計算量	理由
Contains	O(log n)	高さが O(log n) に保たれたBST探索
Insert	O(log n)	探索 + 経路上の再彩色 + 回転は高々2回
Delete	O(log n)	探索 + successor探索 + 回転は高々3回

2. テスト (red_black_tree_test.go)

go test -v -cover ./... で実行。カバレッジは 100.0%。

2.1 不変条件そのものを検証するヘルパー

• checkRedBlackProperties: 全ノードを再帰的に辿り、 「赤の子は黒」「左右の黒高さが一致」「parentポインタの整合性」を 直接検証する、赤黒木の定義そのものをチェックする最重要ヘルパー。 戻り値として黒高さを返すため、左右の部分木で黒高さが食い違った 瞬間に t.Fatalf で落ちる。
• assertValid: 上記に加えて「根が黒であること」「番兵nullが 黒のままであること」も確認してからまとめて呼ぶラッパー。

これらは各テストの中で操作のたびに呼び出し、性質が 壊れていないことをステップ単位で確認する。

2.2 個別機能のユニットテスト

• TestInsertAndContains / TestInsertDuplicateReturnsFalse: 基本の挿入と、重複挿入時に false を返しサイズが増えないことの確認。
• TestInsertKeepsBalanceSequential: 昇順・降順という 単純なBSTなら最悪ケース(一直線)になる入力でも、 赤黒木なら高さが理論上限 2*log2(n+1) を超えないことを確認する 回帰テスト。
• TestDeleteLeafNode / TestDeleteNodeWithOneChild / TestDeleteNodeWithTwoChildren: 削除の3分岐それぞれを 明示的に踏むケース。
• TestDeleteNonExistentReturnsFalse: 存在しない値の削除は false を返しサイズも変わらないことの確認。
• TestDeleteAllKeepsPropertiesAndOrder: ランダム順に全要素を挿入した 木から、さらに別のランダム順で1つずつ削除し、削除の毎ステップで assertValid を呼んで性質が壊れていないかを確認する。 削除は挿入よりも修復パターンが複雑(4ケース×左右対称)なため、 厚めに確認している。
• TestInorderIsSorted: 中順巡回がソート済み(重複なし)になるという 「赤黒木もBSTである」という性質の確認。

2.3 プロパティベースの統合テスト

• TestRandomOperationsAgainstReferenceSet: 標準ライブラリの map[int]bool を「正解」として使い、同じ乱数シードで5000回の Insert/Delete を赤黒木とmapの両方に適用し続ける。各ステップで:
  • 戻り値(true/false)がmapと一致するか
  • assertValid が常に成り立つか

  最後に:

  • size がmapの要素数と一致するか
  • Contains が全要素について一致するか
  • 中順巡回が厳密に昇順か

  を検証する。個別のユニットテストでは踏みにくい「挿入と削除が ランダムに混ざった経路」を大量に生成し、 どんな操作列でも5つの性質が破れないことを高い確度で確認する。

3. ベンチマーク (red_black_tree_bench_test.go)

go test -bench=. -benchmem -run=^$ ./... で実行 (-run=^$ で通常テストをスキップしベンチマークだけ走らせる)。

3.1 計測項目と狙い

• BenchmarkInsert: 0..b.N-1 をランダムな順序(重複なしの順列)で 空の木に挿入するコスト。あえて rand.Intn の重複ありの値を 使わなかった理由: 既存値と衝突した Insert は探索だけで false を返して NewNode を呼ばずに終わってしまい、 挿入コストを過小評価してしまうため(実際、最初はこれで 0 allocs/op という誤った計測値が出た。後述3.3参照)。
• BenchmarkInsertSequential: 昇順の値を挿入し続けるベンチマーク。 単純なBSTなら片重りしてO(n)に劣化する入力だが、 赤黒木なら回転によって高さが抑えられ、どの程度コストが変わるかを 見るための対比用ベンチマーク。
• BenchmarkContains: あらかじめ 100 / 1,000 / 10,000 / 100,000 件を 挿入した木に対する検索コストをサイズ別に計測。 O(log n) であれば、サイズが10倍になっても所要時間は 一定幅(木の高さの増分)しか伸びないはず。
• BenchmarkDelete: 既存木のサイズを一定に保ったまま削除コストを計測する。 木にまだ存在しない値(負の数)を1つ挿入してすぐ削除するペア操作を 繰り返すことで、木のサイズnを固定したままdelete単体のコストを 計測している(b.StopTimer/b.StartTimerを毎回呼ぶとタイマー呼び出し 自体のオーバーヘッドで計測が破綻するため使わない)。

3.2 実測結果(参考値)

実行環境: AMD Ryzen 5 5500U (12論理コア)、go test -bench=. -benchmem

BenchmarkInsert-12              1,650,078 回      837.6 ns/op      48 B/op   1 allocs/op
BenchmarkInsertSequential-12   12,138,400 回      116.9 ns/op      48 B/op   1 allocs/op
BenchmarkContains/100-12       32,679,680 回       36.8 ns/op       0 B/op   0 allocs/op
BenchmarkContains/1k-12        22,261,239 回       53.8 ns/op       0 B/op   0 allocs/op
BenchmarkContains/10k-12       13,953,355 回       86.5 ns/op       0 B/op   0 allocs/op
BenchmarkContains/100k-12       5,117,442 回      233.9 ns/op       0 B/op   0 allocs/op
BenchmarkDelete/100-12          19,232,476 回       56.0 ns/op      48 B/op   1 allocs/op
BenchmarkDelete/1k-12           18,899,344 回       58.3 ns/op      48 B/op   1 allocs/op
BenchmarkDelete/10k-12          17,400,606 回       62.3 ns/op      48 B/op   1 allocs/op

3.3 結果の読み方

• Contains: サイズが 100 → 1k → 10k → 100k(1000倍)と増えても、 1回あたりの所要時間は 36.8ns → 53.8ns → 86.5ns → 233.9ns と 約6.4倍にしかなっていない。これは O(log n) の挙動と整合する (log2(1000) ≈ 10、定数倍を考慮すればおおむね妥当な伸び方)。
• Delete: 100 → 1k → 10k で 56.0ns → 58.3ns → 62.3ns と、 ほぼ横ばいに近い緩やかな伸びに収まっている。1 allocs/op (48 B/op)は各反復で対になるInsertが確保するNode分であり、 Delete自体は追加のアロケーションをしない。
• Insert vs InsertSequential: 完全ランダム順の挿入(837.6ns)は 昇順連続挿入(116.9ns)よりかなり遅い。InsertSequentialは木のサイズが b.Nと等しい状態まで育つ点は同じだが、新しいノードが毎回 「一番右」に追加されるため探索経路のキャッシュ局所性が高く 分岐予測も効きやすい。一方BenchmarkInsertはランダムな順列で b.N件(ベンチによっては100万件超)を挿入するため、木全体に ランダムアクセスすることになりキャッシュミスが支配的になる。
• 0 allocs/op の罠(デバッグの記録): 当初BenchmarkInsertは rng.Intn(b.N*10+1)で値を生成しており、b.Nが大きくなると 一定割合が既存値と衝突してInsertがNewNodeを呼ばずにfalseを 返していた。Goのベンチマーク集計は総アロケーション数 / b.Nを 整数除算するため、衝突で発生しない反復が混ざると 平均が1未満になり0 allocs/opという(実態と異なる)表示になっていた。 値を0..b.N-1の重複なし順列に変えたことで1 allocs/op (=Insertのたびに必ず1個Nodeを確保する)という正しい値になった。

4. 実行方法まとめ

# ユニットテスト(詳細出力)
go test -v ./...

# カバレッジ
go test -cover ./...

# ベンチマーク(通常テストは実行しない)
go test -bench=. -benchmem -run=^$ ./...

# 特定のベンチマークだけ、反復回数を指定して実行
go test -bench=BenchmarkDelete -benchmem -run=^$ -benchtime=200000x ./...

AVL Tree go

en.wikipedia.org

package avl_tree

import "errors"

type Node struct {
        value  int
        height int
        left   *Node
        right  *Node
}

func NewNode(value int) *Node {
        return &Node{
                value:  value,
                height: 1,
        }
}

type AVLTree struct {
        root *Node
        size int
}

func (a *AVLTree) Insert(value int) bool {
        if a.Contains(value) {
                return false
        }

        a.root = a.insertRecursive(a.root, value)
        a.size++

        return true
}

func (a *AVLTree) insertRecursive(node *Node, value int) *Node {
        if node == nil {
                return NewNode(value)
        }

        if value < node.value {
                node.left = a.insertRecursive(node.left, value)
        } else {
                node.right = a.insertRecursive(node.right, value)
        }

        // 現在のノードの高さを更新
        a.updateHeight(node)

        // 木が傾いていれば回転して平衡化する
        return a.rebalance(node)
}

func (a *AVLTree) Contains(value int) bool {
        current := a.root

        for current != nil {
                if value < current.value {
                        current = current.left
                } else if value > current.value {
                        current = current.right
                } else {
                        return true
                }
        }

        return false
}

func (a *AVLTree) Delete(value int) bool {
        if !a.Contains(value) {
                return false
        }

        a.root = a.deleteRecursive(a.root, value)
        a.size--

        return true
}

func (a *AVLTree) deleteRecursive(node *Node, value int) *Node {
        if node == nil {
                return nil
        }

        if value < node.value {
                node.left = a.deleteRecursive(node.left, value)
        } else if value > node.value {
                node.right = a.deleteRecursive(node.right, value)
        } else {
                // 削除対象を発見

                if node.left == nil {
                        return node.right
                }

                if node.right == nil {
                        return node.left
                }

                // 左右に子がある場合
                successor := a.findMinNode(node.right)
                node.value = successor.value
                node.right = a.deleteRecursive(
                        node.right,
                        successor.value,
                )
        }

        // 削除後に高さを更新
        a.updateHeight(node)

        return a.rebalance(node)
}

func (a *AVLTree) rebalance(node *Node) *Node {
        balanceFactor := a.getBalanceFactor(node)

        // 左側が重い
        if balanceFactor > 1 {
                if a.getBalanceFactor(node.left) < 0 {
                        node.left = a.rotateLeft(node.left)
                }
                return a.rotateRight(node)
        }

        // 右側が重い
        if balanceFactor < -1 {
                if a.getBalanceFactor(node.right) > 0 {
                        node.right = a.rotateRight(node.right)
                }
                return a.rotateLeft(node)
        }

        // すでに平衡
        return node
}

// 右回転
func (a *AVLTree) rotateRight(y *Node) *Node {
        x := y.left
        subtree := x.right

        // 回転
        x.right = y
        y.left = subtree

        a.updateHeight(y)
        a.updateHeight(x)

        return x
}

// 左回転
func (a *AVLTree) rotateLeft(x *Node) *Node {
        y := x.right
        subtree := y.left

        // 回転
        y.left = x
        x.right = subtree

        a.updateHeight(x)
        a.updateHeight(y)

        return y
}

func (a *AVLTree) updateHeight(node *Node) {
        node.height = 1 + max(a.getHeight(node.left), a.getHeight(node.right))
}

func (a *AVLTree) getHeight(node *Node) int {
        if node == nil {
                return 0
        }
        return node.height
}

func (a *AVLTree) getBalanceFactor(node *Node) int {
        if node == nil {
                return 0
        }

        return a.getHeight(node.left) - a.getHeight(node.right)
}

func (a *AVLTree) Min() (int, error) {
        if a.root == nil {
                return 0, errors.New("tree is empty")
        }

        return a.findMinNode(a.root).value, nil
}

func (a *AVLTree) findMinNode(node *Node) *Node {
        current := node

        for current.left != nil {
                current = current.left
        }

        return current
}

func (a *AVLTree) Max() (int, error) {
        if a.root == nil {
                return 0, errors.New("tree is empty")
        }

        current := a.root

        for current.right != nil {
                current = current.right
        }

        return current.value, nil
}

func (a *AVLTree) Height() int {
        return a.getHeight(a.root)
}

func (a *AVLTree) Size() int {
        return a.size
}

func (a *AVLTree) IsEmpty() bool {
        return a.root == nil
}

func (a *AVLTree) Clear() {
        a.root = nil
        a.size = 0
}

func (a *AVLTree) Inorder() []int {
        return a.inorderRecursive(a.root, []int{})
}

func (a *AVLTree) inorderRecursive(node *Node, result []int) []int {
        if node == nil {
                return result
        }

        result = a.inorderRecursive(node.left, result)
        result = append(result, node.value)
        result = a.inorderRecursive(node.right, result)
        return result
}

func (a *AVLTree) Preorder() []int {
        return a.preorderRecursive(a.root, []int{})
}

func (a *AVLTree) preorderRecursive(node *Node, result []int) []int {
        if node == nil {
                return result
        }

        result = append(result, node.value)
        result = a.preorderRecursive(node.left, result)
        result = a.preorderRecursive(node.right, result)
        return result
}

func (a *AVLTree) Postorder() []int {
        return a.postorderRecursive(a.root, []int{})
}

func (a *AVLTree) postorderRecursive(node *Node, result []int) []int {
        if node == nil {
                return result
        }

        result = a.postorderRecursive(node.left, result)
        result = a.postorderRecursive(node.right, result)
        result = append(result, node.value)
        return result
}

追記

AIでベンチ取ってもらった

AVL木 (AVL Tree)

avl_tree.go は自己平衡二分探索木の一種である AVL木 の実装。 挿入・削除のたびに左右部分木の高さの差(平衡係数)をチェックし、 崩れていれば回転によって再平衡化することで、常に高さ O(log n) を維持する。

1. アルゴリズムの解説

1.1 なぜ平衡が必要か

単純な二分探索木(BST)は、1, 2, 3, 4, 5, ... のようにソート済みの順で 値を挿入すると、枝分かれせず一直線に伸びた木(連結リストと同じ形)になる。 この場合、探索・挿入・削除はすべて O(n) に劣化してしまう。

AVL木は「どのノードでも左右部分木の高さの差(平衡係数)が -1, 0, +1 のいずれかである」という不変条件を常に保つことで、 木の高さを O(log n) に抑える。これにより最悪ケースでも 探索・挿入・削除が O(log n) で済む。

1.2 各ノードが持つ情報

type Node struct {
    value  int
    height int   // このノードを根とする部分木の高さ
    left   *Node
    right  *Node
}

height はノードごとにキャッシュしておき、子の高さから 毎回 1 + max(左の高さ, 右の高さ) で更新する(updateHeight)。 これにより平衡係数を O(1) で計算できる(getBalanceFactor)。

func (a *AVLTree) getBalanceFactor(node *Node) int {
    return a.getHeight(node.left) - a.getHeight(node.right)
}

• balanceFactor > 1 → 左が重い(Left-Heavy)
• balanceFactor < -1 → 右が重い(Right-Heavy)
• それ以外 → 平衡が取れている

1.3 挿入 (Insert / insertRecursive)

通常のBSTと同じく、値の大小で左右どちらに潜るかを再帰的に決めて リーフ位置に新しいノードを追加する。その後、再帰の帰りがけに 通過した各ノードで次の2つを行う:

1. updateHeight で高さを更新
2. rebalance で平衡係数をチェックし、崩れていれば回転

再帰的に「一番深い場所から根に向かって」平衡を直していくため、 1回の挿入で崩れる箇所は根までの経路上のどこか1箇所だけになる。

1.4 回転 (rotateLeft / rotateRight)

回転は「二分探索木としての順序関係(inorder)を保ったまま、 木の形だけを変える」操作。AVL木の不平衡は4パターンに分類され、 それぞれ次のように直す(rebalance 内のロジック):

平衡係数	子の平衡係数	パターン	対処
> 1 (左重い)	子の係数 >= 0	Left-Left	右回転1回
> 1 (左重い)	子の係数 < 0	Left-Right	左の子を左回転 → 右回転
< -1 (右重い)	子の係数 <= 0	Right-Right	左回転1回
< -1 (右重い)	子の係数 > 0	Right-Left	右の子を右回転 → 左回転

1.4.1 右回転 rotateRight の詳細

左側が重すぎる(LL型)ときに使う。付け替えるポインタは3本だけで、 inorder順序はまったく変わらない。

回転前:                     回転後:
        y                          x
       / \                        / \
      x   T3                    T1   y
     / \        rotateRight(y)      / \
    T1  T2      ------------>     T2  T3

• y は「重すぎる」ノード(不平衡の起点)
• x は y.left。回転後はこの x が新しい根になる
• T2(= x.right)は宙に浮く部分木。x が根に上がるので居場所を失い、 新しい y の左(y.left)に付け替える。T1 は x.left のまま動かない

func (a *AVLTree) rotateRight(y *Node) *Node {
    x := y.left        // x を取り出す(将来の新しい根)
    subtree := x.right // T2 を退避しておく

    // --- ポインタの付け替え ---
    x.right = y         // x の右に y をぶら下げる(x が親、y が子)
    y.left = subtree    // y の左に T2 を移す

    a.updateHeight(y) // ① 先に「下」になった y の高さを更新
    a.updateHeight(x) // ② 次に「上」になった x の高さを更新

    return x            // この部分木の新しい根は x
}

具体例(3, 2, 1 の順で挿入するとLL型不平衡が発生する):

1を挿入後:  (3)          <- balanceFactor(3) = 2 (左が重い)
           /
         (2)
         /
       (1)

rotateRight(3):
  y = 3, x = 2, subtree = x.right = nil (T2はなし)
  x.right = y  →  2.right = 3
  y.left = subtree → 3.left = nil

結果:
       (2)
      /   \
    (1)   (3)

1,2,3 のinorderは回転の前後で変わらない。回転は形だけを変える操作であることが確認できる。

1.4.2 左回転 rotateLeft は右回転の鏡像

回転前:                     回転後:
    x                              y
   / \                            / \
  T1  y                         x    T3
     / \      rotateLeft(x)    / \
   T2  T3     ------------>  T1  T2

func (a *AVLTree) rotateLeft(x *Node) *Node {
    y := x.right       // y を取り出す(将来の新しい根)
    subtree := y.left  // T2 を退避

    y.left = x          // y の左に x をぶら下げる
    x.right = subtree   // x の右に T2 を移す

    a.updateHeight(x) // 下になった x が先
    a.updateHeight(y) // 上になった y が後
    return y
}

1.4.3 高さ更新の順序がなぜ重要か

updateHeight は子の高さから自分の高さを計算する:

func (a *AVLTree) updateHeight(node *Node) {
    node.height = 1 + max(a.getHeight(node.left), a.getHeight(node.right))
}

回転で親子関係が入れ替わるため、x の高さは(新しく子になった)y の 高さに依存する。必ず「下になった側→上になった側」の順で呼ばないと、 上になった側が古い(誤った)高さを参照してしまう。

• rotateRight: y(下)→ x(上)
• rotateLeft: x(下)→ y(上)

1.4.4 二重回転(Left-Right / Right-Left)

LL・RRのように「まっすぐ」傾いている場合は1回の回転で直せるが、 LR・RLのように「く」の字に折れている場合は先に子を回転して まっすぐな形に矯正してから、もう一度回転する必要がある。

Left-Right型:
   z              z                 y
  /              /                / \
 x        →     y        →      x   z
  \             /
   y           x

1. rotateLeft(x)  … z.left を回転し、「く」の字をLL型に矯正
2. rotateRight(z) … LL型を解消、y が新しい根になる

具体例: 1, 3, 2 の順で挿入すると 1 → 3 → 2 という「く」の字ができる。 rebalance は rotateLeft(1) → rotateRight(3) の順で呼び、 最終的に 2 が新しい根になる。

Right-Left型はこの鏡像で、rotateRight(子) で「く」の字をRR型に 矯正してから rotateLeft(自分) で解消する。

func (a *AVLTree) rebalance(node *Node) *Node {
    balanceFactor := a.getBalanceFactor(node)

    if balanceFactor > 1 { // 左が重い
        if a.getBalanceFactor(node.left) < 0 {
            // Left-Right型: 先に左の子を左回転してLL型に矯正
            node.left = a.rotateLeft(node.left)
        }
        return a.rotateRight(node) // LL型はここで解消
    }

    if balanceFactor < -1 { // 右が重い
        if a.getBalanceFactor(node.right) > 0 {
            // Right-Left型: 先に右の子を右回転してRR型に矯正
            node.right = a.rotateRight(node.right)
        }
        return a.rotateLeft(node) // RR型はここで解消
    }

    return node
}

1.4.5 まとめ

操作	動かすポインタ	高さ更新順序
rotateRight(y)	x.right, y.left	y→x
rotateLeft(x)	y.left, x.right	x→y
Left-Right	rotateLeft(子) → rotateRight(自分)	—
Right-Left	rotateRight(子) → rotateLeft(自分)	—

コード中の subtree := x.right(右回転の場合)のような 「宙に浮く部分木の退避」が回転の核心。これを忘れるとポインタが 上書きされ、部分木がまるごと失われて木が壊れる。

1.5 削除 (Delete / deleteRecursive)

BSTと同じ3ケース分岐に加え、削除後の再平衡が必要になる:

1. 子がない/片方だけ: そのノードを、存在する方の子(またはnil)で置き換える
2. 両方に子がある: 右部分木の最小値(中順後続, successor)を探し、 値をコピーしてから、右部分木からその successor を再帰的に削除する

挿入と同様、削除の再帰から戻る際に経路上の全ノードで updateHeight → rebalance を行う。削除は挿入と異なり、 1回の削除で 根までの経路上、複数箇所 で回転が必要になりうる (これは挿入との重要な違い)。

1.6 計算量

操作	計算量	理由
Contains	O(log n)	高さがO(log n)に保たれたBST探索
Insert	O(log n)	探索 + 経路上の高々O(log n)回の回転
Delete	O(log n)	探索 + successor探索 + 経路上の回転
Min / Max	O(log n)	左(右)端まで一直線に辿るだけ
Inorder / Preorder / Postorder	O(n)	全ノードを1回ずつ訪問

2. テスト (avl_tree_test.go)

go test -v ./... で実行。方針は大きく分けて3つ。

2.1 不変条件そのものを検証するヘルパー

• isBalanced(node): 全ノードを再帰的に辿り、|左の高さ - 右の高さ| <= 1 が常に成り立っているかを直接検証する。AVL木の定義そのものをチェックする、 最も重要なヘルパー。
• heightsAreConsistent(node): node.height フィールドが、実際の子の 高さから計算される値と一致しているかを検証する。回転やコピー処理の バグで「値は正しいが高さの更新を忘れる」ような不整合を検出できる。

これらは単体テストの中で繰り返し呼び出し、操作のたびに木の整合性が 壊れていないことを確認する。

2.2 個別機能のユニットテスト

• TestInsertAndContains / TestInsertDuplicateReturnsFalse: 基本の 挿入・重複挿入時の挙動(falseを返しサイズが増えないこと)。
• TestInsertKeepsBalance / TestInsertKeepsBalanceDescending: 昇順・降順という単純なBSTなら最悪ケースになる入力でも、 AVL木なら高さが O(log n) に収まることを確認する回帰テスト。 (1000要素挿入して高さが20以下、など)
• TestDeleteLeafNode / TestDeleteNodeWithOneChild / TestDeleteNodeWithTwoChildren: 削除の3分岐それぞれを 明示的に踏むケース。特に two-children のケースは successor による 値のコピー処理が絡むため個別にテストしている。
• TestDeleteAllKeepsBalanceAndOrder: 構築した木からランダム順に 全要素を1つずつ削除し、削除の毎ステップで isBalanced / heightsAreConsistent / サイズを検証する。削除は挿入よりも 再平衡パターンが複雑なため、経路上の複数回転が正しく効いているかを 厚めに確認している。
• TestMinMax(OnEmptyTree) / TestHeightAndSizeAndIsEmpty / TestClear: 補助APIとエッジケース(空の木)。
• TestInorderIsSorted / TestPreorderRootFirst / TestPostorderRootLast / TestTraversalsOnEmptyTree: 3種の巡回。特に Inorder は ソート列になっているはずというAVL木=BSTとしての性質を検証する (これは以前、巡回結果がスライスの値渡しにより呼び出し元に反映されない バグで壊れていた箇所でもある)。

2.3 プロパティベースの統合テスト

• TestRandomOperationsAgainstReferenceSet: 標準ライブラリの map[int]bool を「正解」として使い、同じ乱数シードで5000回の Insert/Delete を AVL木とmapの両方に適用し続ける。各ステップで:
  • 戻り値(true/false)がmapと一致するか
  • isBalanced が常に成り立つか 最後に:
  • Size() がmapの要素数と一致するか
  • Contains が全要素について一致するか
  • Inorder() が厳密に昇順(重複なし)か

  を検証する。個別のユニットテストでは踏みにくい「挿入と削除が ランダムに混ざった経路」を大量に生成し、平衡条件がどんな操作列でも 破れないことを高い確度で確認するためのテスト。

カバレッジは go test -cover ./... で 98.2% (未到達は主に到達がほぼ不可能なエラー分岐など)。

3. ベンチマーク (avl_tree_bench_test.go)

go test -bench=. -benchmem -run=^$ ./... で実行 (-run=^$ で通常テストをスキップしベンチマークだけ走らせる)。

3.1 計測項目と狙い

• BenchmarkInsert: ランダムな値を空の木から1件ずつ挿入するコスト。 木が育つにつれて O(log n) で挿入コストが伸びることを示す。
• BenchmarkInsertSequential: 昇順の値を挿入し続けるベンチマーク。 単純なBSTなら片重りして O(n) に劣化する入力だが、AVL木なら 回転によって高さが抑えられ、Insert と同程度のコストに収まることを 示すための対比用ベンチマーク。
• BenchmarkContains: あらかじめ 100 / 1,000 / 10,000 / 100,000 件を 挿入した木に対する検索コストを木のサイズ別に計測。 O(log n) であれば、サイズが10倍になっても所要時間は 一定幅(木の高さの増分)しか伸びないはず。
• BenchmarkDelete: 既存木のサイズを一定に保ったまま削除コストを計測する。 素朴に「都度木を作り直す」と削除以外のコストが混ざってしまうため、 木にまだ存在しない値(負の数)を1つ挿入してすぐ削除する ペア操作を繰り返すことで、木のサイズ n を固定したまま delete 単体のコストを計測している。 (b.StopTimer()/b.StartTimer() を毎回呼ぶ実装も試したが、 タイマー呼び出し自体のオーバーヘッドで計測が破綻し、 ベンチマークが極端に遅くなったため、このアプローチに変更した)
• BenchmarkInorder: 全件走査のコストを木のサイズ別に計測。 O(n) であることが期待値なので、サイズ10倍でおよそ10倍の 時間・メモリになることを確認できる。

3.2 実測結果(参考値)

実行環境: AMD Ryzen 5 5500U (12論理コア)、go test -bench=. -benchmem

BenchmarkInsert-12              1,352,491 回      961.0 ns/op      30 B/op   0 allocs/op
BenchmarkInsertSequential-12    6,053,409 回      213.3 ns/op      32 B/op   1 allocs/op
BenchmarkContains/100-12       32,212,252 回       36.5 ns/op       0 B/op   0 allocs/op
BenchmarkContains/1k-12        23,377,689 回       51.3 ns/op       0 B/op   0 allocs/op
BenchmarkContains/10k-12       15,008,078 回       79.3 ns/op       0 B/op   0 allocs/op
BenchmarkContains/100k-12       7,084,486 回      173.3 ns/op       0 B/op   0 allocs/op
BenchmarkDelete/100-12           9,593,146 回      119.6 ns/op      32 B/op   1 allocs/op
BenchmarkDelete/1k-12            7,725,520 回      157.2 ns/op      32 B/op   1 allocs/op
BenchmarkDelete/10k-12           6,316,480 回      193.9 ns/op      32 B/op   1 allocs/op
BenchmarkInorder/100-12            936,962 回     1,200 ns/op     2,040 B/op   8 allocs/op
BenchmarkInorder/1k-12            109,360 回    10,519 ns/op    25,208 B/op  12 allocs/op
BenchmarkInorder/10k-12              7,002 回   150,312 ns/op   357,626 B/op  19 allocs/op
BenchmarkInorder/100k-12               224 回 5,214,445 ns/op 4,101,386 B/op  28 allocs/op

3.3 結果の読み方

• Contains: サイズが 100 → 1k → 10k → 100k(1000倍)と増えても、 1回あたりの所要時間は 36.5ns → 51.3ns → 79.3ns → 173.3ns と 約4.7倍にしかなっていない。これは O(log n) の挙動と整合する (log2(1000) ≈ 10、定数倍を考慮すればおおむね妥当な伸び方)。 仮に片重りしたBSTだった場合はサイズに対して線形に近い伸びになるはずで、 この結果はAVL木が実際に平衡を維持できていることの裏付けになっている。
• Delete: 100 → 1k → 10k で 119.6ns → 157.2ns → 193.9ns と、 こちらも対数的な伸びに収まっている。1 allocs/op は挿入時の NewNode によるヒープ確保で、削除自体は追加のアロケーションをしない。
• Insert vs InsertSequential: ランダム挿入(961ns)より 昇順連続挿入(213ns)の方が速い。これは昇順挿入では新しいノードが 毎回「一番右」に追加されるため、探索経路のキャッシュ局所性が高く 分岐予測も効きやすいため。どちらも同程度のオーダーで済んでおり、 「昇順入力で木が崩壊する」という単純BSTの弱点をAVL木が 克服できていることが確認できる。
• Inorder: サイズ10倍ごとに所要時間・メモリ確保量がおおむね 10倍になっており(1.2μs→10.5μs→150μs→5.2ms、メモリも同程度の比率)、 O(n) の全走査として期待通りの挙動を示している。

4. 実行方法まとめ

# ユニットテスト(詳細出力)
go test -v ./...

# カバレッジ
go test -cover ./...

# ベンチマーク(通常テストは実行しない)
go test -bench=. -benchmem -run=^$ ./...

# 特定のベンチマークだけ、反復回数を指定して実行
go test -bench=BenchmarkDelete -benchmem -run=^$ -benchtime=200000x ./...

Binary Search go

en.wikipedia.org

package binary_search_tree2

import "errors"

type Node struct {
        value int
        left  *Node
        right *Node
}

func NewNode(value int) *Node {
        return &Node{value: value}
}

type BinarySearchTree struct {
        root *Node
        size int
}

func (b *BinarySearchTree) Insert(value int) bool {
        // rootがからの場合
        if b.root == nil {
                b.root = NewNode(value)
                b.size++
                return true
        }

        current := b.root

        for {
                // あたいが小さい場合
                if value < current.value {
                        if current.left == nil {
                                current.left = NewNode(value)
                                b.size++
                                return true
                        }
                        current = current.left

                        // 値が大きい場合
                } else if value > current.value {
                        if current.right == nil {
                                current.right = NewNode(value)
                                b.size++
                                return true
                        }

                        current = current.right
                } else {
                        return false
                }
        }
}

func (b *BinarySearchTree) Contains(value int) bool {
        return b.findNode(value) != nil
}

func (b *BinarySearchTree) findNode(value int) *Node {
        current := b.root

        for current != nil {
                if value < current.value {
                        current = current.left
                } else if value > current.value {
                        current = current.right
                } else {
                        return current
                }
        }

        return nil
}

func (b *BinarySearchTree) Delete(value int) bool {
        if !b.Contains(value) {
                return false
        }

        b.root = b.deleteRecursive(b.root, value)
        b.size--
        return true
}

func (b *BinarySearchTree) deleteRecursive(node *Node, value int) *Node {
        if node == nil {
                return nil
        }

        // node.valueより小さい場合
        if value < node.value {
                node.left = b.deleteRecursive(node.left, value)
                return node
        }

        // node.valueより大きい場合
        if value > node.value {
                node.right = b.deleteRecursive(node.right, value)
                return node
        }

        // ここに来た時点で削除対象ノードが見つかっている

        if node.left == nil {
                return node.right
        }

        if node.right == nil {
                return node.left
        }

        // 左右両方に子がある場合
        successor := b.findMinNode(node.right)
        node.value = successor.value
        node.right = b.deleteRecursive(
                node.right,
                successor.value,
        )

        return node
}

func (b *BinarySearchTree) MinValue() (int, error) {
        if b.root == nil {
                return 0, errors.New("tree is empty")
        }

        return b.findMinNode(b.root).value, nil
}

func (b *BinarySearchTree) findMinNode(node *Node) *Node {
        current := node

        for current.left != nil {
                current = current.left
        }

        return current
}

func (b *BinarySearchTree) MaxValue() (int, error) {
        if b.root == nil {
                return 0, errors.New("tree is empty")
        }

        current := b.root
        for current.right != nil {
                current = current.right
        }

        return current.value, nil
}

// 昇順
func (b *BinarySearchTree) Inorder() []int {
        return b.inorderRecursive(b.root, []int{})
}

func (b *BinarySearchTree) inorderRecursive(node *Node, result []int) []int {
        if node == nil {
                return result
        }

        result = b.inorderRecursive(node.left, result)
        result = append(result, node.value)
        result = b.inorderRecursive(node.right, result)
        return result
}

func (b *BinarySearchTree) Preorder() []int {
        return b.preorderRecursive(b.root, []int{})
}

func (b *BinarySearchTree) preorderRecursive(node *Node, result []int) []int {
        if node == nil {
                return result
        }

        result = append(result, node.value)
        result = b.preorderRecursive(node.left, result)
        result = b.preorderRecursive(node.right, result)

        return result
}

func (b *BinarySearchTree) Postorder() []int {
        return b.postorderRecursive(b.root, []int{})
}

func (b *BinarySearchTree) postorderRecursive(node *Node, result []int) []int {
        if node == nil {
                return result
        }

        result = b.postorderRecursive(node.left, result)
        result = b.postorderRecursive(node.right, result)
        result = append(result, node.value)
        return result
}

func (b *BinarySearchTree) Height() int {
        return b.heightRecursive(b.root)
}

func (b *BinarySearchTree) heightRecursive(node *Node) int {
        if node == nil {
                return 0
        }

        leftHeight := b.heightRecursive(node.left)
        rightHeight := b.heightRecursive(node.right)
        return 1 + max(leftHeight, rightHeight)
}

func (b *BinarySearchTree) Size() int {
        return b.size
}

func (b *BinarySearchTree) IsEmpty() bool {
        return b.root == nil
}

func (b *BinarySearchTree) Clear() {
        b.root = nil
        b.size = 0
}

追記

AIでベンチ取った

二分探索木 (Binary Search Tree) — 実装解説・テスト・ベンチマーク

binary_search_tree2.go に実装された、自己平衡を行わない素朴な二分探索木 (BST) について、アルゴリズムの解説と、テスト・ベンチマークの実行結果をまとめる。

1. 二分探索木とは

二分探索木は各ノードが最大2つの子(left, right)を持ち、以下のBST性質を満たす木構造。

• ノード n の左部分木に含まれる値は、すべて n.value より小さい
• ノード n の右部分木に含まれる値は、すべて n.value より大きい
• この性質は木全体のすべてのノードに対して再帰的に成り立つ

この性質により、中間順(inorder)走査で値を昇順に取り出せる。また、探索・挿入・削除は「木の高さ h に比例する計算量 O(h)」で行える。木が平衡していれば h = O(log n) だが、この実装は挿入順序に依存して偏る可能性がある(後述のベンチマークで実際に観測される)。

2. 各操作の解説

Insert(value) — 挿入: O(h)

func (b *BinarySearchTree) Insert(value int) bool

ルートから始めて、value と現在ノードの値を比較しながら左右どちらかへ降りていく。nil に到達したらそこに新ノードを作る。すでに同じ値が存在する場合は false を返し重複を許さない。ループで実装されており再帰呼び出しのオーバーヘッドがない。

Contains(value) / findNode(value) — 探索: O(h)

value と一致するノードが見つかるまで、大小比較に応じて左右に降りるだけ。BST性質のおかげで毎回探索範囲が(平衡していれば)半分程度に絞られる。

Delete(value) — 削除: O(h)

deleteRecursive で削除対象を探しながら再帰的に木を再構築する。削除対象ノードが見つかった時点で3パターンに分岐する。

1. 子がない/片方だけ: 該当ノード自身を、存在する方の子(または nil)で置き換える
2. 子が2つ: 右部分木の最小値(中間順での直後の値 = successor)をコピーしてノードの値とし、右部分木側でその successor を再帰的に削除する

この「successor で値を置き換えて右部分木を再帰削除する」方式は実装がシンプルな反面、削除を繰り返すと木が偏りやすいという性質が知られている(後述のベンチマークで確認)。

MinValue / MaxValue: O(h)

左(右)に辿れるだけ辿った先のノードが最小値(最大値)。空の木では errors.New でエラーを返す。

Inorder / Preorder / Postorder — 走査: O(n)

再帰でスライスに値を積んでいく。特に Inorder は昇順ソート済みのスライスを返すため、BSTを「ソート済み配列を作る」用途にも使える(ただし各再帰呼び出しで append によるスライスの再割り当てが起こりうる)。

Height: O(n)

全ノードを再帰的に辿り、左右部分木の高さの大きい方 + 1 を返す。キャッシュしていないため、呼び出すたびに木全体を走査する点に注意。

Size / IsEmpty / Clear: O(1)

挿入・削除のたびに size フィールドを増減させているため、Size() はキャッシュされた値を返すだけで木を走査しない。

3. 計算量まとめ

操作	平均(平衡時)	最悪(偏った時)
Insert	O(log n)	O(n)
Contains	O(log n)	O(n)
Delete	O(log n)	O(n)
MinValue / MaxValue	O(log n)	O(n)
Inorder / Preorder / Postorder	O(n)	O(n)
Height	O(n)	O(n)
Size / IsEmpty / Clear	O(1)	O(1)

この実装には自己平衡機構(AVL木や赤黒木のような回転処理)がないため、挿入順序や削除の繰り返し方によっては最悪ケースの O(n) に近づいていく。これは後述のベンチマーク結果にも表れている。

4. テスト

binary_search_tree2_test.go に以下のテストを追加し、全件PASS(カバレッジ 98.9%)を確認した。

• TestInsertAndContains: 挿入・重複拒否・存在確認・サイズ
• TestDelete: 存在しない値/葉ノード/子1つ/子2つの削除パターン、削除後の中間順の整合性
• TestMinMaxValue: 空の木でのエラー、最小値・最大値
• TestTraversals: Inorder/Preorder/Postorderの走査結果
• TestHeight: 空・単一ノード・一直線に伸びた木の高さ
• TestSizeIsEmptyClear: サイズ管理とクリア
• TestRandomInsertDeleteConsistency: 乱数シード固定で500件挿入→整列性・存在確認→全削除→空であることを検証するプロパティテスト

$ go test -v ./...
=== RUN   TestInsertAndContains
--- PASS: TestInsertAndContains (0.00s)
=== RUN   TestDelete
--- PASS: TestDelete (0.00s)
=== RUN   TestMinMaxValue
--- PASS: TestMinMaxValue (0.00s)
=== RUN   TestTraversals
--- PASS: TestTraversals (0.00s)
=== RUN   TestHeight
--- PASS: TestHeight (0.00s)
=== RUN   TestSizeIsEmptyClear
--- PASS: TestSizeIsEmptyClear (0.00s)
=== RUN   TestRandomInsertDeleteConsistency
--- PASS: TestRandomInsertDeleteConsistency (0.00s)
PASS
ok      example.com/binary_search_tree2 0.003s

$ go test -cover ./...
ok      example.com/binary_search_tree2 0.003s  coverage: 98.9% of statements

5. ベンチマーク

binary_search_tree2_bench_test.go に、木のサイズ (100 / 1,000 / 10,000 ノード) ごとの Insert / Contains / Delete / Inorder を計測するベンチマークを追加した。値は math/rand(固定シード)でランダムに生成し、各サイズの木を事前に構築してから計測している。

実行環境: go1.26.1 linux/amd64、CPU: AMD Ryzen 5 5500U with Radeon Graphics

$ go test -bench=. -benchmem -run=^$ ./...
goos: linux
goarch: amd64
pkg: example.com/binary_search_tree2
cpu: AMD Ryzen 5 5500U with Radeon Graphics
BenchmarkInsert/100-12            29397541        40.54 ns/op        0 B/op    0 allocs/op
BenchmarkInsert/1k-12             17126144        69.62 ns/op        0 B/op    0 allocs/op
BenchmarkInsert/10k-12             8832976       134.2 ns/op         0 B/op    0 allocs/op
BenchmarkContains/100-12          51256532        23.70 ns/op        0 B/op    0 allocs/op
BenchmarkContains/1k-12           27934609        42.43 ns/op        0 B/op    0 allocs/op
BenchmarkContains/10k-12          16227289        73.88 ns/op        0 B/op    0 allocs/op
BenchmarkDelete/100-12            12282766        94.66 ns/op       24 B/op    1 allocs/op
BenchmarkDelete/1k-12              2291061       522.6 ns/op        24 B/op    1 allocs/op
BenchmarkDelete/10k-12              252987      4851  ns/op        24 B/op    1 allocs/op
BenchmarkInorder/100-12            1000000      1191  ns/op      2040 B/op    8 allocs/op
BenchmarkInorder/1k-12              116946     10295  ns/op     25208 B/op   12 allocs/op
BenchmarkInorder/10k-12               6517    166245  ns/op    357625 B/op   19 allocs/op
PASS
ok      example.com/binary_search_tree2        17.743s

結果の考察

• Insert / Contains はノード数が10倍(100→1k→10k)になっても、所要時間はおよそ1.7〜2倍程度の増加に留まっている。これはランダムな値で構築した木がおおむね平衡に近い状態を保ち、O(log n) の理論通りの挙動をしていることを示す。
• Inorder はノード数に比例して増加している(100→1kで約8.6倍、1k→10kで約16倍)。全ノードを走査し append でスライスを構築するため O(n) であり、allocs/op もノード数増加に伴って増えている(スライスの再割り当て)。

• Delete が最も興味深い結果を示している。100→1kで約5.5倍、1k→10kで約9.3倍と、O(log n) から予想される増加(1.5倍・1.3倍程度)を大きく超えて、ほぼノード数に比例するペースで遅くなっている。

  これは「削除→同じ値を再挿入」を繰り返すベンチマークの性質上、deleteRecursive が採用している 「右部分木の最小値(successor)で置き換える」削除戦略によって、木が徐々に偏っていくためだと考えられる。この戦略は実装は単純だが、削除を左右対称に行わない(常に右部分木から successor を取る)ため、削除を繰り返すたびに木が右に浅く・左に深くなる方向へ統計的に偏っていくことが知られている。今回のベンチマークはまさにその偏りを可視化した形になっている。

  実務的には、この偏りを避けるには「左部分木の最大値(predecessor)と右部分木の最小値(successor)を交互に使う」「削除のたびに回転で再平衡する(AVL木・赤黒木)」といった対策が必要になる。この実装にはそのような自己平衡機構がないため、削除を繰り返す用途では木の高さが理論値より大きくなりうる点に注意する。

6. まとめ

• 実装自体は教科書的な二分探索木として正しく動作しており、テストは全件PASS(カバレッジ98.9%)。
• Insert/Contains はランダムなデータに対しては対数時間に近い挙動を示す。
• 一方でDeleteを繰り返すと、successorベースの削除戦略の偏りにより性能が劣化していく傾向が実測でも確認できた。長時間にわたって削除・挿入を繰り返すユースケースでは、AVL木や赤黒木など自己平衡BSTの採用を検討すべきである。

Redisのようなもの go

ここ最近の記事で書いたアルゴリズムをまとめてredisのような物をAIで作ってベンチ取ってみました

redispoi 開発レポート

自作アルゴリズム実装群を RESP プロトコル(Redis 互換プロトコル)サーバーへ統合し、 ユニットテストとベンチマークを整備した記録。

検証環境: AMD Ryzen 5 5500U (12 論理コア) / Go 1.26.1 linux-amd64 / Redis 7.0.15

---

目次

1. 経緯
2. アーキテクチャ概要
3. 各アルゴリズムの解説
4. テストの解説
5. ベンチマークの解説
6. まとめ

---

1. 経緯

go-algo リポジトリには、Redis の内部実装を模した学習用アルゴリズムが ディレクトリ単位でバラバラに実装されていた(incremental_hash, skiplist_d, radix_tree, quick_list, simple_list_pack_fast, hyper_log_log, bloom_filter, cuckoo_filter, count_min_sketch, heavy_keeper_top_k_fast, t_digest の 11 個)。それぞれは単体の main.go / *_test.go として動くデモ実装で、相互には何のつながりもなかった。

今回の作業は大きく 3 段階で進めた。

1. サーバー化: この 11 個を redispoi/internal/ 配下にコピーし、実際に コマンドを処理する RESP サーバーとして統合した。
2. テスト整備: 各 internal/ パッケージに、元の *_test.go を新 API に 合わせて移植・拡充したユニットテストを追加した。この過程で 実際のバグを 1 件発見・修正した(zset.Range の負インデックス未対応、詳細は §4)。
3. ベンチマーク整備と本家 Redis 比較: 各パッケージに Go ベンチマークを 追加し、実際の redis-server を用意して redis-benchmark で比較した。 この過程で 2 件目の実バグ(性能バグ)を発見・修正した(LRANGE が 範囲サイズによらずリスト全体を毎回展開していた問題、詳細は §5)。

「アルゴリズムを移植して終わり」ではなく、実際に動かして負荷をかけることで 初めて見つかる不具合が 2 件あった、というのがこの作業全体を通しての一番の 収穫と言える。

---

2. アーキテクチャ概要

redispoi/
├── main.go                 # エントリポイント (TCPリッスン)
├── server/                 # RESPプロトコル・コマンドディスパッチ・Store
│   ├── resp.go             # RESPパーサ(multibulk + inline両対応)
│   ├── reply.go            # RESPレスポンス構築
│   ├── store.go            # 統一キースペース(dict + radixの二重管理)
│   └── commands_*.go       # コマンドグループ別ハンドラ
└── internal/                # 移植した11アルゴリズム
    ├── dict/                # incremental_hash → キースペース本体
    ├── radix/               # radix_tree → KEYSのための補助インデックス
    ├── zset/                # skiplist_d → ZADD/ZRANGE
    ├── quicklist/           # quick_list → LPUSH/RPUSH/LRANGE
    ├── listpack/             # simple_list_pack_fast → LPACK.*
    ├── hyperloglog/         # hyper_log_log → PFADD/PFCOUNT
    ├── bloom/               # bloom_filter → BF.*
    ├── cuckoo/              # cuckoo_filter → CF.*
    ├── cms/                 # count_min_sketch → CMS.*
    ├── topk/                # heavy_keeper_top_k_fast → TOPK.*
    └── tdigest/             # t_digest → TDIGEST.*

すべてのキーは dict(incremental_hash 由来のインクリメンタルリハッシュ ハッシュテーブル)1 つに集約されており、型ごとに kind タグを持つ RedisValue を格納する。これは実際の Redis がキースペース全体を単一の dict で管理しているのと同じ設計思想で、STRING/LIST/ZSET/LISTPACKは もちろん HLL/BLOOM/CUCKOO/CMS/TOPK/TDIGEST も同じキースペースの 住民として DEL/EXISTS/TYPE/KEYS から一様に扱える。

コマンド実行は Store が持つ単一の sync.Mutex で直列化しており、 1コマンド = 1クリティカルセクションといういたってシンプルな並行性モデルに なっている(§5 のベンチマーク考察で触れる通り、これが本家 Redis との比較で 無視できない差になっている)。

---

3. 各アルゴリズムの解説

3.1 incremental_hash → internal/dict(キースペース本体)

Redis の dict が採用している「インクリメンタルリハッシュ」をそのまま 実装したもの。テーブルを拡張する瞬間に全要素を一括で移し替える(stop-the-world) のではなく、tables[0](旧)と tables[1](新)を同時に持ち、Put/Get/ Remove などの操作 1 回につき 1 バケット分だけ移行を進める。これにより 巨大なハッシュテーブルでもリハッシュ中に長い停止が発生しない。

• 移植時の変更点: 元コードは var tables [2][]*Node というパッケージレベル グローバル変数を使っており、プロセス内にインスタンスが 1 つしか存在 できないという欠陥があった(2 個目の NewIncrementalHashTable を作ると 1 個目のテーブルを共有・破壊してしまう)。Dict 構造体のフィールドに 移すことで、キーごとに独立したハッシュテーブルを持てるようにした。
• 値の型を int から any に変更し、文字列・リスト・zset など任意の Redis 値を格納できるようにした。
• Keys() を新規追加し、KEYS コマンドの全件列挙に対応。

3.2 skiplist_d → internal/zset(ソート済みセット)

スキップリストは平衡木を使わずに O(log n) の検索・挿入・削除を実現する 確率的データ構造。各ノードがコイン投げ(確率 p=0.5)でランダムに「上位 レベルへの近道」を持つことで、実質的に二分探索的な高速検索を実現する。

• 移植時の変更点が最も大きいパッケージ。元コードは int の集合を扱う 単純なスキップリストで、var head *Node がやはりグローバル変数 だった(同時に 1 つしか使えない)。ZSET は「メンバー文字列 + スコア」の ペアをスコア昇順、同スコアはメンバー名の辞書順で並べる必要があるため、 ノードの比較キーを Entry{Score, Member} に変更し、さらに map[string]float64 を横に持たせて ZSCORE を O(1) にした。これは 実際の Redis の zset 実装(スキップリスト + dict の組み合わせ)と同じ 構成。
• Range(start, stop) はランク(順位)範囲を返す。当初 Redis 流の負 インデックス(-1 = 最後の要素)を解釈できておらず、テストで発見・修正 した(§4 参照)。

3.3 radix_tree → internal/radix(KEYS 用補助インデックス)

パトリシアトライ(基数木)。共通接頭辞を持つ文字列同士でノードを共有する ことで、単純なトライよりメモリ効率よく前方一致検索ができる。例えば "romane", "romanus", "romulus" を挿入すると、"rom" は共有ノードに なり、そこから "ane", "anus", "ulus" に枝分かれする。

• こちらも var root *Node がグローバル変数だったのを構造体フィールドに 修正。
• Delete(ノードの isWord フラグを倒すだけの簡易実装)と CollectWithPrefix(前方一致するキーを全列挙)を新規追加した。これは KEYS prefix* のようなプレフィックス検索をサーバー側で提供するために 必要だったため。

3.4 quick_list → internal/quicklist(LIST 型)

Redis の実際のリスト型と同じ「双方向連結リスト + 各ノードが固定容量の 配列を持つ」構造(quicklist)。単純な連結リスト(ノード 1 個 = 要素 1 個) よりポインタのオーバーヘッドが小さく、単純な配列(1 本の巨大スライス)より 先頭挿入が軽い、という両者の折衷案になっている。

• 元コードはすでにグローバル変数を使っておらず、そのまま移植できた。
• Len() と ToSlice() を追加し、後から気づいた性能バグの修正で Range(start, stop) を追加した(§5 参照)。

3.5 simple_list_pack_fast → internal/listpack(コンパクトリスト型)

Redis の listpack(小さいリストや小さい zset を圧縮してメモリ効率よく 保持するためのバイト列エンコーディング)を再現したもの。要素をポインタの 配列ではなく、[type(1B)][長さ/値][末尾からの逆算用長さ(4B)] という 自己記述的なバイト列としてフラットな []byte に直列化する。末尾の 「逆算用長さ」フィールドのおかげで、末尾から前方向にもジャンプしながら 辿れる(Backward()/Last() が O(1) に近いコストで動く理由)。 本サーバーでは LPACK.* という独立したコマンド群として公開している。

• 元々完全にグローバル変数フリーで、パッケージ名変更以外はほぼ無改造で 移植できた。

3.6 hyper_log_log → internal/hyperloglog(PFADD/PFCOUNT)

確率的濃度推定アルゴリズム。要素をハッシュ化し、ハッシュ値の上位ビットで レジスタ(バケット)を選び、残りビット列の「先頭から連続する 0 の個数+1」 (=順位/rank)をそのレジスタの最大値として記録する。まれにしか出現しない 高順位の観測ほど「レジスタ数が多いはず」という統計的性質を使い、全レジスタ の調和平均から濃度を逆算する。メモリは要素数に依存せず 2^precision 個のレジスタ(1バイト)だけで済むのが最大の特徴。

• 元コードは add(value any)(int/string 両対応)と estimate() が 非公開メソッドで、パッケージ外(サーバー側)から呼べなかった。 PFADD の要素は常に文字列なので、Add(string)/Estimate() int という 薄いエクスポート版を追加した(int 対応は今回のユースケースでは不要と 判断し削除)。

3.7 bloom_filter → internal/bloom(BF.*)

確率的メンバーシップ判定。ビット配列に対し、1 要素につき複数のハッシュ 位置のビットを立てる。判定時はそれらのビットが全て立っていれば 「たぶん含まれる」、1つでも立っていなければ「確実に含まれない」と判定する (偽陽性はあるが偽陰性は絶対にない、という非対称性が特徴)。

• addString/mightContainString が非公開だったのを Add/MightContain としてエクスポート。

3.8 cuckoo_filter → internal/cuckoo(CF.*)

Bloom フィルタと同じくメンバーシップ判定の確率的構造だが、削除が 可能という違いがある。各要素をフィンガープリント(短いハッシュ値)に 変換し、2 つの候補バケットのどちらかに格納する。両方満杯なら既存の フィンガープリントを蹴り出して玉突き的に別バケットへ移す(cuckoo hashing と同じ発想)。

• 元々グローバル変数なし・API 完全エクスポート済みで、パッケージ名変更 以外はほぼ無改造。

3.9 count_min_sketch → internal/cms(CMS.*)

出現頻度を近似カウントする構造。depth 行 × width 列のカウンタ表を 持ち、要素を depth 個の異なるハッシュ関数で各行に写像してカウンタを インクリメントする。問い合わせ時は該当する depth 個のカウンタの 最小値を答えとする(ハッシュ衝突は必ずカウントを水増しする方向にしか 働かないため、最小値を取ることで過大評価はあっても過小評価は絶対にない、 という保証が成り立つ)。

• add(value, count) が非公開だったため、任意の増分を指定できる IncrBy(value string, count int) を新規追加(CMS.INCRBY 用)。

3.10 heavy_keeper_top_k_fast → internal/topk(TOPK.*)

「今どのアイテムが上位 k 件か」を近似的に答える構造。Count-Min Sketch に 似た多段ハッシュ表を持つが、各バケットは「フィンガープリント + カウント」 を 1 組だけ保持し、異なるフィンガープリントが来た際は指数的に減衰する 確率でカウントを減らす(decay^count の確率で -1 する)、というのが Count-Min との最大の違い。これにより低頻度アイテムのノイズが自然に 消えていき、真に頻出するアイテムだけが生き残りやすくなる。上位 k 件の 候補集合自体は最小ヒープで管理し、TopK()/IsTopK() を O(k log k) で 提供する。

• 元コードの Result 型は GetItem()/GetEstimatedCount()/String()/ Equals()/Hash() を持つオブジェクト指向的な作りだったが、サーバー側 では単に Item/EstimatedCount を読めれば十分なため、素直な構造体の 公開フィールドに簡略化した。

3.11 t_digest → internal/tdigest(TDIGEST.*)

分位点(パーセンタイル)を近似計算する構造。値を「重心(mean, weight)」の 集合(セントロイド)に圧縮していくが、圧縮の粒度は分布の端(0%/100% 付近)ほど細かく、中央(50%付近)ほど粗くなるように設計されている (maxCentroidWeight(q) = 4 * totalWeight * q * (1-q) / compression という 放物線状の重み上限関数がその役割を担う)。これにより p50 の精度を多少 犠牲にしてでも、監視ツールなどで重要になりがちな p99/p999 のような裾の 分位点を高精度に保てる。

• API がすでに完全にエクスポートされており、パッケージ名変更以外はほぼ 無改造で移植できた。

---

4. テストの解説

各 internal/ パッケージに、元の *_test.go(存在するもの)を新 API に 合わせて移植し、radix(元々テストが存在しなかった)は新規に書いた。

パッケージ	テスト数(概算)	備考
dict	12	incremental_hash から移植。グローバル変数リセット処理は不要になったため削除
radix	16	新規作成(Insert/Search/StartsWith/Delete/CollectWithPrefix)
zset	15	skiplist_d の int ベーステストを (member, score) API 向けに全面書き換え
quicklist	10	quick_list から移植 + Range 用テスト追加
listpack	17	simple_list_pack_fast からほぼそのまま移植
hyperloglog	16	int 対応部分を除いて移植
bloom	15	Add/MightContain に改名して移植
cuckoo	9	ほぼそのまま移植
cms	15	IncrBy 用テストを追加
topk	14	Result の平フィールド化に合わせて移植
tdigest	30	ほぼそのまま移植

go test -race ./... を含め全パッケージがパスすることを確認済み。

発見したバグ 1: zset.Range が負インデックスを解釈できていなかった

ZRANGE key start stop は Redis の仕様上、-1 が「最後の要素」を指す 負インデックスをサポートする。当初この変換はコマンド層 (commands_zset.go)だけで行っており、zset.Range 自体は非負の 絶対インデックスしか受け付けない実装だった。

zset パッケージ単体でユニットテストを書いた際、z.Range(0, -1) が 本来「全件取得」を期待する自然な呼び方であるにもかかわらず、 start(0) > stop(-1) という比較で弾かれて常に空を返すことが判明した。 サーバー経由(ZRANGE key 0 -1)ではコマンド層が事前に変換していたため 問題が表面化していなかった、まさに「ユニットテストで初めて見つかる」 典型的な不具合だった。

修正は変換ロジックそのものを zset.Range に移動し、コマンド層の重複 コードを削除する形で行った。これは Redis 風の負インデックス解釈が そもそも zset の意味論に属する話であり、呼び出し側ではなく zset パッケージ自身が知っているべき、という設計判断でもある。

---

5. ベンチマークの解説

5.1 各 internal/ パッケージの Go ベンチマーク(自己比較)

go test -bench=. -benchmem ./internal/... の結果(抜粋、代表的なもの)。

パッケージ	操作	1回あたり	メモリ/回	alloc/回
dict	Put	312 ns	116 B	3
dict	Get	135 ns	15 B	1
dict	Mixed(Put/Get/Remove混在)	76.6 ns	10 B	0
radix	Insert	296 ns	96 B	3
radix	Search(hit)	166 ns	15 B	1
radix	StartsWith	4.8 ns	0 B	0
radix	CollectWithPrefix(1万件中)	66.4 µs	68.6 KB	1123
zset	Add	833 ns	334 B	5
zset	Score	132 ns	15 B	1
zset	Range(100件)	1.6 µs	2.7 KB	1
quicklist	PushBack	144 ns	51 B	2
quicklist	PushFront	689 ns	1132 B	4
quicklist	PopBack/PopFront	3.2〜3.4 ns	0 B	0
listpack	AddSequential	29.6 ns	71 B	0
listpack	Values(全件展開)	399 µs	744 KB	9762
hyperloglog	Add	14.3 ns	0 B	0
hyperloglog	Estimate(precision=14, 16384レジスタ走査)	70.5 µs	0 B	0
bloom	Add	236 ns	0 B	0
bloom	MightContain	192 ns	23 B	1
cuckoo	Add	261 ns	0 B	0
cuckoo	MightContain(miss)	40.2 ns	0 B	0
cms	Add	56.7 ns	0 B	0
cms	EstimateCount	49.0 ns	0 B	0
topk	AddString(width=4096/depth=5)	207 ns	16 B	1
topk	TopK()(k=200時)	23.9 µs	4.9 KB	1
tdigest	AddByValue(compression=100)	265 ns	100 B	3
tdigest	Compress(compression=500)	619 µs	147 KB	5294
tdigest	Quantile問い合わせ	306 ns	0 B	0

傾向として妥当な結果: - PopBack/PopFront/StartsWith/Mix64 のような単純なポインタ操作は ナノ秒未満〜数ナノ秒。 - HyperLogLog.Estimate() はレジスタ数(precision=14 → 16384個)に比例して 約 70µs かかり、レジスタ数を増やすほど精度と引き換えに遅くなるという 設計通りの傾向が出ている。 - TDigest.Compress は compression を上げる(centroid数の上限を上げる) ほど 1 回あたりのコストが跳ね上がる(25→619µs換算で約31倍)。精度と 引き換えという設計が数値にも表れている。

⚠️ ベンチマーク実装上の注意(教訓): quicklist.PopBack/PopFront は O(1) 操作のため Go の testing パッケージが目標実行時間(既定1秒)に 到達させようと b.N を非常に大きく(数億〜)自動で引き上げる。素朴に 「タイマー計測外で b.N 個 push してからタイマー計測で b.N 回 pop する」 という書き方をすると、計測対象外の準備コストがメモリを食い尽くして OOM Kill されるという事故が実際に発生した(このレポート作成中に踏んだ)。 対策として、一定サイズ(10万件)の作業セットを維持し、空になったら (計測を止めた上で)再充填する形に書き換えて解決した。

5.2 発見したバグ 2: LRANGE が範囲サイズに関わらず O(リスト全長) だった

本家 Redis との比較ベンチマーク(§5.3)を取ったところ、単純な SET/GET/LPUSH 等は本家と大差ない結果だった一方で、LRANGE だけ 突出して遅い(後述の通り本家の約 50〜65 分の1)ことが判明した。

原因は server/commands_list.go の cmdLRange が、要求された範囲が たとえ LRANGE key 0 99(先頭100件)であっても、quicklist.ToSlice()で リスト全体を一旦フルスライスに展開してから Go のスライス構文で 切り出していたこと。本家 Redis の LRANGE は範囲の近い方の端から辿る ことで O(オフセット + 範囲長) で済むのに対し、この実装は常に O(リスト全長) になっていた。10 万要素のリストから先頭 100 件を取り出す だけなのに、10 万要素分のコピーが毎回発生していた計算になる。

修正として quicklist.Range(start, stop) を新設した。Get(index) が 既に採用している「head/tail のうち近い方から辿る」という戦略を範囲取得 にも適用し、リスト全体を経由せず該当範囲だけを収集するようにした。

修正前後の内部ベンチマーク比較(10万要素のリストから先頭100件を取得):

実装	1回あたり	メモリ/回
修正前 (ToSlice() → スライス)	1,010,650 ns (≒1.01 ms)	1.6 MB
修正後 (Range())	1,596 ns	1.8 KB

約 633 倍の高速化。範囲サイズが固定(100件)なのにリスト全長 (10万件)に比例していたコストが、範囲サイズだけに比例するコストへ 改善されたことになる。

quicklist.Range の実装

修正前は cmdLRange(server/commands_list.go)が list.ToSlice()で 全件コピーしてから Go のスライス構文 sliced[start:stop+1] で切り出す 実装だった。ToSlice() は head から tail まで無条件に全ノードを辿るため、 start/stop の値に関係なく常に size 分のコピーが発生していた。

新設した quicklist.Range(start, stop) は、Get(index) が採用していた 「head/tail のうち近い方から辿る」戦略を単一要素ではなく範囲取得に 拡張したもので、次の3点で無駄なコピーを排除している。

1. 走査の起点を選ぶ: start(先頭からの距離)と size-1-stop (末尾からの距離)を比較し、近い方の端からノードを辿る。LRANGE 0 99 のような先頭寄りの範囲は head から、LRANGE -100 -1 のような 末尾寄りの範囲は tail から辿るので、無関係な側のノードには触れない。
2. 範囲の終端に達した時点で打ち切る: head 側から辿る場合は offset == stop になった瞬間に return し、それ以降の要素は ループにすら入らない。tail 側から辿る場合も offset == start で 同様に打ち切る。
3. tail 側から辿った場合の順序調整: tail からの走査は逆順に 要素を積むことになるため、reverse() で一度だけ反転して Redis が期待する昇順に戻す。

呼び出し側の cmdLRange は、Redis 流の負インデックス解決 (§4 の zset.Range 修正と同じパターン)を済ませたあと list.Range(start, stop) を呼ぶだけになり、ToSlice() は LRANGE の経路からは呼ばれなくなった。結果として計算量は 「常に O(リスト全長)」から「O(近い方の端までの距離 + 範囲長)」 に変わり、本家 Redis の LRANGE と同じ考え方になっている。

5.3 本家 Redis との比較ベンチマーク

redis-benchmark -n 100000 -c 50(同時接続50、リクエスト数10万、 パイプラインなし)で計測。redispoi はポート 6380、redis-server (永続化無効・save ""/appendonly no)はポート 6399 で待受け。

LRANGE 修正前 vs 修正後 vs 本家Redis

コマンド	redispoi(修正前)	redispoi(修正後)	本家Redis
LRANGE_100(先頭100件)	874 req/s (p50 57.1ms)	41,068 req/s (p50 0.615ms)	46,773 req/s (p50 0.583ms)
LRANGE_300	869 req/s (p50 57.4ms)	26,254 req/s (p50 0.943ms)	30,221 req/s (p50 0.879ms)
LRANGE_500	854 req/s (p50 58.4ms)	16,918 req/s (p50 1.479ms)	22,336 req/s (p50 1.167ms)
LRANGE_600	848 req/s (p50 58.8ms)	14,957 req/s (p50 1.687ms)	20,161 req/s (p50 1.335ms)

修正前は本家の 約 50〜65 分の1 だったスループットが、修正後は 本家の 74〜88% 程度まで縮まった。残る差は quicklist の探索方向選択 (近い方の端から辿る)は同じでも、Redis 本体は C 実装かつオブジェクト アロケーションがさらに少なく調律されている分と考えられる。

その他のコマンド(修正後、パイプラインなし)

コマンド	redispoi	本家Redis	redispoiの比率
PING_INLINE	53,419 req/s	55,036 req/s	97%
SET	53,050 req/s	57,241 req/s	93%
GET	53,277 req/s	57,405 req/s	93%
INCR	53,191 req/s	59,102 req/s	90%
LPUSH	52,056 req/s	57,904 req/s	90%
RPUSH	53,107 req/s	57,937 req/s	92%
LPOP	52,994 req/s	57,770 req/s	92%
RPOP	53,476 req/s	57,703 req/s	93%
ZADD	53,447 req/s	57,770 req/s	93%

このスケール(c=50, パイプラインなし)では、単純なコマンドはローカル ループバック越しの往復レイテンシがボトルネックになりやすく、本家Redisの シングルスレッドイベントループと redispoi のゴルーチン+グローバルロック 方式との差が 10%前後に収まっている。

パイプライン時(-P 16)の参考値

コマンド	redispoi	本家Redis
SET	292,398 req/s (p50 1.375ms)	265,957 req/s (p50 2.943ms)
GET	297,619 req/s (p50 1.359ms)	280,112 req/s (p50 2.815ms)
LRANGE_100	111,235 req/s (p50 3.095ms)	103,413 req/s (p50 6.591ms)

パイプライン(1接続で複数コマンドをまとめ送り)時は、むしろ redispoi の 方が高いスループットと低い p50 を記録した。これは 1 回の局所実行・ 1 回の測定に基づく参考値であり、本家Redis側の追加機能(キー期限管理、 キースペース通知、AOF/RDB 周りのフック等、save ""/appendonly no でも 完全にはゼロにならないオーバーヘッド)がここで相対的に効いてきた可能性が 考えられるが、断定はできない。

5.4 比較の公平性に関する注意

以下の非対称性があるため、この比較は「同じ土俵での性能勝負」ではなく 「学習実装がどの水準にあるかの目安」として読むべきもの。

• 並行性モデルが違う: 本家Redisはシングルスレッドイベントループ (I/O多重化 + 非ブロッキング)。redispoiは「接続ごとに1ゴルーチン + コマンド実行は単一の sync.Mutex で直列化」という、単純だが ロック競合や goroutine 切り替えのオーバーヘッドを抱える方式。
• 実装言語と最適化のレベルが違う: 本家Redisは十数年に渡ってCで チューニングされた実装。redispoiは今回作成したばかりの学習用Go実装で、 メモリアロケーションのチューニングなどは行っていない。
• 機能差: 本家Redisが持つ永続化(RDB/AOF)・レプリケーション・ キー有効期限・キースペース通知・ACL 等は redispoi には一切ない (save ""/appendonly no で計測時は無効化しているが、本家側コード パスに完全にゼロコストというわけではない)。
• プロービング系コマンド(BF.*/CF.*/CMS.*/TOPK.*/TDIGEST.*)は 本稿執筆時点では本家 RedisBloom モジュールが未導入だったため 比較対象から除外していた。今回の比較は両者が共通して持つ STRING/LIST/ZSET のコマンドに限定している(後日 RedisBloom を導入して 比較した結果は §5.5 参照)。
• 計測はいずれも 1 回のローカル実行に基づく。負荷変動や JIT・GC ウォームアップ差などのノイズを平均化する複数回試行はしていないため、 数値は目安として扱うべきである。

5.5 本家 RedisBloom モジュールとの比較(BF/CF/CMS/TOPK/TDIGEST)

§5.4 で除外していたプロービング系コマンドについて、後日 RedisBloom 本体(標準単体配布として最後のリリースである v2.8.23。それ以降は Redis 8 本体にデータ構造ごと統合され単体配布は終了している)をソースから ビルドし、redis-server --loadmodule で本家 redis-server(§5.3 と同じ ポート 6399、save ""/appendonly no)にロードして追加比較を行った。 redispoi は同じくポート 6380。各データ構造は両者で同一パラメータで 事前に確保した上で計測している:

コマンド群	事前確保パラメータ
BF.*	BF.RESERVE key 0.01 1000000(誤検出率1%、容量100万)
CF.*	CF.RESERVE key 1000000(容量100万)
CMS.*	CMS.INITBYDIM key 2000 5(width=2000, depth=5)
TOPK.*	TOPK.RESERVE key 100 8 4 0.9(k=100, width=8, depth=4, decay=0.9)
TDIGEST.*	TDIGEST.CREATE key COMPRESSION 100

redis-benchmark -n 100000 -c 50 -r 1000000(ランダムキー、パイプライン なし)での結果:

コマンド	redispoi	本家Redis+RedisBloom	redispoiの比率
BF.ADD	52,715 req/s (p50 0.479ms)	52,966 req/s (p50 0.527ms)	100%
BF.EXISTS	52,411 req/s (p50 0.479ms)	53,022 req/s (p50 0.527ms)	99%
CF.ADD	52,083 req/s (p50 0.479ms)	52,826 req/s (p50 0.527ms)	99%
CF.EXISTS	52,549 req/s (p50 0.471ms)	53,447 req/s (p50 0.527ms)	98%
CMS.INCRBY	52,192 req/s (p50 0.479ms)	52,743 req/s (p50 0.527ms)	99%
CMS.QUERY	52,493 req/s (p50 0.479ms)	52,715 req/s (p50 0.527ms)	100%
TOPK.ADD	52,493 req/s (p50 0.471ms)	53,735 req/s (p50 0.519ms)	98%
TOPK.QUERY	52,659 req/s (p50 0.479ms)	52,659 req/s (p50 0.527ms)	100%
TDIGEST.ADD	52,549 req/s (p50 0.479ms)	53,419 req/s (p50 0.527ms)	98%
TDIGEST.QUANTILE	53,135 req/s (p50 0.463ms)	52,165 req/s (p50 0.535ms)	102%

パイプライン時(-P 16)での結果:

コマンド	redispoi	本家Redis+RedisBloom	redispoiの比率
BF.ADD	289,017 req/s (p50 1.375ms)	181,324 req/s (p50 4.455ms)	159%
BF.EXISTS	294,551 req/s (p50 1.367ms)	184,502 req/s (p50 4.367ms)	160%
CF.ADD	286,123 req/s (p50 1.399ms)	185,185 req/s (p50 4.423ms)	155%
CF.EXISTS	291,971 req/s (p50 1.375ms)	187,091 req/s (p50 4.223ms)	156%
CMS.INCRBY	284,495 req/s (p50 1.407ms)	181,984 req/s (p50 4.519ms)	156%
CMS.QUERY	287,770 req/s (p50 1.399ms)	187,970 req/s (p50 4.295ms)	153%
TOPK.ADD	287,356 req/s (p50 1.399ms)	181,653 req/s (p50 4.415ms)	158%
TOPK.QUERY	290,698 req/s (p50 1.383ms)	189,215 req/s (p50 4.327ms)	154%
TDIGEST.ADD	292,826 req/s (p50 1.383ms)	180,995 req/s (p50 4.535ms)	162%
TDIGEST.QUANTILE	280,112 req/s (p50 1.407ms)	163,132 req/s (p50 5.119ms)	172%

傾向: パイプラインなしでは §5.3 の STRING/LIST/ZSET と同様、5つの データ構造すべてで redispoi は本家の 98〜102% とほぼ互角(ループバックの 往復レイテンシが支配的なため、実装差が埋もれる)。一方パイプライン時は 5つのデータ構造すべてで redispoi が本家の 153〜172% と一貫して上回った。 §5.3 でSET/GETパイプライン時に見えた「redispoiの方が速い」傾向が、 プロービング系コマンドではさらに大きな差(約1.1倍→約1.5〜1.7倍)として 再現している。考えられる要因:

• 本家RedisBloomは「スケーラブルフィルタ」(容量超過時に新しいサブ フィルタ/バケット配列を追加するExpansion rate機構)を持ち、BF.ADD やCF.ADDのたびに複数サブ構造を辿る分岐処理が挟まる。redispoiの実装は いずれも単一の固定長配列のみを持ち、この種の分岐がない。
• RedisModule API経由のコマンド実行は、引数のパース・型チェック・ RedisModule_ReplyWith*呼び出しなど、redispoiの薄いディスパッチ層 より層が厚い。
• 本家側はRDB永続化(モジュール型のAUX_SAVE/RDB_LOADフック)や ACL・キースペース通知との連携のための追加コストを恒常的に抱えている。

いずれも実装を読んで検証したわけではない推測であり、§5.4と同様 1回のローカル実行に基づく参考値である点に注意。

---

6. まとめ

• 11個のバラバラなアルゴリズム学習実装を、実際に RESP プロトコルで 会話できる 1 つの Redis 風サーバーに統合した。
• 全 internal/ パッケージにユニットテストを追加し、テストを書く過程で zset.Range の負インデックス未対応バグを発見・修正した。
• 全 internal/ パッケージに Go ベンチマークを追加し、実行過程で ベンチマーク自体の設計不備(O(1)操作に対する b.N 自動スケーリングが 引き起こす OOM)にも遭遇し修正した。
• 本家 redis-server との比較ベンチマークを通じて、LRANGE が O(リスト全長) になっている性能バグを発見し、quicklist.Range の 新設によって修正した(該当ケースで約633倍の高速化、本家比で 50〜65分の1 → 74〜88%程度まで改善)。
• 単純なコマンドについては、学習用のシンプルな実装であっても本家の 90%台のスループットに達しており、この規模の学習実装としては妥当な 水準にあると言える。
• 後日、本家 RedisBloom モジュール(v2.8.23)をビルド・導入して BF.*/CF.*/CMS.*/TOPK.*/TDIGEST.* も比較したところ、 パイプラインなしでは本家の98〜102%とほぼ互角、パイプライン時は 5構造すべてで redispoi が本家の153〜172%という結果になった (詳細は§5.5)。

t_digest go

package t_digest

import (
        "math"
        "sort"
)

type Centroid struct {
        mean   float64
        weight int
}

func NewCentroid(mean float64, weight int) *Centroid {
        return &Centroid{
                mean:   mean,
                weight: weight,
        }
}

func (c *Centroid) add(value float64, additionalWeight int) {
        newWeight := c.weight + additionalWeight
        c.mean += (value - c.mean) * float64(additionalWeight) / float64(newWeight)
        c.weight = newWeight
}

func (c *Centroid) getMean() float64 {
        return c.mean
}

func (c *Centroid) getWeight() int {
        return c.weight
}

type TDigest struct {
        compression float64
        centroids   []*Centroid
        buffer      []*Centroid
        bufferLimit int
        totalWeight int
        minf        float64
        maxf        float64
}

func NewTDigest() *TDigest {
        return NewTDigestByCompression(100.0)
}

func NewTDigestByCompression(compression float64) *TDigest {
        if math.IsInf(compression, 0) || math.IsNaN(compression) || compression <= 0.0 {
                panic("compression must be finite and greater than 0")
        }

        return &TDigest{
                compression: compression,
                centroids:   []*Centroid{},
                buffer:      []*Centroid{},
                minf:        math.Inf(1),
                maxf:        math.Inf(-1),
                bufferLimit: int(math.Max(20, math.Ceil(compression*8.0))),
        }
}

func (t *TDigest) AddByValue(value float64) {
        t.AddByValueAndWeight(value, 1)
}

func (t *TDigest) AddByValueAndWeight(value float64, weight int) {
        if math.IsNaN(value) || math.IsInf(value, 0) {
                panic("value must be finite")
        }

        if weight <= 0 {
                panic("weight must be greater than 0")
        }

        t.buffer = append(t.buffer, NewCentroid(value, weight))
        t.totalWeight = safeAdd(
                t.totalWeight,
                weight,
        )

        t.minf = math.Min(t.minf, value)
        t.maxf = math.Max(t.maxf, value)

        if len(t.buffer) >= t.bufferLimit {
                t.compress()
        }
}

func (t *TDigest) Merge(other *TDigest) {
        if other == nil {
                panic("other must not be null")
        }

        if other.totalWeight == 0 {
                return
        }

        other.compress()

        for _, centroid := range other.centroids {
                t.buffer = append(t.buffer, NewCentroid(centroid.mean, centroid.weight))
        }

        t.totalWeight = safeAdd(
                t.totalWeight,
                other.totalWeight,
        )

        t.minf = math.Min(t.minf, other.minf)
        t.maxf = math.Max(t.maxf, other.maxf)

        t.compress()
}

func (t *TDigest) Quantile(q float64) float64 {
        if math.IsNaN(q) || q < 0.0 || q > 1.0 {
                panic("q must be between 0 and 1")
        }

        if t.totalWeight == 0 {
                return math.NaN()
        }

        if q == 0.0 {
                return t.minf
        }

        if q == 1.0 {
                return t.maxf
        }

        t.compress()

        if len(t.centroids) == 1 {
                return t.centroids[0].mean
        }

        target := q * (float64(t.totalWeight) - 1.0)
        cumulative := 0.0

        for i := 0; i < len(t.centroids); i++ {
                current := t.centroids[i]
                center := cumulative + (float64(current.weight)-1.0)/2.0

                if target <= center {
                        if i == 0 {
                                return interpolate(
                                        t.minf,
                                        current.mean,
                                        target,
                                        center,
                                )
                        }
                        previous := t.centroids[i-1]
                        previousCenter := cumulative - float64(previous.weight) + (float64(previous.weight)-1.0)/2.0

                        return interpolateBetweenCenters(
                                previous.mean,
                                current.mean,
                                previousCenter,
                                center,
                                target,
                        )
                }

                cumulative += float64(current.weight)
        }

        last := t.centroids[len(t.centroids)-1]
        lastCenter := float64(t.totalWeight) - 1.0 - (float64(last.weight)-1.0)/2.0

        return interpolate(
                last.mean,
                t.maxf,
                target-lastCenter, (float64(t.totalWeight)-1.0)-lastCenter,
        )
}

func (t *TDigest) Size() int {
        return t.totalWeight
}

func (t *TDigest) CentroidCount() int {
        t.compress()

        return len(t.centroids)
}

func (t *TDigest) Min() float64 {
        if t.totalWeight == 0 {
                return math.NaN()
        }

        return t.minf
}

func (t *TDigest) Max() float64 {
        if t.totalWeight == 0 {
                return math.NaN()
        }

        return t.maxf
}

func (t *TDigest) Compression() float64 {
        return t.compression
}

func (t *TDigest) compress() {
        if len(t.buffer) == 0 {
                return
        }

        all := make([]*Centroid, 0, len(t.centroids)+len(t.buffer))

        for _, centroid := range t.centroids {
                all = append(all, NewCentroid(centroid.mean, centroid.weight))
        }

        for _, centroid := range t.buffer {
                all = append(all, NewCentroid(centroid.mean, centroid.weight))
        }

        sort.Slice(all, func(i, j int) bool {
                return all[i].getMean() < all[j].getMean()
        })

        t.buffer = t.buffer[:0]

        if len(all) == 0 {
                t.centroids = []*Centroid{}
                return
        }

        merged := []*Centroid{}
        current := NewCentroid(all[0].mean, all[0].weight)
        cumulativeBefore := 0

        for i := 1; i < len(all); i++ {
                next := all[i]
                proposedWeight := safeAdd(
                        current.weight,
                        next.weight,
                )

                q := (float64(cumulativeBefore) + float64(proposedWeight)/2.0) / float64(t.totalWeight)
                maxWeight := t.maxCentroidWeight(q)

                if float64(proposedWeight) <= math.Max(1.0, maxWeight) {
                        current.add(next.mean, next.weight)
                } else {
                        merged = append(merged, current)

                        cumulativeBefore = safeAdd(
                                cumulativeBefore,
                                current.weight,
                        )

                        current = NewCentroid(next.mean, next.weight)
                }
        }

        merged = append(merged, current)
        t.centroids = merged
}

func (t *TDigest) maxCentroidWeight(q float64) float64 {
        q = math.Max(0.0, math.Min(1.0, q))
        return 4.0 * float64(t.totalWeight) * q * (1.0 - q) / t.compression
}

func interpolateBetweenCenters(leftValue, rightValue, leftPosition, rightPosition, target float64) float64 {
        distance := rightPosition - leftPosition

        if distance <= 0.0 {
                return (leftValue + rightValue) / 2.0
        }

        ratio := (target - leftPosition) / distance
        ratio = math.Max(0.0, math.Min(1.0, ratio))

        return leftValue + ratio*(rightValue-leftValue)
}

func interpolate(start, end, position, distance float64) float64 {
        if distance <= 0.0 {
                return (start + end) / 2.0
        }

        ratio := position / distance
        ratio = math.Max(0.0, math.Min(1.0, ratio))

        return start + ratio*(end-start)
}

func safeAdd(left, right int) int {
        if math.MaxInt-left < right {
                panic("weight overflow")
        }

        return left + right
}

追記

AIでベンチを取った

t-digest

概要

t-digest は、ストリーム上のデータから分位数(quantile / percentile)を近似的に求めるためのデータ構造。

• すべての値を保持してソートすれば正確な分位数が求まるが、メモリが要素数 n に比例して増え続ける。
• 固定幅ヒストグラムはメモリを一定に抑えられるが、ビン幅を分布の裾(tail)に合わせて細かくすると中央部のビン数が爆発し、逆に中央部に合わせるとp99/p999のような裾の分位数の精度が粗くなる。
• t-digest はこのトレードオフを、値の集まり(centroid)を分位数上の位置によって不均一なサイズに保つことで解決する。中央付近(q≈0.5)では粗く(centroidが多くの点を吸収してよい)、両端(q≈0/1)では細かく(centroidはほぼ1点のまま)保つ。中央値の精度をやや犠牲にしてでも、SLA監視やレイテンシ計測で重要になるp99・p999のような裾の分位数を高精度に保つ、という設計思想。

このリポジトリの実装(t_digest.go)は AddByValue(追加)・Merge(複数digestの統合)・Quantile(分位数の問い合わせ)の3操作を持つオンライン(逐次追加可能)な近似構造。

仕組み

データ構造

type Centroid struct {
    mean   float64 // このcentroidに属する点の重み付き平均値
    weight int      // このcentroidに属する点の総重み(=件数、AddByValueAndWeightで重み指定可)
}

type TDigest struct {
    compression float64     // 精度パラメータ(大きいほど高精度・centroid数増加)
    centroids   []*Centroid // 平均値でソート済みの、圧縮済みcentroid列
    buffer      []*Centroid // まだ圧縮(マージ)されていない、追加直後の1点1centroid
    bufferLimit int         // bufferがこの件数に達したら自動的にcompress()を実行
    totalWeight int         // 追加された全要素の総重み
    minf, maxf  float64     // 追加された値の最小値・最大値(centroidの平均とは別に厳密値を保持)
}

• Add のたびに毎回全体を再圧縮すると O(n log n) を毎回払うことになるため、まず buffer に「1点=1centroid」として溜めておき、bufferLimit(= max(20, ceil(compression * 8)))件溜まった時点でまとめて compress() する。これにより1回あたりの追加コストが償却 O(1) に近づく。
• minf/maxf は centroid の平均値とは別に厳密な最小値・最大値として保持される。Quantile(0)/Quantile(1) はこれをそのまま返し、境界のcentroidの補間もこの値を使うことで、centroidの平均に丸められることによる境界での誤差を避けている。

圧縮(compress)とスケール関数

compress() は centroids と buffer を合わせてソートし、隣接するcentroidを先頭から順に走査しながら、「マージ後の重みがそのcentroidの位置で許される上限を超えない限り」貪欲にマージしていく。

func (t *TDigest) maxCentroidWeight(q float64) float64 {
    q = math.Max(0.0, math.Min(1.0, q))
    return 4.0 * float64(t.totalWeight) * q * (1.0 - q) / t.compression
}

q*(1-q) は q=0.5 で最大値 0.25 を取り、q=0 または q=1 で 0 になる放物線。したがって maxCentroidWeight(q) は次のような形になる。

weight上限
   ^
   |          ____
   |        _/    \_
   |      _/        \_
   |    _/            \_
   |  _/                \_
   +----------------------------> q (0 → 1)
   0                        1
   (裾: 上限は小さい)    (裾: 上限は小さい)
        (中央: 上限は大きい)

• 裾(q が 0 や 1 に近い): 上限がほぼ 0 になるため、実質 max(1.0, maxWeight) = 1.0 が効き、centroidはほとんど1点のまま残る → 裾の分位数はほぼ実データそのもの(高精度)。
• 中央(q が 0.5 付近): 上限が最大 totalWeight / compression 程度まで許されるため、大量の点を1つのcentroidに吸収できる → centroid数を大きく削減できる(メモリ効率が良い代わりに中央の分位数はやや粗い)。

この 4nq(1-q)/δ という式は、原論文(Ted Dunning, "Computing Extremely Accurate Quantiles Using t-Digests")で言及される複数のスケール関数(k0〜k3、多くは arcsin ベース)の考え方を簡略化した二次関数版で、「両端で急激に細かく・中央で粗く」という定性的な性質は保ちつつ、asin を使わない分計算が軽い。

なお、マージ条件は centroid の値ではなく重み・位置(q)だけで決まる点に注意。同じ値を大量に追加しても、centroid が1つに収束するとは限らない(値が同じである以上、いくつに分かれていても補間結果は変わらないため実害はない)。詳細は TestCentroidWeightGrowsTowardMedian と TestQuantileRepeatedValueIsExact を参照。

追加(AddByValue / AddByValueAndWeight)

1. 値の有効性を検証(NaN/Inf は panic、weight <= 0 も panic)。
2. buffer に新しい1点centroidとして追加。
3. totalWeight(safeAdd でオーバーフロー検出。オーバーフロー時は panic)、minf/maxf を更新。
4. len(buffer) >= bufferLimit になったら compress() を呼び、buffer を centroids に統合する。

分位数の問い合わせ(Quantile)

1. q の範囲チェック(0 <= q <= 1、NaN 不可。範囲外は panic)。
2. totalWeight == 0 なら NaN。q == 0/q == 1 ならそれぞれ minf/maxf を返す。
3. compress() を呼んで buffer を確実に centroids へ反映させる。
4. centroids が1個だけならその平均値をそのまま返す。
5. それ以外は target = q * (totalWeight - 1)(全体の重み空間上での目標位置、いわゆる k1 スタイルのインデックス指定)を求め、各centroidの「中心位置」(center = 累積重み + (weight-1)/2)を先頭から走査。target が現在のcentroidの中心以下になった時点で、前後のcentroid間(または先頭は minf、末尾は maxf との間)で線形補間する。

これは centroid の平均値を「その centroid が代表する重み区間の中心」に固定した近似であり、centroid内部の実際の分布は分からないため線形補間で埋める、という設計。

マージ(Merge)

1. other を compress() して確定済みcentroid列にする。
2. その各centroidを新しい1点centroidとして自分の buffer に追加(centroid単位でマージするため、other 側で既に失われた解像度以上には戻らない)。
3. totalWeight・minf・maxf を合算・更新し、compress() を呼んで統合する。

複数ノードで集計した t-digest を1つにまとめる、分散環境での近似統計の集約に使える操作。

他の分位数近似手法との比較

	全件ソート	固定幅ヒストグラム	リザーバサンプリング	t-digest
メモリ	O(n)	O(バケット数)(固定)	O(サンプルサイズ)(固定)	O(compression) 程度(固定、centroid数は増加が抑制される)
精度の分布	常に厳密	ビン幅に依存(裾は粗くなりがち)	全体に一様な誤差(裾も中央も同程度)	裾ほど高精度、中央はやや粗い
オンライン追加	不可(再ソートが必要)	可能	可能	可能(償却 O(1))
複数ノードのマージ	結合してソートし直し	対応するビン同士を加算するだけ	別途アルゴリズムが必要(合成が非自明)	Merge で単純にcentroidを統合
典型用途	小規模データのバッチ処理	分布の形が事前にある程度わかっている場合	一様な誤差で十分な統計サマリ	p99/p999 のようなSLA・レイテンシ監視、裾の精度が重要な用途

t-digest はこのリポジトリにある count_min_sketch(頻度推定)や bloom_filter/cuckoo_filter(存在判定)と同じ「固定メモリで近似的な統計量を答える」データ構造の一種だが、答える質問が「分布上の位置(分位数)」である点が異なる。

このリポジトリの実装 (t_digest.go)

• NewTDigest(): compression = 100.0 で初期化。
• NewTDigestByCompression(compression float64): 圧縮率を指定して初期化(Inf/NaN/<= 0 は panic)。
• AddByValue(value float64) / AddByValueAndWeight(value float64, weight int): 値の追加。
• Merge(other *TDigest): 別の t-digest を統合(nil は panic)。
• Quantile(q float64) float64: 0 <= q <= 1 の分位数を返す。
• Size() int: 追加された総重み(totalWeight)。
• CentroidCount() int: 現在のcentroid数(呼び出し時に compress() が走る)。
• Min() / Max(): 追加された値の最小値・最大値(空なら NaN)。
• Compression() float64: 設定された圧縮率。

内部関数:

• compress(): buffer と centroids を統合・再圧縮。
• maxCentroidWeight(q): 上記のスケール関数。
• interpolate / interpolateBetweenCenters: 2点間の線形補間(距離0のときは平均値にフォールバック、範囲外は [0, 1] にクランプ)。
• safeAdd(left, right int) int: int オーバーフロー時に panic する加算(他リポジトリの count_min_sketch.safeAdd は飽和させて返す実装だが、こちらは panic する点が異なる)。

テスト (t_digest_test.go)

テスト	目的
TestNewCentroid	Centroid のコンストラクタ・getterが正しいこと。
TestCentroidAdd / TestCentroidAddWeighted	Centroid.add の加重平均計算が正しいこと。
TestNewTDigestDefaults	NewTDigest() のデフォルト値(compression=100、bufferLimit=800)。
TestNewTDigestByCompressionBufferLimit	bufferLimit = max(20, ceil(compression*8)) の計算が式通りであること。
TestNewTDigestByCompressionPanicsOnInvalidArgs	compression が 0/負/Inf/NaN のとき panic すること。
TestAddByValuePanicsOnNonFiniteValue	value が NaN/Inf のとき panic すること。
TestAddByValueAndWeightPanicsOnNonPositiveWeight	weight <= 0 のとき panic すること。
TestAddByValueUpdatesMinMaxAndSize	追加のたびに Min/Max/Size が正しく更新されること。
TestMinMaxNaNWhenEmpty	空の digest で Min/Max が NaN を返すこと。
TestAddByValueAndWeightAccumulatesWeight	同一値への重み付き追加が Size に正しく積算されること。
TestSafeAddOverflowPanicsThroughAdd	totalWeight が math.MaxInt 近辺で AddByValue が panic すること。
TestBufferFlushesOnceLimitReached	buffer が bufferLimit に達するまで centroids が空のままで、達した瞬間に compress() が走って buffer が空になること。
TestQuantileEmptyReturnsNaN	空の digest で Quantile が NaN を返すこと。
TestQuantilePanicsOnInvalidQ	q が範囲外・NaN のとき panic すること。
TestQuantileZeroAndOneReturnMinMax	Quantile(0)/Quantile(1) が Min()/Max() と一致すること。
TestQuantileSingleValue	値を1つだけ追加した場合、どの q でもその値が返ること。
TestQuantileRepeatedValueIsExact	同一値を大量に追加した場合、centroidが複数に分かれても(下記「核心的な性質」参照)Quantile は常に厳密にその値を返すこと。
TestQuantileMonotonic	q を 0→1 に増やしたとき Quantile(q) が単調非減少であること。
TestQuantileAccuracyUniformDistribution	一様分布データに対し、Quantile の推定値が真の分位数(線形補間によるexact percentile)から許容誤差内に収まること。
TestCentroidWeightGrowsTowardMedian	核心的な性質: 十分なデータ量のもとで、中央付近のcentroidの重みが両端のcentroidの重みより大きいこと(スケール関数が意図通り機能していることの検証)。
TestCentroidCountBoundedByCompression	大量データを追加してもcentroid数がデータ件数よりはるかに少なく抑えられること。
TestCompressionGetter	Compression() が設定値を返すこと。
TestMergePanicsOnNil	Merge(nil) が panic すること。
TestMergeEmptyOtherIsNoop	空の digest をマージしても状態が変化しないこと。
TestMergeCombinesWeightAndRange	2つの digest をマージした結果の Size/Min/Max/中央値が正しいこと。
TestMaxCentroidWeight / TestMaxCentroidWeightClampsOutOfRangeQ	スケール関数の値と、q が [0, 1] 外のときのクランプ挙動。
TestInterpolate / TestInterpolateBetweenCenters	線形補間関数の境界条件(距離0でのフォールバック、範囲外のクランプ)。
TestSafeAdd / TestSafeAddPanicsOnOverflow	safeAdd の通常加算とオーバーフロー時の panic。

「核心的な性質」として2つを重点的に検証している。1つは TestCentroidWeightGrowsTowardMedian で、スケール関数が「裾は細かく・中央は粗く」という設計通りに動作していること。もう1つは TestQuantileAccuracyUniformDistribution で、圧縮によって情報を落としているにもかかわらず全体としての分位数推定精度が実用的な範囲に収まっていること。

実行結果

$ go test -v ./...

全34件 PASS。

ベンチマーク (t_digest_bench_test.go)

ベンチマーク	内容
BenchmarkAddByValue	compression(25/100/500)ごとの AddByValue 1回あたりのコスト(compress() の償却コストを含む)。
BenchmarkAddByValueAndWeight	重み付き追加のコスト。
BenchmarkCompress	buffer が bufferLimit まで溜まった状態からの compress() 単体のコスト(compression ごと)。
BenchmarkQuantile	buffer が空(圧縮済み)の状態での Quantile の問い合わせコスト。
BenchmarkMerge	圧縮済みの t-digest 同士(各50,000件由来、compression=100)のマージコスト。
BenchmarkCentroidCount	buffer が空の状態での CentroidCount 呼び出しコスト(compress() が即座に返るケース)。

実測結果 (AMD Ryzen 5 5500U, 12論理コア, go test -bench=. -benchmem)

ベンチマーク	1回あたりの時間	メモリ/回	alloc回数/回
AddByValue (compression=25)	256.3 ns	104 B	4
AddByValue (compression=100)	265.7 ns	100 B	3
AddByValue (compression=500)	281.3 ns	99 B	3
AddByValueAndWeight	266.0 ns	100 B	3
Compress (compression=25)	20,132 ns	7,032 B	271
Compress (compression=100)	92,793 ns	27,896 B	1,065
Compress (compression=500)	613,816 ns	147,097 B	5,294
Quantile	309.3 ns	0 B	0
Merge (50,000件 × 2, compression=100)	129,527 ns (≈130 µs)	63,152 B	2,208
CentroidCount(圧縮済み)	2.181 ns	0 B	0

読み方のポイント

• AddByValue は compression を上げても1回あたりのコストがほぼ横ばい(256〜281ns)。これは bufferLimit(compression*8)も同時に大きくなり、compress() の呼び出し頻度が下がるため、1回あたりの compress() コストの増加分がほぼ相殺されるから。compress() 自体(下記)は compression にほぼ比例して重くなるが、それを bufferLimit 件の Add で割った償却コストはあまり変わらない、という設計になっている。
• Compress は compression を 25→100→500 と20倍にすると、コストは約30倍(20µs→614µs)、メモリも約21倍(7KB→147KB)に増える。これは compress() の対象件数がほぼ bufferLimit(compressionにほぼ比例)であり、内部のソート(O(m log m))とcentroidの再構築コストがそこに乗るため。高圧縮率(高精度)を求めるほど、1回あたりの compress() は重くなる点がトレードオフとして表れている。
• Quantile はゼロアロケーションで約309ns。内部で毎回 compress() を呼んでいるが、buffer が空であれば即座に返る(CentroidCount の2.18nsがその実測値)ため、実際のコストは centroids を先頭から線形走査して補間する部分が支配的。compression=100 程度ではcentroid数が数百のオーダーに収まるため、この走査コストは小さい。
• Merge は50,000件由来の圧縮済みcentroid列同士(centroid数自体はcompression=100により数百程度に抑えられている)を統合するコストで、約130µs。マージされるのはcentroid数(≒compression)であって元の件数(50,000)ではないため、Compress(compression=100) の約93µsに近い(centroidをbufferへコピーしてから再度compress()する分、やや重い)。件数がどれだけ増えても、マージコストは圧縮後のcentroid数にしか依存しないというt-digestの性質がここに表れている。
• CentroidCount は「buffer が空なら compress() は即座に返る」経路を測っており、2.18nsとほぼ関数呼び出しのオーバーヘッドのみ。compress() を明示的に呼び出すコストは、buffer が実際に溜まっているかどうかで数桁変わることがこの数字からもわかる。