src/lsm/lsm_tree.cpp - master - smychkov/SStorage

src/lsm/lsm_tree.cpp
910 строк32 KB
Андрей
feat: top N keys + scan stream (range/top/all) with pagination
14 июл 2026, 12:56
14 июл 2026, 12:5656c6264
#include "lsm_tree.hpp"
#include "compaction.hpp"
#include "../util/utils.hpp"

#include <algorithm>
#include <chrono>
#include <cstdio>
#include <dirent.h>
#include <fcntl.h>
#include <iostream>
#include <sstream>
#include <stdexcept>
#include <sys/stat.h>
#include <unistd.h>

namespace sstorage {

namespace {

//============================================================================
// Распарсить имя SSTable-файла: sst_L<level>_<seq>.sst
// Возвращает (level, seqNo) или nullopt
//============================================================================
std::optional<std::pair<int, uint64_t>> parseSSTableName(const std::string& name) {
    // "sst_L0_00000042.sst"
    if (name.compare(0, 5, "sst_L") != 0) return std::nullopt;
    size_t underscore = name.find('_', 5);
    if (underscore == std::string::npos) return std::nullopt;
    size_t dot = name.rfind(".sst");
    if (dot == std::string::npos || dot <= underscore + 1) return std::nullopt;

    try {
        int level = std::stoi(name.substr(5, underscore - 5));
        uint64_t seq = std::stoull(name.substr(underscore + 1, dot - underscore - 1));
        return std::make_pair(level, seq);
    } catch (...) {
        return std::nullopt;
    }
}

//============================================================================
// listDirectory — простое перечисление файлов в директории
//============================================================================
std::vector<std::string> listDirectory(const std::string& dir) {
    std::vector<std::string> files;
    DIR* d = ::opendir(dir.c_str());
    if (!d) return files;
    struct dirent* entry;
    while ((entry = ::readdir(d)) != nullptr) {
        std::string name = entry->d_name;
        if (name == "." || name == "..") continue;
        files.push_back(name);
    }
    ::closedir(d);
    return files;
}

//============================================================================
// fsyncDirectory — важно для атомарности rename
//============================================================================
void fsyncDirectory(const std::string& dir) {
    int fd = ::open(dir.c_str(), O_RDONLY);
    if (fd >= 0) {
        ::fsync(fd);
        ::close(fd);
    }
}

// Maximal upper-bound string для in-memory MemTable::scan — лексикографически
// больше любого реального ключа (ключи ограничены maxKeySize, по умолчанию 4 КБ).
// Используется только для внутреннего range-scan'а MemTable (требует `to`),
// при scanAll (без верхней отсечки по контракту S1 §5).
const std::string& maxUpperBound() {
    static const std::string s(65536, '\xFF');
    return s;
}

// Потоковый k-way merge поверх снимков MemTable и уровней SSTable.
// Обход БЕЗ полной материализации: SSTable читаются поблочно через
// MergingIterator, MemTable отдаются in-memory (они и так в RAM).
// Дедуп latest-wins по seqNo, tombstone исключаются, диапазон включительный.
// См. contract S1 §3-§6.
size_t runStreamMerge(
    const std::shared_ptr<MemTable>& active,
    const std::vector<std::shared_ptr<MemTable>>& immutables,
    const std::vector<std::vector<SSTablePtr>>& levelSnap,
    const std::string& fromKey,
    const std::string& toKey,
    bool hasUpper,
    size_t limit,
    const ScanCallback& callback) {
    const std::string& mtUpper = hasUpper ? toKey : maxUpperBound();

    std::vector<std::vector<Record>> inMemStreams;
    inMemStreams.reserve(1 + immutables.size());
    inMemStreams.push_back(active->scan(fromKey, mtUpper, 0));
    for (const auto& mt : immutables) {
        inMemStreams.push_back(mt->scan(fromKey, mtUpper, 0));
    }

    std::vector<SSTablePtr> readers;
    for (const auto& lvl : levelSnap) {
        for (const auto& t : lvl) {
            if (hasUpper) {
                if (t->overlaps(fromKey, toKey)) readers.push_back(t);
            } else {
                if (t->maxKey() >= fromKey) readers.push_back(t);
            }
        }
    }

    MergingIterator memIt(std::move(inMemStreams));
    MergingIterator sstIt(std::move(readers));

    size_t count = 0;
    std::string prevKey;
    bool havePrev = false;

    while (true) {
        if (!memIt.valid() && !sstIt.valid()) break;
        bool fromMem;
        if (!memIt.valid()) {
            fromMem = false;
        } else if (!sstIt.valid()) {
            fromMem = true;
        } else {
            const Record& mr = memIt.current();
            const Record& sr = sstIt.current();
            if (mr.key() == sr.key()) {
                fromMem = (mr.seqNo() >= sr.seqNo());
            } else {
                fromMem = (mr.key() < sr.key());
            }
        }
        const Record& cur = fromMem ? memIt.current() : sstIt.current();

        if (havePrev && cur.key() == prevKey) {
            if (fromMem) memIt.next(); else sstIt.next();
            continue;
        }
        if (cur.key() < fromKey) {
            prevKey = cur.key();
            havePrev = true;
            if (fromMem) memIt.next(); else sstIt.next();
            continue;
        }
        if (hasUpper && cur.key() > toKey) break;

        prevKey = cur.key();
        havePrev = true;

        if (cur.isTombstone()) {
            if (fromMem) memIt.next(); else sstIt.next();
            continue;
        }

        if (!callback(cur.key(), cur.value())) {
            ++count;
            return count;
        }
        ++count;
        if (fromMem) memIt.next(); else sstIt.next();
        if (limit > 0 && count >= limit) return count;
    }
    return count;
}

}

//============================================================================
// Конструктор / деструктор
//============================================================================
LSMTree::LSMTree(std::string dataDir, LSMOptions options)
    : dataDir_(std::move(dataDir)), options_(options),
      activeMemtable_(std::make_shared<MemTable>()) {
    // Создаём директорию если её нет
    ::mkdir(dataDir_.c_str(), 0755);

    // Инициализируем уровни
    for (size_t i = 0; i < options_.maxLevels; ++i) {
        levels_.emplace_back(static_cast<int>(i));
    }
}

LSMTree::~LSMTree() {
    close();
}

//============================================================================
// Открытие: recovery + запуск фоновых потоков
//============================================================================
bool LSMTree::open() {
    if (opened_) return true;

    if (!recover()) return false;

    // Открываем текущий WAL
    uint64_t walNum = nextWalNumber();
    wal_ = std::make_unique<WriteAheadLog>(dataDir_, walNum);
    if (!wal_->open()) return false;

    // Запускаем фоновые потоки
    running_ = true;
    flushThread_ = std::thread(&LSMTree::flushLoop, this);
    compactionThread_ = std::thread(&LSMTree::compactionLoop, this);

    opened_ = true;
    return true;
}

//============================================================================
// Закрытие
//============================================================================
void LSMTree::close() {
    if (!opened_) return;

    // Финальный flush активной MemTable, чтобы не потерять данные
    flush();

    running_ = false;
    flushCv_.notify_all();
    compactionCv_.notify_all();
    if (flushThread_.joinable()) flushThread_.join();
    if (compactionThread_.joinable()) compactionThread_.join();

    if (wal_) {
        wal_->close();
        wal_.reset();
    }
    opened_ = false;
}

//============================================================================
// put / remove / get / scan
//============================================================================
//============================================================================
// Потокобезопасные снимки MemTable
//============================================================================
// Эти функции получают shared_ptr под мьютексом — после возврата из функции
// лок освобождён, но сам объект MemTable thread-safe (shared_mutex внутри),
// поэтому работать с ним без блокировки memtableMutex_ безопасно.
// Это защищает от race condition между put/get и ротацией MemTable.
//============================================================================
std::shared_ptr<MemTable> LSMTree::snapshotActiveMemtable() const {
    std::lock_guard<std::mutex> lk(memtableMutex_);
    return activeMemtable_;
}

std::vector<std::shared_ptr<MemTable>> LSMTree::snapshotImmutableMemtables() const {
    std::lock_guard<std::mutex> lk(memtableMutex_);
    std::vector<std::shared_ptr<MemTable>> result;
    result.reserve(immutableMemtables_.size());
    for (const auto& [mt, _] : immutableMemtables_) {
        result.push_back(mt);
    }
    return result;
}

bool LSMTree::put(const std::string& key, const std::string& value) {
    if (!opened_) return false;

    uint64_t seq = nextSeqNo();
    Record r(key, value, seq);

    // 1. WAL (блокируется до fsync)
    if (!wal_->append(r)) return false;

    // 2. MemTable — получаем безопасный ��нимок под локом
    auto mt = snapshotActiveMemtable();
    mt->put(r);

    // 3. Возможно пора ротировать
    maybeRotateMemtable();
    return true;
}

bool LSMTree::remove(const std::string& key) {
    if (!opened_) return false;

    uint64_t seq = nextSeqNo();
    Record r = Record::makeTombstone(key, seq);

    if (!wal_->append(r)) return false;
    auto mt = snapshotActiveMemtable();
    mt->put(r);

    maybeRotateMemtable();
    return true;
}

std::optional<std::string> LSMTree::get(const std::string& key) {
    if (!opened_) return std::nullopt;

    // 1. Active MemTable — самый свежий источник.
    // Снимок shared_ptr под локом: после возврата указатель валиден,
    // даже если параллельно произойдёт rotate.
    auto active = snapshotActiveMemtable();
    if (auto r = active->get(key)) {
        return r->isTombstone() ? std::nullopt : std::optional<std::string>(r->value());
    }

    // 2. Immutable MemTables (от новых к старым). Снимок списка под локом —
    // мы НЕ удерживаем memtableMutex_ во время обхода (избегаем lock contention
    // с put/flushLoop).
    auto immutables = snapshotImmutableMemtables();
    for (auto it = immutables.rbegin(); it != immutables.rend(); ++it) {
        if (auto r = (*it)->get(key)) {
            return r->isTombstone() ? std::nullopt : std::optional<std::string>(r->value());
        }
    }

    // 3. L0 — все файлы от новых к старым
    std::vector<SSTablePtr> l0Tables;
    std::vector<std::vector<SSTablePtr>> higherLevels;
    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        l0Tables = levels_[0].tables();
        higherLevels.reserve(levels_.size() - 1);
        for (size_t i = 1; i < levels_.size(); ++i) {
            higherLevels.push_back(levels_[i].tables());
        }
    }

