解析 Go 实现的 LSM-Tree 存储引擎：如何通过计算 Amplify = rac{Total Bytes Written}{User Bytes Written}$ 优化写入放大？

各位同仁，下午好！

今天，我们将深入探讨Go语言实现的LSM-Tree存储引擎，并着重分析一个核心问题：如何通过计算和优化“写入放大”（Write Amplification, WA），即 Amplify = Total Bytes Written / User Bytes Written，来提升存储系统的性能和寿命。作为一名编程专家，我将从理论基础出发，结合Go语言的实现细节，为大家剖析LSM-Tree的内部机制，并提出一系列行之有效的优化策略。

1. LSM-Tree：高性能存储的基石

Log-Structured Merge-Tree (LSM-Tree) 是一种为写密集型工作负载优化的数据结构，它通过将随机写转换为顺序写来提高吞吐量，并有效利用现代存储介质（特别是SSD）的特性。LSM-Tree的核心思想是将数据写入内存中的一个可变结构（MemTable），当MemTable达到一定大小后，将其冻结并刷写（flush）到磁盘上的不可变文件（SSTable）。同时，磁盘上的多个SSTable会周期性地进行合并（compaction），以消除冗余、回收空间并优化读性能。

1.1 Go语言与LSM-Tree

Go语言以其简洁的语法、优秀的并发原语（Goroutines和Channels）以及强大的标准库，成为实现高性能存储引擎的理想选择。在LSM-Tree的实现中，Go语言能够轻松处理并发写入、异步刷写、后台压缩等任务，同时保持较低的GC开销。

// kv.go: 定义存储的键值对结构
package storage

import "time"

// EntryType 定义条目类型：普通键值对或删除标记
type EntryType byte

const (
    EntryPut EntryType = iota
    EntryDelete
)

// Entry 代表一个键值对条目
type Entry struct {
    Key       []byte
    Value     []byte
    Timestamp int64 // 用于MVCC或最新版本判断
    Type      EntryType
}

// NewEntry 创建一个新的键值对条目
func NewEntry(key, value []byte, entryType EntryType) Entry {
    return Entry{
        Key:       key,
        Value:     value,
        Timestamp: time.Now().UnixNano(),
        Type:      entryType,
    }
}

2. 什么是写入放大 (Write Amplification, WA)？

写入放大是衡量存储系统效率的一个关键指标。它定义为：

Amplify = Total Bytes Written / User Bytes Written

其中：

• User Bytes Written：用户实际写入的数据量，通常指len(key) + len(value)的总和。
• Total Bytes Written：存储系统为持久化这些用户数据而在物理存储介质上实际写入的总字节数。这包括写入WAL、MemTable刷写到SSTable、以及所有后续SSTable合并过程中产生的额外写入。

一个高的写入放大意味着存储介质需要进行更多的物理写入操作来持久化相同量的用户数据。这会带来几个负面影响：

1. 性能下降：额外的写入会消耗存储介质的I/O带宽，降低吞吐量。
2. 寿命缩短：闪存存储（如SSD）具有有限的擦写周期。高的写入放大加速了这些周期的消耗，从而缩短了SSD的寿命。
3. 能耗增加：更多的物理写入意味着更多的能量消耗。

因此，优化写入放大是LSM-Tree存储引擎设计中的一项核心挑战。

3. LSM-Tree的核心组件与写入放大来源

为了理解写入放大，我们首先要回顾LSM-Tree的几个关键组件及其数据流。

3.1 MemTable (内存表)

LSM-Tree的写操作首先进入MemTable。MemTable是一个内存中的数据结构，通常使用跳表（SkipList）或B-Tree实现，以支持高效的插入和有序遍历。所有新的写入和更新都在这里进行。

// memtable.go: 简化版MemTable实现，使用一个内部的跳表
package storage

import (
    "bytes"
    "sync"

    "github.com/huandu/skiplist" // 假设使用一个开源的Go跳表库
)

// MemTable 是一个内存中的键值存储，通常使用跳表实现
type MemTable struct {
    mu       sync.RWMutex
    data     *skiplist.SkipList // 存储Entry，按Key排序
    size     int64              // 当前MemTable占用的近似字节数
    maxSize  int64              // MemTable的最大容量
}

// NewMemTable 创建一个新的MemTable
func NewMemTable(maxSize int64) *MemTable {
    return &MemTable{
        data:    skiplist.New(skiplist.Bytes), // 使用字节比较器
        maxSize: maxSize,
    }
}

// Add 将一个Entry添加到MemTable
func (mt *MemTable) Add(entry Entry) {
    mt.mu.Lock()
    defer mt.mu.Unlock()

    // 计算当前Entry的近似大小
    entrySize := int64(len(entry.Key) + len(entry.Value) + 8 + 1) // Key+Value+Timestamp+Type
    if oldEntry := mt.data.Get(entry.Key); oldEntry != nil {
        // 如果是更新操作，需要扣除旧Entry的大小
        oldVal := oldEntry.Value.(Entry)
        mt.size -= int64(len(oldVal.Key) + len(oldVal.Value) + 8 + 1)
    }

    mt.data.Set(entry.Key, entry)
    mt.size += entrySize
}

// Get 从MemTable获取一个Entry
func (mt *MemTable) Get(key []byte) (Entry, bool) {
    mt.mu.RLock()
    defer mt.mu.RUnlock()

    if elem := mt.data.Get(key); elem != nil {
        return elem.Value.(Entry), true
    }
    return Entry{}, false
}

// Size 返回MemTable的当前大小
func (mt *MemTable) Size() int64 {
    mt.mu.RLock()
    defer mt.mu.RUnlock()
    return mt.size
}

// ShouldFlush 判断MemTable是否已满，需要刷写
func (mt *MemTable) ShouldFlush() bool {
    mt.mu.RLock()
    defer mt.mu.RUnlock()
    return mt.size >= mt.maxSize
}

// Iterator 返回一个MemTable的迭代器
func (mt *MemTable) Iterator() *skiplist.Iterator {
    mt.mu.RLock() // 迭代过程中可能需要读锁
    return mt.data.Iterator()
}

3.2 WAL (Write-Ahead Log)

为了保证数据持久性，在数据写入MemTable之前，通常会先写入一个Write-Ahead Log (WAL)。WAL是一个顺序写入的文件，记录了所有的数据修改操作。即使系统崩溃，也可以通过重放WAL来恢复MemTable中的数据。

WAL是写入放大的第一个来源：每次用户写入都会导致一次WAL的物理写入。如果WAL配置为每次写入都强制刷盘（fsync），则开销更大。

// wal.go: 简单的WAL实现
package storage

import (
    "bufio"
    "encoding/binary"
    "io"
    "os"
    "path/filepath"
    "sync"
)

// WALEntryType 定义WAL中记录的条目类型
type WALEntryType byte

const (
    WALEntryPut WALEntryType = iota
    WALEntryDelete
)

// WALEntry 是WAL中存储的单个操作记录
type WALEntry struct {
    Type      WALEntryType
    Key       []byte
    Value     []byte // 对于Delete操作，Value为空
    Timestamp int64
}

// WAL 是Write-Ahead Log的结构
type WAL struct {
    file   *os.File
    writer *bufio.Writer
    mu     sync.Mutex // 保护写入操作
    path   string
}

// NewWAL 创建或打开一个WAL文件
func NewWAL(dir string, filename string) (*WAL, error) {
    path := filepath.Join(dir, filename)
    file, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
    if err != nil {
        return nil, err
    }
    return &WAL{
        file:   file,
        writer: bufio.NewWriter(file),
        path:   path,
    }, nil
}

