src/sstable/sstable_reader.cpp - master - smychkov/SStorage

src/sstable/sstable_reader.cpp
324 строки13 KB
Андрей Смычков
docs: перевод оставшихся комментариев на русский + актуализация AGENTS
25 апр 2026, 10:08
25 апр 2026, 10:08f2b9fd9
#include "sstable_reader.hpp"
#include "block.hpp"
#include "../util/varint.hpp"

#include <algorithm>
#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>

namespace sstorage {

//============================================================================
// Вспомогательная функция: pread-полностью (с защитой от частичных чтений)
//============================================================================
namespace {

bool readAt(int fd, uint64_t offset, size_t len, std::string& out) {
    out.resize(len);
    size_t total = 0;
    while (total < len) {
        ssize_t n = ::pread(fd, out.data() + total, len - total, offset + total);
        if (n < 0) {
            if (errno == EINTR) continue;
            return false;
        }
        if (n == 0) return false;   // EOF раньше чем ожидали
        total += static_cast<size_t>(n);
    }
    return true;
}

}

//============================================================================
// Конструктор / деструктор
//============================================================================
SSTableReader::SSTableReader(std::string path) : path_(std::move(path)) {
}

SSTableReader::~SSTableReader() {
    if (fd_ >= 0) ::close(fd_);
}

//============================================================================
// Открытие файла и чтение метаданных в память
//============================================================================
bool SSTableReader::open() {
    // O_NOFOLLOW — защита от symlink attack (файл не должен быть симлинком)
    fd_ = ::open(path_.c_str(), O_RDONLY | O_NOFOLLOW);
    if (fd_ < 0) return false;

    struct stat st;
    if (::fstat(fd_, &st) != 0) {
        ::close(fd_);
        fd_ = -1;
        return false;
    }
    fileSize_ = static_cast<uint64_t>(st.st_size);

    if (fileSize_ < kFooterSize) {
        ::close(fd_);
        fd_ = -1;
        return false;
    }

    if (!readFooter()) return false;
    if (!readIndex()) return false;
    if (!readBloom()) return false;

    // minKey = firstKey первого блока (хранится как lastKey "предыдущего",
    // поэтому нужно загрузить первый блок и взять firstKey).
    // Упрощение: minKey = первый lastKey минус... на самом деле в блоке
    // записи отсортированы, minKey первого блока — это первая запись.
    if (!indexEntries_.empty()) {
        std::vector<Record> firstBlock;
        if (loadBlock(0, firstBlock) && !firstBlock.empty()) {
            minKey_ = firstBlock.front().key();
        }
        maxKey_ = indexEntries_.back().lastKey;
    }

    opened_ = true;
    return true;
}

//============================================================================
// Чтение footer
//============================================================================
bool SSTableReader::readFooter() {
    std::string footerBytes;
    if (!readAt(fd_, fileSize_ - kFooterSize, kFooterSize, footerBytes)) {
        return false;
    }
    return footer_.deserialize(footerBytes.data(), footerBytes.size());
}

//============================================================================
// Чтение index
//============================================================================
bool SSTableReader::readIndex() {
    if (footer_.indexSize == 0) return true;   // пустой SSTable

    // Жёсткий лимит на размер index block (256 МБ — очень щедро даже для
    // гигантских SSTable с миллионами блоков). Защита от вредоносного footer.
    constexpr uint64_t kMaxIndexSize = 256ULL * 1024 * 1024;
    if (footer_.indexSize > kMaxIndexSize) return false;

    // Проверка границ: index не должен выходить за пределы файла
    if (footer_.indexOffset > fileSize_) return false;
    if (footer_.indexSize > fileSize_ - footer_.indexOffset) return false;

    std::string indexBytes;
    if (!readAt(fd_, footer_.indexOffset, footer_.indexSize, indexBytes)) {
        return false;
    }

    // Лимит на размер ключа в index entry (защита от вредоносного varint)
    constexpr uint64_t kMaxIndexKeySize = 4ULL * 1024 * 1024;   // 4 МБ

    size_t offset = 0;
    uint64_t numEntries;
    size_t n;
    if (!util::decodeVarint(indexBytes.data() + offset, indexBytes.size() - offset,
                           numEntries, n)) return false;
    offset += n;
    if (offset > indexBytes.size()) return false;

    // Защита от невалидного числа
    if (numEntries > 10000000) return false;

    indexEntries_.reserve(static_cast<size_t>(numEntries));
    for (uint64_t i = 0; i < numEntries; ++i) {
        IndexEntry e;

        uint64_t keyLen;
        if (!util::decodeVarint(indexBytes.data() + offset,
                               indexBytes.size() - offset, keyLen, n)) return false;
        offset += n;
        if (offset > indexBytes.size()) return false;
        if (keyLen > kMaxIndexKeySize) return false;
        if (keyLen > indexBytes.size() - offset) return false;
        e.lastKey.assign(indexBytes.data() + offset, static_cast<size_t>(keyLen));
        offset += static_cast<size_t>(keyLen);

        if (!util::decodeVarint(indexBytes.data() + offset,
                               indexBytes.size() - offset, e.blockOffset, n)) return false;
        offset += n;
        if (offset > indexBytes.size()) return false;
        if (!util::decodeVarint(indexBytes.data() + offset,
                               indexBytes.size() - offset, e.blockSize, n)) return false;
        offset += n;
        if (offset > indexBytes.size()) return false;

        // Валидация смещений блока: должны быть внутри файла
        if (e.blockOffset > fileSize_) return false;
        if (e.blockSize > fileSize_ - e.blockOffset) return false;
        // Блок не должен "залезать" в index block
        if (e.blockOffset + e.blockSize > footer_.indexOffset) return false;

        indexEntries_.push_back(std::move(e));
    }
    return true;
}

//============================================================================
// Чтение bloom
//============================================================================
bool SSTableReader::readBloom() {
    if (footer_.bloomSize == 0) return true;

    // Лимит и проверка границ файла
    constexpr uint64_t kMaxBloomSize = 128ULL * 1024 * 1024;
    if (footer_.bloomSize > kMaxBloomSize) return false;
    if (footer_.bloomOffset > fileSize_) return false;
    if (footer_.bloomSize > fileSize_ - footer_.bloomOffset) return false;

    std::string bloomBytes;
    if (!readAt(fd_, footer_.bloomOffset, footer_.bloomSize, bloomBytes)) {
        return false;
    }

    size_t consumed;
    return bloom_.deserialize(bloomBytes.data(), bloomBytes.size(), consumed);
}

//============================================================================
// Загрузка data block
//============================================================================
bool SSTableReader::loadBlock(size_t idx, std::vector<Record>& out) const {
    if (idx >= indexEntries_.size()) return false;
    const auto& e = indexEntries_[idx];

    std::string onDisk;
    if (!readAt(fd_, e.blockOffset, e.blockSize, onDisk)) return false;

    std::string uncompressed;
    if (!readBlockFromDisk(onDisk.data(), onDisk.size(), uncompressed)) return false;

    return parseBlockRecords(uncompressed.data(), uncompressed.size(), out);
}

//============================================================================
// Поиск индекса блока для ключа
//============================================================================
// Ищем первый блок, где lastKey >= key.
// Если key > lastKey всех блоков — возвращаем indexEntries_.size().
//============================================================================
size_t SSTableReader::findBlockIndex(const std::string& key) const {
    auto it = std::lower_bound(indexEntries_.begin(), indexEntries_.end(), key,
        [](const IndexEntry& e, const std::string& k) {
            return e.lastKey < k;
        });
    return static_cast<size_t>(it - indexEntries_.begin());
}

//============================================================================
// Поиск записи по ключу
//============================================================================
std::optional<Record> SSTableReader::find(const std::string& key) const {
    if (!opened_) return std::nullopt;
    if (indexEntries_.empty()) return std::nullopt;

    // Быстрые отсечения
    if (key < minKey_ || key > maxKey_) return std::nullopt;
    if (!bloom_.mayContain(key)) return std::nullopt;

    size_t blockIdx = findBlockIndex(key);
    if (blockIdx >= indexEntries_.size()) return std::nullopt;

    std::vector<Record> records;
    if (!loadBlock(blockIdx, records)) return std::nullopt;

    // Записи в блоке сортированы: key ASC, при равных — seqNo DESC.
    // Находим первую запись с таким key (это будет самая свежая версия).
    auto it = std::lower_bound(records.begin(), records.end(), key,
        [](const Record& r, const std::string& k) {
            return r.key() < k;
        });
    if (it == records.end() || it->key() != key) return std::nullopt;
    return *it;
}

//============================================================================
// Диапазонный запрос (scan): все записи с fromKey <= key <= toKey
//============================================================================
std::vector<Record> SSTableReader::scan(const std::string& fromKey,
                                        const std::string& toKey,
                                        size_t limit) const {
    std::vector<Record> result;
    if (!opened_ || indexEntries_.empty()) return result;

    // Нет пересечения диапазонов
    if (toKey < minKey_ || fromKey > maxKey_) return result;

    size_t startBlock = findBlockIndex(fromKey);

    for (size_t i = startBlock; i < indexEntries_.size(); ++i) {
        // Если предыдущий блок заканчивался lastKey <= toKey — продолжаем.
        // Если текущий firstKey > toKey — останавливаемся. Для простоты
        // проверим после загрузки блока.

        std::vector<Record> records;
        if (!loadBlock(i, records)) break;

        for (const auto& r : records) {
            if (r.key() < fromKey) continue;
            if (r.key() > toKey) {
                // Все последующие будут больше (блоки отсортированы)
                return result;
            }
            result.push_back(r);
            if (limit > 0 && result.size() >= limit) return result;
        }

        // Если lastKey блока уже > toKey — следующий блок не нужен
        if (indexEntries_[i].lastKey > toKey) break;
    }
    return result;
}

//============================================================================
// readAll — для compaction (малые файлы, тесты)
//============================================================================
std::vector<Record> SSTableReader::readAll() const {
    std::vector<Record> result;
    if (!opened_) return result;
    result.reserve(static_cast<size_t>(footer_.numRecords));
    for (size_t i = 0; i < indexEntries_.size(); ++i) {
        std::vector<Record> block;
        if (!loadBlock(i, block)) break;
        for (auto& r : block) result.push_back(std::move(r));
    }
    return result;
}

//============================================================================
// Streaming API для compaction
//============================================================================
// Число блоков в файле — используется итератором чтобы знать, когда остановиться.
size_t SSTableReader::blockCount() const {
    return indexEntries_.size();
}

// Public-обёртка над loadBlock — нужна MergingIterator'у для постепенной
// загрузки блоков без чтения всего файла в память.
bool SSTableReader::loadBlockAt(size_t idx, std::vector<Record>& out) const {
    return loadBlock(idx, out);
}

//============================================================================
// Диапазонная проверка
//============================================================================
bool SSTableReader::overlaps(const std::string& from, const std::string& to) const {
    if (!opened_ || indexEntries_.empty()) return false;
    return !(to < minKey_ || from > maxKey_);
}

bool SSTableReader::mayContain(const std::string& key) const {
    if (!opened_ || indexEntries_.empty()) return false;
    if (key < minKey_ || key > maxKey_) return false;
    return bloom_.mayContain(key);
}

}