    for (const auto& t : l0Tables) {
        if (auto r = t->find(key)) {
            return r->isTombstone() ? std::nullopt : std::optional<std::string>(r->value());
        }
    }

    // 4. L1+ — один файл на уровень (бинпоиск)
    for (size_t i = 0; i < higherLevels.size(); ++i) {
        for (const auto& t : higherLevels[i]) {
            if (t->mayContain(key)) {
                if (auto r = t->find(key)) {
                    return r->isTombstone() ? std::nullopt : std::optional<std::string>(r->value());
                }
            }
        }
    }

    return std::nullopt;
}

std::vector<std::pair<std::string, std::string>> LSMTree::scan(
    const std::string& fromKey, const std::string& toKey, size_t limit) {

    if (!opened_) return {};

    // Для простоты: собираем все подходящие записи изо всех источников
    // (MemTable, immutable, L0, L1+), сортируем по key, дедуплицируем по key
    // с предпочтением max seqNo.
    std::vector<Record> all;

    auto append = [&](std::vector<Record> src) {
        for (auto& r : src) all.push_back(std::move(r));
    };

    // Снимок active и immutable MemTables под локом — далее работаем без лока
    auto active = snapshotActiveMemtable();
    append(active->scan(fromKey, toKey, 0));

    auto immutables = snapshotImmutableMemtables();
    for (const auto& mt : immutables) {
        append(mt->scan(fromKey, toKey, 0));
    }

    std::vector<Level> snapshot;
    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        snapshot = levels_;
    }

    for (const auto& lvl : snapshot) {
        for (const auto& t : lvl.tables()) {
            append(t->scan(fromKey, toKey, 0));
        }
    }

    // Сортируем (key ASC, seqNo DESC)
    std::sort(all.begin(), all.end());

    // Дедуп: берём первую запись для каждого key (самая свежая)
    std::vector<std::pair<std::string, std::string>> result;
    std::string prevKey;
    bool havePrev = false;
    for (const auto& r : all) {
        if (havePrev && r.key() == prevKey) continue;
        prevKey = r.key();
        havePrev = true;
        if (r.isTombstone()) continue;
        result.emplace_back(r.key(), r.value());
        if (limit > 0 && result.size() >= limit) break;
    }
    return result;
}