// WriteEntry 将一个Entry写入WAL
func (w *WAL) WriteEntry(entry Entry) error {
    w.mu.Lock()
    defer w.mu.Unlock()

    walEntryType := WALEntryPut
    if entry.Type == EntryDelete {
        walEntryType = WALEntryDelete
    }

    // 编码格式: Type(1 byte) | KeyLen(4 bytes) | Key | ValueLen(4 bytes) | Value | Timestamp(8 bytes)
    err := w.writer.WriteByte(byte(walEntryType))
    if err != nil { return err }

    err = binary.Write(w.writer, binary.LittleEndian, uint32(len(entry.Key)))
    if err != nil { return err }
    _, err = w.writer.Write(entry.Key)
    if err != nil { return err }

    err = binary.Write(w.writer, binary.LittleEndian, uint32(len(entry.Value)))
    if err != nil { return err }
    _, err = w.writer.Write(entry.Value)
    if err != nil { return err }

    err = binary.Write(w.writer, binary.LittleEndian, entry.Timestamp)
    if err != nil { return err }

    return nil
}

// Sync 将WAL缓冲区的内容刷写到磁盘
func (w *WAL) Sync() error {
    w.mu.Lock()
    defer w.mu.Unlock()
    if err := w.writer.Flush(); err != nil {
        return err
    }
    return w.file.Sync()
}

// Close 关闭WAL文件
func (w *WAL) Close() error {
    w.mu.Lock()
    defer w.mu.Unlock()
    if err := w.writer.Flush(); err != nil {
        return err
    }
    return w.file.Close()
}

// Recover 从WAL文件中恢复数据，并重建MemTable
func (w *WAL) Recover(mt *MemTable) error {
    w.mu.Lock()
    defer w.mu.Unlock()

    // 确保所有数据都已写入文件
    if err := w.writer.Flush(); err != nil {
        return err
    }

    // 从文件开头读取
    _, err := w.file.Seek(0, io.SeekStart)
    if err != nil {
        return err
    }

    reader := bufio.NewReader(w.file)
    for {
        // 读取Type
        typeByte, err := reader.ReadByte()
        if err == io.EOF { break }
        if err != nil { return err }
        walEntryType := WALEntryType(typeByte)

        // 读取Key长度和Key
        var keyLen uint32
        err = binary.Read(reader, binary.LittleEndian, &keyLen)
        if err != nil { return err }
        key := make([]byte, keyLen)
        _, err = io.ReadFull(reader, key)
        if err != nil { return err }

        // 读取Value长度和Value
        var valueLen uint32
        err = binary.Read(reader, binary.LittleEndian, &valueLen)
        if err != nil { return err }
        value := make([]byte, valueLen)
        _, err = io.ReadFull(reader, value)
        if err != nil { return err }

        // 读取Timestamp
        var timestamp int64
        err = binary.Read(reader, binary.LittleEndian, &timestamp)
        if err != nil { return err }

        entryType := EntryPut
        if walEntryType == WALEntryDelete {
            entryType = EntryDelete
        }
        mt.Add(Entry{Key: key, Value: value, Timestamp: timestamp, Type: entryType})
    }
    return nil
}

3.3 SSTable (Sorted String Table)

当MemTable达到预设大小后，它会被标记为不可变（immutable MemTable），并异步刷写到磁盘上，形成一个SSTable文件。SSTable是不可变的，且内部数据按键有序存储。

MemTable刷写是写入放大的第二个来源：MemTable中的所有数据（包括其元数据，如Bloom Filter、索引块等）都会被顺序写入SSTable。

// sstable.go: 简化版SSTable的写入和读取
package storage

import (
    "bufio"
    "bytes"
    "encoding/binary
    "fmt"
    "hash/crc32"
    "io"
    "os"
    "path/filepath"

    "github.com/bits-and-blooms/bloom" // 假设使用一个Go的Bloom Filter库
)

const (
    BlockSize       = 4 * 1024 // 4KB 数据块
    ChecksumSize    = 4        // CRC32校验和大小
    BlockMetaSize   = 8 + 4    // Offset (uint64) + Size (uint32)
    BloomFilterBits = 10 * 1024 // Bloom Filter大小
)

// BlockMeta 存储数据块的元数据
type BlockMeta struct {
    Offset uint64 // 块在文件中的偏移
    Size   uint32 // 块的原始大小
    FirstKey []byte // 块中第一个键，用于稀疏索引
}

// SSTableWriter 用于构建和写入SSTable文件
type SSTableWriter struct {
    file         *os.File
    writer       *bufio.Writer
    blockBuf     bytes.Buffer // 用于暂存当前数据块的数据
    blockMetas   []BlockMeta  // 存储所有数据块的元数据
    bloomFilter  *bloom.BloomFilter // Bloom Filter
    currentBlockOffset uint64
    lastEntryKey []byte
}

// NewSSTableWriter 创建一个新的SSTableWriter
func NewSSTableWriter(dir string, filename string) (*SSTableWriter, error) {
    path := filepath.Join(dir, filename)
    file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
    if err != nil {
        return nil, err
    }
    return &SSTableWriter{
        file:        file,
        writer:      bufio.NewWriter(file),
        bloomFilter: bloom.NewWithEstimates(BloomFilterBits, 0.01), // 10k items, 1% false positive
    }, nil
}

// AddEntry 将一个Entry添加到SSTableWriter
func (w *SSTableWriter) AddEntry(entry Entry) error {
    // 将Entry编码到blockBuf
    // 格式: KeyLen(varint) | Key | ValueLen(varint) | Value | Timestamp(8 bytes) | Type(1 byte)
    buf := &bytes.Buffer{}
    binary.Write(buf, binary.LittleEndian, uint32(len(entry.Key)))
    buf.Write(entry.Key)
    binary.Write(buf, binary.LittleEndian, uint32(len(entry.Value)))
    buf.Write(entry.Value)
    binary.Write(buf, binary.LittleEndian, entry.Timestamp)
    buf.WriteByte(byte(entry.Type))

    // 如果添加此Entry会导致当前块过大，则先刷写当前块
    if w.blockBuf.Len()+buf.Len() > BlockSize && w.blockBuf.Len() > 0 {
        if err := w.flushBlock(); err != nil {
            return err
        }
    }

    // 将Entry数据写入当前块缓冲区
    _, err := w.blockBuf.Write(buf.Bytes())
    if err != nil {
        return err
    }

    // 更新Bloom Filter
    w.bloomFilter.Add(entry.Key)

    // 记录当前块的第一个Key
    if len(w.blockMetas) == 0 || len(w.blockMetas[len(w.blockMetas)-1].FirstKey) == 0 || bytes.Compare(entry.Key, w.blockMetas[len(w.blockMetas)-1].FirstKey) < 0 {
        // This logic is tricky. For a new block, the first key needs to be set.
        // For an existing block, first key is already set.
        // Simplified: we'll set it when flushing the block.
    }
    if w.lastEntryKey == nil {
        w.lastEntryKey = entry.Key
    }

    return nil
}

// flushBlock 将当前缓冲区的数据作为SSTable的一个数据块写入文件
func (w *SSTableWriter) flushBlock() error {
    if w.blockBuf.Len() == 0 {
        return nil
    }

    blockData := w.blockBuf.Bytes()

    // 计算校验和
    checksum := crc32.ChecksumIEEE(blockData)

    // 写入数据块
    _, err := w.writer.Write(blockData)
    if err != nil { return err }

    // 写入校验和
    err = binary.Write(w.writer, binary.LittleEndian, checksum)
    if err != nil { return err }

    // 记录块元数据
    w.blockMetas = append(w.blockMetas, BlockMeta{
        Offset:   w.currentBlockOffset,
        Size:     uint32(len(blockData)),
        FirstKey: w.lastEntryKey, // 存储当前块的第一个key
    })

    w.currentBlockOffset += uint64(len(blockData) + ChecksumSize)
    w.blockBuf.Reset()
    w.lastEntryKey = nil // Reset for next block
    return nil
}

