2 files changed, 95 insertions, 4 deletions

diff --git a/src/CursorManager.cpp b/src/CursorManager.cpp
index 6a44767..91a0183 100644
--- a/src/CursorManager.cpp
+++ b/src/CursorManager.cpp
@@ -1,4 +1,5 @@
 #include "CursorManager.h"
+#include "ObjectManager.h"
 #include "TupleCodec.h"
 
 namespace nt
@@ -29,7 +30,20 @@ namespace nt
     {
         if (handle == nullptr || handle->object == nullptr)
             return nullptr;
-        if (handle->object->head->type->label != RELATION)
+
+        const auto label = handle->object->head->type->label;
+
+        if (label == EPHEMERAL_RELATION)
+        {
+            // Cursor is created exhausted — the JOIN operator writes args and
+            // resets the state before the first probe.
+            auto* c = new cursor();
+            c->handle = handle;
+            c->exhausted = true;
+            return c;
+        }
+
+        if (label != RELATION)
             return nullptr;
 
         auto* c = new cursor();
@@ -53,9 +67,21 @@ namespace nt
 
         if (c->exhausted) return nullptr;
 
+        if (c->handle->object->head->type->label == EPHEMERAL_RELATION)
+        {
+            auto* etype = static_cast<ObjectManager::ephemeral_object_type*>(
+                c->handle->object->head->type.get());
+            c->page = etype->generator(c->args, c->fetch_offset, PAGE_SIZE);
+            c->page_position = 0;
+            c->fetch_offset += c->page.size();
+            if (c->page.size() < PAGE_SIZE)
+                c->exhausted = true;
+            if (c->page.empty()) return nullptr;
+            return &c->page[c->page_position++];
+        }
+
         LoadPage(c, c->handle->object->head->path);
         if (c->page.empty()) return nullptr;
-
         return &c->page[c->page_position++];
     }
 
diff --git a/src/VM.cpp b/src/VM.cpp
index b5824d0..1f1ed7f 100644
--- a/src/VM.cpp
+++ b/src/VM.cpp
@@ -4,15 +4,80 @@ namespace nt
 {
     VM::VM(CursorManager& cursors) : cursors_(cursors) {}
 
+    std::vector<std::string> VM::ResolveArgs(const std::vector<PathArg>& tmpl, Tuple* from)
+    {
+        std::vector<std::string> resolved;
+        resolved.reserve(tmpl.size());
+        for (const auto& arg : tmpl)
+        {
+            if (arg.kind == PathArg::Kind::Const)
+            {
+                resolved.push_back(arg.name);
+                continue;
+            }
+            std::string val;
+            for (const auto& attr : from->attrs())
+                if (attr.name == arg.name) { val = attr.value; break; }
+            resolved.push_back(std::move(val));
+        }
+        return resolved;
+    }
+
+    void VM::ResetInner(CursorManager::cursor* c, std::vector<std::string> args)
+    {
+        c->args          = std::move(args);
+        c->page.clear();
+        c->page_position = 0;
+        c->fetch_offset  = 0;
+        c->exhausted     = false;
+    }
+
+    Tuple* VM::MergeInto(PlanNode* node, Tuple* left, Tuple* right)
+    {
+        std::vector<Attribute> merged;
+        for (const auto& a : left->attrs())  merged.push_back(a);
+        for (const auto& a : right->attrs()) merged.push_back(a);
+        node->join_buffer.emplace(std::move(merged));
+        return &*node->join_buffer;
+    }
+
     Tuple* VM::Next(PlanNode* node)
     {
-        if (node == nullptr)
-            return nullptr;
+        if (node == nullptr) return nullptr;
 
         switch (node->op)
         {
             case PlanNode::Op::SCAN:
                 return cursors_.Next(node->scan_cursor);
+
+            case PlanNode::Op::JOIN:
+            {
+                while (true)
+                {
+                    if (node->join_left == nullptr)
+                    {
+                        node->join_left = Next(node->left);
+                        if (node->join_left == nullptr) return nullptr;
+
+                        auto args = ResolveArgs(node->right->scan_args, node->join_left);
+                        ResetInner(node->right->scan_cursor, std::move(args));
+                    }
+
+                    Tuple* right = Next(node->right);
+                    if (right != nullptr)
+                        return MergeInto(node, node->join_left, right);
+
+                    node->join_left = nullptr;
+                }
+            }
+
+            case PlanNode::Op::TAKE:
+            {
+                if (node->take_count >= node->take_limit) return nullptr;
+                Tuple* t = Next(node->left);
+                if (t) ++node->take_count;
+                return t;
+            }
         }
 
         return nullptr;