//============================================================================
// scanStream / scanAll / top — потоковый обход. См. contract S1.
//============================================================================
size_t LSMTree::scanStream(const std::string& fromKey,
                           const std::string& toKey,
                           size_t limit,
                           const ScanCallback& callback) {
    if (!opened_) return 0;
    if (!callback) return 0;

    auto active = snapshotActiveMemtable();
    auto immutables = snapshotImmutableMemtables();

    std::vector<std::vector<SSTablePtr>> levelSnap;
    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        levelSnap.reserve(levels_.size());
        for (const auto& lvl : levels_) {
            levelSnap.push_back(lvl.tables());
        }
    }

    return runStreamMerge(active, immutables, levelSnap,
                          fromKey, toKey, true, limit, callback);
}

size_t LSMTree::scanAll(size_t limit, const ScanCallback& callback) {
    if (!opened_) return 0;
    if (!callback) return 0;

    auto active = snapshotActiveMemtable();
    auto immutables = snapshotImmutableMemtables();

    std::vector<std::vector<SSTablePtr>> levelSnap;
    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        levelSnap.reserve(levels_.size());
        for (const auto& lvl : levels_) {
            levelSnap.push_back(lvl.tables());
        }
    }

    return runStreamMerge(active, immutables, levelSnap,
                          std::string(), std::string(), false, limit, callback);
}

