TsDB2/DB/core/index/btree/validate.go

package btree

import (
	"TsDB2/DB/core/page"
	"fmt"
)

// ValidateBTree validates all B+ tree invariants
func ValidateBTree(rootPID uint32, handle page.HandleDb, isPrimary bool) error {
	if rootPID == 0 {
		return fmt.Errorf("empty tree")
	}

	validator := &btreeValidator{
		handle:    handle,
		isPrimary: isPrimary,
		errors:    []string{},
	}

	// 1. Validate root node
	root := validator.loadNode(rootPID)
	if root == nil {
		return fmt.Errorf("failed to load root page %d", rootPID)
	}

	// 2. Traverse and validate all nodes
	var leafKeys [][]byte
	validator.validateNode(root, nil, nil, &leafKeys)

	// 3. Validate leaf linked list
	validator.validateLeafList(leafKeys)

	if len(validator.errors) > 0 {
		return fmt.Errorf("B+ tree validation failed:\n%s",
			formatErrors(validator.errors))
	}
	return nil
}

type btreeValidator struct {
	handle    page.HandleDb
	isPrimary bool
	errors    []string
}

func (v *btreeValidator) loadNode(pid uint32) *btNode {
	if v.isPrimary {
		return btLoad(v.handle, pid)
	}
	// For secondary/string, we need a generic loader
	// For now, assume primary format
	return btLoad(v.handle, pid)
}

func (v *btreeValidator) validateNode(n *btNode, minKey, maxKey []byte, leafKeys *[][]byte) {
	if n == nil {
		v.errors = append(v.errors, "nil node encountered")
		return
	}

	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)

	if cnt == 0 {
		v.errors = append(v.errors,
			fmt.Sprintf("page %d has 0 entries", n.pageID()))
		return
	}

	if n.isLeaf() {
		v.validateLeaf(n, minKey, maxKey, el, cnt, leafKeys)
	} else {
		v.validateInternal(n, minKey, maxKey, el, cnt, leafKeys)
	}
}

func (v *btreeValidator) validateLeaf(n *btNode, minKey, maxKey []byte, el, cnt int, leafKeys *[][]byte) {
	pid := n.pageID()

	for i := 0; i < cnt; i++ {
		entry := btEntry(n, i)
		key := btKey(entry)

		// Check ordering within leaf
		// 注意: 二级索引允许重复 key, 相邻条目键相等是合法的
		if i > 0 {
			prevKey := btKey(btEntry(n, i-1))
			if btCmp(prevKey, key) > 0 {
				v.errors = append(v.errors,
					fmt.Sprintf("leaf page %d: entries not sorted at pos %d (el=%d cnt=%d useSize=%d hdrType=%d prev=%x cur=%x prevFull=%x curFull=%x)",
						pid, i, el, cnt, n.info.Hdr.UseSize, n.info.Hdr.Type, prevKey, key,
						btEntry(n, i-1), btEntry(n, i)))
			}
		}

		// Check range constraints
		if minKey != nil && btCmp(key, minKey) < 0 {
			v.errors = append(v.errors,
				fmt.Sprintf("leaf page %d: key %s < minKey %s at pos %d",
					pid, string(key), string(minKey), i))
		}
		// 上界取"下一个兄弟子树的最小键", 该键在重复 key 跨叶时可能与本叶的尾键相等,
		// 所以这里必须是 > maxKey 才算越界。
		if maxKey != nil && btCmp(key, maxKey) > 0 {
			v.errors = append(v.errors,
				fmt.Sprintf("leaf page %d: key %s > maxKey %s at pos %d",
					pid, string(key), string(maxKey), i))
		}

		*leafKeys = append(*leafKeys, key)
	}
}