// Finalize 完成SSTable的写入，包括写入索引和Bloom Filter
func (w *SSTableWriter) Finalize() error {
    // 刷写最后一个可能未满的块
    if err := w.flushBlock(); err != nil {
        return err
    }

    // 写入索引块
    indexBuf := &bytes.Buffer{}
    for _, meta := range w.blockMetas {
        binary.Write(indexBuf, binary.LittleEndian, meta.Offset)
        binary.Write(indexBuf, binary.LittleEndian, meta.Size)
        binary.Write(indexBuf, binary.LittleEndian, uint32(len(meta.FirstKey))) // Key len
        indexBuf.Write(meta.FirstKey) // First Key
    }
    indexData := indexBuf.Bytes()
    indexChecksum := crc32.ChecksumIEEE(indexData)

    indexOffset := w.currentBlockOffset // 索引块的起始偏移
    _, err := w.writer.Write(indexData)
    if err != nil { return err }
    err = binary.Write(w.writer, binary.LittleEndian, indexChecksum)
    if err != nil { return err }
    indexEndOffset := w.currentBlockOffset + uint64(len(indexData)+ChecksumSize)

    // 写入Bloom Filter
    bloomData, err := w.bloomFilter.MarshalBinary()
    if err != nil { return err }
    bloomChecksum := crc32.ChecksumIEEE(bloomData)

    bloomOffset := indexEndOffset // Bloom Filter的起始偏移
    _, err = w.writer.Write(bloomData)
    if err != nil { return err }
    err = binary.Write(w.writer, binary.LittleEndian, bloomChecksum)
    if err != nil { return err }
    bloomEndOffset := indexEndOffset + uint64(len(bloomData)+ChecksumSize)

    // 写入文件尾部元数据 (Footer)
    // 格式: IndexOffset(8 bytes) | BloomOffset(8 bytes) | Magic(4 bytes)
    footerBuf := &bytes.Buffer{}
    binary.Write(footerBuf, binary.LittleEndian, indexOffset)
    binary.Write(footerBuf, binary.LittleEndian, bloomOffset)
    binary.Write(footerBuf, binary.LittleEndian, uint32(0xCAFEF00D)) // Magic number
    _, err = w.writer.Write(footerBuf.Bytes())
    if err != nil { return err }

    if err := w.writer.Flush(); err != nil {
        return err
    }
    return w.file.Close()
}

// SSTableReader 用于读取SSTable文件
type SSTableReader struct {
    file         *os.File
    indexMetas   []BlockMeta
    bloomFilter  *bloom.BloomFilter
    footerOffset int64 // Footer的起始偏移
}

// NewSSTableReader 打开并读取SSTable的元数据
func NewSSTableReader(path string) (*SSTableReader, error) {
    file, err := os.Open(path)
    if err != nil {
        return nil, err
    }

    // 读取文件大小
    stat, err := file.Stat()
    if err != nil { file.Close(); return nil, err }
    fileSize := stat.Size()

    // 读取Footer
    footerSize := int64(8 + 8 + 4) // IndexOffset + BloomOffset + Magic
    if fileSize < footerSize { file.Close(); return nil, fmt.Errorf("sstable: file too small for footer") }

    _, err = file.Seek(-footerSize, io.SeekEnd)
    if err != nil { file.Close(); return nil, err }

    footerBuf := make([]byte, footerSize)
    _, err = io.ReadFull(file, footerBuf)
    if err != nil { file.Close(); return nil, err }

    reader := bytes.NewReader(footerBuf)
    var indexOffset, bloomOffset uint64
    var magic uint32
    binary.Read(reader, binary.LittleEndian, &indexOffset)
    binary.Read(reader, binary.LittleEndian, &bloomOffset)
    binary.Read(reader, binary.LittleEndian, &magic)

    if magic != 0xCAFEF00D { file.Close(); return nil, fmt.Errorf("sstable: invalid magic number") }

    // 读取Bloom Filter
    bloomLen := int64(indexOffset - bloomOffset - ChecksumSize)
    _, err = file.Seek(int64(bloomOffset), io.SeekStart)
    if err != nil { file.Close(); return nil, err }
    bloomData := make([]byte, bloomLen)
    _, err = io.ReadFull(file, bloomData)
    if err != nil { file.Close(); return nil, err }
    // 校验Bloom Filter的Checksum... (略)

    bf := bloom.New(1,1) // Placeholder
    err = bf.UnmarshalBinary(bloomData)
    if err != nil { file.Close(); return nil, err }

    // 读取索引块
    indexLen := int64(bloomOffset - indexOffset - ChecksumSize)
    _, err = file.Seek(int64(indexOffset), io.SeekStart)
    if err != nil { file.Close(); return nil, err }
    indexData := make([]byte, indexLen)
    _, err = io.ReadFull(file, indexData)
    if err != nil { file.Close(); return nil, err }
    // 校验索引块的Checksum... (略)

    indexReader := bytes.NewReader(indexData)
    var indexMetas []BlockMeta
    for indexReader.Len() > 0 {
        var offset uint64
        var size, keyLen uint32
        binary.Read(indexReader, binary.LittleEndian, &offset)
        binary.Read(indexReader, binary.LittleEndian, &size)
        binary.Read(indexReader, binary.LittleEndian, &keyLen)
        firstKey := make([]byte, keyLen)
        _, err = io.ReadFull(indexReader, firstKey)
        if err != nil { file.Close(); return nil, err }
        indexMetas = append(indexMetas, BlockMeta{Offset: offset, Size: size, FirstKey: firstKey})
    }

    return &SSTableReader{
        file:        file,
        indexMetas:  indexMetas,
        bloomFilter: bf,
        footerOffset: fileSize - footerSize,
    }, nil
}

// Get 从SSTable中获取一个Entry
func (r *SSTableReader) Get(key []byte) (Entry, bool, error) {
    if !r.bloomFilter.Test(key) {
        return Entry{}, false, nil // Bloom Filter 快速判断不存在
    }

    // 使用二分查找在索引中找到可能包含key的块
    blockIndex := r.findBlockIndex(key)
    if blockIndex == -1 {
        return Entry{}, false, nil
    }
    meta := r.indexMetas[blockIndex]

    // 读取数据块
    blockData, err := r.readBlock(meta.Offset, meta.Size)
    if err != nil {
        return Entry{}, false, err
    }

    // 在数据块内查找Entry (这里可以进一步优化为块内二分查找，简化为线性扫描)
    reader := bytes.NewReader(blockData)
    for reader.Len() > 0 {
        var keyLen uint32
        err = binary.Read(reader, binary.LittleEndian, &keyLen)
        if err != nil { return Entry{}, false, err }
        currentKey := make([]byte, keyLen)
        _, err = io.ReadFull(reader, currentKey)
        if err != nil { return Entry{}, false, err }

        var valueLen uint32
        err = binary.Read(reader, binary.LittleEndian, &valueLen)
        if err != nil { return Entry{}, false, err }
        currentValue := make([]byte, valueLen)
        _, err = io.ReadFull(reader, currentValue)
        if err != nil { return Entry{}, false, err }

        var timestamp int64
        err = binary.Read(reader, binary.LittleEndian, &timestamp)
        if err != nil { return Entry{}, false, err }

        var entryTypeByte byte
        entryTypeByte, err = reader.ReadByte()
        if err != nil { return Entry{}, false, err }
        currentType := EntryType(entryTypeByte)

        if bytes.Equal(currentKey, key) {
            return Entry{Key: currentKey, Value: currentValue, Timestamp: timestamp, Type: currentType}, true, nil
        }
    }

    return Entry{}, false, nil
}