std::vector<std::pair<std::string, std::string>> LSMTree::top(size_t n) {
    std::vector<std::pair<std::string, std::string>> result;
    if (!opened_ || n == 0) return result;

    result.reserve(n);
    scanAll(n, [&](const std::string& key, const std::string& value) {
        result.emplace_back(key, value);
        return true;
    });
    return result;
}

//============================================================================
// maybeRotateMemtable — если active переполнен, делаем immutable и сигналим flush
//============================================================================
void LSMTree::maybeRotateMemtable() {
    // Быстрая проверка без лока (snapshot). Если размер ниже порога —
    // ничего не делаем. Ложноположительные срабатывания здесь безопасны:
    // мы перепроверим под локом.
    {
        auto active = snapshotActiveMemtable();
        if (active->approximateSize() < options_.memtableSizeBytes) {
            return;
        }
    }

    std::lock_guard<std::mutex> lk(memtableMutex_);
    // Повторная проверка под локом: между двумя чтениями другой поток
    // мог уже выполнить ротацию.
    if (activeMemtable_->approximateSize() < options_.memtableSizeBytes) {
        return;
    }

    // Замораживаем active, создаём новую
    activeMemtable_->freeze();
    uint64_t walNum = wal_->logNumber();
    immutableMemtables_.emplace_back(activeMemtable_, walNum);

    activeMemtable_ = std::make_shared<MemTable>();

    // Ротируем WAL: закрываем старый, открываем новый
    wal_->close();
    uint64_t newWalNum = nextWalNumber();
    wal_ = std::make_unique<WriteAheadLog>(dataDir_, newWalNum);
    wal_->open();

    flushCv_.notify_one();
}

