SSTables and LSM Trees

package storage

import (
    "math"
    "math/rand"
    "sync"
)

const (
    MaxLevel = 16
    pFactor  = 0.5
)

type SkipNode struct {
    key   []byte
    value []byte
    next  []*SkipNode
}

type SkipList struct {
    head     *SkipNode
    level    int
    mu       sync.RWMutex
    size     int64
}

// NewSkipList creates empty skip list
func NewSkipList() *SkipList {
    head := &SkipNode{
        key:  nil,
        next: make([]*SkipNode, MaxLevel),
    }
    return &SkipList{
        head:  head,
        level: 1,
        size:  0,
    }
}

// randomLevel generates random level for new node
func randomLevel() int {
    level := 1
    for level < MaxLevel && rand.Float64() < pFactor {
        level++
    }
    return level
}

// Put inserts or updates a key
func (sl *SkipList) Put(key, value []byte) {
    sl.mu.Lock()
    defer sl.mu.Unlock()
    
    level := randomLevel()
    update := make([]*SkipNode, MaxLevel)
    current := sl.head
    
    // Find update nodes at each level
    for i := sl.level - 1; i >= 0; i-- {
        for current.next[i] != nil &&
            bytesLess(current.next[i].key, key) {
            current = current.next[i]
        }
        update[i] = current
    }
    
    // Check if key exists
    if current.next[0] != nil && bytesEqual(current.next[0].key, key) {
        current.next[0].value = value
        return
    }
    
    // Create new node
    newNode := &SkipNode{
        key:  key,
        value: value,
        next: make([]*SkipNode, level),
    }
    
    // Insert at each level
    for i := 0; i < level; i++ {
        newNode.next[i] = update[i].next[i]
        update[i].next[i] = newNode
    }
    
    // Update level if needed
    if level > sl.level {
        for i := sl.level; i < level; i++ {
            sl.head.next[i] = newNode
        }
        sl.level = level
    }
    
    sl.size++
}

// Memtable with auto-flush
type Memtable struct {
    active      *SkipList
    immutable   *SkipList
    mu          sync.RWMutex
    
    maxSize     int64
    currentSize int64
    
    flushCh     chan *SkipList
}

// Put with auto-flush
func (m *Memtable) Put(key, value []byte) error {
    m.mu.Lock()
    
    // Check if should flush
    newSize := m.currentSize + int64(len(key)+len(value))
    if newSize > m.maxSize {
        // Swap memtables
        m.immutable = m.active
        m.active = NewSkipList()
        m.currentSize = 0
        
        // Signal flush
        select {
        case m.flushCh <- m.immutable:
        default:
            // Flush queue full, apply backpressure
            m.mu.Unlock()
            m.flushCh <- m.immutable
            m.mu.Lock()
        }
    }
    
    m.active.Put(key, value)
    m.currentSize = newSize
    m.mu.Unlock()
    
    return nil
}

// Get checks both memtables
func (m *Memtable) Get(key []byte) ([]byte, bool) {
    m.mu.RLock()
    defer m.mu.RUnlock()
    
    // Check active
    if v, ok := m.active.Get(key); ok {
        return v, ok
    }
    
    // Check immutable
    if v, ok := m.immutable.Get(key); ok {
        return v, ok
    }
    
    return nil, false
}

// StartFlusher starts background flushing
func (db *DB) StartFlusher() {
    go func() {
        for sl := range db.memtable.flushCh {
            if err := db.flushMemtable(sl); err != nil {
                log.Printf("flush error: %v", err)
            }
        }
    }()
}

// flushMemtable converts memtable to SSTable
func (db *DB) flushMemtable(mt *SkipList) error {
    // Create SSTable writer
    sstPath := filepath.Join(db.dataDir, 
        fmt.Sprintf("sstable-%d.sst", time.Now().UnixNano()))
    
    writer, err := NewSSTWriter(sstPath)
    if err != nil {
        return err
    }
    defer writer.Close()
    
    // Iterate and write all entries
    iter := mt.NewIterator()
    for iter.Next() {
        writer.Add(iter.Key(), iter.Value())
    }
    
    return writer.Finish()
}