tests/test_scan_stream.cpp - master - smychkov/SStorage

tests/test_scan_stream.cpp
601 строка20 KB
Андрей
feat: top N keys + scan stream (range/top/all) with pagination
14 июл 2026, 12:56
14 июл 2026, 12:5656c6264
//============================================================================
// Тесты для потокового read-path API: scanStream / scanAll / top.
// Контракт: docs/contracts/S1-core-scan-streaming.md
//============================================================================
#include "../src/core/database.hpp"
#include <algorithm>
#include <cstdlib>
#include <filesystem>
#include <iostream>
#include <map>
#include <string>
#include <sys/stat.h>
#include <unistd.h>
#include <vector>

using namespace sstorage;

static int g_passed = 0;
static int g_failed = 0;

#define CHECK(cond) do { \
    if (cond) { ++g_passed; } \
    else { ++g_failed; std::cerr << "FAIL: " #cond " at line " << __LINE__ << "\n"; } \
} while (0)

static std::string makeTempDir(const std::string& prefix) {
    std::string p = "/tmp/sstorage_scan_" + prefix + "_" +
                    std::to_string(::getpid()) + "_" +
                    std::to_string(rand());
    ::mkdir(p.c_str(), 0755);
    return p;
}

static void cleanupDir(const std::string& dir) {
    std::error_code ec;
    std::filesystem::remove_all(dir, ec);
}

// Хелпер: собирает записи через callback в вектор
static bool collectAll(const std::string& key, const std::string& value,
                       std::vector<std::pair<std::string,std::string>>& out) {
    out.push_back({key, value});
    return true;
}

// Хелпер: собирает записи через callback в вектор, останавливаясь на 1-й
static bool collectOne(const std::string& key, const std::string& value,
                       std::vector<std::pair<std::string,std::string>>& out) {
    out.push_back({key, value});
    return false;
}

// Хелпер: форматирует key_NNNN с ведущими нулями
static std::string keyN(int n) {
    char buf[16];
    std::snprintf(buf, sizeof(buf), "key_%04d", n);
    return std::string(buf);
}