//============================================================================
// flush — синхронный
//============================================================================
bool LSMTree::flush() {
    if (!opened_) return false;

    std::shared_ptr<MemTable> mt;
    uint64_t oldWalNum = 0;
    {
        std::lock_guard<std::mutex> lk(memtableMutex_);
        if (activeMemtable_->empty() && immutableMemtables_.empty()) {
            return true;
        }
        if (!activeMemtable_->empty()) {
            activeMemtable_->freeze();
            oldWalNum = wal_->logNumber();
            immutableMemtables_.emplace_back(activeMemtable_, oldWalNum);
            activeMemtable_ = std::make_shared<MemTable>();
            wal_->close();
            uint64_t newWalNum = nextWalNumber();
            wal_ = std::make_unique<WriteAheadLog>(dataDir_, newWalNum);
            wal_->open();
        }
    }

    // Ждём пока immutable очередь опустеет
    flushCv_.notify_one();
    while (true) {
        {
            std::lock_guard<std::mutex> lk(memtableMutex_);
            if (immutableMemtables_.empty()) break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    return true;
}

//============================================================================
// compact — принудительный compaction всех уровней
//============================================================================
bool LSMTree::compact() {
    if (!opened_) return false;

    // Делаем flush сначала
    flush();

    // Выполняем compaction пока возможно
    while (true) {
        int level = pickLevelForCompaction();
        if (level < 0) break;
        if (!doCompaction(level)) return false;
    }
    return true;
}

//============================================================================
// Фоновый flush-поток
//============================================================================
void LSMTree::flushLoop() {
    while (running_) {
        std::shared_ptr<MemTable> mt;
        uint64_t walNum = 0;
        {
            std::unique_lock<std::mutex> lk(memtableMutex_);
            if (immutableMemtables_.empty()) {
                lk.unlock();
                std::unique_lock<std::mutex> cvLk(flushCvMutex_);
                flushCv_.wait_for(cvLk, std::chrono::milliseconds(100));
                continue;
            }
            mt = immutableMemtables_.front().first;
            walNum = immutableMemtables_.front().second;
        }

        if (mt) {
            doFlushOne(mt, walNum);
            {
                std::lock_guard<std::mutex> lk(memtableMutex_);
                if (!immutableMemtables_.empty()) {
                    immutableMemtables_.pop_front();
                }
            }
            compactionCv_.notify_one();
        }
    }
}

//============================================================================
// doFlushOne — записать MemTable как L0 SSTable
//============================================================================
bool LSMTree::doFlushOne(std::shared_ptr<MemTable> mt, uint64_t walLogNumber) {
    if (mt->empty()) {
        // Пустая — просто удаляем WAL
        WriteAheadLog::remove(WriteAheadLog::buildFilename(dataDir_, walLogNumber));
        return true;
    }

    auto records = mt->drainSorted();

    uint64_t seq = nextFileSeqNo();
    std::string path = sstablePath(0, seq);

    SSTableWriter writer(path, options_.blockSize,
                        options_.compression, records.size(),
                        options_.bloomBitsPerKey);
    if (!writer.open()) return false;

    for (const auto& r : records) {
        if (!writer.add(r)) return false;
    }
    if (!writer.finish()) return false;

    // fsync директории — гарантирует, что rename() виден после крэша
    fsyncDirectory(dataDir_);

    // Добавляем в L0
    auto reader = std::make_shared<SSTableReader>(path);
    if (!reader->open()) return false;

    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        levels_[0].addTable(reader);
    }

    // Удаляем WAL (данные уже на диске)
    WriteAheadLog::remove(WriteAheadLog::buildFilename(dataDir_, walLogNumber));

    return true;
}

//============================================================================
// Фоновый цикл compaction-потока
//============================================================================
void LSMTree::compactionLoop() {
    while (running_) {
        int level = pickLevelForCompaction();
        if (level < 0) {
            std::unique_lock<std::mutex> lk(compactionCvMutex_);
            compactionCv_.wait_for(lk, std::chrono::milliseconds(200));
            continue;
        }
        doCompaction(level);
    }
}

//============================================================================
// Выбор уровня для очередного compaction
//============================================================================
int LSMTree::pickLevelForCompaction() const {
    std::lock_guard<std::mutex> lk(levelsMutex_);

    // L0: по числу файлов
    if (levels_[0].fileCount() >= options_.l0CompactionTrigger) {
        return 0;
    }

    // L1+: по суммарному размеру (грубо — по totalRecords * avg-size-estimate)
    // Для простоты используем fileCount * estimate.
    for (size_t i = 1; i + 1 < levels_.size(); ++i) {
        // Упрощённая оценка: targetSizeForLevel в файлах среднего размера
        uint64_t targetSize = targetSizeForLevel(static_cast<int>(i));
        uint64_t estimatedSize = levels_[i].fileCount() * options_.l1SizeBytes /
                                 std::max<size_t>(1, options_.levelRatio);
        if (estimatedSize > targetSize) {
            return static_cast<int>(i);
        }
    }
    return -1;
}

uint64_t LSMTree::targetSizeForLevel(int level) const {
    if (level == 0) return 0;  // L0 триггерится по файлам
    uint64_t size = options_.l1SizeBytes;
    for (int i = 1; i < level; ++i) {
        size *= options_.levelRatio;
    }
    return size;
}

//============================================================================
// Выполнение одной операции compaction для заданного уровня
//============================================================================
bool LSMTree::doCompaction(int level) {
    std::vector<SSTablePtr> inputs;
    int targetLevel = level + 1;
    if (static_cast<size_t>(targetLevel) >= levels_.size()) {
        return false;
    }

    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        if (level == 0) {
            inputs = levels_[0].tables();
            if (inputs.empty()) return false;
            // Диапазон всех L0 файлов
            auto [mn, mx] = keyRange(inputs);
            for (const auto& t : levels_[targetLevel].overlapping(mn, mx)) {
                inputs.push_back(t);
            }
        } else {
            // Берём один (первый) файл из level
            if (levels_[level].tables().empty()) return false;
            auto t = levels_[level].tables().front();
            inputs.push_back(t);
            for (const auto& nt : levels_[targetLevel].overlapping(t->minKey(), t->maxKey())) {
                inputs.push_back(nt);
            }
        }
    }

    // Предикат для tombstone elimination:
    // Если targetLevel — последний уровень ИЛИ ниже нет этого key — можно удалить
    auto shouldDropTombstone = [this, targetLevel](const std::string& key) -> bool {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        for (size_t i = targetLevel + 1; i < levels_.size(); ++i) {
            for (const auto& t : levels_[i].tables()) {
                if (t->mayContain(key)) return false;
            }
        }
        return true;
    };

    CompactionOptions copts;
    copts.blockSize = options_.blockSize;
    copts.bloomBitsPerKey = options_.bloomBitsPerKey;
    copts.compression = options_.compression;
    copts.targetFileSize = targetSizeForLevel(targetLevel) / 4;
    if (copts.targetFileSize < 1024 * 1024) copts.targetFileSize = 1024 * 1024;

    std::vector<std::string> newFiles;
    bool ok = runCompaction(
        inputs, targetLevel, copts,
        [this](int lvl, uint64_t seq) { return sstablePath(lvl, seq); },
        [this]() { return nextFileSeqNo(); },
        shouldDropTombstone,
        newFiles
    );
    if (!ok) return false;

    // Открываем новые SSTable
    std::vector<SSTablePtr> newReaders;
    for (const auto& p : newFiles) {
        auto r = std::make_shared<SSTableReader>(p);
        if (!r->open()) return false;
        newReaders.push_back(r);
    }

    // Собираем пути старых
    std::vector<std::string> oldPaths;
    for (const auto& t : inputs) oldPaths.push_back(t->path());

    // Атомарный swap: сначала fsync директории (rename уже сделан в writer)
    fsyncDirectory(dataDir_);

    // Определяем с каких уровней удалять
    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        // Удаляем inputs из levels[level] и levels[targetLevel]
        for (const auto& t : inputs) {
            auto parsed = parseSSTableName(
                t->path().substr(t->path().find_last_of('/') + 1));
            if (parsed.has_value()) {
                levels_[parsed->first].removeTable(t->path());
            }
        }
        for (const auto& r : newReaders) {
            levels_[targetLevel].addTable(r);
        }
    }

    // Удаляем старые файлы с диска
    for (const auto& p : oldPaths) {
        ::unlink(p.c_str());
    }
    fsyncDirectory(dataDir_);
    return true;
}