func (v *btreeValidator) validateInternal(n *btNode, minKey, maxKey []byte, el, cnt int, leafKeys *[][]byte) {
	pid := n.pageID()

	for i := 0; i < cnt; i++ {
		entry := btEntry(n, i)
		sepKey := btKey(entry)
		childPID := btPID(entry)

		// Load child node
		child := v.loadNode(childPID)
		if child == nil {
			v.errors = append(v.errors,
				fmt.Sprintf("internal page %d: failed to load child %d at pos %d",
					pid, childPID, i))
			continue
		}

		// Determine child's key range
		var childMin, childMax []byte
		if i == 0 {
			childMin = minKey
		} else {
			// Child's min should be >= previous separator
			prevSep := btKey(btEntry(n, i-1))
			childMin = prevSep
		}

		if i == cnt-1 {
			childMax = maxKey
		} else {
			// 子节点 i 的上界 = 下一个子节点的分隔键(即下一个子树的最小键)。
			// 这里必须取 i+1 的分隔键: sepKey 是子节点 i 自己的最小键,
			// 用它当上界会把本子树所有条目都误判为越界。
			childMax = btKey(btEntry(n, i+1))
		}

		// Validate separator ordering
		// 重复 key 跨叶时相邻子树可能拥有相同的最小键, 故允许相等
		if i > 0 {
			prevSep := btKey(btEntry(n, i-1))
			if btCmp(prevSep, sepKey) > 0 {
				v.errors = append(v.errors,
					fmt.Sprintf("internal page %d: separators not sorted at pos %d (%s > %s)",
						pid, i, string(prevSep), string(sepKey)))
			}
		}

		// Recursively validate child
		v.validateNode(child, childMin, childMax, leafKeys)
	}
}

func (v *btreeValidator) validateLeafList(allKeys [][]byte) {
	// Check that all keys are globally sorted (允许相等: 二级索引有重复 key)
	for i := 1; i < len(allKeys); i++ {
		if btCmp(allKeys[i-1], allKeys[i]) > 0 {
			v.errors = append(v.errors,
				fmt.Sprintf("global key order violation: %s > %s",
					string(allKeys[i-1]), string(allKeys[i])))
		}
	}
}

func formatErrors(errors []string) string {
	result := ""
	for i, err := range errors {
		if i > 0 {
			result += "\n"
		}
		result += fmt.Sprintf("  [%d] %s", i+1, err)
	}
	return result
}

// DumpBTree prints the B+ tree structure for debugging
func DumpBTree(rootPID uint32, handle page.HandleDb, isPrimary bool, maxDepth int) {
	if rootPID == 0 {
		fmt.Println("Empty tree")
		return
	}

	dumper := &btreeDumper{
		handle:    handle,
		isPrimary: isPrimary,
		maxDepth:  maxDepth,
	}

	fmt.Println("=== B+ Tree Structure ===")
	dumper.dumpNode(rootPID, 0)
	fmt.Println("=========================")
}

type btreeDumper struct {
	handle    page.HandleDb
	isPrimary bool
	maxDepth  int
}

func (d *btreeDumper) dumpNode(pid uint32, depth int) {
	if depth > d.maxDepth {
		fmt.Printf("%s... (max depth reached)\n", indent(depth))
		return
	}

	n := btLoad(d.handle, pid)
	if n == nil {
		fmt.Printf("%sPage %d: FAILED TO LOAD\n", indent(depth), pid)
		return
	}

	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)

	nodeType := "LEAF"
	if !n.isLeaf() {
		nodeType = "INTERNAL"
	}

	fmt.Printf("%sPage %d [%s] count=%d entryLen=%d\n",
		indent(depth), pid, nodeType, cnt, el)

	if n.isLeaf() {
		for i := 0; i < cnt && i < 5; i++ { // Show first 5 entries
			entry := btEntry(n, i)
			key := btKey(entry)
			dataPID := btPID(entry)
			fmt.Printf("%s  [%d] key='%s' dataPage=%d\n",
				indent(depth+1), i, string(key), dataPID)
		}
		if cnt > 5 {
			fmt.Printf("%s  ... and %d more entries\n", indent(depth+1), cnt-5)
		}
	} else {
		for i := 0; i < cnt && i < 5; i++ { // Show first 5 children
			entry := btEntry(n, i)
			sepKey := btKey(entry)
			childPID := btPID(entry)
			fmt.Printf("%s  [%d] sep='%s' -> child=%d\n",
				indent(depth+1), i, string(sepKey), childPID)
			d.dumpNode(childPID, depth+1)
		}
		if cnt > 5 {
			fmt.Printf("%s  ... and %d more children\n", indent(depth+1), cnt-5)
		}
	}
}

func indent(depth int) string {
	s := ""
	for i := 0; i < depth; i++ {
		s += "  "
	}
	return s
}