add nested loop join strategy to VM and tests

11 files changed, 886 insertions, 16 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 97301cc..8343a25 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -13,6 +13,7 @@ project ("RelationalNT")
 
 find_package(unofficial-sqlite3 CONFIG REQUIRED)
 find_path(PICOSHA2_INCLUDE_DIRS "picosha2.h")
+find_package(Catch2 3 REQUIRED)
 
 # Add source to this project's executable.
 add_executable (RelationalNT
@@ -40,4 +41,30 @@ endif()
 target_link_libraries(RelationalNT PRIVATE unofficial::sqlite3::sqlite3)
 target_include_directories(RelationalNT PRIVATE ${PICOSHA2_INCLUDE_DIRS})
 
-# TODO: Add tests and install targets if needed.
+enable_testing()
+
+set(RNT_CORE_SOURCES
+    "src/ObjectManager.cpp"
+    "src/PermissionsManager.cpp"
+    "src/IdentityManager.cpp"
+    "src/LifecycleManager.cpp"
+    "src/HandlerManager.cpp"
+    "src/CursorManager.cpp"
+    "src/TupleCodec.cpp"
+    "src/InMemoryBackend.cpp"
+    "src/NamespaceReferenceManager.cpp"
+    "src/VM.cpp"
+)
+
+add_executable(RNT_tests
+    "test/VM.cpp"
+    ${RNT_CORE_SOURCES}
+)
+target_include_directories(RNT_tests PRIVATE "include")
+target_include_directories(RNT_tests PRIVATE ${PICOSHA2_INCLUDE_DIRS})
+target_compile_definitions(RNT_tests PRIVATE NT_CONSOLE_APP)
+target_link_libraries(RNT_tests PRIVATE Catch2::Catch2WithMain)
+set_property(TARGET RNT_tests PROPERTY CXX_STANDARD 20)
+
+include(Catch)
+catch_discover_tests(RNT_tests DISCOVERY_MODE PRE_TEST)
diff --git a/docs/join-strategies.md b/docs/join-strategies.md
new file mode 100644
index 0000000..049aad7
--- /dev/null
+++ b/docs/join-strategies.md
@@ -0,0 +1,305 @@
+# Join Strategies in the Tarski FOL Runtime
+
+This document covers the join operators available or planned for the Tarski
+Volcano engine, their memory and I/O characteristics, and the role that backend
+indexes play in making large-relation joins practical.
+
+---
+
+## Current implementation: nested-loop join
+
+The current `JOIN` node is a classic nested-loop join.
+
+```
+for each outer (left) tuple L:
+    reset inner (right) cursor to offset 0
+    for each inner tuple R:
+        if join condition holds: emit merge(L, R)
+```
+
+**Memory**: O(1). The left cursor holds one page (one tuple at `PAGE_SIZE = 1`).
+The right cursor also holds one page at a time; `ResetInner` clears the page and
+resets `fetch_offset` to 0 so the backend re-reads from the start for each outer
+tuple. Neither side is ever materialized in full.
+
+**I/O**: O(n × m) backend reads. The right relation is scanned once per outer
+tuple, so a billion-row right side costs a billion backend reads per outer tuple.
+
+**Best case**: ephemeral relations (builtins like `eq`, `lt`, etc.) on the right
+side. These run the generator function in O(1) with no backend I/O. The join
+degenerates into a streamed filter.
+
+**Worst case**: two large stored relations with no index. Acceptable only when
+one side is small, or when the right side fits in the OS page cache.
+
+---
+
+## Index nested-loop join
+
+The most practical improvement. If Sakura exposes a B+ tree index on the join
+attribute of the right relation, the inner loop becomes a point lookup instead
+of a full scan.
+
+```
+for each outer tuple L:
+    key = L[join_attr]
+    R = backend.Lookup(right_relation, join_attr, key)   // O(log m)
+    if R exists: emit merge(L, R)
+```
+
+**Memory**: O(1). Same as nested-loop — no buffering on either side.
+
+**I/O**: O(n log m). Each outer tuple triggers one B+ tree traversal on the right.
+
+**Backend requirement**: `IStorageBackend` would need a new method:
+
+```cpp
+virtual std::vector<std::string> TupleHashesByKey(
+    const std::vector<std::string>& relation_path,
+    const std::string& attribute,
+    const std::string& value,
+    std::size_t offset,
+    std::size_t limit) = 0;
+```
+
+Sakura already maintains a per-relation B+ tree of tuple hashes. An additional
+B+ tree keyed by attribute value is the natural extension and mirrors how most
+production storage engines implement secondary indexes.
+
+This is the correct long-term default join strategy for stored × stored joins.
+
+---
+
+## Sort-merge join
+
+Both inputs are scanned once in sorted order, advancing a cursor on each side
+in lock-step.
+
+```
+L = Next(left)
+R = Next(right)
+while L and R:
+    if L[key] == R[key]: emit merge(L, R); advance both
+    if L[key] <  R[key]: L = Next(left)
+    if L[key] >  R[key]: R = Next(right)
+```
+
+**Memory**: O(1) if both inputs arrive pre-sorted. The Volcano model handles
+page boundaries transparently — each `Next()` call fetches the next page from
+the backend when the current page is exhausted, independent of the other cursor.
+There is no synchronization problem at the page level; the operator logic only
+ever holds two tuples at a time.
+
+**I/O**: O(n + m). Each relation is scanned exactly once.
+
+**Requirement**: both inputs must be sorted on the join attribute before the
+merge step. This means either:
+- The backend returns tuples in sorted attribute order (sorted scan support), or
+- A `SORT` operator precedes each input in the plan tree.
+
+Sorting an unsorted relation requires O(n log n) time and — unless done via an
+external merge sort that stays within a fixed memory budget — O(n) memory.
+Sakura's B+ tree already imposes an ordering on stored tuple hashes; whether
+that ordering corresponds to an attribute value ordering depends on the index
+structure.
+
+---
+
+## Hash join
+
+Build a hash table from the smaller relation, probe with the larger.
+
+```
+phase 1 — build:  for each R: insert R into hash_table[R[key]]
+phase 2 — probe:  for each L: for each match in hash_table[L[key]]: emit merge(L, R)
+```
+
+**Memory**: O(min(n, m)). The build side must fit entirely in memory. Fails
+under memory pressure for large relations.
+
+**I/O**: O(n + m) — each relation scanned once.
+
+This is the right choice when the build side is small and fits in RAM. Not
+suitable as a general-purpose operator in a constant-memory runtime.
+
+---
+
+## Grace hash join (partitioned hash join)
+
+An extension of hash join that handles larger-than-memory relations by
+partitioning both sides into buckets that individually fit in memory.
+
+```
+phase 1 — partition: hash each side into k buckets written to temporary storage
+phase 2 — join:      for each bucket pair (left_i, right_i): in-memory hash join
+```
+
+**Memory**: O(√(n + m)) with optimal bucket count. Bounded but not O(1).
+
+**I/O**: O(3(n + m)) — three full passes over the data.
+
+Requires a writable temporary storage area (spill buffer). Relevant only if
+memory pressure prevents a standard hash join and sort-merge is not available.
+Not planned for the near term.
+
+---
+
+## Symmetric hash join
+
+Builds hash tables for both sides simultaneously as tuples arrive from either
+input. Suitable for streaming or pipelined inputs where one side may block
+before producing all its tuples.
+
+**Memory**: O(n + m) in the worst case — both hash tables may grow to full size.
+
+Not planned. Mentioned because Sakura streams tuples lazily; if the OCaml side
+ever introduces back-pressure on one relation while the other produces freely,
+symmetric hashing becomes relevant.
+
+---
+
+## LeapFrog TrieJoin
+
+LeapFrog TrieJoin (Veldhuizen, 2012) is the state of the art for multi-way
+joins in logic programming and Datalog engines. It is **worst-case optimal**:
+it runs in time proportional to the AGM bound — the tight theoretical upper
+bound on the output size of a conjunctive query given the sizes of its input
+relations. Traditional pairwise binary joins applied sequentially can produce
+intermediate results that exceed this bound, making them suboptimal for cyclic
+or highly connected queries.
+
+### How it works
+
+Rather than joining two relations at a time, LeapFrog TrieJoin joins all
+relations in a query **simultaneously**. Each relation is represented as a
+trie — a sorted, prefix-keyed index over its attribute values. One cursor per
+relation is maintained, all pointing into their respective tries at the same
+attribute depth.
+
+The algorithm iterates over one shared variable at a time. At each step:
+
+1. Find the relation whose cursor currently points to the **maximum** value
+   among all cursors.
+2. **Seek** all other cursors forward to that maximum (or the next value ≥ it).
+3. If all cursors agree on the same value: emit the binding and descend to the
+   next variable.
+4. If a cursor cannot find a value ≥ the maximum: backtrack and advance the
+   previous variable.
+
+The seek operation is what makes this efficient. Instead of scanning linearly
+through values that cannot match, a cursor jumps directly to the next candidate
+using the sorted order of the trie. No intermediate results are materialized.
+
+**Memory**: O(k) where k is the number of relations — one cursor per relation,
+plus one tuple per active binding level. Constant with respect to relation size.
+
+**I/O**: O(n^τ\*) where τ\* is the fractional edge cover of the query
+hypergraph. For acyclic joins this matches the Yannakakis bound; for cyclic
+joins it beats any binary join plan.
+
+### Why this fits the Tarski/Karuta model
+
+Karuta is WAM-based. The WAM already operates on term tries during unification,
+advancing through sorted clause indexes when resolving goals. LeapFrog TrieJoin
+is the relational analogue: the trie structure and the seek-based iteration
+pattern are the same. When Sakura emits a plan to Tarski, the multi-way join
+over several relations maps directly onto a LeapFrog node whose cursors are
+driven by the same pull model as every other Volcano operator.
+
+### Current state of Sakura
+
+Reading `lib/algebra.ml`, the current join implementation materializes the
+entire right-hand relation into memory before beginning the left-hand scan:
+
+```ocaml
+let right_tuples = drain right_gen in   (* ← full materialization *)
+List.filter_map (fun rt -> ...) right_tuples   (* ← linear scan per left tuple *)
+```
+
+This is a documented TODO in the codebase. Beyond the memory cost, it is a
+strictly binary join — three or more relations require multiple sequential
+passes with growing intermediate materialization.
+
+The generator type (`int option -> result`) is forward-only and position-counted.
+It has no seek primitive: you cannot ask a generator to jump to the next value
+≥ a target without scanning every value in between.
+
+Sakura does have a B+ tree binding (`lib/bplustree.ml`, FFI to a C library) but
+it is currently wired only to the KV storage layer for tuple hash persistence,
+not to any per-attribute ordered index.
+
+### What would need to change in Sakura
+
+**1. Per-attribute sorted indexes**
+
+Each stored relation needs a secondary index mapping attribute values to tuple
+hashes, maintained in sorted order. The existing B+ tree binding is the natural
+substrate for this — it already supports ordered key storage and would expose
+range traversal and seek as cursor operations.
+
+The index structure per relation would be:
+
+```
+relation_path / attribute_name → B+ tree of (value, tuple_hash_set)
+```
+
+**2. A seek-capable cursor type**
+
+The generator type needs to be extended (or complemented) with a cursor
+abstraction that supports:
+
+- `current()` — the value at the current position
+- `seek(target)` — advance to the first value ≥ target; return whether one exists
+- `next()` — advance by one step
+
+A generator that only supports `next()` cannot implement LeapFrog without
+scanning every skipped value. The seek operation is the essential primitive.
+
+**3. A multi-way join operator in `algebra.ml`**
+
+The current `equijoin` function is binary. LeapFrog TrieJoin takes an arbitrary
+list of relations and a variable ordering. A new `leapfrog_join` function would
+replace the sequential application of `equijoin` for multi-relation conjunctive
+queries.
+
+### What would need to change in Tarski (this codebase)
+
+On the C++ side, the `CursorManager` would need a `Seek` operation alongside
+`Next`:
+
+```cpp
+/** Advances the cursor to the first tuple whose join-attribute value is ≥ target.
+ *  Returns nullptr if no such tuple exists (relation exhausted). */
+Tuple* Seek(cursor* c, const std::string& attribute, const std::string& target);
+```
+
+The plan tree would gain a `LEAPFROG` node holding N child SCAN cursors and a
+variable ordering list. `VM::Next` on a `LEAPFROG` node drives the seek loop
+across all cursors rather than delegating to nested-loop iteration.
+
+---
+
+## Summary
+
+| Strategy | Memory | I/O | Requires |
+|---|---|---|---|
+| Nested-loop (current) | O(1) | O(n × m) | Nothing |
+| Index nested-loop | O(1) | O(n log m) | Per-attribute B+ tree index |
+| Sort-merge | O(1)* | O(n + m) | Pre-sorted inputs |
+| Hash join | O(min(n,m)) | O(n + m) | Build side fits in RAM |
+| Grace hash join | O(√(n+m)) | O(3(n+m)) | Spill storage |
+| LeapFrog TrieJoin | O(k) | O(n^τ\*) | Seek-capable trie cursors |
+
+\* O(1) only if both inputs arrive sorted. Sorting in-place requires O(n) memory.
+
+The practical roadmap for this runtime:
+
+1. **Now**: nested-loop covers all ephemeral-relation joins in O(1) memory.
+2. **Next**: index nested-loop once Sakura exposes per-attribute B+ tree reads —
+   covers the large stored × stored binary join case without buffering.
+3. **Target**: LeapFrog TrieJoin once Sakura adds seek-capable cursors and
+   per-attribute sorted indexes. This replaces both index nested-loop and
+   sort-merge for multi-relation conjunctive queries and is the correct
+   long-term join primitive for a Datalog/WAM-oriented engine.
+4. **Only if needed**: hash join with an explicit memory budget, for ad-hoc
+   joins where no usable index exists.
diff --git a/docs/tarski-sakura-separation.md b/docs/tarski-sakura-separation.md
new file mode 100644
index 0000000..a33e2e3
--- /dev/null
+++ b/docs/tarski-sakura-separation.md
@@ -0,0 +1,98 @@
+# Tarski / Sakura Separation of Concerns
+
+This document describes how query analysis responsibilities are divided between
+the **Sakura** OCaml frontend and the **Tarski** FOL runtime inside the C++ VM.
+
+## The Two Runtimes
+
+The VM hosts two distinct execution cores:
+
+| Core | Language | Responsibility |
+|---|---|---|
+| **Tarski** | C++ (this repo) | Mechanical FOL plan execution via Volcano pull model |
+| **Sakura** | OCaml | Query parsing, cardinality analysis, plan emission, physical storage |
+
+Sakura is the authoritative source of data and of query semantics. The SQLite
+backend in this repository exists only for isolated testing; in production,
+CursorManager streams data from Sakura over a well-defined interface.
+
+## What Sakura Is Responsible For
+
+### Cardinality analysis and constraint propagation
+
+Every relation in Sakura carries a cardinality tag: `Finite`, `ConstrainedFinite`,
+`AlephZero`, or `Continuum`. When a query joins multiple relations, Sakura
+propagates these tags through the plan to determine the cardinality of the result.
+
+This analysis is non-trivial. Consider:
+
+```karuta
+natural[A], natural[B], natural[C], plus[A,B,C], eq[C, 10].
+```
+
+Each of `natural`, `plus`, and `eq` is individually `AlephZero`. But Sakura can
+determine that binding `C = 10` via `eq` constrains `plus[A,B,10]` to a finite
+set of natural-number pairs, making the overall result finite. Replacing
+`natural` with `integer` changes the result to infinite, because negative pairs
+also satisfy `plus[A,B,10]`.
+
+Tarski does not repeat this analysis. It receives a plan that is already
+annotated with cardinality metadata and executes it mechanically.
+
+### Termination guarantees
+
+If a query produces an infinite result set, Sakura adds a `TAKE` node to the
+plan before emission. Tarski enforces the `TAKE` limit but makes no independent
+judgement about whether a plan terminates.
+
+### Join ordering
+
+Sakura orders the children of each JOIN so that variables are bound from the
+outer (left) side before the inner (right) side is probed. This ensures that
+builtin relations — which may have `AlephZero` cardinality when arguments are
+unbound — are always probed with fully or partially bound arguments, enabling
+O(1) membership validation rather than unbounded enumeration.
+
+### Physical storage
+
+Sakura owns the physical storage layer. CursorManager pulls paginated tuple
+streams from Sakura; the cursor abstraction hides whether the underlying data
+comes from a stored relation or an ephemeral builtin generator.
+
+## What Tarski Is Responsible For
+
+### Mechanical plan execution
+
+Tarski receives a complete, annotated plan tree from Sakura and executes it via
+the Volcano pull model: `VM::Next(PlanNode*)` propagates demand from the root
+to the leaves, returning one tuple at a time.
+
+### Cardinality-aware execution strategy
+
+Each `PlanNode` carries the cardinality tag Sakura attached at plan time. Tarski
+uses this tag only to choose an execution strategy — never to reason about
+termination:
+
+- **Finite / ConstrainedFinite inner side** → safe to materialize for
+  nested-loop join.
+- **AlephZero inner side, all args bound** → probe via `membership_criteria`
+  in O(1); the generator is never called.
+- **AlephZero inner side, unbound args** → should not occur; Sakura must have
+  added a `TAKE` node or reordered the join. Tarski may assert in debug builds.
+
+### TAKE enforcement
+
+`TAKE` is a first-class node in the plan tree. When `VM::Next` passes through a
+`TAKE` node it decrements a counter and returns `nullptr` once the limit is
+reached, regardless of what the subtree would produce.
+
+## Why This Division Is Correct
+
+The cardinality analysis described above requires understanding the **semantics**
+of each relation — what `plus` means over naturals versus integers, how `eq`
+interacts with a functional relation, and so on. That knowledge lives in Sakura.
+Duplicating it in Tarski would couple two layers that must remain independent and
+would place type-level reasoning inside a mechanical executor.
+
+Tarski's contract is simple: given a safe plan, produce correct tuples. Sakura's
+contract is: emit only safe plans.
diff --git a/include/CursorManager.h b/include/CursorManager.h
index 5636d8d..092229c 100644
--- a/include/CursorManager.h
+++ b/include/CursorManager.h
@@ -46,6 +46,12 @@ namespace nt
          */
         struct cursor {
             HandlerManager::handle* handle = nullptr;
+            /**
+             * Bound argument values for EPHEMERAL_RELATION cursors.
+             * Written by the JOIN operator before each probe; ignored for
+             * stored RELATION cursors.
+             */
+            std::vector<std::string> args;
             std::vector<Tuple> page;
             std::size_t page_position = 0;
             std::size_t fetch_offset  = 0;   // next TupleHashes() call starts here
diff --git a/include/ObjectManager.h b/include/ObjectManager.h
index 4b537f7..32b3ae9 100644
--- a/include/ObjectManager.h
+++ b/include/ObjectManager.h
@@ -4,6 +4,7 @@
 #include "Types.h"
 
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <set>
 #include <string>
@@ -73,6 +74,44 @@ namespace nt
              * @todo Wire into LifecycleManager::Contention.
              */
             bool exclusive = false;
+
+            virtual ~object_type() = default;
+        };
+
+        /**
+         * @brief Object type descriptor for EPHEMERAL_RELATION objects.
+         *
+         * An ephemeral relation has no physical storage. Its tuples are produced
+         * on demand by a generator function. Cardinality may be finite (e.g. a
+         * projected stored relation) or AlephZero (e.g. the eq/lt builtins).
+         *
+         * The generator receives the bound argument values that were written into
+         * the cursor by the JOIN operator before each probe, together with a
+         * pagination offset and limit. It must return at most `limit` tuples
+         * starting at logical position `offset`.
+         *
+         * For membership-probe builtins (all args ground), the generator returns
+         * at most one tuple. For enumeration over AlephZero relations the generator
+         * maps `offset` to the appropriate pair via a bijective enumeration scheme
+         * (e.g. Cantor pairing); callers must bound such scans with a TAKE node.
+         */
+        struct NT_API ephemeral_object_type : object_type
+        {
+            enum class Cardinality { Finite, ConstrainedFinite, AlephZero, Continuum };
+
+            /**
+             * @brief Produces a page of tuples for the given bound arguments.
+             * @param args   Bound argument values written by the JOIN before probing.
+             * @param offset Zero-based logical tuple offset (for pagination).
+             * @param limit  Maximum number of tuples to return.
+             */
+            using Generator = std::function<std::vector<Tuple>(
+                const std::vector<std::string>& args,
+                std::size_t offset,
+                std::size_t limit)>;
+
+            Cardinality cardinality;
+            Generator   generator;
         };
 
         /**
diff --git a/include/Types.h b/include/Types.h
index c6f4b9f..3e38829 100644
--- a/include/Types.h
+++ b/include/Types.h
@@ -12,12 +12,19 @@
 enum OBJECT_TYPE {
     /** A database snapshot grouping relations and tuples. */
     MULTIGROUP,
-    /** A relation object. */
+    /** A relation object backed by physical storage. */
     RELATION,
     /** An attribute object (schema metadata). */
     ATTRIBUTE,
     /** A transaction object. */
-    TRANSACTION
+    TRANSACTION,
+    /**
+     * A relation whose tuples are produced on demand by a generator function
+     * rather than read from physical storage. Cardinality may be finite (e.g.
+     * a projected stored relation) or AlephZero (e.g. the eq builtin). The
+     * object_type for this label must be an ephemeral_object_type.
+     */
+    EPHEMERAL_RELATION
 };
 
 /** @brief Operations that may be supported by an object type. */
