1 files changed, 40 insertions, 24 deletions

diff --git a/src/CursorManager.cpp b/src/CursorManager.cpp
index f80ac54..6a44767 100644
--- a/src/CursorManager.cpp
+++ b/src/CursorManager.cpp
@@ -1,50 +1,66 @@
 #include "CursorManager.h"
-
-#include <numeric>
+#include "TupleCodec.h"
 
 namespace nt
 {
-    static std::string JoinPath(const std::vector<std::string>& path)
+    CursorManager::CursorManager(IStorageBackend& backend)
+        : backend_(backend)
+    {}
+
+    void CursorManager::LoadPage(cursor* c, const std::vector<std::string>& path)
     {
-        return std::accumulate(
-            std::next(path.begin()), path.end(), path.front(),
-            [](std::string a, const std::string& b) { return std::move(a) + "/" + b; });
+        auto hashes = backend_.TupleHashes(path, c->fetch_offset, PAGE_SIZE);
+        c->page.clear();
+        c->page_position = 0;
+
+        for (const auto& hash : hashes)
+        {
+            auto bytes = backend_.Get(hash);
+            if (!bytes) continue;
+            c->page.emplace_back(TupleCodec::Deserialize(*bytes));
+        }
+
+        c->fetch_offset += hashes.size();
+        if (hashes.size() < PAGE_SIZE)
+            c->exhausted = true;
     }
 
     CursorManager::cursor* CursorManager::Open(HandlerManager::handle* handle)
     {
         if (handle == nullptr || handle->object == nullptr)
             return nullptr;
-
         if (handle->object->head->type->label != RELATION)
             return nullptr;
 
-        const std::string key = JoinPath(handle->object->head->path);
-        auto it = mock_store_.find(key);
-        if (it == mock_store_.end())
-            return nullptr;
-
         auto* c = new cursor();
         c->handle = handle;
-        for (const auto& row : it->second)
-            c->tuples.emplace_back(row);
-        return c;
-    }
+        LoadPage(c, handle->object->head->path);
 
-    Tuple* CursorManager::Next(cursor* cursor)
-    {
-        if (cursor == nullptr || cursor->position >= cursor->tuples.size())
+        if (c->page.empty())
+        {
+            delete c;
             return nullptr;
-        return &cursor->tuples[cursor->position++];
+        }
+        return c;
     }
 
-    void CursorManager::Close(cursor* cursor)
+    Tuple* CursorManager::Next(cursor* c)
     {
-        delete cursor;
+        if (c == nullptr) return nullptr;
+
+        if (c->page_position < c->page.size())
+            return &c->page[c->page_position++];
+
+        if (c->exhausted) return nullptr;
+
+        LoadPage(c, c->handle->object->head->path);
+        if (c->page.empty()) return nullptr;
+
+        return &c->page[c->page_position++];
     }
 
-    void CursorManager::MockInsert(std::vector<std::string> path, std::vector<Attribute> tuple)
+    void CursorManager::Close(cursor* c)
     {
-        mock_store_[JoinPath(path)].push_back(std::move(tuple));
+        delete c;
     }
 }