//============================================================================
// Recovery
//============================================================================
bool LSMTree::recover() {
    cleanupTempFiles();
    if (!scanAndLoadSSTables()) return false;
    if (!replayWALs()) return false;
    return true;
}

void LSMTree::cleanupTempFiles() {
    auto files = listDirectory(dataDir_);
    for (const auto& f : files) {
        // Удаляем недоделанные tmp_*.sst и *.sst.tmp (artefacts writer'а)
        if (f.compare(0, 4, "tmp_") == 0 ||
            (f.size() > 4 && f.compare(f.size() - 4, 4, ".tmp") == 0)) {
            ::unlink((dataDir_ + "/" + f).c_str());
        }
    }
}

bool LSMTree::scanAndLoadSSTables() {
    auto files = listDirectory(dataDir_);

    uint64_t maxSeqNo = 0;

    for (const auto& f : files) {
        auto parsed = parseSSTableName(f);
        if (!parsed.has_value()) continue;
        int level = parsed->first;
        uint64_t seq = parsed->second;

        if (level < 0 || static_cast<size_t>(level) >= levels_.size()) continue;

        std::string fullPath = dataDir_ + "/" + f;
        auto reader = std::make_shared<SSTableReader>(fullPath);
        if (!reader->open()) {
            // Битый файл — переносим в .corrupt (не удаляем, для диагностики)
            ::rename(fullPath.c_str(), (fullPath + ".corrupt").c_str());
            continue;
        }

        levels_[level].addTable(reader);
        if (seq > maxSeqNo) maxSeqNo = seq;
    }

    // Устанавливаем следующий seqNo для новых SSTable
    nextFileSeqNo_ = maxSeqNo + 1;

    // Сортируем и разрешаем перекрытия
    for (auto& lvl : levels_) {
        lvl.sortTables();
    }
    std::vector<std::string> toDelete;
    for (size_t i = 1; i < levels_.size(); ++i) {
        levels_[i].resolveOverlaps(toDelete);
    }
    for (const auto& p : toDelete) {
        ::unlink(p.c_str());
    }

    return true;
}