@@ -76,6 +83,9 @@ namespace nt
 
         void Reset() { position_ = 0; }
 
+        /** @brief Direct read-only access to all attributes without affecting position. */
+        const std::vector<Attribute>& attrs() const { return attributes_; }
+
     private:
         std::vector<Attribute> attributes_;
         size_t position_ = 0;
diff --git a/include/VM.h b/include/VM.h
index d6402b3..c560414 100644
--- a/include/VM.h
+++ b/include/VM.h
@@ -4,6 +4,10 @@
 #include "CursorManager.h"
 #include "Types.h"
 
+#include <optional>
+#include <string>
+#include <vector>
+
 /**
  * @file VM.h
  * @brief Declares the logic execution boundary for relational and higher-order workloads.
@@ -47,24 +51,87 @@ namespace nt
      *   snapshots pinned through LifecycleManager::Monitor only for the minimum
      *   time required.
      */
+
+    /**
+     * @brief One segment of a parameterized relation path in a SCAN node.
+     *
+     * Var segments are resolved at runtime from the outer tuple of a JOIN:
+     * the JOIN writes the bound value into the cursor's args vector before
+     * each probe. Const segments are ground literals fixed at plan construction.
+     *
+     * The resolved args vector is passed to the ephemeral relation's generator
+     * (for EPHEMERAL_RELATION cursors) or ignored (for stored RELATION cursors,
+     * which need no per-probe parameterization).
+     */
+    struct NT_API PathArg
+    {
+        enum class Kind { Var, Const };
+
+        Kind        kind;
+        std::string name; /**< Attribute name to look up (Var), or literal value (Const). */
+
+        static PathArg Var(std::string attr)  { return { Kind::Var,   std::move(attr) }; }
+        static PathArg Const(std::string val) { return { Kind::Const, std::move(val) }; }
+    };
+
     /**
      * @brief A node in the Volcano operator tree executed by the Tarski (FOL) runtime.
      *
      * Each node represents one relational algebra operator. The tree is demand-driven:
      * VM::Next() propagates pull requests from root to leaves.
      *
-     * TODO: Add FILTER, PROJECT, and JOIN operators as the query layer grows.
+     * Supported operators:
+     * - SCAN  : pulls tuples from a pre-opened cursor (stored or ephemeral relation).
+     * - JOIN  : nested-loop join. For each outer (left) tuple, resolves scan_args on
+     *           the inner (right) SCAN, resets its cursor, and probes it. Works for
+     *           both stored relations (full re-scan of inner) and ephemeral relations
+     *           (O(1) membership probe when all args are bound).
+     * - TAKE  : passes at most take_limit tuples through, then returns nullptr. Use
+     *           this to bound scans over AlephZero ephemeral relations.
+     *
+     * TODO: Add PROJECT operator as the query layer grows.
      */
     struct NT_API PlanNode
     {
-        enum class Op { SCAN };
+        enum class Op { SCAN, JOIN, TAKE };
 
-        Op op;
-        PlanNode* left  = nullptr;  /**< Left child (binary operators). */
-        PlanNode* right = nullptr;  /**< Right child (binary operators). */
+        Op        op;
+        PlanNode* left  = nullptr;
+        PlanNode* right = nullptr;
 
-        /** Cursor to scan. Set when op == SCAN. */
+        /**
+         * SCAN: pre-opened cursor. For ephemeral relations this cursor is
+         * initially exhausted; the JOIN resets it with resolved args before
+         * each probe.
+         */
         CursorManager::cursor* scan_cursor = nullptr;
+
+        /**
+         * SCAN: argument template for parameterized relations.
+         * Var entries are resolved from the outer JOIN tuple; Const entries
+         * are used as-is. The resolved values are written into scan_cursor->args
+         * by the JOIN before it calls Next() on this node.
+         */
+        std::vector<PathArg> scan_args;
+
+        /** TAKE: maximum number of tuples to emit before returning nullptr. */
+        std::size_t take_limit = 0;
+        std::size_t take_count = 0; /**< Runtime counter; mutable during execution. */
+
+        /**
+         * JOIN runtime state.
+         *
+         * join_left holds the current outer tuple while probing the inner
+         * (right) side. The pointer is stable for the lifetime of one outer
+         * iteration because Next(left) is only called when join_left is reset
+         * to nullptr, which happens only after the inner side is exhausted.
+         *
+         * join_buffer holds the merged output tuple and is overwritten on each
+         * successful join. Callers must consume the returned pointer before
+         * calling Next() again — same contract as a cursor page pointer.
+         */
+        Tuple*               join_left   = nullptr;
+        std::optional<Tuple> join_buffer;
     };
 
     class NT_API VM
