Joins with provenance and lineage (#11)

8 files changed, 1376 insertions, 66 deletions

diff --git a/docs/features.org b/docs/features.org
index 945bfaf..98350c5 100644
--- a/docs/features.org
+++ b/docs/features.org
@@ -58,6 +58,18 @@
     - [X] Hash consistency validation
     - [X] Iterator protocol tests
     - [X] Retraction and truncate tests
+    - [X] Join operation tests
+    - [X] Provenance tracking tests
+
+*** DONE Attribute-Level Provenance Tracking
+    CLOSED: [2025-01-22]
+    - [X] Nested metadata structure (meta.provenance)
+    - [X] Track immediate source of each attribute
+    - [X] Always enabled - every tuple includes metadata
+    - [X] Automatic provenance merging in joins
+    - [X] Conflict resolution for duplicate attributes
+    - [X] Clean separation of data attributes and metadata
+    - [X] Foundation for tuple-level lineage tracking (implemented)
 
 ** Planned RM/T Extensions
 
@@ -146,15 +158,22 @@
     - [ ] Unrestricted concatenation
     - [ ] Basis for joins
 
-*** TODO Natural Join (⋈)
+*** DONE Natural Join (⋈)
+    CLOSED: [2025-01-22]
     Join on common attributes.
-    - [ ] Equi-join with duplicate removal
-    - [ ] Used in normalization
-
-*** TODO Theta Join (⋈θ)
+    - [X] Equi-join with duplicate removal (equijoin_iterator)
+    - [X] Nested loop join strategy
+    - [X] Attribute conflict resolution (right_ prefix)
+    - [X] Attribute-level provenance tracking
+    - [X] Comprehensive test coverage
+
+*** DONE Theta Join (⋈θ)
+    CLOSED: [2025-01-22]
     Join with arbitrary comparison.
-    - [ ] Support all theta operators
-    - [ ] Generalizes natural join
+    - [X] Support all theta operators via predicate functions
+    - [X] Generalizes natural join
+    - [X] Nested loop join with custom predicates
+    - [X] Provenance merging from both sides
 
 *** TODO Union (∪)
     Set union of compatible relations.
@@ -179,6 +198,145 @@
 
 ** Advanced Features
 
+*** DONE Tuple-Level Lineage Tracking
+    CLOSED: [2025-01-23]
+    Track complete operational history of each tuple through query pipelines.
+
+**** Status: IMPLEMENTED
+
+**** API Usage
+
+    Metadata tracking is ALWAYS enabled. All tuples include the 'meta' field:
+    #+BEGIN_SRC erlang
+    %% Simple query (metadata automatically included)
+    Iter = operations:get_tuples_iterator(DB, employees, #{}).
+
+    %% With constraints (metadata still included)
+    Iter = operations:get_tuples_iterator(DB, employees,
+        #{constraints => #{age => {gt, 30}}}).
+
+    %% Results always have:
+    %% #{id => 1, name => "Alice",
+    %%   meta => #{
+    %%     provenance => #{id => {employees, id}, name => {employees, name}},
+    %%     lineage => {base, employees}
+    %%   }}
+    #+END_SRC
+
+**** Provenance vs. Lineage
+
+    - *Provenance* (IMPLEMENTED): Tracks immediate source of each attribute
+      #+BEGIN_SRC erlang
+      meta => #{
+        provenance => #{
+          name => {employees, name},
+          dept_name => {departments, dept_name}
+        }
+      }
+      #+END_SRC
+
+    - *Lineage* (IMPLEMENTED): Tracks complete operation tree that created the tuple
+      #+BEGIN_SRC erlang
+      meta => #{
+        provenance => #{...},  % Attribute sources
+        lineage => {project, [name, dept_name],
+                     {join, dept_id,
+                       {select, fun(T) -> maps:get(age, T) > 30 end,
+                         {base, employees}},
+                       {base, departments}}}
+      }
+      #+END_SRC
+
+**** Type Definition
+
+    #+BEGIN_SRC erlang
+    -type lineage_op() ::
+        {base, atom()}                                    % Base relation
+      | {select, fun(), lineage_op()}                     % Filter operation
+      | {project, [atom()], lineage_op()}                 % Projection
+      | {join, atom(), lineage_op(), lineage_op()}        % Join on attribute
+      | {theta_join, fun(), lineage_op(), lineage_op()}   % Join with predicate
+      | {sort, fun(), lineage_op()}                       % Sort operation
+      | {take, pos_integer(), lineage_op()}               % Limit operation
+      | {aggregate, atom(), atom(), lineage_op()}.        % Aggregation (future)
+    #+END_SRC
+
+**** Use Cases
+
+    - [ ] *Updatable Views*: Propagate updates back to base relations
+      - Example: =UPDATE view SET name = "Bob"= → trace lineage → =UPDATE employees SET name = "Bob"=
+
+    - [ ] *Incremental View Maintenance*: Identify affected tuples when base data changes
+      - Example: If employees tuple changes, lineage identifies which view tuples need recomputation
+
+    - [ ] *Audit Trails*: Track how sensitive data was transformed
+      - Example: "This salary came from: employees → join(departments) → aggregate(avg)"
+
+    - [ ] *Query Explanation*: Debug complex pipelines
+      - Example: "Why is this tuple here?" → inspect lineage to see operations applied
+
+    - [ ] *Provenance Queries*: Query the lineage itself
+      - Example: "Show all tuples derived from employees.id=5"
+
+**** Implementation Plan
+
+    1. [X] Extend =tuple_metadata()= type with lineage field
+    2. [X] Add lineage construction to each iterator operator
+       - [X] select_iterator: ={select, Predicate, ChildLineage}=
+       - [X] project_iterator: ={project, Attributes, ChildLineage}=
+       - [X] equijoin_iterator: ={join, Attr, LeftLineage, RightLineage}=
+       - [X] theta_join_iterator: ={theta_join, Pred, LeftLineage, RightLineage}=
+       - [X] sort_iterator: ={sort, CompareFun, ChildLineage}=
+       - [X] take_iterator: ={take, N, ChildLineage}=
+    3. [X] Implement lineage merging in joins (combine operation trees)
+    4. [ ] Add lineage querying API for provenance analysis (future)
+    5. [ ] Use lineage for updatable view update propagation (future)
+    6. [X] Add comprehensive lineage tests
+
+**** Relation to Existing Code
+
+    Relation-level provenance already exists in =#relation.provenance=:
+    #+BEGIN_SRC erlang
+    -type provenance() ::
+        {base, atom()}
+      | {join, provenance(), provenance()}
+      | {select, provenance(), constraints()}
+      | {project, provenance(), [atom()]}
+      | {take, provenance(), pos_integer()}.
+    #+END_SRC
+
+    This tracks where *relations* come from. Lineage is the *tuple-level*
+    equivalent, tracking operational history of individual tuples through
+    query pipelines.
+
+**** Example: Complete Lineage Chain
+
+    #+BEGIN_SRC erlang
+    % Query: employees → select(age>30) → join(departments) → project([name, dept_name])
+
+    % Resulting tuple with full lineage:
+    #{name => "Alice",
+      dept_name => "Engineering",
+      meta => #{
+        provenance => #{
+          name => {employees, name},
+          dept_name => {departments, dept_name}
+        },
+        lineage => {
+          project, [name, dept_name],
+          {join, dept_id,
+            {select, fun(T) -> maps:get(age, T) > 30 end,
+              {base, employees}},
+            {base, departments}}}
+      }}
+
+    % From this lineage, we can:
+    % 1. Trace "Alice" back to employees.name
+    % 2. Know she passed the age>30 filter
+    % 3. See she was joined with departments on dept_id
+    % 4. Understand the projection removed other attributes
+    #+END_SRC
+
 *** TODO Null Value Handling
     Three-valued logic for incomplete information.
     - [ ] "Value unknown" null type
diff --git a/include/operations.hrl b/include/operations.hrl
index 6d481ad..93b4a74 100644
--- a/include/operations.hrl
+++ b/include/operations.hrl
@@ -65,9 +65,40 @@
 
 %%% Provenance Types (for updatable views)
 
+%% Relation-level provenance (tracks operation lineage)
 -type provenance() :: undefined
                     | {base, atom()}       % Base relation
                     | {join, provenance(), provenance()}
                     | {select, provenance(), constraints()}
                     | {project, provenance(), [atom()]}
                     | {take, provenance(), pos_integer()}.