int main() {
    //==========================================================================
    // UNIT-ТЕСТЫ (Database, один MemTable, без flush)
    //==========================================================================

    // --- top(2): три ключа, первые два в ASC ---
    // §7: b=2, a=1, c=3 → top(2) → [(a,1),(b,2)]
    {
        auto dir = makeTempDir("top2");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("b", "2"));
        CHECK(db.put("a", "1"));
        CHECK(db.put("c", "3"));

        auto res = db.top(2);
        CHECK(res.size() == 2);
        CHECK(res[0].first == "a" && res[0].second == "1");
        CHECK(res[1].first == "b" && res[1].second == "2");

        db.close();
        cleanupDir(dir);
    }

    // --- top(0): пустой результат ---
    // §7: b=2, a=1, c=3 → top(0) → []
    {
        auto dir = makeTempDir("top0");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("b", "2"));
        CHECK(db.put("a", "1"));
        CHECK(db.put("c", "3"));

        auto res = db.top(0);
        CHECK(res.empty());

        db.close();
        cleanupDir(dir);
    }

    // --- top(100): больше чем есть → все живые ---
    // §7: b=2, a=1, c=3 → top(100) → [(a,1),(b,2),(c,3)]
    {
        auto dir = makeTempDir("top100");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("b", "2"));
        CHECK(db.put("a", "1"));
        CHECK(db.put("c", "3"));

        auto res = db.top(100);
        CHECK(res.size() == 3);
        CHECK(res[0].first == "a" && res[0].second == "1");
        CHECK(res[1].first == "b" && res[1].second == "2");
        CHECK(res[2].first == "c" && res[2].second == "3");

        db.close();
        cleanupDir(dir);
    }

    // --- top после remove: tombstone исключается ---
    // §7: a=1, b=2, remove(a) → top(10) → [(b,2)]
    {
        auto dir = makeTempDir("top_rm");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("a", "1"));
        CHECK(db.put("b", "2"));
        CHECK(db.remove("a"));

        auto res = db.top(10);
        CHECK(res.size() == 1);
        CHECK(res[0].first == "b" && res[0].second == "2");

        db.close();
        cleanupDir(dir);
    }

    // --- top после upsert: latest-wins ---
    // §7: a=1, a=9 (upsert), b=2 → top(10) → [(a,9),(b,2)]
    {
        auto dir = makeTempDir("top_upsert");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("a", "1"));
        CHECK(db.put("a", "9"));
        CHECK(db.put("b", "2"));

        auto res = db.top(10);
        CHECK(res.size() == 2);
        CHECK(res[0].first == "a" && res[0].second == "9");
        CHECK(res[1].first == "b" && res[1].second == "2");

        db.close();
        cleanupDir(dir);
    }

    // --- scanAll(0,·) собрать всё ---
    // §7: b=2, a=1, c=3 → [(a,1),(b,2),(c,3)], return==3
    {
        auto dir = makeTempDir("scanAll_all");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("b", "2"));
        CHECK(db.put("a", "1"));
        CHECK(db.put("c", "3"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanAll(0, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n == 3);
        CHECK(collected.size() == 3);
        CHECK(collected[0].first == "a" && collected[0].second == "1");
        CHECK(collected[1].first == "b" && collected[1].second == "2");
        CHECK(collected[2].first == "c" && collected[2].second == "3");

        db.close();
        cleanupDir(dir);
    }

    // --- scanAll(2,·) → ровно 2 первых ---
    // §7: b=2, a=1, c=3 → [(a,1),(b,2)], return==2
    {
        auto dir = makeTempDir("scanAll_2");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("b", "2"));
        CHECK(db.put("a", "1"));
        CHECK(db.put("c", "3"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanAll(2, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n == 2);
        CHECK(collected.size() == 2);
        CHECK(collected[0].first == "a" && collected[0].second == "1");
        CHECK(collected[1].first == "b" && collected[1].second == "2");

        db.close();
        cleanupDir(dir);
    }

    // --- scanAll(5,·) при 3 живых → 3, return==3 ---
    // §7: b=2, a=1, c=3 → [(a,1),(b,2),(c,3)], return==3
    {
        auto dir = makeTempDir("scanAll_5");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("b", "2"));
        CHECK(db.put("a", "1"));
        CHECK(db.put("c", "3"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanAll(5, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n == 3);
        CHECK(collected.size() == 3);
        CHECK(collected[0].first == "a" && collected[0].second == "1");
        CHECK(collected[1].first == "b" && collected[1].second == "2");
        CHECK(collected[2].first == "c" && collected[2].second == "3");

        db.close();
        cleanupDir(dir);
    }

    // --- scanAll(0,·) после remove: tombstone исключается ---
    // §7: a=1, b=2, remove(b) → [(a,1)], return==1
    {
        auto dir = makeTempDir("scanAll_rm");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("a", "1"));
        CHECK(db.put("b", "2"));
        CHECK(db.remove("b"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanAll(0, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n == 1);
        CHECK(collected.size() == 1);
        CHECK(collected[0].first == "a" && collected[0].second == "1");

        db.close();
        cleanupDir(dir);
    }

    // --- scanAll(0,·) пустая БД → return==0, [] ---
    // §7: (пусто) → [], return==0
    {
        auto dir = makeTempDir("scanAll_empty");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanAll(0, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n == 0);
        CHECK(collected.empty());

        db.close();
        cleanupDir(dir);
    }

    // --- scanStream("","z",0,cb) с cb-false на 1-й записи ---
    // §7: a=1,b=2,c=3; cb false на 1-й → ровно 1 запись, return==1
    {
        auto dir = makeTempDir("ss_false1_lim0");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("a", "1"));
        CHECK(db.put("b", "2"));
        CHECK(db.put("c", "3"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanStream("", "z", 0,
            [&](const std::string& k, const std::string& v) {
                return collectOne(k, v, collected);
            });
        CHECK(n == 1);
        CHECK(collected.size() == 1);
        CHECK(collected[0].first == "a" && collected[0].second == "1");

        db.close();
        cleanupDir(dir);
    }

    // --- scanStream("","z",5,cb) с cb-false на 1-й ---
    // §7: cb-false приоритетнее limit → ровно 1 запись, return==1 (§4.5)
    {
        auto dir = makeTempDir("ss_false1_lim5");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("a", "1"));
        CHECK(db.put("b", "2"));
        CHECK(db.put("c", "3"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanStream("", "z", 5,
            [&](const std::string& k, const std::string& v) {
                return collectOne(k, v, collected);
            });
        CHECK(n == 1);
        CHECK(collected.size() == 1);
        CHECK(collected[0].first == "a" && collected[0].second == "1");

        db.close();
        cleanupDir(dir);
    }

    // --- scanStream("a","b",0,·) → включительный диапазон [a,b] ---
    // §7: [(a,1),(b,2)], return==2
    {
        auto dir = makeTempDir("ss_range");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());

        CHECK(db.put("a", "1"));
        CHECK(db.put("b", "2"));
        CHECK(db.put("c", "3"));

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n = db.scanStream("a", "b", 0,
            [&](const std::string& k, const std::string& v) {
                return collectAll(k, v, collected);
            });
        CHECK(n == 2);
        CHECK(collected.size() == 2);
        CHECK(collected[0].first == "a" && collected[0].second == "1");
        CHECK(collected[1].first == "b" && collected[1].second == "2");

        db.close();
        cleanupDir(dir);
    }

    // --- Гейт: до open() → scanStream/scanAll возвращают 0, top → {} ---
    // §4.9
    {
        auto dir = makeTempDir("gate_before");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        // НЕ вызываем open()

        std::vector<std::pair<std::string,std::string>> collected;
        size_t n1 = db.scanStream("a", "z", 0,
            [&](const std::string& k, const std::string& v) {
                return collectAll(k, v, collected);
            });
        CHECK(n1 == 0);

        size_t n2 = db.scanAll(0, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n2 == 0);

        auto topRes = db.top(10);
        CHECK(topRes.empty());

        cleanupDir(dir);
    }

    // --- Гейт: после close() → scanStream/scanAll возвращают 0, top → {} ---
    // §4.9
    {
        auto dir = makeTempDir("gate_after");
        Config cfg;
        cfg.setDataDirectory(dir);
        Database db(cfg);
        CHECK(db.open());
        CHECK(db.put("a", "1"));
        db.close();

        // После close()
        std::vector<std::pair<std::string,std::string>> collected;
        size_t n1 = db.scanStream("a", "z", 0,
            [&](const std::string& k, const std::string& v) {
                return collectAll(k, v, collected);
            });
        CHECK(n1 == 0);

        size_t n2 = db.scanAll(0, [&](const std::string& k, const std::string& v) {
            return collectAll(k, v, collected);
        });
        CHECK(n2 == 0);

        auto topRes = db.top(10);
        CHECK(topRes.empty());

        cleanupDir(dir);
    }

    //==========================================================================
    // INTEGRATION-ТЕСТЫ (насыщенное состояние, реальные SSTable на диске)
    //==========================================================================

    // --- Насыщенное состояние: scanAll(0,cb) == все живые, top(k), limit, паритет ---
    {
        auto dir = makeTempDir("integration");
        Config cfg;
        cfg.setDataDirectory(dir);
        auto& lsmOpts = cfg.lsmOptions();
        lsmOpts.memtableSizeBytes = 8 * 1024;  // 8 КБ — малый, чтобы flush произошёл

        Database db(cfg);
        CHECK(db.open());

        const int TOTAL_KEYS = 200;
        const int UPSERT_FROM = 100;  // key_0100..key_0149 перезаписаны
        const int UPSERT_TO = 150;
        const int REMOVE_FROM = 160;  // key_0160..key_0179 удалены
        const int REMOVE_TO = 180;

        // Детерминированное вычисление эталона
        // Структура: map[key] = value (последняя операция), tombstone = удалён
        std::map<std::string, std::string> expectedLive;

        // Фаза 1: вставка всех ключей
        for (int i = 0; i < TOTAL_KEYS; ++i) {
            std::string k = keyN(i);
            std::string v = "val_" + std::to_string(i);
            CHECK(db.put(k, v));
            expectedLive[k] = v;
        }

        // Фаза 2: upsert для подмножества (latest-wins)
        for (int i = UPSERT_FROM; i < UPSERT_TO; ++i) {
            std::string k = keyN(i);
            std::string v = "updated_" + std::to_string(i);
            CHECK(db.put(k, v));
            expectedLive[k] = v;
        }

        // Фаза 3: remove для подмножества
        for (int i = REMOVE_FROM; i < REMOVE_TO; ++i) {
            std::string k = keyN(i);
            CHECK(db.remove(k));
            expectedLive.erase(k);
        }

        // Принудительный flush, чтобы данные попали на диск
        CHECK(db.flush());

        // Ещё вставим данные после flush — они останутся в active MemTable
        const int POST_FLUSH_KEYS = 50;
        for (int i = TOTAL_KEYS; i < TOTAL_KEYS + POST_FLUSH_KEYS; ++i) {
            std::string k = keyN(i);
            std::string v = "post_flush_" + std::to_string(i);
            CHECK(db.put(k, v));
            expectedLive[k] = v;
        }

        // Опционально compact
        CHECK(db.compact());

        // Формируем эталонный вектор (отсортирован, т.к. std::map ordered)
        std::vector<std::pair<std::string,std::string>> expectedVec(
            expectedLive.begin(), expectedLive.end());
        size_t liveCount = expectedVec.size();
        CHECK(liveCount > 0);

        // --- scanAll(0,cb) → все живые ключи, каждый один раз, ASC ---
        {
            std::vector<std::pair<std::string,std::string>> collected;
            size_t n = db.scanAll(0, [&](const std::string& k, const std::string& v) {
                return collectAll(k, v, collected);
            });
            CHECK(n == liveCount);
            CHECK(collected.size() == liveCount);
            // Проверяем порядок и содержимое
            bool orderOk = true;
            bool contentOk = true;
            for (size_t i = 0; i < collected.size(); ++i) {
                if (i > 0 && collected[i].first <= collected[i-1].first) {
                    orderOk = false;
                }
                if (collected[i] != expectedVec[i]) {
                    contentOk = false;
                }
            }
            CHECK(orderOk);
            CHECK(contentOk);
        }

        // --- top(0) → пусто ---
        {
            auto res = db.top(0);
            CHECK(res.empty());
        }

        // --- top(k) при 0 < k < liveCount → первые k ---
        {
            size_t k = 5;
            auto res = db.top(k);
            CHECK(res.size() == k);
            for (size_t i = 0; i < k; ++i) {
                CHECK(res[i] == expectedVec[i]);
            }
        }

        // --- top(k) при k >= liveCount → все live ---
        {
            size_t k = liveCount + 100;
            auto res = db.top(k);
            CHECK(res.size() == liveCount);
            for (size_t i = 0; i < liveCount; ++i) {
                CHECK(res[i] == expectedVec[i]);
            }
        }

        // --- scanAll(k,cb) при 0 < k < liveCount → ровно k, return==k ---
        {
            size_t k = 10;
            std::vector<std::pair<std::string,std::string>> collected;
            size_t n = db.scanAll(k, [&](const std::string& key, const std::string& val) {
                return collectAll(key, val, collected);
            });
            CHECK(n == k);
            CHECK(collected.size() == k);
            for (size_t i = 0; i < k; ++i) {
                CHECK(collected[i] == expectedVec[i]);
            }
        }

        // --- scanAll(k,cb) при k >= liveCount → все live, return==liveCount ---
        {
            size_t k = liveCount + 50;
            std::vector<std::pair<std::string,std::string>> collected;
            size_t n = db.scanAll(k, [&](const std::string& key, const std::string& val) {
                return collectAll(key, val, collected);
            });
            CHECK(n == liveCount);
            CHECK(collected.size() == liveCount);
        }

        // --- Паритет (§5): scanAll(0,cb) == scan(firstKey, lastKey, 0) ---
        // Эталон: вектор из scanAll(0,cb) побитово равен scan(firstKey, lastKey, 0)
        {
            std::string firstKey = expectedVec.front().first;
            std::string lastKey = expectedVec.back().first;

            std::vector<std::pair<std::string,std::string>> viaStream;
            db.scanAll(0, [&](const std::string& k, const std::string& v) {
                return collectAll(k, v, viaStream);
            });

            auto viaScan = db.scan(firstKey, lastKey, 0);

            CHECK(viaStream.size() == viaScan.size());
            CHECK(viaStream == viaScan);
        }

        db.close();
        cleanupDir(dir);
    }

    std::cout << "test_scan_stream: passed=" << g_passed
              << " failed=" << g_failed << "\n";
    return g_failed == 0 ? 0 : 1;
}