// findBlockIndex 使用二分查找找到可能包含给定键的块索引
func (r *SSTableReader) findBlockIndex(key []byte) int {
    // 简单的线性查找示例，实际应为二分查找
    for i, meta := range r.indexMetas {
        if bytes.Compare(key, meta.FirstKey) >= 0 { // Key >= current block's first key
            if i+1 < len(r.indexMetas) {
                if bytes.Compare(key, r.indexMetas[i+1].FirstKey) < 0 { // Key < next block's first key
                    return i
                }
            } else {
                return i // Last block
            }
        }
    }
    return -1
}

// readBlock 读取指定偏移和大小的数据块
func (r *SSTableReader) readBlock(offset uint64, size uint32) ([]byte, error) {
    blockData := make([]byte, size)
    _, err := r.file.ReadAt(blockData, int64(offset))
    if err != nil {
        return nil, err
    }
    // TODO: 校验checksum
    return blockData, nil
}

// Close 关闭SSTable文件
func (r *SSTableReader) Close() error {
    return r.file.Close()
}

// TableInfo 结构用于存储SSTable的元数据，便于管理
type TableInfo struct {
    ID   int // SSTable的唯一标识
    Path string
    MinKey []byte
    MaxKey []byte
    Size   int64 // 文件大小
    Level  int   // 所属层级
    Reader *SSTableReader // 运行时持有Reader对象
}

3.4 Compaction (合并)

LSM-Tree的核心是Compaction（合并）过程。随着MemTable不断刷写，磁盘上会累积大量的SSTable文件，尤其是L0层级。为了优化读性能、回收被删除数据的空间以及减少SSTable文件的数量，后台会周期性地选择一些SSTable进行合并。合并过程通常涉及读取多个SSTable，将它们的数据进行排序、合并，然后写入一个新的SSTable文件。

Compaction是写入放大的主要来源：数据在合并过程中会被反复读取和写入。一个键值对可能因为多次更新或为了合并到更低层级而经历多次重写。

// lsm.go: 简化版LSM-Tree的Put和Flush逻辑
package storage

import (
    "fmt"
    "io/ioutil"
    "os"
    "path/filepath"
    "sync"
    "sync/atomic"
    "time"
)

// LSM 是LSM-Tree存储引擎的核心结构
type LSM struct {
    mu           sync.RWMutex
    activeMemTable   *MemTable
    immutableMemTables []*MemTable // 待刷写的MemTable队列
    wal            *WAL
    baseDir        string
    nextTableID    int32 // 用于生成SSTable文件名

    levels []*Level // SSTable的层级结构

    flushChan    chan *MemTable // 用于触发异步刷写
    compactionChan chan struct{}  // 用于触发异步合并
    stopChan     chan struct{}

    config LSMConfig
}

// LSMConfig 存储LSM-Tree的配置
type LSMConfig struct {
    MemTableMaxSize int64
    MaxLevels       int
    LevelRatio      int // 每个层级比上一层级大多少倍
    WalDir          string
    SSTableDir      string
    CompactionInterval time.Duration // 合并检查间隔
}

// NewLSM 创建并初始化LSM-Tree
func NewLSM(config LSMConfig) (*LSM, error) {
    // ... 目录创建，WAL初始化 ...
    walPath := filepath.Join(config.WalDir, "current.wal")
    wal, err := NewWAL(config.WalDir, "current.wal")
    if err != nil { return nil, err }

    lsm := &LSM{
        activeMemTable: NewMemTable(config.MemTableMaxSize),
        wal:            wal,
        baseDir:        config.SSTableDir,
        levels:         make([]*Level, config.MaxLevels),
        flushChan:      make(chan *MemTable, 10), // 缓冲区可容纳10个待刷写MemTable
        compactionChan: make(chan struct{}, 1),
        stopChan:       make(chan struct{}),
        config:         config,
    }

    for i := 0; i < config.MaxLevels; i++ {
        lsm.levels[i] = NewLevel(i, config.SSTableDir)
    }

    // 恢复WAL
    if err := lsm.wal.Recover(lsm.activeMemTable); err != nil {
        fmt.Printf("WAL recovery failed: %v, attempting to continue...n", err)
        // 严重错误，可能需要停止或进行更复杂的恢复
    }

    go lsm.flushLoop()
    go lsm.compactionLoop()
    go lsm.startCompactionScheduler()

    return lsm, nil
}

// Put 写入一个键值对
func (l *LSM) Put(key, value []byte) error {
    entry := NewEntry(key, value, EntryPut)

    l.mu.Lock()
    defer l.mu.Unlock()

    // 写入WAL
    if err := l.wal.WriteEntry(entry); err != nil {
        return err
    }
    // 写入MemTable
    l.activeMemTable.Add(entry)

    // 如果MemTable已满，则触发刷写
    if l.activeMemTable.ShouldFlush() {
        l.flushMemTable()
    }
    return nil
}

// Delete 删除一个键值对
func (l *LSM) Delete(key []byte) error {
    entry := NewEntry(key, nil, EntryDelete) // Value为nil表示删除

    l.mu.Lock()
    defer l.mu.Unlock()

    if err := l.wal.WriteEntry(entry); err != nil {
        return err
    }
    l.activeMemTable.Add(entry) // 标记为删除的Entry也会进入MemTable

    if l.activeMemTable.ShouldFlush() {
        l.flushMemTable()
    }
    return nil
}

// Get 从LSM-Tree中获取一个键值对
func (l *LSM) Get(key []byte) ([]byte, bool, error) {
    l.mu.RLock()
    defer l.mu.RUnlock()

    // 1. 优先从active MemTable查找
    if entry, ok := l.activeMemTable.Get(key); ok {
        if entry.Type == EntryDelete { return nil, false, nil } // 已被删除
        return entry.Value, true, nil
    }

    // 2. 其次从immutable MemTables查找 (从最新到最旧)
    for i := len(l.immutableMemTables) - 1; i >= 0; i-- {
        if entry, ok := l.immutableMemTables[i].Get(key); ok {
            if entry.Type == EntryDelete { return nil, false, nil }
            return entry.Value, true, nil
        }
    }

    // 3. 最后从SSTable层级查找 (从L0到最高层)
    for i := 0; i < len(l.levels); i++ {
        level := l.levels[i]
        entry, found, err := level.Get(key)
        if err != nil { return nil, false, err }
        if found {
            if entry.Type == EntryDelete { return nil, false, nil }
            return entry.Value, true, nil
        }
    }

    return nil, false, nil // 未找到
}

// flushMemTable 将当前的active MemTable转变为immutable，并创建一个新的active MemTable
func (l *LSM) flushMemTable() {
    // 1. 将当前的active MemTable变为immutable
    immutable := l.activeMemTable
    l.immutableMemTables = append(l.immutableMemTables, immutable)

    // 2. 创建一个新的active MemTable
    l.activeMemTable = NewMemTable(l.config.MemTableMaxSize)

    // 3. 切换WAL文件，旧的WAL可以被清理（在SSTable刷写完成后）
    // 实际实现中，通常是创建一个新的WAL段，或者重置当前WAL。简化处理为继续使用
    // TODO: 实现WAL滚动和清理

    // 4. 异步触发刷写操作
    l.flushChan <- immutable
}

