aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorMarcos Magueta <maguetamarcos@gmail.com>2025-11-24 22:40:22 -0300
committerGitHub <noreply@github.com>2025-11-24 22:40:22 -0300
commit62a017ce95ad2f6d9445b5b96f7aeba3f7c0d42f (patch)
treed67f93407ad049e4dc39fae1002146a8d03e4505
parent09be36e5f5e2c5714b9542df8ccc363ab86ac6da (diff)
downloadSakura-62a017ce95ad2f6d9445b5b96f7aeba3f7c0d42f.tar.gz
Sakura-62a017ce95ad2f6d9445b5b96f7aeba3f7c0d42f.zip
Joins with provenance and lineage (#11)
-rw-r--r--docs/features.org172
-rw-r--r--include/operations.hrl31
-rw-r--r--src/generators.erl26
-rw-r--r--src/operations.erl422
-rw-r--r--test/function_relations_test.erl70
-rw-r--r--test/infinite_relations_test.erl22
-rw-r--r--test/join_test.erl279
-rw-r--r--test/provenance_test.erl420
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))
+ ].