@@ -82,8 +149,13 @@ namespace nt
         /**
          * @brief Pulls the next tuple from the plan tree (Tarski/FOL core).
          *
-         * Propagates a Next() call down the operator tree. The call chain
-         * terminates at a SCAN node, which delegates to CursorManager.
+         * Propagates a Next() call down the operator tree:
+         * - SCAN  delegates to CursorManager::Next.
+         * - JOIN  drives the nested-loop: for each outer tuple it resolves
+         *         scan_args on the inner SCAN, resets the inner cursor, and
+         *         probes until a match is found or the inner is exhausted.
+         * - TAKE  counts emitted tuples and returns nullptr once the limit
+         *         is reached.
          *
          * @param node Root of the plan subtree to evaluate.
          * @return Next matching tuple, or nullptr when the plan is exhausted.
@@ -92,5 +164,19 @@ namespace nt
 
     private:
         CursorManager& cursors_;
+
+        /**
+         * Resolves a scan_args template against an outer tuple, producing the
+         * concrete args vector to write into the inner cursor before probing.
+         */
+        static std::vector<std::string> ResolveArgs(const std::vector<PathArg>& tmpl,
+                                                    Tuple* from);
+
+        /** Resets the inner cursor state and writes resolved args into it. */
+        static void ResetInner(CursorManager::cursor* c,
+                               std::vector<std::string> args);
+
+        /** Merges left and right tuple attributes into node->join_buffer. */
+        static Tuple* MergeInto(PlanNode* node, Tuple* left, Tuple* right);
     };
 }
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;
diff --git a/test/VM.cpp b/test/VM.cpp
new file mode 100644
index 0000000..ed98be0
--- /dev/null
+++ b/test/VM.cpp
@@ -0,0 +1,207 @@
+#include <catch2/catch_test_macros.hpp>
+
+#include "CursorManager.h"
+#include "HandlerManager.h"
+#include "IdentityManager.h"
+#include "InMemoryBackend.h"
+#include "LifecycleManager.h"
+#include "ObjectManager.h"
+#include "PermissionsManager.h"
+#include "TupleCodec.h"
+#include "VM.h"
+
+#include <algorithm>
+#include <string>
+#include <vector>
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+static void store_tuple(nt::InMemoryBackend& backend,
+                        const std::vector<std::string>& path,
+                        std::vector<nt::Attribute> attrs)
+{
+    auto bytes = nt::TupleCodec::Serialize(attrs);
+    auto hash  = backend.Put(std::move(bytes));
+    backend.LinkTuple(path, hash);
+}
+
+static std::unique_ptr<nt::ObjectManager::object_type> make_relation_type()
+{
+    auto t = std::make_unique<nt::ObjectManager::object_type>();
+    t->label      = RELATION;
+    t->disposable = false;
+    t->methods    = { OPEN, CLOSE };
+    return t;
+}
+
+// Generator for eq[A, B]: returns one tuple when A == B, nothing otherwise.
+// Treats both args as integers when parseable, strings otherwise.
+static std::unique_ptr<nt::ObjectManager::ephemeral_object_type> make_eq_type()
+{
+    auto t = std::make_unique<nt::ObjectManager::ephemeral_object_type>();
+    t->label       = EPHEMERAL_RELATION;
+    t->disposable  = false;
+    t->methods     = { OPEN, CLOSE };
+    t->cardinality = nt::ObjectManager::ephemeral_object_type::Cardinality::AlephZero;
+    t->generator   = [](const std::vector<std::string>& args,
+                        std::size_t offset,
+                        std::size_t limit) -> std::vector<nt::Tuple>
+    {
+        if (args.size() < 2 || offset > 0 || limit == 0) return {};
+        if (args[0] != args[1]) return {};
+        return { nt::Tuple({ { "left", args[0] }, { "right", args[1] } }) };
+    };
+    return t;
+}
+
+// ---------------------------------------------------------------------------
+// Fixture: shared manager stack
+// ---------------------------------------------------------------------------
+
+struct Fixture
+{
+    nt::InMemoryBackend  backend;
+    nt::ObjectManager    objects;
+    nt::PermissionsManager permissions;
+    nt::IdentityManager  identities;
+    nt::LifecycleManager lifecycles;
+    nt::HandlerManager   handler { objects, permissions, identities, lifecycles };
+    nt::CursorManager    cursors { backend };
+    int conn = 1;
+};
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+TEST_CASE("SCAN returns all tuples from a stored relation", "[scan]")
+{
+    Fixture f;
+    const std::vector<std::string> path = { "relations", "villager" };
+
+    f.objects.Register(path, std::make_unique<nt::ObjectManager::Relation>(), make_relation_type());
+    store_tuple(f.backend, path, { { "name", "John" }, { "age", "30" } });
+    store_tuple(f.backend, path, { { "name", "Peter"   }, { "age", "25" } });
+
+    auto* handle = f.handler.Open(path, &f.conn);
+    REQUIRE(handle != nullptr);
+    auto* cursor = f.cursors.Open(handle);
+    REQUIRE(cursor != nullptr);
+
+    nt::PlanNode plan;
+    plan.op          = nt::PlanNode::Op::SCAN;
+    plan.scan_cursor = cursor;
+
+    nt::VM vm(f.cursors);
+    int count = 0;
+    while (vm.Next(&plan)) ++count;
+
+    REQUIRE(count == 2);
+
+    f.cursors.Close(cursor);
+    f.handler.Close(handle);
+}
+
+TEST_CASE("JOIN filters a stored relation with eq ephemeral", "[join]")
+{
+    Fixture f;
+    const std::vector<std::string> people_path = { "relations", "people" };
+    const std::vector<std::string> eq_path     = { "builtins", "eq"     };
+
+    // Register stored relation: people(name, age)
+    f.objects.Register(people_path,
+                       std::make_unique<nt::ObjectManager::Relation>(),
+                       make_relation_type());
+
+    store_tuple(f.backend, people_path, { { "name", "John"  }, { "age", "30" } });
+    store_tuple(f.backend, people_path, { { "name", "Peter" }, { "age", "25" } });
+    store_tuple(f.backend, people_path, { { "name", "Jude"  }, { "age", "25" } });
+
+    // Register ephemeral eq relation
+    f.objects.Register(eq_path,
+                       std::make_unique<nt::ObjectManager::Relation>(),
+                       make_eq_type());
+
+    // Open handles and cursors
+    auto* people_handle = f.handler.Open(people_path, &f.conn);
+    REQUIRE(people_handle != nullptr);
+    auto* eq_handle = f.handler.Open(eq_path, &f.conn);
+    REQUIRE(eq_handle != nullptr);
+
+    auto* people_cursor = f.cursors.Open(people_handle);
+    REQUIRE(people_cursor != nullptr);
+    auto* eq_cursor = f.cursors.Open(eq_handle); // starts exhausted — JOIN resets it
+    REQUIRE(eq_cursor != nullptr);
+
+    // Plan: JOIN( SCAN(people), SCAN(eq[ Var("age"), Const("25") ]) )
+    nt::PlanNode right;
+    right.op          = nt::PlanNode::Op::SCAN;
+    right.scan_cursor = eq_cursor;
+    right.scan_args   = { nt::PathArg::Var("age"), nt::PathArg::Const("25") };
+
+    nt::PlanNode left;
+    left.op          = nt::PlanNode::Op::SCAN;
+    left.scan_cursor = people_cursor;
+
+    nt::PlanNode root;
+    root.op    = nt::PlanNode::Op::JOIN;
+    root.left  = &left;
+    root.right = &right;
+
+    // Execute and collect matched names
+    nt::VM vm(f.cursors);
+    std::vector<std::string> matched;
+    while (nt::Tuple* t = vm.Next(&root))
+    {
+        for (const auto& attr : t->attrs())
+            if (attr.name == "name")
+                matched.push_back(attr.value);
+    }
+
+    // Jude (age=25) and Peter (age=25) should match; Jogn (age=30) should not
+    REQUIRE(matched.size() == 2);
+    REQUIRE(std::find(matched.begin(), matched.end(), "Jude")  != matched.end());
+    REQUIRE(std::find(matched.begin(), matched.end(), "Peter") != matched.end());
+    REQUIRE(std::find(matched.begin(), matched.end(), "John")  == matched.end());
+
+    f.cursors.Close(people_cursor);
+    f.cursors.Close(eq_cursor);
+    f.handler.Close(people_handle);
+    f.handler.Close(eq_handle);
+}
+
+TEST_CASE("TAKE limits tuples emitted from a SCAN", "[take]")
+{
+    Fixture f;
+    const std::vector<std::string> path = { "relations", "items" };
+
+    f.objects.Register(path, std::make_unique<nt::ObjectManager::Relation>(), make_relation_type());
+    store_tuple(f.backend, path, { { "id", "1" } });
+    store_tuple(f.backend, path, { { "id", "2" } });
+    store_tuple(f.backend, path, { { "id", "3" } });
+
+    auto* handle = f.handler.Open(path, &f.conn);
+    REQUIRE(handle != nullptr);
+    auto* cursor = f.cursors.Open(handle);
+    REQUIRE(cursor != nullptr);
+
+    nt::PlanNode scan;
+    scan.op          = nt::PlanNode::Op::SCAN;
+    scan.scan_cursor = cursor;
+
+    nt::PlanNode take;
+    take.op         = nt::PlanNode::Op::TAKE;
+    take.left       = &scan;
+    take.take_limit = 2;
+
+    nt::VM vm(f.cursors);
+    int count = 0;
+    while (vm.Next(&take)) ++count;
+
+    REQUIRE(count == 2);
+
+    f.cursors.Close(cursor);
+    f.handler.Close(handle);
+}
diff --git a/vcpkg.json b/vcpkg.json
index afa340a..679fd2e 100644
--- a/vcpkg.json
+++ b/vcpkg.json
@@ -2,6 +2,7 @@
   "builtin-baseline": "99a97de2cb371449d4fb9dc970f2ac562d689ec2",
   "dependencies": [
     "sqlite3",
-    "picosha2"
+    "picosha2",
+    "catch2"
   ]
 }