// flushLoop 负责将immutable MemTables刷写到L0层级的SSTable
func (l *LSM) flushLoop() {
    for {
        select {
        case mt := <-l.flushChan:
            fmt.Printf("Flushing MemTable to L0, size: %d bytesn", mt.Size())

            // 为新的SSTable分配ID和文件名
            tableID := atomic.AddInt32(&l.nextTableID, 1)
            filename := fmt.Sprintf("sstable-%06d.sst", tableID)

            writer, err := NewSSTableWriter(filepath.Join(l.baseDir, l.config.SSTableDir), filename)
            if err != nil {
                fmt.Printf("Error creating SSTable writer: %vn", err)
                // 错误处理：可能需要重试或报警
                continue
            }

            // 遍历MemTable并写入SSTable
            iter := mt.Iterator()
            for iter.Next() {
                entry := iter.Value().(Entry)
                if err := writer.AddEntry(entry); err != nil {
                    fmt.Printf("Error writing entry to SSTable: %vn", err)
                    writer.Close() // 尝试关闭文件
                    // 错误处理
                    goto cleanupImmutable // 跳转到清理immutable MemTable
                }
            }

            if err := writer.Finalize(); err != nil {
                fmt.Printf("Error finalizing SSTable: %vn", err)
                // 错误处理
                goto cleanupImmutable
            }
            writer.Close() // 确保文件句柄关闭

            // 创建SSTableReader并添加到L0层级
            reader, err := NewSSTableReader(filepath.Join(l.baseDir, l.config.SSTableDir, filename))
            if err != nil {
                fmt.Printf("Error creating SSTable reader for new file: %vn", err)
                goto cleanupImmutable
            }

            // 获取MinKey和MaxKey
            var minKey, maxKey []byte
            if len(reader.indexMetas) > 0 {
                minKey = reader.indexMetas[0].FirstKey
                // 为了获取MaxKey，通常需要迭代器或者在SSTableWriter中记录最后一个Key
                // 简化处理：假设SSTableWriter记录了最后一个Entry的Key
                // TODO: 完善SSTableReader获取MaxKey的逻辑
            }

            table := &TableInfo{
                ID:   int(tableID),
                Path: filepath.Join(l.baseDir, l.config.SSTableDir, filename),
                MinKey: minKey,
                MaxKey: minKey, // Placeholder for MaxKey
                Size:   mt.Size(), // 近似大小，实际应为文件大小
                Level:  0,
                Reader: reader,
            }

            l.levels[0].AddTable(table)
            fmt.Printf("SSTable %s flushed to L0. Size: %d bytesn", filename, table.Size)

        cleanupImmutable:
            // 移除已刷写的immutable MemTable
            l.mu.Lock()
            // 找到并移除mt
            for i, imt := range l.immutableMemTables {
                if imt == mt {
                    l.immutableMemTables = append(l.immutableMemTables[:i], l.immutableMemTables[i+1:]...)
                    break
                }
            }
            l.mu.Unlock()

            // 触发一次compaction检查
            select {
            case l.compactionChan <- struct{}{}:
            default: // 防止阻塞
            }

        case <-l.stopChan:
            return
        }
    }
}

// compactionLoop 负责后台合并SSTable文件
func (l *LSM) compactionLoop() {
    for {
        select {
        case <-l.compactionChan:
            fmt.Println("Compaction triggered...")
            // 实现合并策略
            l.runCompaction()
        case <-l.stopChan:
            return
        }
    }
}

// startCompactionScheduler 定时触发compaction
func (l *LSM) startCompactionScheduler() {
    ticker := time.NewTicker(l.config.CompactionInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            select {
            case l.compactionChan <- struct{}{}: // 尝试触发合并
            default: // 防止阻塞
            }
        case <-l.stopChan:
            return
        }
    }
}

// runCompaction 具体的合并逻辑 (简化版，仅L0到L1的Leveled Compaction)
func (l *LSM) runCompaction() {
    l.mu.Lock()
    defer l.mu.Unlock()

    // 检查是否有需要合并的L0 SSTables
    if l.levels[0].Size() == 0 {
        return // L0为空，无需合并
    }

    // 简单的策略：将L0所有SSTable与L1的重叠SSTable合并
    // 实际生产级LSM-Tree有更复杂的合并选择策略，例如LevelDB/RocksDB的Size-Tiered或Leveled

    // 获取L0的所有SSTable
    l0Tables := l.levels[0].GetAllTables()
    if len(l0Tables) == 0 { return }

    // 找出L0所有SSTable的Key范围
    minKey, maxKey := l0Tables[0].MinKey, l0Tables[0].MaxKey
    for _, t := range l0Tables {
        if bytes.Compare(t.MinKey, minKey) < 0 { minKey = t.MinKey }
        if bytes.Compare(t.MaxKey, maxKey) > 0 { maxKey = t.MaxKey }
    }

    // 获取L1中与L0范围重叠的SSTable
    l1Tables := l.levels[1].GetOverlappingTables(minKey, maxKey)

    // 合并输入：L0所有SSTable + L1重叠SSTable
    inputTables := append(l0Tables, l1Tables...)

    // 创建一个合并迭代器，用于遍历所有输入SSTable的有序数据
    // 这是一个核心组件，需要将多个SSTable的迭代器合并为一个有序迭代器
    mergedIterator := NewMergedIterator(inputTables) // 假设存在这样一个合并迭代器

    // 创建新的SSTablewriter，写入到L1
    newTableID := atomic.AddInt32(&l.nextTableID, 1)
    newFilename := fmt.Sprintf("sstable-%06d.sst", newTableID)
    newWriter, err := NewSSTableWriter(filepath.Join(l.baseDir, l.config.SSTableDir), newFilename)
    if err != nil {
        fmt.Printf("Error creating new SSTable writer for compaction: %vn", err)
        return
    }

    var newTableMinKey, newTableMaxKey []byte
    var firstEntry = true

    for mergedIterator.Next() {
        entry := mergedIterator.Current()
        // 跳过旧版本的Entry和被删除的Entry（只有当所有引用都消失时才物理删除）
        // 合并的核心逻辑：只保留最新版本的Entry
        if entry.Type == EntryDelete {
            // 在合并时，如果遇到删除标记，并且在当前合并的SSTable范围中没有更旧的版本，则该Entry不会被写入新SSTable
            // 复杂逻辑：需要检查所有输入SSTable中是否有比当前删除标记更早但未被删除的Entry
            // 简化：如果发现删除标记，则不写入新SSTable
            continue
        }

        if firstEntry {
            newTableMinKey = entry.Key
            firstEntry = false
        }
        newTableMaxKey = entry.Key // 记录最后一个键

        if err := newWriter.AddEntry(entry); err != nil {
            fmt.Printf("Error adding entry during compaction: %vn", err)
            newWriter.Close()
            return
        }
    }

    if err := newWriter.Finalize(); err != nil {
        fmt.Printf("Error finalizing new SSTable during compaction: %vn", err)
        newWriter.Close()
        return
    }
    newWriter.Close()

    // 将新的SSTable添加到L1层级
    newReader, err := NewSSTableReader(filepath.Join(l.baseDir, l.config.SSTableDir, newFilename))
    if err != nil {
        fmt.Printf("Error creating reader for new compacted SSTable: %vn", err)
        return
    }

    newTableInfo := &TableInfo{
        ID:   int(newTableID),
        Path: filepath.Join(l.baseDir, l.config.SSTableDir, newFilename),
        MinKey: newTableMinKey,
        MaxKey: newTableMaxKey, // 实际应从SSTableReader获取
        Size:   newReader.footerOffset, // 近似文件大小
        Level:  1,
        Reader: newReader,
    }
    l.levels[1].AddTable(newTableInfo)

    // 移除旧的SSTable文件
    l.levels[0].RemoveTables(l0Tables)
    l.levels[1].RemoveTables(l1Tables) // 移除L1被合并的SSTable

    // 物理删除文件
    for _, t := range inputTables {
        os.Remove(t.Path)
        t.Reader.Close()
    }

    fmt.Printf("Compaction completed: L0 + L1 -> new L1. New table: %sn", newFilename)
}