bool LSMTree::replayWALs() {
    auto walFiles = WriteAheadLog::listLogFiles(dataDir_);
    uint64_t maxWalNum = 0;
    uint64_t maxSeq = 0;

    // replay WAL происходит синхронно ДО запуска фоновых потоков,
    // поэтому race с другими потоками невозможен — доступ к activeMemtable_
    // безопасен. Тем не менее используем snapshot для единообразия кода.
    auto mt = snapshotActiveMemtable();
    for (const auto& [num, path] : walFiles) {
        if (num > maxWalNum) maxWalNum = num;
        auto records = WriteAheadLog::replay(path);
        for (auto& r : records) {
            if (r.seqNo() > maxSeq) maxSeq = r.seqNo();
            mt->put(r);
        }
        // Файл удалим после успешного flush; пока оставляем
    }

    nextWalNumber_ = maxWalNum + 1;
    nextSeqNo_ = maxSeq + 1;

    // Если восстановленная MemTable не пуста — форсим её flush (и удаление WAL)
    // Выполнится в потоке flush после open()
    return true;
}

//============================================================================
// Служебное
//============================================================================
std::string LSMTree::sstablePath(int level, uint64_t seqNo) const {
    char buf[64];
    std::snprintf(buf, sizeof(buf), "/sst_L%d_%010llu.sst",
                  level, static_cast<unsigned long long>(seqNo));
    return dataDir_ + buf;
}

std::pair<std::string, std::string> LSMTree::keyRange(
    const std::vector<SSTablePtr>& tables) {
    std::string mn, mx;
    bool have = false;
    for (const auto& t : tables) {
        if (!have) {
            mn = t->minKey();
            mx = t->maxKey();
            have = true;
        } else {
            if (t->minKey() < mn) mn = t->minKey();
            if (t->maxKey() > mx) mx = t->maxKey();
        }
    }
    return {mn, mx};
}

LSMStats LSMTree::stats() const {
    LSMStats s;
    // Снимок active под локом — избегаем data race с ротацией
    auto active = snapshotActiveMemtable();
    s.memtableSize = active->approximateSize();
    {
        std::lock_guard<std::mutex> lk(memtableMutex_);
        s.immutableCount = immutableMemtables_.size();
    }
    {
        std::lock_guard<std::mutex> lk(levelsMutex_);
        s.levelSizes.resize(levels_.size());
        s.levelFileCounts.resize(levels_.size());
        for (size_t i = 0; i < levels_.size(); ++i) {
            s.levelSizes[i] = levels_[i].totalRecords();
            s.levelFileCounts[i] = levels_[i].fileCount();
            s.totalRecords += s.levelSizes[i];
        }
    }
    return s;
}

}