+
+%% Attribute-level provenance (tracks data lineage for each attribute)
+%% Maps attribute name to its source (relation, original_attribute_name)
+-type attribute_provenance() :: #{atom() => {atom(), atom()}}.
+
+%% Tuple-level lineage (tracks complete operational history)
+%% Operation tree showing how the tuple was derived
+-type lineage_op() ::
+    {base, atom()}                                      % Base relation
+  | {select, fun(), lineage_op()}                       % Filter operation
+  | {project, [atom()], lineage_op()}                   % Projection
+  | {join, atom(), lineage_op(), lineage_op()}          % Join on attribute
+  | {theta_join, fun(), lineage_op(), lineage_op()}     % Join with predicate
+  | {sort, fun(), lineage_op()}                         % Sort operation
+  | {take, pos_integer(), lineage_op()}                 % Limit operation
+  | {aggregate, atom(), atom(), lineage_op()}.          % Aggregation (future)
+
+%% Tuple metadata (contains provenance, lineage, and other system-managed fields)
+-type tuple_metadata() :: #{
+    provenance => attribute_provenance(),
+    lineage => lineage_op()
+    % Future: timestamp, version, etc.
+}.
+
+%% Tuples with metadata use a nested 'meta' field:
+%% #{id => 1, name => "Alice",
+%%   meta => #{
+%%     provenance => #{id => {employees, id}, name => {employees, name}},
+%%     lineage => {select, fun(T) -> maps:get(age, T) > 30 end, {base, employees}}
+%%   }}
diff --git a/src/generators.erl b/src/generators.erl
index 1dc363c..9e682d3 100644
--- a/src/generators.erl
+++ b/src/generators.erl
@@ -90,12 +90,27 @@ naturals(Constraints) ->
     case extract_bounds(Constraints, value) of
         {range, Min, Max} when Min >= 0 ->
             make_range_generator(Min, Max, fun(N) -> #{value => N} end);
+        {range, Min, Max} when Min < 0, Max >= 0 ->
+            %% Negative min for naturals: adjust to start at 0
+            make_range_generator(0, Max, fun(N) -> #{value => N} end);
+        {range, _Min, Max} when Max < 0 ->
+            %% Entire range is negative: no naturals exist
+            fun(_) -> done end;
         {gte, Min} when Min >= 0 ->
             make_unbounded_generator(Min, fun(N) -> #{value => N} end);
+        {gte, Min} when Min < 0 ->
+            %% Negative lower bound for naturals: start at 0
+            make_unbounded_generator(0, fun(N) -> #{value => N} end);
         {lte, Max} when Max >= 0 ->
             make_range_generator(0, Max, fun(N) -> #{value => N} end);
+        {lte, Max} when Max < 0 ->
+            %% Upper bound is negative: no naturals exist
+            fun(_) -> done end;
         {eq, Value} when Value >= 0 ->
             make_singleton_generator(#{value => Value});
+        {eq, _Value} ->
+            %% Negative value: not a natural
+            fun(_) -> done end;
         {in, List} ->
             make_membership_generator(List, fun(N) -> #{value => N} end);
         unbounded ->
@@ -474,14 +489,19 @@ make_plus_b_sum_gen([{B, Sum} | Rest]) ->
     end.
 
 %% Plus: only sum constrained - generate all non-negative pairs
-make_plus_sum_only_generator(SumValue) ->
+make_plus_sum_only_generator(SumValue) when SumValue >= 0 ->
     Pairs = [{A, SumValue - A} || A <- lists:seq(0, SumValue)],
-    make_plus_pairs_gen(Pairs).
+    make_plus_pairs_gen(Pairs);
+make_plus_sum_only_generator(_SumValue) ->
+    %% Negative sum: no non-negative pairs exist
+    fun(_) -> done end.
 
 %% Plus: sum range constrained
 make_plus_sum_range_generator(SMin, SMax) ->
+    %% Adjust range to non-negative values (naturals only)
+    AdjustedMin = max(0, SMin),
     Pairs = [{A, B, S} ||
-             S <- lists:seq(SMin, SMax),
+             S <- lists:seq(AdjustedMin, SMax),
              A <- lists:seq(0, S),
              B <- [S - A]],
     make_plus_triple_gen(Pairs).
diff --git a/src/operations.erl b/src/operations.erl
index 58beae8..d91eec9 100644
--- a/src/operations.erl
+++ b/src/operations.erl
@@ -65,6 +65,9 @@
 	 project_iterator/2,
 	 sort_iterator/2,
 	 take_iterator/2,
+	 %% Join operators
+	 equijoin_iterator/3,
+	 theta_join_iterator/3,
 	 %% Materialization (eager)
 	 materialize/3]).
 
@@ -690,6 +693,24 @@ take(Database, RelationName, N) when is_record(Database, database_state), is_int
 %% Extended version of get_tuples_iterator/2 that accepts boundedness constraints.
 %% For infinite relations, constraints are mandatory to ensure termination.
 %%
+%% <b>All tuples include metadata tracking</b> with a nested `meta' field containing:
+%% <ul>
+%%   <li>`provenance' - Tracks immediate source of each attribute</li>
+%%   <li>`lineage' - Tracks complete operational history</li>
+%% </ul>
+%%
+%% == Options Format ==
+%%
+%% Nested format (preferred):
+%% ```
+%% #{constraints => #{age => {gt, 30}}}
+%% '''
+%%
+%% Flat format (backward compatible):
+%% ```
+%% #{age => {gt, 30}}
+%% '''
+%%
 %% == Constraints ==
 %%
 %% Constraints bound the iteration over infinite relations:
@@ -703,35 +724,49 @@ take(Database, RelationName, N) when is_record(Database, database_state), is_int
 %%
 %% == Example ==
 %% ```
-%% %% Iterate naturals in range [0, 100]
-%% Iterator = operations:get_tuples_iterator(DB, naturals,
-%%     #{value => {range, 0, 100}}).
+%% %% Iterate employees with age > 30 (metadata always included)
+%% Iterator = operations:get_tuples_iterator(DB, employees,
+%%     #{constraints => #{age => {gt, 30}}}).
+%%
+%% %% Results will have:
+%% %% #{id => 1, name => "Alice", age => 35,
+%% %%   meta => #{
+%% %%     provenance => #{id => {employees, id}, ...},
+%% %%     lineage => {select, Fun, {base, employees}}
+%% %%   }}
 %% '''
 %%
 %% @param Database `#database_state{}' record
 %% @param RelationName Atom naming the relation to iterate
-%% @param Constraints Map of attribute constraints
+%% @param Options Map of constraints (nested or flat format)
 %% @returns Pid of iterator process
 %%
 %% @see get_tuples_iterator/2
 %% @see take/3
-get_tuples_iterator(Database, RelationName, Constraints) when is_record(Database, database_state) ->
+get_tuples_iterator(Database, RelationName, Options) when is_record(Database, database_state) ->
     case maps:get(RelationName, Database#database_state.relations, error) of
         error ->
             erlang:error({error_tuple_iterator_init, RelationName});
         RelationHash ->
             [Relation] = mnesia:dirty_read(relation, RelationHash),
 
+            %% Extract constraints from options (support both nested and flat format)
+            Constraints = case maps:is_key(constraints, Options) of
+                true -> maps:get(constraints, Options);
+                false -> Options  %% Flat format: entire map is constraints
+            end,
+
+            %% Metadata tracking is ALWAYS enabled
             case Relation#relation.cardinality of
                 {finite, _} ->
                     %% Finite relation: use tuple hashes
                     TupleHashes = hashes_from_tuple(RelationHash),
-                    spawn(fun() -> tuple_iterator_loop(TupleHashes) end);
+                    spawn(fun() -> tuple_iterator_loop(TupleHashes, RelationName, true) end);
 
                 _ ->
                     %% Infinite relation: use generator
                     Generator = instantiate_generator(Relation#relation.generator, Constraints),
-                    spawn(fun() -> generator_iterator_loop(Generator) end)
+                    spawn(fun() -> generator_iterator_loop(Generator, RelationName, true) end)
             end
     end.
 
@@ -819,20 +854,24 @@ close_iterator(IteratorPid) ->
 %% </ul>
 %%
 %% @param TupleHashes List of binary tuple hashes to stream
-tuple_iterator_loop([]) ->
+%% @param RelationName Name of the source relation (for provenance)
+%% @param EnableProvenance Boolean flag to enable provenance tracking
+tuple_iterator_loop([], _RelationName, _EnableProvenance) ->
     receive
         {next, Caller} ->
             Caller ! done;
         stop ->
             ok
     end;
-tuple_iterator_loop([TupleHash | Rest]) ->
+tuple_iterator_loop([TupleHash | Rest], RelationName, EnableProvenance) ->
     receive
         {next, Caller} ->
             %% Resolve tuple on demand
             ResolvedTuple = resolve_tuple(TupleHash),
-            Caller ! {tuple, ResolvedTuple},
-            tuple_iterator_loop(Rest);
+            %% Add provenance tracking (always enabled)
+            FinalTuple = add_base_provenance(ResolvedTuple, RelationName),
+            Caller ! {tuple, FinalTuple},
+            tuple_iterator_loop(Rest, RelationName, EnableProvenance);
         stop ->
             ok
     end.
@@ -901,7 +940,9 @@ select_loop(ChildPid, Predicate) ->
                 {ok, Tuple} ->
                     case Predicate(Tuple) of
                         true ->
-                            Caller ! {tuple, Tuple},
+                            %% Wrap lineage with select operation
+                            TupleWithLineage = wrap_lineage(Tuple, {select, Predicate}),
+                            Caller ! {tuple, TupleWithLineage},
                             select_loop(ChildPid, Predicate);
                         false ->
                             %% Skip this tuple, get next
@@ -940,8 +981,30 @@ project_loop(ChildPid, Attributes) ->
         {next, Caller} ->
             case next_tuple(ChildPid) of
                 {ok, Tuple} ->
+                    % Project data attributes
                     ProjectedTuple = maps:with(Attributes, Tuple),
-                    Caller ! {tuple, ProjectedTuple},
+
+                    % Preserve and update metadata
+                    case maps:get(meta, Tuple, undefined) of
+                        undefined ->
+                            % No metadata, just return projected tuple
+                            Caller ! {tuple, ProjectedTuple};
+                        Meta ->
+                            % Filter provenance to only projected attributes
+                            OldProv = maps:get(provenance, Meta, #{}),
+                            NewProv = maps:with(Attributes, OldProv),
+
+                            % Build new metadata
+                            UpdatedMeta = case NewProv of
+                                Empty when map_size(Empty) =:= 0 -> maps:remove(provenance, Meta);
+                                _ -> maps:put(provenance, NewProv, Meta)
+                            end,
+
+                            % Add metadata back and wrap lineage
+                            TupleWithMeta = maps:put(meta, UpdatedMeta, ProjectedTuple),
+                            TupleWithLineage = wrap_lineage(TupleWithMeta, {project, Attributes}),
+                            Caller ! {tuple, TupleWithLineage}
+                    end,
                     project_loop(ChildPid, Attributes);
                 done ->
                     close_iterator(ChildPid),
@@ -980,8 +1043,11 @@ sort_iterator(ChildIterator, CompareFun) when is_function(CompareFun, 2) ->
         %% Phase 2: Sort in memory
         SortedTuples = lists:sort(CompareFun, AllTuples),
 
-        %% Phase 3: Emit sorted tuples on demand
-        sorted_emit_loop(SortedTuples)
+        %% Phase 3: Wrap lineage for all tuples
+        TuplesWithLineage = [wrap_lineage(T, {sort, CompareFun}) || T <- SortedTuples],
+
+        %% Phase 4: Emit sorted tuples on demand
+        sorted_emit_loop(TuplesWithLineage)
     end).
 
 sorted_emit_loop([]) ->
@@ -1013,15 +1079,17 @@ sorted_emit_loop([Tuple | Rest]) ->
 %% @param N Maximum number of tuples to emit
 %% @returns Iterator process (Pid)
 take_iterator(ChildIterator, N) when is_integer(N), N > 0 ->
-    spawn(fun() -> take_iter_loop(ChildIterator, N, 0) end).
+    spawn(fun() -> take_iter_loop(ChildIterator, N, 0, N) end).
 
-take_iter_loop(ChildPid, Limit, Count) when Count < Limit ->
+take_iter_loop(ChildPid, Limit, Count, OriginalN) when Count < Limit ->
     receive
         {next, Caller} ->
             case next_tuple(ChildPid) of
                 {ok, Tuple} ->
-                    Caller ! {tuple, Tuple},
-                    take_iter_loop(ChildPid, Limit, Count + 1);
+                    %% Wrap lineage with take operation
+                    TupleWithLineage = wrap_lineage(Tuple, {take, OriginalN}),
+                    Caller ! {tuple, TupleWithLineage},
+                    take_iter_loop(ChildPid, Limit, Count + 1, OriginalN);
                 done ->
                     close_iterator(ChildPid),
                     Caller ! done
@@ -1030,7 +1098,7 @@ take_iter_loop(ChildPid, Limit, Count) when Count < Limit ->
             close_iterator(ChildPid),
             ok
     end;
-take_iter_loop(ChildPid, _Limit, _Count) ->
+take_iter_loop(ChildPid, _Limit, _Count, _OriginalN) ->
     %% Reached limit - close child and signal done
     close_iterator(ChildPid),
     receive
@@ -1039,6 +1107,59 @@ take_iter_loop(ChildPid, _Limit, _Count) ->
     end.
 
 %%% ============================================================================
+%%% Join Operators
+%%% ============================================================================
+
+%% @doc Equijoin iterator: Natural join on matching attribute values.
+%%
+%% Performs an equijoin between two relations on a common attribute.
+%% For each tuple from the left iterator, finds all matching tuples from
+%% the right iterator where the specified attributes are equal.
+%%
+%% The result tuples merge attributes from both sides. If attribute names
+%% conflict, right side attributes are prefixed with "right_".
+%%
+%% Implementation: Nested loop join (simple but works for small-medium datasets).
+%% Future: Hash join for better performance.
+%%
+%% == Example ==
+%% ```
+%% %% Join employees with departments on dept_id
+%% LeftIter = operations:get_tuples_iterator(DB, employees, #{}),
+%% RightIter = operations:get_tuples_iterator(DB, departments, #{}),
+%% JoinIter = operations:equijoin_iterator(LeftIter, RightIter, dept_id)
+%% '''
+%%
+%% @param LeftIterator Iterator for left relation
+%% @param RightIterator Iterator for right relation
+%% @param JoinAttribute Attribute name to join on (must exist in both relations)
+%% @returns Iterator process that emits joined tuples
+equijoin_iterator(LeftIterator, RightIterator, JoinAttribute) ->
+    spawn(fun() -> equijoin_loop(LeftIterator, RightIterator, JoinAttribute) end).
+
+%% @doc Theta-join iterator: Join with arbitrary predicate.
+%%
+%% Performs a theta-join between two relations using a user-provided
+%% predicate function. For each pair of tuples (left, right), the predicate
+%% is evaluated. If true, the tuples are merged and emitted.
+%%
+%% == Example ==
+%% ```
+%% %% Join where employee.age > department.min_age
+%% Predicate = fun(EmpTuple, DeptTuple) ->
+%%     maps:get(age, EmpTuple) > maps:get(min_age, DeptTuple)
+%% end,
+%% JoinIter = operations:theta_join_iterator(LeftIter, RightIter, Predicate)
+%% '''
+%%
+%% @param LeftIterator Iterator for left relation
+%% @param RightIterator Iterator for right relation
+%% @param Predicate Function `fun(LeftTuple, RightTuple) -> boolean()'
+%% @returns Iterator process that emits joined tuples
+theta_join_iterator(LeftIterator, RightIterator, Predicate) ->
+    spawn(fun() -> theta_join_loop(LeftIterator, RightIterator, Predicate) end).
+
+%%% ============================================================================
 %%% Materialization (Eager - Relation Creation)
 %%% ============================================================================
 
@@ -1157,20 +1278,275 @@ instantiate_generator(Other, _) ->
 %% of tuple hashes.
 %%
 %% @param Generator Generator function
-generator_iterator_loop(Generator) ->
+generator_iterator_loop(Generator, RelationName, EnableProvenance) ->
     receive
         {next, Caller} ->
             case Generator(next) of
                 done ->
                     Caller ! done;
                 {value, Tuple, NextGen} ->
-                    Caller ! {tuple, Tuple},
-                    generator_iterator_loop(NextGen);
+                    %% Add provenance tracking (always enabled)
+                    FinalTuple = add_base_provenance(Tuple, RelationName),
+                    Caller ! {tuple, FinalTuple},
+                    generator_iterator_loop(NextGen, RelationName, EnableProvenance);
+                {error, Reason} ->
+                    Caller ! {error, Reason}
+            end;
+        stop ->
+            ok
+    end.
+
+%% @private
+%% @doc Equijoin loop: nested loop join on attribute equality.
+%%
+%% Strategy:
+%% 1. Collect all tuples from right iterator into memory
+%% 2. For each left tuple, scan right tuples for matches
+%% 3. Emit merged tuple when join attribute values match
+%%
+%% @param LeftIter Left iterator process
+%% @param RightIter Right iterator process
+%% @param JoinAttr Attribute to join on
+equijoin_loop(LeftIter, RightIter, JoinAttr) ->
+    % Materialize right side (for nested loop)
+    RightTuples = collect_all(RightIter),
+    equijoin_emit_loop(LeftIter, RightTuples, JoinAttr).
+
+equijoin_emit_loop(LeftIter, RightTuples, JoinAttr) ->
+    receive
+        {next, Caller} ->
+            case next_tuple(LeftIter) of
+                done ->
+                    Caller ! done,
+                    ok;  % Stop after sending done
+                {ok, LeftTuple} ->
+                    % Find all right tuples that match on join attribute
+                    case maps:get(JoinAttr, LeftTuple, undefined) of
+                        undefined ->
+                            % Left tuple doesn't have join attribute - try next
+                            self() ! {next, Caller},
+                            equijoin_emit_loop(LeftIter, RightTuples, JoinAttr);
+                        LeftValue ->
+                            % Find matching right tuples
+                            Matches = [merge_tuples(LeftTuple, RightTuple, {equijoin, JoinAttr})
+                                      || RightTuple <- RightTuples,
+                                         maps:get(JoinAttr, RightTuple, undefined) =:= LeftValue],
+                            case Matches of
+                                [] ->
+                                    % No matches - try next left tuple
+                                    self() ! {next, Caller},
+                                    equijoin_emit_loop(LeftIter, RightTuples, JoinAttr);
+                                [FirstMatch | RestMatches] ->
+                                    % Emit first match, buffer rest
+                                    Caller ! {tuple, FirstMatch},
+                                    equijoin_emit_buffered(LeftIter, RightTuples, JoinAttr, RestMatches)
+                            end
+                    end;
+                {error, Reason} ->
+                    Caller ! {error, Reason},
+                    ok  % Stop after sending error
+            end;
+        stop ->
+            close_iterator(LeftIter),
+            ok
+    end.
+
+equijoin_emit_buffered(LeftIter, RightTuples, JoinAttr, [Match | Rest]) ->
+    receive
+        {next, Caller} ->
+            Caller ! {tuple, Match},
+            equijoin_emit_buffered(LeftIter, RightTuples, JoinAttr, Rest);
+        stop ->
+            close_iterator(LeftIter),
+            ok
+    end;
+equijoin_emit_buffered(LeftIter, RightTuples, JoinAttr, []) ->
+    % Buffer empty - continue with next left tuple
+    equijoin_emit_loop(LeftIter, RightTuples, JoinAttr).
+
+%% @private
+%% @doc Theta-join loop: nested loop join with predicate.
+%%
+%% @param LeftIter Left iterator process
+%% @param RightIter Right iterator process
+%% @param Pred Predicate function fun(LeftTuple, RightTuple) -> boolean()
+theta_join_loop(LeftIter, RightIter, Pred) ->
+    % Materialize right side
+    RightTuples = collect_all(RightIter),
+    theta_join_emit_loop(LeftIter, RightTuples, Pred).
+
+theta_join_emit_loop(LeftIter, RightTuples, Pred) ->
+    receive
+        {next, Caller} ->
+            case next_tuple(LeftIter) of
+                done ->
+                    Caller ! done;
+                {ok, LeftTuple} ->
+                    % Find all right tuples that satisfy predicate
+                    Matches = [merge_tuples(LeftTuple, RightTuple, {theta_join, Pred})
+                              || RightTuple <- RightTuples,
+                                 Pred(LeftTuple, RightTuple)],
+                    case Matches of
+                        [] ->
+                            % No matches - continue
+                            self() ! {next, Caller},
+                            theta_join_emit_loop(LeftIter, RightTuples, Pred);
+                        [FirstMatch | RestMatches] ->
+                            Caller ! {tuple, FirstMatch},
+                            theta_join_emit_buffered(LeftIter, RightTuples, Pred, RestMatches)
+                    end;
                 {error, Reason} ->
                     Caller ! {error, Reason}
             end;
         stop ->
+            close_iterator(LeftIter),
+            ok
+    end.
+
+theta_join_emit_buffered(LeftIter, RightTuples, Pred, [Match | Rest]) ->
+    receive
+        {next, Caller} ->
+            Caller ! {tuple, Match},
+            theta_join_emit_buffered(LeftIter, RightTuples, Pred, Rest);
+        stop ->
+            close_iterator(LeftIter),
             ok
+    end;
+theta_join_emit_buffered(LeftIter, RightTuples, Pred, []) ->
+    theta_join_emit_loop(LeftIter, RightTuples, Pred).
+
+%% @private
+%% @doc Wrap existing lineage with a new operation.
+%%
+%% Takes a tuple with optional lineage and wraps it with a new operation node.
+%% If the tuple has no lineage, does nothing.
+%%
+%% @param Tuple Tuple potentially with meta.lineage
+%% @param OpInfo Operation info: {select, Pred} | {project, Attrs} | {sort, Cmp} | {take, N}
+wrap_lineage(Tuple, OpInfo) ->
+    case maps:get(meta, Tuple, undefined) of
+        undefined ->
+            % No metadata, return as-is
+            Tuple;
+        Meta ->
+            case maps:get(lineage, Meta, undefined) of
+                undefined ->
+                    % No lineage, return as-is
+                    Tuple;
+                ChildLineage ->
+                    % Wrap existing lineage with new operation
+                    NewLineage = case OpInfo of
+                        {select, Pred} -> {select, Pred, ChildLineage};
+                        {project, Attrs} -> {project, Attrs, ChildLineage};
+                        {sort, CmpFun} -> {sort, CmpFun, ChildLineage};
+                        {take, N} -> {take, N, ChildLineage}
+                    end,
+                    maps:put(meta, maps:put(lineage, NewLineage, Meta), Tuple)
+            end
+    end.
+
+%% @private
+%% @doc Add base provenance and lineage to a tuple from a relation.
+%%
+%% Creates nested meta fields with both provenance and lineage.
+%% Example: add_base_provenance(#{id => 1, name => "Alice"}, employees)
+%%   => #{id => 1, name => "Alice",
+%%        meta => #{provenance => #{id => {employees, id}, name => {employees, name}},
+%%                  lineage => {base, employees}}}
+add_base_provenance(Tuple, RelationName) ->
+    % Build provenance map for all attributes (excluding existing meta)
+    DataAttrs = maps:remove(meta, Tuple),
+    Provenance = maps:fold(
+        fun(Key, _Value, Acc) ->
+            maps:put(Key, {RelationName, Key}, Acc)
+        end,
+        #{},
+        DataAttrs
+    ),
+    % Add metadata with provenance and lineage
+    maps:put(meta, #{
+        provenance => Provenance,
+        lineage => {base, RelationName}
+    }, Tuple).
+
+%% @private
+%% @doc Merge two tuples for join result, including metadata.
+%%
+%% If attribute names conflict, right side attributes are prefixed with "right_".
+%% Metadata (including provenance and lineage) is also merged.
+%%
+%% @param LeftTuple Left side tuple
+%% @param RightTuple Right side tuple
+%% @param JoinInfo Join information: {equijoin, Attr} | {theta_join, Predicate}
+merge_tuples(LeftTuple, RightTuple, JoinInfo) ->
+    % Extract metadata from both sides (if present)
+    LeftMeta = maps:get(meta, LeftTuple, #{}),
+    RightMeta = maps:get(meta, RightTuple, #{}),
+
+    % Extract provenance and lineage from metadata
+    LeftProv = maps:get(provenance, LeftMeta, #{}),
+    RightProv = maps:get(provenance, RightMeta, #{}),
+    LeftLineage = maps:get(lineage, LeftMeta, undefined),
+    RightLineage = maps:get(lineage, RightMeta, undefined),
+
+    % Remove metadata from tuples for data merging
+    LeftData = maps:remove(meta, LeftTuple),
+    RightData = maps:remove(meta, RightTuple),
+
+    % Merge data attributes
+    MergedData = maps:fold(
+        fun(Key, Value, Acc) ->
+            case maps:is_key(Key, Acc) of
+                true ->
+                    % Conflict - prefix right attribute
+                    RightKey = list_to_atom("right_" ++ atom_to_list(Key)),
+                    maps:put(RightKey, Value, Acc);
+                false ->
+                    maps:put(Key, Value, Acc)
+            end
+        end,
+        LeftData,
+        RightData
+    ),
+
+    % Merge provenance (same conflict resolution)
+    MergedProv = maps:fold(
+        fun(Key, Source, Acc) ->
+            case maps:is_key(Key, Acc) of
+                true ->
+                    % Conflict - prefix right attribute provenance
+                    RightKey = list_to_atom("right_" ++ atom_to_list(Key)),
+                    maps:put(RightKey, Source, Acc);
+                false ->
+                    maps:put(Key, Source, Acc)
+            end
+        end,
+        LeftProv,
+        RightProv
+    ),
+
+    % Build merged lineage if both sides have lineage
+    MergedLineage = case {LeftLineage, RightLineage, JoinInfo} of
+        {undefined, undefined, _} -> undefined;
+        {L, R, {equijoin, Attr}} when L =/= undefined, R =/= undefined ->
+            {join, Attr, L, R};
+        {L, R, {theta_join, Pred}} when L =/= undefined, R =/= undefined ->
+            {theta_join, Pred, L, R};
+        _ -> undefined
+    end,
+
+    % Build merged metadata
+    MergedMeta = case {maps:size(MergedProv), MergedLineage} of
+        {0, undefined} -> #{};
+        {0, Lineage} -> #{lineage => Lineage};
+        {_, undefined} -> #{provenance => MergedProv};
+        {_, Lineage} -> #{provenance => MergedProv, lineage => Lineage}
+    end,
+
+    % Add merged metadata to result (only if non-empty)
+    case maps:size(MergedMeta) of
+        0 -> MergedData;
+        _ -> maps:put(meta, MergedMeta, MergedData)
     end.
 
 %% @private
diff --git a/test/function_relations_test.erl b/test/function_relations_test.erl
index 78c9ec3..a9164bf 100644
--- a/test/function_relations_test.erl
+++ b/test/function_relations_test.erl
@@ -87,6 +87,14 @@ setup() ->
 cleanup(_DB) ->
     ok.
 
+%%% Helper Functions
+
+%% Strip metadata from tuples for comparison
+strip_meta(Tuple) when is_map(Tuple) ->
+    maps:remove(meta, Tuple);
+strip_meta(Tuples) when is_list(Tuples) ->
+    [strip_meta(T) || T <- Tuples].
+
 %%% Plus Relation Tests
 
 test_plus_both_operands_constrained(DB) ->
@@ -98,17 +106,18 @@ test_plus_both_operands_constrained(DB) ->
         b => {range, 2, 4}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     % should get 9 tuples total (3 × 3 = 9)
     % like (1,2,3), (1,3,4), (1,4,5), (2,2,4), ..., (3,4,7)
     [
      ?_assertEqual(9, length(Results)),
-     ?_assert(lists:member(#{a => 1, b => 2, sum => 3}, Results)),
-     ?_assert(lists:member(#{a => 2, b => 3, sum => 5}, Results)),
-     ?_assert(lists:member(#{a => 3, b => 4, sum => 7}, Results)),
+     ?_assert(lists:member(#{a => 1, b => 2, sum => 3}, DataOnly)),
+     ?_assert(lists:member(#{a => 2, b => 3, sum => 5}, DataOnly)),
+     ?_assert(lists:member(#{a => 3, b => 4, sum => 7}, DataOnly)),
      ?_assert(lists:all(fun(#{a := A, b := B, sum := S}) ->
                            S =:= A + B
-                       end, Results))
+                       end, DataOnly))
     ].
 
 test_plus_a_and_sum_constrained(DB) ->
@@ -120,6 +129,7 @@ test_plus_a_and_sum_constrained(DB) ->
         sum => {range, 5, 7}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     % Expected tuples: (1,4,5), (1,5,6), (1,6,7), (2,3,5), (2,4,6), (2,5,7), (3,2,5), (3,3,6), (3,4,7)
     % (filtered to only keep where b >= 0)
@@ -127,9 +137,9 @@ test_plus_a_and_sum_constrained(DB) ->
      ?_assert(length(Results) > 0),
      ?_assert(lists:all(fun(#{a := A, b := B, sum := S}) ->
                            S =:= A + B andalso B >= 0
-                       end, Results)),
-     ?_assert(lists:member(#{a => 1, b => 4, sum => 5}, Results)),
-     ?_assert(lists:member(#{a => 3, b => 4, sum => 7}, Results))
+                       end, DataOnly)),
+     ?_assert(lists:member(#{a => 1, b => 4, sum => 5}, DataOnly)),
+     ?_assert(lists:member(#{a => 3, b => 4, sum => 7}, DataOnly))
     ].
 
 test_plus_b_and_sum_constrained(DB) ->
@@ -141,14 +151,15 @@ test_plus_b_and_sum_constrained(DB) ->
         sum => {range, 5, 6}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     [
      ?_assert(length(Results) > 0),
      ?_assert(lists:all(fun(#{a := A, b := B, sum := S}) ->
                            S =:= A + B andalso A >= 0
-                       end, Results)),
-     ?_assert(lists:member(#{a => 3, b => 2, sum => 5}, Results)),
-     ?_assert(lists:member(#{a => 2, b => 4, sum => 6}, Results))
+                       end, DataOnly)),
+     ?_assert(lists:member(#{a => 3, b => 2, sum => 5}, DataOnly)),
+     ?_assert(lists:member(#{a => 2, b => 4, sum => 6}, DataOnly))
     ].
 
 test_plus_only_sum_constrained(DB) ->
@@ -158,15 +169,16 @@ test_plus_only_sum_constrained(DB) ->
         sum => {eq, 5}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     [
      ?_assertEqual(6, length(Results)),
-     ?_assert(lists:member(#{a => 0, b => 5, sum => 5}, Results)),
-     ?_assert(lists:member(#{a => 2, b => 3, sum => 5}, Results)),
-     ?_assert(lists:member(#{a => 5, b => 0, sum => 5}, Results)),
+     ?_assert(lists:member(#{a => 0, b => 5, sum => 5}, DataOnly)),
+     ?_assert(lists:member(#{a => 2, b => 3, sum => 5}, DataOnly)),
+     ?_assert(lists:member(#{a => 5, b => 0, sum => 5}, DataOnly)),
      ?_assert(lists:all(fun(#{a := A, b := B, sum := S}) ->
                            S =:= A + B andalso S =:= 5
-                       end, Results))
+                       end, DataOnly))
     ].
 
 test_plus_sum_range_constrained(DB) ->
@@ -175,15 +187,16 @@ test_plus_sum_range_constrained(DB) ->
         sum => {range, 2, 3}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     % For sum=2: (0,2), (1,1), (2,0) → 3 pairs
     % For sum=3: (0,3), (1,2), (2,1), (3,0) → 4 pairs
     % Total = 7
     [
      ?_assertEqual(7, length(Results)),
-     ?_assert(lists:all(fun(#{sum := S}) -> S >= 2 andalso S =< 3 end, Results)),
-     ?_assert(lists:member(#{a => 0, b => 2, sum => 2}, Results)),
-     ?_assert(lists:member(#{a => 3, b => 0, sum => 3}, Results))
+     ?_assert(lists:all(fun(#{sum := S}) -> S >= 2 andalso S =< 3 end, DataOnly)),
+     ?_assert(lists:member(#{a => 0, b => 2, sum => 2}, DataOnly)),
+     ?_assert(lists:member(#{a => 3, b => 0, sum => 3}, DataOnly))
     ].
 
 test_plus_insufficient_constraints(DB) ->
@@ -204,16 +217,17 @@ test_times_both_operands_constrained(DB) ->
         b => {range, 3, 5}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     % 3×3 = 9 tuples
     [
      ?_assertEqual(9, length(Results)),
-     ?_assert(lists:member(#{a => 2, b => 3, product => 6}, Results)),
-     ?_assert(lists:member(#{a => 3, b => 4, product => 12}, Results)),
-     ?_assert(lists:member(#{a => 4, b => 5, product => 20}, Results)),
+     ?_assert(lists:member(#{a => 2, b => 3, product => 6}, DataOnly)),
+     ?_assert(lists:member(#{a => 3, b => 4, product => 12}, DataOnly)),
+     ?_assert(lists:member(#{a => 4, b => 5, product => 20}, DataOnly)),
      ?_assert(lists:all(fun(#{a := A, b := B, product := P}) ->
                            P =:= A * B
-                       end, Results))
+                       end, DataOnly))
     ].
 
 test_times_insufficient_constraints(DB) ->
@@ -235,15 +249,16 @@ test_minus_both_operands_constrained(DB) ->
         b => {range, 2, 3}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     % 3 × 2 = 6 tuples
     [
      ?_assertEqual(6, length(Results)),
-     ?_assert(lists:member(#{a => 5, b => 2, difference => 3}, Results)),
-     ?_assert(lists:member(#{a => 7, b => 3, difference => 4}, Results)),
+     ?_assert(lists:member(#{a => 5, b => 2, difference => 3}, DataOnly)),
+     ?_assert(lists:member(#{a => 7, b => 3, difference => 4}, DataOnly)),
      ?_assert(lists:all(fun(#{a := A, b := B, difference := D}) ->
                            D =:= A - B
-                       end, Results))
+                       end, DataOnly))
     ].
 
 %%% Divide (integer division) Tests
@@ -255,15 +270,16 @@ test_divide_both_operands_constrained(DB) ->
         b => {range, 3, 4}
     }),
     Results = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Results),
 
     % 3 × 2 = 6 tuples
     [
      ?_assertEqual(6, length(Results)),
-     ?_assert(lists:member(#{a => 10, b => 3, quotient => 3, remainder => 1}, Results)),
-     ?_assert(lists:member(#{a => 12, b => 4, quotient => 3, remainder => 0}, Results)),
+     ?_assert(lists:member(#{a => 10, b => 3, quotient => 3, remainder => 1}, DataOnly)),
+     ?_assert(lists:member(#{a => 12, b => 4, quotient => 3, remainder => 0}, DataOnly)),
      ?_assert(lists:all(fun(#{a := A, b := B, quotient := Q, remainder := R}) ->
                            A =:= B * Q + R andalso R >= 0 andalso R < abs(B)
-                       end, Results))
+                       end, DataOnly))
     ].
 
 test_divide_avoids_zero(DB) ->
diff --git a/test/infinite_relations_test.erl b/test/infinite_relations_test.erl
index cafd325..39fe29a 100644
--- a/test/infinite_relations_test.erl
+++ b/test/infinite_relations_test.erl
@@ -26,6 +26,14 @@ setup() ->
 cleanup(_DB) ->
     ok.
 
+%%% Helper Functions
+
+%% Strip metadata from tuples for comparison
+strip_meta(Tuple) when is_map(Tuple) ->
+    maps:remove(meta, Tuple);
+strip_meta(Tuples) when is_list(Tuples) ->
+    [strip_meta(T) || T <- Tuples].
+
 %%% Tests
 
 test_create_naturals(DB) ->
@@ -59,13 +67,14 @@ test_naturals_with_constraints(DB) ->
     %% Query with range constraint [0, 9]
     Iterator = operations:get_tuples_iterator(DB1, naturals, #{value => {range, 0, 9}}),
     Tuples = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Tuples),
 
     %% Should get exactly 10 tuples: {0, 1, 2, ..., 9}
     [
      ?_assertEqual(10, length(Tuples)),
-     ?_assert(lists:member(#{value => 0}, Tuples)),
-     ?_assert(lists:member(#{value => 5}, Tuples)),
-     ?_assert(lists:member(#{value => 9}, Tuples))
+     ?_assert(lists:member(#{value => 0}, DataOnly)),
+     ?_assert(lists:member(#{value => 5}, DataOnly)),
+     ?_assert(lists:member(#{value => 9}, DataOnly))
     ].
 
 test_integers_generator(DB) ->
@@ -80,14 +89,15 @@ test_integers_generator(DB) ->
     %% Query with range constraint [-5, 5]
     Iterator = operations:get_tuples_iterator(DB1, integers, #{value => {range, -5, 5}}),
     Tuples = operations:collect_all(Iterator),
+    DataOnly = strip_meta(Tuples),
 
     %% Should get 11 tuples: {-5, -4, ..., 0, ..., 4, 5}
     [
      ?_assertEqual(aleph_zero, Integers#relation.cardinality),
      ?_assertEqual(11, length(Tuples)),
-     ?_assert(lists:member(#{value => 0}, Tuples)),
-     ?_assert(lists:member(#{value => -5}, Tuples)),
-     ?_assert(lists:member(#{value => 5}, Tuples))
+     ?_assert(lists:member(#{value => 0}, DataOnly)),
+     ?_assert(lists:member(#{value => -5}, DataOnly)),
+     ?_assert(lists:member(#{value => 5}, DataOnly))
     ].
 
 test_rationals_generator(DB) ->
diff --git a/test/join_test.erl b/test/join_test.erl
new file mode 100644
index 0000000..8abdc0b
--- /dev/null
+++ b/test/join_test.erl
@@ -0,0 +1,279 @@
+-module(join_test).
+-include_lib("eunit/include/eunit.hrl").
+-include("operations.hrl").
+
+%%% Join Operators Test Suite
+%%%
+%%% Tests for relational join operations:
+%%% - Equijoin (natural join on attribute equality)
+%%% - Theta-join (join with arbitrary predicate)
+
+join_test_() ->
+    {foreach,
+     fun setup/0,
+     fun cleanup/1,
+     [
+      fun test_equijoin_simple/1,
+      fun test_equijoin_multiple_matches/1,
+      fun test_equijoin_no_matches/1,
+      fun test_equijoin_with_conflicts/1,
+      fun test_theta_join_greater_than/1,
+      fun test_theta_join_complex_predicate/1,
+      fun test_join_with_infinite_relation/1,
+      fun test_join_materialize_pipeline/1
+     ]}.
+
+%%% Setup and Cleanup
+
+setup() ->
+    operations:setup(),
+    DB = operations:create_database(test_db),
+
+    % Create employees relation
+    {DB1, _} = operations:create_relation(DB, employees, #{
+        id => integer,
+        name => string,
+        dept_id => integer,
+        age => integer
+    }),
+
+    {DB2, _} = operations:create_tuple(DB1, employees, #{id => 1, name => "Alice", dept_id => 10, age => 30}),
+    {DB3, _} = operations:create_tuple(DB2, employees, #{id => 2, name => "Bob", dept_id => 20, age => 35}),
+    {DB4, _} = operations:create_tuple(DB3, employees, #{id => 3, name => "Carol", dept_id => 10, age => 28}),
+    {DB5, _} = operations:create_tuple(DB4, employees, #{id => 4, name => "Dave", dept_id => 30, age => 42}),
+
+    % Create departments relation
+    {DB6, _} = operations:create_relation(DB5, departments, #{
+        dept_id => integer,
+        dept_name => string,
+        budget => integer
+    }),
+
+    {DB7, _} = operations:create_tuple(DB6, departments, #{dept_id => 10, dept_name => "Engineering", budget => 100000}),
+    {DB8, _} = operations:create_tuple(DB7, departments, #{dept_id => 20, dept_name => "Sales", budget => 80000}),
+    {DB9, _} = operations:create_tuple(DB8, departments, #{dept_id => 30, dept_name => "Marketing", budget => 90000}),
+
+    DB9.
+
+cleanup(_DB) ->
+    ok.
+
+%%% Equijoin Tests
+
+test_equijoin_simple(DB) ->
+    % Join employees with departments on dept_id
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    JoinIter = operations:equijoin_iterator(EmpIter, DeptIter, dept_id),
+    Results = operations:collect_all(JoinIter),
+
+    % Should get 4 joined tuples (one for each employee)
+    [
+     ?_assertEqual(4, length(Results)),
+     % Check Alice joined with Engineering
+     ?_assert(lists:any(fun(T) ->
+                           maps:get(name, T, undefined) =:= "Alice" andalso
+                           maps:get(dept_name, T, undefined) =:= "Engineering"
+                       end, Results)),
+     % Check Bob joined with Sales
+     ?_assert(lists:any(fun(T) ->
+                           maps:get(name, T, undefined) =:= "Bob" andalso
+                           maps:get(dept_name, T, undefined) =:= "Sales"
+                       end, Results)),
+     % All results should have both employee and department attributes
+     ?_assert(lists:all(fun(T) ->
+                           maps:is_key(name, T) andalso
+                           maps:is_key(dept_name, T) andalso
+                           maps:is_key(dept_id, T)
+                       end, Results))
+    ].
+
+test_equijoin_multiple_matches(DB) ->
+    % Multiple employees in same department should produce multiple results
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    JoinIter = operations:equijoin_iterator(EmpIter, DeptIter, dept_id),
+    Results = operations:collect_all(JoinIter),
+
+    % Engineering dept (10) has 2 employees (Alice and Carol)
+    EngineeringResults = lists:filter(fun(T) ->
+        maps:get(dept_name, T, undefined) =:= "Engineering"
+    end, Results),
+
+    [
+     ?_assertEqual(2, length(EngineeringResults)),
+     ?_assert(lists:any(fun(T) -> maps:get(name, T) =:= "Alice" end, EngineeringResults)),
+     ?_assert(lists:any(fun(T) -> maps:get(name, T) =:= "Carol" end, EngineeringResults))
+    ].
+
+test_equijoin_no_matches(DB) ->
+    % Create a department with no employees
+    {DB1, _} = operations:create_tuple(DB, departments, #{dept_id => 40, dept_name => "HR", budget => 70000}),
+
+    % Filter to only dept_id=40 which has no employees
+    DeptIter = operations:get_tuples_iterator(DB1, departments, #{}),
+    FilteredDept = operations:select_iterator(DeptIter, fun(D) -> maps:get(dept_id, D) =:= 40 end),
+
+    EmpIter = operations:get_tuples_iterator(DB1, employees, #{}),
+
+    JoinIter = operations:equijoin_iterator(EmpIter, FilteredDept, dept_id),
+    Results = operations:collect_all(JoinIter),
+
+    % No employees in HR, so no results
+    [
+     ?_assertEqual(0, length(Results))
+    ].
+
+test_equijoin_with_conflicts(DB) ->
+    % Create relations with conflicting attribute names
+    {DB1, _} = operations:create_relation(DB, table_a, #{
+        id => integer,
+        value => integer
+    }),
+    {DB2, _} = operations:create_tuple(DB1, table_a, #{id => 1, value => 100}),
+
+    {DB3, _} = operations:create_relation(DB2, table_b, #{
+        id => integer,
+        value => integer  % Same name as table_a.value
+    }),
+    {DB4, _} = operations:create_tuple(DB3, table_b, #{id => 1, value => 200}),
+
+    AIter = operations:get_tuples_iterator(DB4, table_a, #{}),
+    BIter = operations:get_tuples_iterator(DB4, table_b, #{}),
+
+    JoinIter = operations:equijoin_iterator(AIter, BIter, id),
+    Results = operations:collect_all(JoinIter),
+
+    % Should have one result with both values
+    % Right side 'value' should be prefixed as 'right_value'
+    [
+     ?_assertEqual(1, length(Results)),
+     ?_assertMatch([#{id := 1, value := 100, right_value := 200}], Results)
+    ].
+
+%%% Theta-Join Tests
+
+test_theta_join_greater_than(DB) ->
+    % Join where employee.age > department.budget/1000
+    % (contrived example for testing)
+
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    Predicate = fun(Emp, Dept) ->
+        Age = maps:get(age, Emp),
+        Budget = maps:get(budget, Dept),
+        Age > (Budget div 2000)  % Age > Budget/2000
+    end,
+
+    JoinIter = operations:theta_join_iterator(EmpIter, DeptIter, Predicate),
+    Results = operations:collect_all(JoinIter),
+
+    % Should get some results based on predicate
+    [
+     ?_assert(length(Results) > 0),
+     % Verify predicate holds for all results
+     ?_assert(lists:all(fun(T) ->
+                           Age = maps:get(age, T),
+                           Budget = maps:get(budget, T),
+                           Age > (Budget div 2000)
+                       end, Results))
+    ].
+
+test_theta_join_complex_predicate(DB) ->
+    % Join with multiple conditions
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    % Join where employee is in department AND age > 30
+    Predicate = fun(Emp, Dept) ->
+        maps:get(dept_id, Emp) =:= maps:get(dept_id, Dept) andalso
+        maps:get(age, Emp) > 30
+    end,
+
+    JoinIter = operations:theta_join_iterator(EmpIter, DeptIter, Predicate),
+    Results = operations:collect_all(JoinIter),
+
+    % Should get Bob (35) and Dave (42), not Alice (30) or Carol (28)
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) -> maps:get(age, T) > 30 end, Results)),
+     ?_assert(lists:any(fun(T) -> maps:get(name, T) =:= "Bob" end, Results)),
+     ?_assert(lists:any(fun(T) -> maps:get(name, T) =:= "Dave" end, Results))
+    ].
+
+%%% Integration Tests
+
+test_join_with_infinite_relation(DB) ->
+    % Create integers relation
+    {DB1, _} = operations:create_infinite_relation(DB, #{
+        name => integers,
+        schema => #{},
+        cardinality => aleph_zero,
+        generator => {primitive, integers}
+    }),
+
+    % Get small set of integers
+    IntIter = operations:get_tuples_iterator(DB1, integers, #{value => {range, 1, 5}}),
+
+    % Get employees
+    EmpIter = operations:get_tuples_iterator(DB1, employees, #{}),
+
+    % Join where employee.id matches integer.value
+    Predicate = fun(Emp, Int) ->
+        maps:get(id, Emp) =:= maps:get(value, Int)
+    end,
+
+    JoinIter = operations:theta_join_iterator(EmpIter, IntIter, Predicate),
+    Results = operations:collect_all(JoinIter),
+
+    % Should get employees with id in [1,5] (all 4 employees)
+    [
+     ?_assertEqual(4, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Id = maps:get(id, T),
+                           Value = maps:get(value, T),
+                           Id =:= Value andalso Id >= 1 andalso Id =< 5
+                       end, Results))
+    ].
+
+test_join_materialize_pipeline(DB) ->
+    % Complex pipeline: join, filter, project, materialize
+
+    % 1. Join employees with departments
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+    JoinIter = operations:equijoin_iterator(EmpIter, DeptIter, dept_id),
+
+    % 2. Filter to high-budget departments
+    FilteredIter = operations:select_iterator(JoinIter, fun(T) ->
+        maps:get(budget, T) >= 90000
+    end),
+
+    % 3. Project to [name, dept_name, budget]
+    ProjectedIter = operations:project_iterator(FilteredIter, [name, dept_name, budget]),
+
+    % 4. Materialize
+    {DB1, HighBudgetEmps} = operations:materialize(DB, ProjectedIter, high_budget_employees),
+
+    % Verify materialized relation
+    ResultIter = operations:get_tuples_iterator(DB1, high_budget_employees, #{}),
+    Results = operations:collect_all(ResultIter),
+
+    % Engineering (100k) has Alice and Carol
+    % Marketing (90k) has Dave
+    % Sales (80k) is excluded
+    [
+     ?_assertEqual({finite, 3}, HighBudgetEmps#relation.cardinality),
+     ?_assertEqual(3, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           maps:is_key(name, T) andalso
+                           maps:is_key(dept_name, T) andalso
+                           maps:is_key(budget, T) andalso
+                           not maps:is_key(id, T) andalso  % Projected out
+                           not maps:is_key(age, T)  % Projected out
+                       end, Results)),
+     ?_assert(lists:all(fun(T) -> maps:get(budget, T) >= 90000 end, Results))
+    ].
diff --git a/test/provenance_test.erl b/test/provenance_test.erl
new file mode 100644
index 0000000..c98e372
--- /dev/null
+++ b/test/provenance_test.erl
@@ -0,0 +1,420 @@
+-module(provenance_test).
+-include_lib("eunit/include/eunit.hrl").
+-include("operations.hrl").
+
+%%% Provenance & Lineage Tracking Test Suite
+%%%
+%%% Tests for tuple metadata tracking in relational operations.
+%%% Metadata is ALWAYS tracked and included in every tuple via the nested 'meta' field:
+%%% - Provenance: Tracks the immediate source of each attribute
+%%% - Lineage: Tracks the complete operational history of tuple creation
+%%%
+%%% NOTE ON TUPLE ORDERING:
+%%% These tests observe tuples in a specific order (e.g., Alice before Bob) because
+%%% collect_all/1 currently returns results in Merkle tree traversal order for finite
+%%% relations. This ordering is an IMPLEMENTATION DETAIL and is NOT guaranteed.
+%%% Relations are unordered sets - the current order is based on how tuples were
+%%% inserted and hashed into the Merkle tree. Future implementations may return tuples
+%%% in different orders. Tests use lists:all/2 to verify properties hold for ALL tuples
+%%% rather than relying on specific ordering.
+
+provenance_test_() ->
+    {foreach,
+     fun setup/0,
+     fun cleanup/1,
+     [
+      fun test_base_provenance/1,
+      fun test_join_provenance/1,
+      fun test_join_with_conflicts_provenance/1,
+      fun test_equijoin_preserves_both_sources/1,
+      fun test_base_lineage/1,
+      fun test_join_lineage/1,
+      fun test_select_lineage/1,
+      fun test_project_lineage/1,
+      fun test_sort_lineage/1,
+      fun test_take_lineage/1,
+      fun test_multi_operation_pipeline_lineage/1,
+      fun test_theta_join_lineage/1,
+      fun test_complex_nested_lineage/1
+     ]}.
+
+%%% Setup and Cleanup
+
+setup() ->
+    operations:setup(),
+    DB = operations:create_database(test_db),
+
+    % Create employees relation
+    {DB1, _} = operations:create_relation(DB, employees, #{
+        id => integer,
+        name => string,
+        dept_id => integer
+    }),
+
+    {DB2, _} = operations:create_tuple(DB1, employees, #{id => 1, name => "Alice", dept_id => 10}),
+    {DB3, _} = operations:create_tuple(DB2, employees, #{id => 2, name => "Bob", dept_id => 20}),
+
+    % Create departments relation
+    {DB4, _} = operations:create_relation(DB3, departments, #{
+        dept_id => integer,
+        dept_name => string,
+        budget => integer
+    }),
+
+    {DB5, _} = operations:create_tuple(DB4, departments, #{dept_id => 10, dept_name => "Engineering", budget => 100000}),
+    {DB6, _} = operations:create_tuple(DB5, departments, #{dept_id => 20, dept_name => "Sales", budget => 80000}),
+
+    DB6.
+
+cleanup(_DB) ->
+    ok.
+
+%%% Base Provenance Tests
+
+test_base_provenance(DB) ->
+    % Query tuples (metadata always included)
+    Iter = operations:get_tuples_iterator(DB, employees, #{}),
+    Results = operations:collect_all(Iter),
+
+    % All tuples should have correct provenance structure
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           maps:is_key(meta, T) andalso
+                           begin
+                               Meta = maps:get(meta, T),
+                               maps:is_key(provenance, Meta)
+                           end
+                       end, Results)),
+     % Check ALL tuples have correct provenance mapping
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Prov = maps:get(provenance, Meta),
+                           % All attributes should map to {employees, attr_name}
+                           maps:get(id, Prov) =:= {employees, id} andalso
+                           maps:get(name, Prov) =:= {employees, name} andalso
+                           maps:get(dept_id, Prov) =:= {employees, dept_id}
+                       end, Results))
+    ].
+
+%%% Join Provenance Tests
+
+test_join_provenance(DB) ->
+    % Create iterators (metadata always included)
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    % Join them
+    JoinIter = operations:equijoin_iterator(EmpIter, DeptIter, dept_id),
+    Results = operations:collect_all(JoinIter),
+
+    % All results should have metadata with provenance
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) -> maps:is_key(meta, T) end, Results)),
+     % Check that ALL tuples have correct provenance structure
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Prov = maps:get(provenance, Meta),
+                           % Employee attributes should come from employees
+                           maps:get(id, Prov) =:= {employees, id} andalso
+                           maps:get(name, Prov) =:= {employees, name} andalso
+                           % Department attributes should come from departments
+                           maps:get(dept_name, Prov) =:= {departments, dept_name} andalso
+                           maps:get(budget, Prov) =:= {departments, budget}
+                       end, Results))
+    ].
+
+test_join_with_conflicts_provenance(DB) ->
+    % Create relations with conflicting attribute names
+    {DB1, _} = operations:create_relation(DB, table_a, #{
+        id => integer,
+        value => integer
+    }),
+    {DB2, _} = operations:create_tuple(DB1, table_a, #{id => 1, value => 100}),
+
+    {DB3, _} = operations:create_relation(DB2, table_b, #{
+        id => integer,
+        value => integer  % Conflict!
+    }),
+    {DB4, _} = operations:create_tuple(DB3, table_b, #{id => 1, value => 200}),
+
+    % Create iterators with provenance
+    AIter = operations:get_tuples_iterator(DB4, table_a, #{}),
+    BIter = operations:get_tuples_iterator(DB4, table_b, #{}),
+
+    JoinIter = operations:equijoin_iterator(AIter, BIter, id),
+    Results = operations:collect_all(JoinIter),
+
+    % Should be exactly one result tuple
+    ?_assertEqual(1, length(Results)),
+
+    % Check provenance handles conflicts correctly
+    [Result] = Results,
+    Meta = maps:get(meta, Result),
+    Prov = maps:get(provenance, Meta),
+
+    [
+     % Left 'value' provenance
+     ?_assertEqual({table_a, value}, maps:get(value, Prov)),
+     % Right 'value' provenance (prefixed)
+     ?_assertEqual({table_b, value}, maps:get(right_value, Prov)),
+     % Join attribute from left side
+     ?_assertEqual({table_a, id}, maps:get(id, Prov)),
+     % Join attribute from right side (prefixed)
+     ?_assertEqual({table_b, id}, maps:get(right_id, Prov))
+    ].
+
+test_equijoin_preserves_both_sources(DB) ->
+    % Even though equijoin only keeps one dept_id value,
+    % provenance should track both sources
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    JoinIter = operations:equijoin_iterator(EmpIter, DeptIter, dept_id),
+    Results = operations:collect_all(JoinIter),
+
+    % Check that ALL tuples have both dept_id sources in provenance
+    [
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Prov = maps:get(provenance, Meta),
+                           % Left side dept_id
+                           maps:is_key(dept_id, Prov) andalso
+                           % Right side dept_id (prefixed)
+                           maps:is_key(right_dept_id, Prov)
+                       end, Results)),
+     % Verify the sources are correct for ALL tuples
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Prov = maps:get(provenance, Meta),
+                           maps:get(dept_id, Prov) =:= {employees, dept_id} andalso
+                           maps:get(right_dept_id, Prov) =:= {departments, dept_id}
+                       end, Results))
+    ].
+
+%%% Lineage Tests
+
+test_base_lineage(DB) ->
+    % Query tuples (metadata always included)
+    Iter = operations:get_tuples_iterator(DB, employees, #{}),
+    Results = operations:collect_all(Iter),
+
+    % All tuples should have lineage showing base relation
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           Lineage =:= {base, employees}
+                       end, Results))
+    ].
+
+test_join_lineage(DB) ->
+    % Create iterators (metadata always included)
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    % Join them
+    JoinIter = operations:equijoin_iterator(EmpIter, DeptIter, dept_id),
+    Results = operations:collect_all(JoinIter),
+
+    % All results should have lineage showing join operation
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           % Should be a join of two base relations
+                           case Lineage of
+                               {join, dept_id, {base, employees}, {base, departments}} -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+%%% Comprehensive Lineage Tests
+
+test_select_lineage(DB) ->
+    % Test lineage through select (filter) operation
+    Iter = operations:get_tuples_iterator(DB, employees, #{}),
+
+    Predicate = fun(T) -> maps:get(id, T, 0) =:= 1 end,
+    FilteredIter = operations:select_iterator(Iter, Predicate),
+    Results = operations:collect_all(FilteredIter),
+
+    % Should have lineage: select wrapping base
+    [
+     ?_assertEqual(1, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {select, Pred, {base, employees}} when is_function(Pred) -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+test_project_lineage(DB) ->
+    % Test lineage through project operation
+    Iter = operations:get_tuples_iterator(DB, employees, #{}),
+    ProjectedIter = operations:project_iterator(Iter, [name, dept_id]),
+    Results = operations:collect_all(ProjectedIter),
+
+    % Should have lineage: project wrapping base
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {project, [name, dept_id], {base, employees}} -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+test_sort_lineage(DB) ->
+    % Test lineage through sort operation
+    Iter = operations:get_tuples_iterator(DB, employees, #{}),
+
+    CompareFun = fun(A, B) -> maps:get(id, A, 0) =< maps:get(id, B, 0) end,
+    SortedIter = operations:sort_iterator(Iter, CompareFun),
+    Results = operations:collect_all(SortedIter),
+
+    % Should have lineage: sort wrapping base
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {sort, Cmp, {base, employees}} when is_function(Cmp) -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+test_take_lineage(DB) ->
+    % Test lineage through take (limit) operation
+    Iter = operations:get_tuples_iterator(DB, employees, #{}),
+    TakeIter = operations:take_iterator(Iter, 1),
+    Results = operations:collect_all(TakeIter),
+
+    % Should have lineage: take wrapping base
+    [
+     ?_assertEqual(1, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {take, 1, {base, employees}} -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+test_multi_operation_pipeline_lineage(DB) ->
+    % Test complex pipeline: employees → select(id=1) → join(departments) → project([name, dept_name])
+
+    % 1. Start with employees base relation
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+
+    % 2. Filter to Alice (id=1)
+    Predicate = fun(T) -> maps:get(id, T, 0) =:= 1 end,
+    FilteredEmp = operations:select_iterator(EmpIter, Predicate),
+
+    % 3. Join with departments
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+    JoinedIter = operations:equijoin_iterator(FilteredEmp, DeptIter, dept_id),
+
+    % 4. Project to [name, dept_name]
+    ProjectedIter = operations:project_iterator(JoinedIter, [name, dept_name]),
+
+    Results = operations:collect_all(ProjectedIter),
+
+    % Expected lineage:
+    % {project, [name, dept_name],
+    %   {join, dept_id,
+    %     {select, Pred, {base, employees}},
+    %     {base, departments}}}
+
+    [
+     ?_assertEqual(1, length(Results)),  % Only Alice
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {project, [name, dept_name],
+                                {join, dept_id,
+                                 {select, Pred, {base, employees}},
+                                 {base, departments}}} when is_function(Pred) -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+test_theta_join_lineage(DB) ->
+    % Test lineage for theta-join operation
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+
+    Predicate = fun(Emp, Dept) ->
+        maps:get(dept_id, Emp) =:= maps:get(dept_id, Dept)
+    end,
+
+    JoinIter = operations:theta_join_iterator(EmpIter, DeptIter, Predicate),
+    Results = operations:collect_all(JoinIter),
+
+    % Should have theta_join lineage
+    [
+     ?_assertEqual(2, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {theta_join, Pred, {base, employees}, {base, departments}}
+                                   when is_function(Pred) -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].
+
+test_complex_nested_lineage(DB) ->
+    % Test deeply nested lineage:
+    % employees → select(id=1) → project([name, dept_id]) → join(departments → select(budget>50000)) → take(1)
+
+    % Left side: employees → select → project
+    EmpIter = operations:get_tuples_iterator(DB, employees, #{}),
+    FilteredEmp = operations:select_iterator(EmpIter, fun(T) -> maps:get(id, T, 0) =:= 1 end),
+    ProjectedEmp = operations:project_iterator(FilteredEmp, [name, dept_id]),
+
+    % Right side: departments → select
+    DeptIter = operations:get_tuples_iterator(DB, departments, #{}),
+    FilteredDept = operations:select_iterator(DeptIter, fun(D) -> maps:get(budget, D, 0) > 50000 end),
+
+    % Join both sides
+    JoinedIter = operations:equijoin_iterator(ProjectedEmp, FilteredDept, dept_id),
+
+    % Take first result
+    TakeIter = operations:take_iterator(JoinedIter, 1),
+
+    Results = operations:collect_all(TakeIter),
+
+    % Expected lineage should be deeply nested
+    [
+     ?_assertEqual(1, length(Results)),
+     ?_assert(lists:all(fun(T) ->
+                           Meta = maps:get(meta, T),
+                           Lineage = maps:get(lineage, Meta),
+                           case Lineage of
+                               {take, 1,
+                                {join, dept_id,
+                                 {project, [name, dept_id],
+                                  {select, _, {base, employees}}},
+                                 {select, _, {base, departments}}}} -> true;
+                               _ -> false
+                           end
+                       end, Results))
+    ].