// Level 管理一个层级的所有SSTable
type Level struct {
    mu     sync.RWMutex
    id     int
    tables []*TableInfo // 该层级的所有SSTable，按MinKey排序
    sstableDir string
}

func NewLevel(id int, sstableDir string) *Level {
    return &Level{
        id:     id,
        tables: make([]*TableInfo, 0),
        sstableDir: sstableDir,
    }
}

func (l *Level) AddTable(table *TableInfo) {
    l.mu.Lock()
    defer l.mu.Unlock()
    // 插入并保持有序
    // TODO: 实际应使用二分查找插入
    l.tables = append(l.tables, table)
    // 简单的排序，实际应插入时保持有序或使用更高效的结构
    // sort.Slice(l.tables, func(i, j int) bool { return bytes.Compare(l.tables[i].MinKey, l.tables[j].MinKey) < 0 })
}

func (l *Level) RemoveTables(tablesToRemove []*TableInfo) {
    l.mu.Lock()
    defer l.mu.Unlock()

    newTables := make([]*TableInfo, 0, len(l.tables))
    removedMap := make(map[int]bool)
    for _, t := range tablesToRemove {
        removedMap[t.ID] = true
    }

    for _, t := range l.tables {
        if !removedMap[t.ID] {
            newTables = append(newTables, t)
        }
    }
    l.tables = newTables
}

func (l *Level) Size() int64 {
    l.mu.RLock()
    defer l.mu.RUnlock()
    var total int64
    for _, t := range l.tables {
        total += t.Size
    }
    return total
}

func (l *Level) GetAllTables() []*TableInfo {
    l.mu.RLock()
    defer l.mu.RUnlock()
    // 返回副本以防止外部修改
    tables := make([]*TableInfo, len(l.tables))
    copy(tables, l.tables)
    return tables
}

// GetOverlappingTables 获取与给定键范围重叠的SSTable
func (l *Level) GetOverlappingTables(minKey, maxKey []byte) []*TableInfo {
    l.mu.RLock()
    defer l.mu.RUnlock()

    var overlapping []*TableInfo
    for _, t := range l.tables {
        // 简化判断：如果两个范围有交集
        // (t.MinKey <= maxKey && t.MaxKey >= minKey)
        if bytes.Compare(t.MinKey, maxKey) <= 0 && bytes.Compare(t.MaxKey, minKey) >= 0 {
            overlapping = append(overlapping, t)
        }
    }
    return overlapping
}

func (l *Level) Get(key []byte) (Entry, bool, error) {
    l.mu.RLock()
    defer l.mu.RUnlock()

    // 从最新（L0）到最旧（最高层）SSTable查找
    // 在一个层级内，SSTable通常按时间或范围排序
    // 这里假设L0内的SSTable是按时间降序（最新刷写的在前面）
    // L1及更高层级，SSTable通常按键范围不重叠且有序

    // 对于L0（SSTable可能重叠），需要从最新的SSTable开始查找
    // 对于L1+（SSTable不重叠），可以二分查找定位SSTable

    // 简化处理：线性扫描所有SSTable
    for _, t := range l.tables {
        entry, found, err := t.Reader.Get(key)
        if err != nil { return Entry{}, false, err }
        if found {
            return entry, true, nil
        }
    }
    return Entry{}, false, nil
}

// MergedIterator 模拟一个合并多个SSTable迭代器的迭代器
type MergedIterator struct {
    iterators []Iterator // 假设SSTableReader可以返回一个Iterator
    currentEntry *Entry
    // 需要一个优先级队列来管理来自不同迭代器的当前元素
    // 以便始终取出最小的键
    pq *MinHeap // 例如，一个存储EntryWrapper的最小堆
}

type EntryWrapper struct {
    Entry
    SourceID int // 标识来源SSTable，用于解决冲突和版本
}

// NewMergedIterator 创建一个合并迭代器
func NewMergedIterator(tables []*TableInfo) *MergedIterator {
    mi := &MergedIterator{
        pq: NewMinHeap(), // 假设已实现一个最小堆
    }
    for i, t := range tables {
        // 假设SSTableReader有GetIterator方法
        // iter := t.Reader.GetIterator()
        // if iter.Next() {
        //  mi.pq.Push(&EntryWrapper{Entry: iter.Current(), SourceID: i})
        // }
        // mi.iterators = append(mi.iterators, iter)
    }
    return mi
}

// Next 移动到下一个元素
func (mi *MergedIterator) Next() bool {
    // 模拟合并迭代器的逻辑：
    // 1. 从最小堆中取出最小的Entry
    // 2. 将该Entry的来源迭代器的下一个Entry放入堆中
    // 3. 解决键冲突（保留最新版本）
    if mi.pq.Len() == 0 {
        mi.currentEntry = nil
        return false
    }

    minWrapper := mi.pq.Pop().(*EntryWrapper)
    mi.currentEntry = &minWrapper.Entry

    // 从minWrapper.SourceID对应的迭代器中取出下一个元素并放入堆中
    // 避免重复键：如果堆顶还有相同key的Entry，则继续弹出，直到不同key或只留下最新版本
    for mi.pq.Len() > 0 {
        nextWrapper := mi.pq.Peek().(*EntryWrapper)
        if bytes.Equal(nextWrapper.Key, mi.currentEntry.Key) {
            // 解决冲突：保留最新版本
            if nextWrapper.Timestamp > mi.currentEntry.Timestamp {
                mi.currentEntry = &nextWrapper.Entry
            }
            mi.pq.Pop() // 弹出旧版本或相同版本的
            // 再次从该迭代器中取出下一个元素并放入堆中
            // (此处需要更完善的迭代器和堆交互逻辑)
        } else {
            break
        }
    }

    return true
}

// Current 返回当前元素
func (mi *MergedIterator) Current() Entry {
    return *mi.currentEntry
}

// Close 关闭所有底层迭代器
func (mi *MergedIterator) Close() error {
    for _, iter := range mi.iterators {
        // iter.Close()
    }
    return nil
}

// MinHeap 最小堆的简化接口 (实际需要实现heap.Interface)
type MinHeap struct {
    entries []*EntryWrapper
}

func NewMinHeap() *MinHeap {
    return &MinHeap{entries: make([]*EntryWrapper, 0)}
}

func (h *MinHeap) Push(x interface{}) {
    entry := x.(*EntryWrapper)
    h.entries = append(h.entries, entry)
    // heap.Push(h, x) // 实际使用container/heap
}

func (h *MinHeap) Pop() interface{} {
    if len(h.entries) == 0 { return nil }
    // 模拟弹出最小元素
    minIdx := 0
    for i := 1; i < len(h.entries); i++ {
        if bytes.Compare(h.entries[i].Key, h.entries[minIdx].Key) < 0 {
            minIdx = i
        }
    }
    minEntry := h.entries[minIdx]
    h.entries = append(h.entries[:minIdx], h.entries[minIdx+1:]...)
    return minEntry
    // return heap.Pop(h) // 实际使用container/heap
}

