Joins with provenance and lineage (#11)

4 files changed, 758 insertions, 33 deletions

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))
+    ].