func (h *MinHeap) Peek() interface{} {
    if len(h.entries) == 0 { return nil }
    minIdx := 0
    for i := 1; i < len(h.entries); i++ {
        if bytes.Compare(h.entries[i].Key, h.entries[minIdx].Key) < 0 {
            minIdx = i
        }
    }
    return h.entries[minIdx]
}

func (h *MinHeap) Len() int {
    return len(h.entries)
}

4. 优化写入放大的策略

理解了写入放大的来源后，我们可以针对性地提出优化策略。

4.1 调整MemTable大小

• 策略：增大MemTable的容量。
• 原理：MemTable越大，在刷写到磁盘形成SSTable之前可以积累越多的用户数据。这意味着更少的MemTable刷写操作，从而减少L0层级的SSTable数量。L0 SSTable数量的减少直接降低了L0到L1层级的合并频率和数据量，进而减少了写入放大。
• Go实现考量：
  • LSMConfig.MemTableMaxSize 配置项。
  • 优点：显著降低写入放大。
  • 缺点：
    • 内存消耗增加：MemTable完全存在于RAM中，过大会占用过多内存。
    • 恢复时间增加：系统崩溃时，WAL需要重放的数据量更大，恢复时间更长。
    • 刷写延迟：刷写操作会消耗I/O带宽，如果MemTable很大，一次刷写可能耗时较长，对写入吞吐量造成瞬时影响。

4.2 选择合适的合并策略

合并策略是写入放大最重要的影响因素。主要有两大类：Leveled Compaction (分层合并) 和 Size-Tiered Compaction (大小分层合并)。

4.2.1 Leveled Compaction (分层合并)

• 原理：数据被组织成严格的层级（L0, L1, L2…）。每个层级都有一个目标大小，通常是其前一层级的N倍。L0层级的SSTable可以相互重叠。从L1开始，每个层级内的SSTable的键范围互不重叠，并且按键排序。合并操作通常是将一个Li层级的SSTable与所有与其键范围重叠的L{i+1}层级SSTable进行合并，然后写入新的L_{i+1}层级SSTable。
• Go实现考量：
  • 在runCompaction中实现更精细的SSTable选择逻辑。
  • 优点：
    • 读性能优异：由于从L1开始，SSTable的键范围不重叠，读取操作只需要在一个层级中查找最多一个SSTable（加上L0的多个SSTable），读放大较低。
    • 空间放大较低：每个键值对的副本数量较少，通常只有1~2个副本。
  • 缺点：
    • 写入放大较高：当一个键值对被更新时，它需要从当前层级被合并到下一层级，可能经历多次重写。在最坏情况下，一个键值对可能在所有层级中都被重写一遍。典型的写入放大可能在10-30倍。
• 优化思路：
  • 增大层级比率 (Level Ratio)：LSMConfig.LevelRatio。如果Li层的总大小是L{i-1}层的10倍，那么一个键值对从L0向下移动到Lmax层，理论上会被重写 `log{Ratio}(TotalDataSize/MemTableSize)` 次。增大比率可以减少合并次数，但会增加每个层级的SSTable数量，从而增加读取时需要检查的SSTable文件数。
  • 小范围合并：优先合并那些键范围重叠较小的SSTable，避免大范围的全面合并。
  • 延迟合并：在写入负载高峰期，可以适度延迟一些非关键的合并，将其推迟到系统空闲时进行。

4.2.2 Size-Tiered Compaction (大小分层合并)

• 原理：没有严格的层级之分。SSTable根据其大小被组织成“层级”或“桶”。当某个“桶”中的SSTable数量达到阈值时，所有这些SSTable会被合并成一个更大的SSTable，并移动到下一个更大的“桶”。新刷写的SSTable（L0）通常是最小的。
• Go实现考量：
  • runCompaction的逻辑将大不相同，需要根据SSTable的大小分组。
  • 优点：
    • 写入放大较低：一个键值对通常只在合并时被重写几次，因为合并操作通常是将一组SSTable合并成一个更大的SSTable，而不是将数据从一层移动到下一层。典型的写入放大可能在2-5倍。
    • 对写入吞吐量影响较小：合并操作往往将相似大小的文件合并，避免了Leveled Compaction中“大合并”的问题。
  • 缺点：
    • 读放大较高：由于SSTable的键范围可能严重重叠，读取一个键时可能需要检查多个（甚至所有）SSTable文件才能找到最新版本。
    • 空间放大较高：由于旧版本数据和删除标记可能在多个SSTable中存在更长时间，直到它们被合并到足够大的SSTable中才被清理，导致磁盘空间利用率较低。
• 优化思路：
  • 合并阈值：调整每个“桶”中SSTable的数量阈值，过低会增加合并频率，过高会增加读放大。
  • 混合策略：可以考虑将L0到L1使用Size-Tiered，而L1及以上使用Leveled，以兼顾写入和读取性能。这被称为Hybrid Compaction。

4.3 数据编码与压缩

• 策略：优化SSTable内的数据布局和编码，并启用压缩。
• 原理：减少每个键值对在磁盘上占用的物理空间，直接减少了每次写入的字节数，从而降低写入放大。
• Go实现考量：
  • Key-Value Separation (KVS)：如果值很大且更新不频繁，可以将值存储在一个单独的日志文件或存储区域，SSTable中只存储键和指向值的指针。这样在合并SSTable时，大值无需被反复重写。
  • Prefix Encoding / Delta Encoding：在SSTable的每个数据块中，键是排序的。可以利用相邻键的相似性进行前缀压缩或增量编码，例如“KeyA001”、“KeyA002”可以存储为“KeyA001”和“002”。
  • 数据压缩：使用Snappy、LZ4、Zstd等快速压缩算法压缩数据块。在SSTableWriter的flushBlock方法中，在写入blockData之前对其进行压缩。在SSTableReader的readBlock方法中，读取后进行解压。
    • 优点：显著减少物理写入量，节省磁盘空间。
    • 缺点：增加CPU开销，压缩/解压需要时间。需要权衡CPU和I/O。

• 代码示例（SSTableWriter 压缩块）：

  // sstable.go (部分修改)
  // ...
  import "github.com/golang/snappy" // 假设使用Snappy压缩
  
  // ...
  // flushBlock 方法中
  func (w *SSTableWriter) flushBlock() error {
      if w.blockBuf.Len() == 0 {
          return nil
      }
  
      rawBlockData := w.blockBuf.Bytes()
  
      // 压缩数据块
      compressedBlockData := snappy.Encode(nil, rawBlockData) // nil表示让snappy分配缓冲区
  
      // 计算校验和 (对压缩后的数据)
      checksum := crc32.ChecksumIEEE(compressedBlockData)
  
      // 写入压缩后的数据块
      _, err := w.writer.Write(compressedBlockData)
      if err != nil { return err }
  
      // 写入校验和
      err = binary.Write(w.writer, binary.LittleEndian, checksum)
      if err != nil { return err }
  
      // 记录块元数据，注意这里的Size应为原始块大小，因为索引查找需要
      w.blockMetas = append(w.blockMetas, BlockMeta{
          Offset:   w.currentBlockOffset,
          Size:     uint32(len(rawBlockData)), // 原始大小
          FirstKey: w.lastEntryKey,
          CompressedSize: uint32(len(compressedBlockData)), // 新增字段记录压缩后大小
      })
  
      w.currentBlockOffset += uint64(len(compressedBlockData) + ChecksumSize)
      w.blockBuf.Reset()
      w.lastEntryKey = nil
      return nil
  }
  
  // BlockMeta 结构也需要修改
  type BlockMeta struct {
      Offset         uint64 // 块在文件中的偏移
      Size           uint32 // 块的原始大小
      CompressedSize uint32 // 块的压缩后大小
      FirstKey       []byte // 块中第一个键，用于稀疏索引
  }
  
  // SSTableReader 读取块时需要解压
  // readBlock 方法中
  func (r *SSTableReader) readBlock(offset uint64, compressedSize uint32) ([]byte, error) {
      compressedData := make([]byte, compressedSize)
      _, err := r.file.ReadAt(compressedData, int64(offset))
      if err != nil {
          return nil, err
      }
      // TODO: 校验checksum
  
      decompressedData, err := snappy.Decode(nil, compressedData)
      if err != nil {
          return nil, err
      }
      return decompressedData, nil
  }

4.4 延迟删除 (Tombstones)

• 策略：删除操作不立即物理移除数据，而是写入一个特殊的“删除标记”（Tombstone）。
• 原理：Tombstone也会像普通键值对一样参与MemTable刷写和SSTable合并。当Tombstone与其所标记的真正键值对的所有旧版本都被合并到同一个SSTable中时，并且该SSTable被合并到更低层级时，Tombstone和旧版本数据才会一起被物理移除。这避免了立即进行随机删除的开销，但意味着Tombstone会短暂地增加写入量和空间占用。
• Go实现考量：
  • EntryType 用于区分 EntryPut 和 EntryDelete。
  • 在MergedIterator中，处理Tombstone的逻辑至关重要：
    • 当迭代器遇到一个EntryDelete类型的Entry时，它会“删除”所有具有相同Key但Timestamp更早的Entry。
    • 只有当某个Key的所有旧版本都被Tombstone覆盖，并且Tombstone本身也足够老（例如，已经通过多轮合并“清理”了所有更旧的物理数据）时，Tombstone才会在最终的SSTable中被移除。
    • 此过程增加了合并的复杂性，但在LSM-Tree中是自然的GC机制。

4.5 并发与批处理

• 策略：通过并发处理和批处理来提高整体写入吞吐量，虽然不直接降低Amplify比率，但可以提高系统容忍写入放大的能力。
• Go实现考量：
  • LSM.Put 和 LSM.Delete 操作是并发安全的，但WAL和MemTable的写入需要互斥锁保护。
  • WAL的写入通常是批量的，即多个用户写入操作被缓冲后一次性刷盘（wal.Sync()）。这减少了fsync的频率，提高了WAL的写入效率。
  • MemTable的刷写（flushLoop）和SSTable的合并（compactionLoop）都应该在后台异步进行，以避免阻塞用户写入路径。Go的Goroutines和Channels非常适合实现这种并发模型。

5. 测量和监控写入放大

优化写入放大离不开准确的测量。我们需要追踪Total Bytes Written和User Bytes Written。

• User Bytes Written：

  • 在Put和Delete方法中，记录每个Entry的len(Key) + len(Value)（对于Delete，Value可能为零）。
  • 使用Prometheus等监控系统累加这些值。

• Total Bytes Written：

  • WAL写入：WAL.WriteEntry 调用时，记录写入的字节数（编码后的Entry大小）。
  • SSTable刷写：SSTableWriter.AddEntry 和 SSTableWriter.Finalize 时，记录实际写入文件的字节数（包括数据块、校验和、索引、Bloom Filter等）。
  • SSTable合并：runCompaction 中，新创建的SSTable的写入量。

// metrics.go: 简化版Go Prometheus指标集成
package storage

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    // 用户写入字节数
    userBytesWritten = promauto.NewCounter(prometheus.CounterOpts{
        Name: "lsm_user_bytes_written_total",
        Help: "Total number of bytes written by users.",
    })

    // 物理写入磁盘总字节数
    totalBytesWritten = promauto.NewCounter(prometheus.CounterOpts{
        Name: "lsm_total_bytes_written_total",
        Help: "Total number of bytes physically written to disk.",
    })

    // MemTable刷写次数
    memtableFlushCount = promauto.NewCounter(prometheus.CounterOpts{
        Name: "lsm_memtable_flush_total",
        Help: "Total number of memtable flushes.",
    })

    // Compaction次数
    compactionCount = promauto.NewCounter(prometheus.CounterOpts{
        Name: "lsm_compaction_total",
        Help: "Total number of compactions.",
    })

    // 当前MemTable大小
    activeMemTableSize = promauto.NewGauge(prometheus.GaugeOpts{
        Name: "lsm_active_memtable_size_bytes",
        Help: "Current size of the active memtable.",
    })
    // Immutable MemTable数量
    immutableMemTableCount = promauto.NewGauge(prometheus.GaugeOpts{
        Name: "lsm_immutable_memtable_count",
        Help: "Number of immutable memtables waiting to be flushed.",
    })
    // 各层级SSTable数量和大小
    sstableCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
        Name: "lsm_sstable_count",
        Help: "Number of SSTables per level.",
    }, []string{"level"})
    sstableSize = promauto.NewGaugeVec(prometheus.GaugeOpts{
        Name: "lsm_sstable_size_bytes",
        Help: "Total size of SSTables per level.",
    }, []string{"level"})
)

// RecordUserWrite 记录用户写入的字节数
func RecordUserWrite(keyLen, valueLen int) {
    userBytesWritten.Add(float64(keyLen + valueLen))
}

// RecordPhysicalWrite 记录物理写入磁盘的字节数
func RecordPhysicalWrite(bytes int64) {
    totalBytesWritten.Add(float64(bytes))
}

// 在LSM.Put中调用:
// RecordUserWrite(len(key), len(value))
// RecordPhysicalWrite(walEntryEncodedSize) // WAL写入

// 在flushLoop中调用:
// memtableFlushCount.Inc()
// RecordPhysicalWrite(sstableFileSize) // SSTable刷写

// 在compactionLoop中调用:
// compactionCount.Inc()
// RecordPhysicalWrite(newSSTableFileSize) // 合并写入

通过这些指标，我们可以定期计算Amplify = totalBytesWritten / userBytesWritten，并可视化其变化趋势，从而评估优化策略的效果。

6. 总结与展望

写入放大是LSM-Tree存储引擎设计中一个固有的权衡。它反映了为了实现高效的顺序写入、优化读取性能以及垃圾回收，数据必须在磁盘上被重写的次数。没有银弹式的解决方案，所有的优化都伴随着新的权衡：

• 写入放大 (WA) vs. 读取放大 (RA)：降低WA通常会增加RA（例如Size-Tiered）。
• 写入放大 (WA) vs. 空间放大 (SA)：降低WA可能意味着旧数据和Tombstone在磁盘上停留更长时间，增加SA。
• 性能 vs. 资源消耗：CPU压缩减少WA但增加CPU开销；大MemTable减少WA但增加内存消耗和恢复时间。

因此，LSM-Tree的优化是一个持续的平衡艺术，需要根据具体的应用场景、工作负载特性（读写比、键值大小分布、更新频率）以及硬件环境（SSD/HDD）进行精细调整。通过Go语言的强大表达力，我们可以灵活地实现各种合并策略、数据编码和监控机制，构建出高性能、高可靠的LSM-Tree存储引擎。未来的研究方向可能包括基于机器学习的自适应合并策略、更高效的并发控制以及针对新型存储硬件的特定优化。