%%% @doc Extended Relational Database Operations (RM/T) %%% %%% This module implements core database operations for Domino, an extended %%% relational database engine based on E. F. Codd's RM/T (1979). %%% %%% == Architecture == %%% %%% The system uses: %%%
%% DB = operations:create_database(my_db),
%% {DB1, _} = operations:create_relation(DB, users, #{name => string, age => integer}),
%% {DB2, _} = operations:create_tuple(DB1, users, #{name => "Alice", age => 30}).
%%
%%
%% @param Database `#database_state{}' record
%% @param RelationName Atom naming the relation
%% @param Tuple Map of attribute names to values (e.g., `#{name => "Alice", age => 30}')
%% @returns `{UpdatedDatabase, UpdatedRelation}' tuple
%%
%% @see create_relation/3
%% @see get_tuples_iterator/2
create_tuple(Database, RelationName, Tuple) when is_map(Tuple), is_record(Database, database_state) ->
%% Check if relation is immutable before attempting insert
CurrentRelationHash = maps:get(RelationName, Database#database_state.relations),
[RelationRecord] = mnesia:dirty_read(relation, CurrentRelationHash),
%% Validate tuple against relation schema before insertion
case constraint:validate_tuple(Tuple, RelationRecord) of
ok ->
create_tuple_internal(Database, RelationName, Tuple, RelationRecord);
{error, Reason} ->
{error, {constraint_violation, Reason}}
end.
%% @private
%% Internal function for tuple creation
create_tuple_internal(Database, RelationName, Tuple, _RelationRecord) ->
F = fun() ->
%% Step 1: Store each attribute value and build attribute_map
AttributeMap = maps:map(fun(_AttrName, Value) ->
ValueHash = hash(Value),
mnesia:write(attribute, #attribute{hash = ValueHash, value = Value}, write),
ValueHash
end, Tuple),
%% Step 2: Compute tuple hash and store tuple
TupleHash = hash(Tuple),
mnesia:write(tuple, #tuple{hash = TupleHash, relation = RelationName, attribute_map = AttributeMap}, write),
%% Step 3: Get current relation hash from database
CurrentRelationHash = maps:get(RelationName, Database#database_state.relations),
%% Step 4: Read current relation
[RelationRecord] = mnesia:read(relation, CurrentRelationHash),
%% Step 5: Insert tuple hash into relation merkle tree
%% We don't need to store the content of a tuple hash, just the key
%% merklet can spot the difference in both keys and values
%% sets:set follows the same strategy of an empty value
NewRelationTree = merklet:insert({TupleHash, <<>>}, RelationRecord#relation.tree),
%% Step 6: Compute new relation hash as the root of the new merkle tree and store
{_,NewRelationTreeHash,_,_} = NewRelationTree,
NewRelationHash = hash({RelationRecord#relation.name, RelationRecord#relation.schema, NewRelationTreeHash}),
%% Update cardinality: increment by 1
NewCardinality = case RelationRecord#relation.cardinality of
{finite, N} -> {finite, N + 1};
Other -> Other % Shouldn't happen for finite relations
end,
UpdatedRelation = #relation{
hash = NewRelationHash,
name = RelationName,
tree = NewRelationTree,
schema = RelationRecord#relation.schema,
constraints = RelationRecord#relation.constraints,
cardinality = NewCardinality,
generator = RelationRecord#relation.generator,
provenance = RelationRecord#relation.provenance,
lineage = RelationRecord#relation.lineage,
membership_criteria = RelationRecord#relation.membership_criteria
},
%% Delete old relation record before writing new one
%% This ensures only one relation record exists per name
mnesia:delete({relation, CurrentRelationHash}),
mnesia:write(relation, UpdatedRelation, write),
%% Step 7: Update database relations map
NewRelations = maps:put(RelationName, NewRelationHash, Database#database_state.relations),
%% Step 8: Update database tree (for diffing)
RelationKey = atom_to_binary(RelationName, utf8),
NewDatabaseTree = merklet:insert({RelationKey, NewRelationHash}, Database#database_state.tree),
%% Step 9: Compute new database hash and store
{_,NewDatabaseHash,_,_} = NewDatabaseTree,
UpdatedDatabase = #database_state{
name = Database#database_state.name,
hash = NewDatabaseHash,
tree = NewDatabaseTree,
relations = NewRelations,
timestamp = erlang:timestamp()
},
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, UpdatedRelation}
end,
{atomic, Result} = mnesia:transaction(F),
Result.
%% @doc Clear all tuples from a relation (truncate).
%%
%% Wrapper for `clear_relation/3' with transactions enabled.
%%
%% @param Database `#database_state{}' record
%% @param RelationHash Binary hash identifying the relation
%% @returns `{atomic, {UpdatedDatabase, UpdatedRelation}}' or `{aborted, Reason}'
%%
%% @see clear_relation/3
%% @see retract_relation/2
clear_relation(Database, RelationHash) ->
clear_relation(Database, RelationHash, true).
%% @doc Clear all tuples from a relation with optional transaction control.
%%
%% Removes all tuples from the relation but preserves its schema. The relation's
%% merkle tree is set to `undefined', and a new relation hash is computed.
%%
%% @param Database `#database_state{}' record
%% @param RelationHash Binary hash identifying the relation
%% @param Transact Boolean - if `true', wraps operation in Mnesia transaction
%% @returns When `Transact =:= true': `{atomic, {UpdatedDatabase, UpdatedRelation}}'
%% When `Transact =:= false': `{UpdatedDatabase, UpdatedRelation}'
%%
%% @see retract_relation/2
clear_relation(Database, RelationHash, Transact) ->
[Relation] = mnesia:dirty_read(relation, RelationHash),
NewRelationHash = hash({Relation#relation.name, Relation#relation.schema, undefined}),
UpdatedRelation =
Relation#relation{hash = NewRelationHash,
tree = undefined},
DatabaseTreeWithoutTuples = merklet:insert({atom_to_binary(Relation#relation.name, utf8), NewRelationHash}, Database#database_state.tree),
{_,NewDatabaseHash,_,_} = DatabaseTreeWithoutTuples,
NewRelations = maps:put(Relation#relation.name, NewRelationHash, Database#database_state.relations),
UpdatedDatabase =
Database#database_state{timestamp = erlang:timestamp(),
hash = NewDatabaseHash,
tree = DatabaseTreeWithoutTuples,
relations = NewRelations},
TX = fun () ->
mnesia:write(relation, UpdatedRelation, write),
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, UpdatedRelation}
end,
case Transact of
true -> mnesia:transaction(TX);
false -> {UpdatedDatabase, UpdatedRelation}
end.
%% @doc Remove a tuple from a relation.
%%
%% Wrapper for `retract_tuple/4' with transactions enabled.
%%
%% @param Database `#database_state{}' record
%% @param RelationHash Binary hash identifying the relation
%% @param TupleHash Binary hash identifying the tuple to remove
%% @returns `{atomic, {UpdatedDatabase, UpdatedRelation}}' or `{aborted, Reason}'
%%
%% @see retract_tuple/4
%% @see create_tuple/3
retract_tuple(Database, RelationHash, TupleHash) ->
retract_tuple(Database, RelationHash, TupleHash, true).
%% @doc Remove a tuple from a relation with optional transaction control.
%%
%% Deletes a specific tuple from the relation's merkle tree and recomputes
%% the relation and database hashes.
%%
%% @param Database `#database_state{}' record
%% @param RelationHash Binary hash identifying the relation
%% @param TupleHash Binary hash identifying the tuple to remove
%% @param Transact Boolean - if `true', wraps operation in Mnesia transaction
%% @returns When `Transact =:= true': `{atomic, {UpdatedDatabase, UpdatedRelation}}'
%% When `Transact =:= false': `{UpdatedDatabase, UpdatedRelation}'
%%
%% @see create_tuple/3
retract_tuple(Database, RelationHash, TupleHash, Transact)
when is_record(Database, database_state) ->
[Relation] = mnesia:dirty_read(relation, RelationHash),
RelationTreeWithoutTuple = merklet:delete(TupleHash, Relation#relation.tree),
{_,NewRelationTreeHash,_,_} = RelationTreeWithoutTuple,
NewRelationHash = hash({Relation#relation.name, Relation#relation.schema, NewRelationTreeHash}),
DatabaseTreeWithoutTuple = merklet:insert({atom_to_binary(Relation#relation.name, utf8), NewRelationHash}, Database#database_state.tree),
{_,NewDatabaseHash,_,_} = DatabaseTreeWithoutTuple,
NewRelations = maps:put(Relation#relation.name, NewRelationHash, Database#database_state.relations),
UpdatedDatabase =
Database#database_state{timestamp = erlang:timestamp(),
hash = NewDatabaseHash,
tree = DatabaseTreeWithoutTuple,
relations = NewRelations},
UpdatedRelation =
Relation#relation{hash = NewRelationHash,
tree = RelationTreeWithoutTuple},
TX = fun () ->
mnesia:write(relation, UpdatedRelation, write),
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, UpdatedRelation}
end,
case Transact of
true -> mnesia:transaction(TX);
false -> {UpdatedDatabase, UpdatedRelation}
end.
%% @doc Remove a relation from the database.
%%
%% Wrapper for `retract_relation/3' with transactions enabled.
%%
%% @param Database `#database_state{}' record
%% @param Name Atom naming the relation to remove
%% @returns `{atomic, UpdatedDatabase}' or `{aborted, Reason}'
%%
%% @see retract_relation/3
%% @see clear_relation/2
retract_relation(Database, Name) ->
retract_relation(Database, Name, true).
%% @doc Remove a relation from the database with optional transaction control.
%%
%% First clears all tuples from the relation, then removes the relation entirely
%% from the database's relation map and merkle tree.
%%
%% @param Database `#database_state{}' record
%% @param Name Atom naming the relation to remove
%% @param Transact Boolean - if `true', wraps operation in Mnesia transaction
%% @returns When `Transact =:= true': `{atomic, UpdatedDatabase}'
%% When `Transact =:= false': `UpdatedDatabase'
%%
%% @see create_relation/3
retract_relation(Database, Name, Transact) ->
RelationHash = maps:get(Name, Database#database_state.relations, {error, retract_relation_could_not_find}),
{DatabaseWithClearedRelation, _ClearedRelation} = clear_relation(Database, RelationHash, false),
NewTree = merklet:delete(atom_to_binary(Name, utf8), DatabaseWithClearedRelation#database_state.tree),
NewHash = case NewTree of
undefined -> <<>>;
{_,H,_,_} -> H
end,
UpdatedDatabase =
DatabaseWithClearedRelation#database_state{relations = maps:remove(Name, DatabaseWithClearedRelation#database_state.relations),
hash = NewHash,
tree = NewTree,
timestamp = erlang:timestamp()},
TX = fun () ->
mnesia:write(database_state, UpdatedDatabase, write),
UpdatedDatabase
end,
case Transact of
true -> mnesia:transaction(TX);
false -> UpdatedDatabase
end.
%% @doc Create a new relation in the database with a schema definition.
%%
%% Creates an empty relation (no tuples) with the specified schema. The schema
%% is stored but not yet enforced - it serves as documentation and provides a
%% foundation for future domain validation.
%%
%% Important: The relation's identity is tied to its name. Two relations
%% with identical schemas but different names are semantically distinct and can
%% contain different tuples.
%%
%% == Example ==
%%
%% DB = operations:create_database(my_db),
%% Schema = #{name => string, age => integer, email => string},
%% {DB1, _Rel} = operations:create_relation(DB, users, Schema).
%%
%%
%% @param Database `#database_state{}' record
%% @param Name Atom naming the new relation
%% @param Definition Map specifying schema (e.g., `#{name => string, age => integer}')
%% @returns `{UpdatedDatabase, NewRelation}' tuple
%%
%% @see create_database/1
%% @see create_tuple/3
%% @see retract_relation/2
create_relation(Database, Name, Definition) when is_record(Database, database_state) ->
%% Validate schema before creating relation
case constraint:validate_schema(Definition) of
{error, Reason} ->
{error, {invalid_schema, Reason}};
ok ->
create_relation_internal(Database, Name, Definition)
end.
%% @private
%% Internal function for relation creation after schema validation
create_relation_internal(Database, Name, Definition) ->
F = fun() ->
%% Create relation with empty tree
%% The meaning of a relation needs to be attached to its name
%% It may sound nominalistic but that is a physical concern.
%% Two relations might share definitions but their name gives another interpretation
%% and therefore different tuples
RelationHash = hash({Name, Definition, undefined}),
%% Returns a generator function (not a PID) that yields tuples one at a time
%% TODO: investigate if a generator should or should not be bound to a specific DATABASE VERSION. If so, create the fun passing the DB as an attribute
GeneratorFun = fun() ->
[CurrentRelation] = mnesia:dirty_index_read(relation, Name, #relation.name),
CurrentHash = CurrentRelation#relation.hash,
TupleHashes = case CurrentRelation#relation.tree of
undefined -> [];
_ -> hashes_from_tuple(CurrentHash)
end,
InnerGen = make_finite_generator(TupleHashes),
spawn(fun() -> generator_iterator_loop(InnerGen) end)
end,
%% Build membership criteria from schema (1OPiC declarations)
MembershipCriteria = constraint:build_membership_criteria(Definition),
NewRelation = #relation{
hash = RelationHash,
name = Name,
tree = undefined,
schema = Definition,
constraints = constraint:empty_constraints(), % No constraints by default
cardinality = {finite, 0}, % Empty relation
generator = GeneratorFun, % Function generator for finite relations
membership_criteria = MembershipCriteria, % Domain membership from schema
provenance = build_base_provenance(Definition, Name), % Base relation provenance
lineage = {base, Name} % Base relation lineage
},
mnesia:write(relation, NewRelation, write),
%% Update database relations map
%% TODO: maybe we can purge this later after including the name on the merkle trees
NewRelations = maps:put(Name, RelationHash, Database#database_state.relations),
%% Update database tree (for diffing)
%% Note that the name here is important as a key because
%% we gotta go from a name to a hash within a database state
NewTree = merklet:insert({atom_to_binary(Name), RelationHash}, Database#database_state.tree),
{_,NewHash,_,_} = NewTree,
UpdatedDatabase = #database_state{
name = Database#database_state.name,
hash = NewHash,
tree = NewTree,
relations = NewRelations,
timestamp = erlang:timestamp()
},
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, NewRelation}
end,
{atomic, Result} = mnesia:transaction(F),
Result.
%% @doc Create a new database state.
%%
%% Initializes an empty database with no relations. The database starts with:
%% %% DB = operations:create_database(my_app). %%%% %% @param Name Atom identifying the database %% @returns `#database_state{}' record %% %% @see setup/0 %% @see create_relation/3 create_database(Name) -> DB = #database_state{ name = Name, hash = <<>>, tree = undefined, relations = #{}, timestamp = erlang:timestamp() }, %% Standard immutable relations (domains and their predicates) StandardTypeRelations = [ %% Boolean domain - finite, immutable #domain{ name = boolean, schema = #{value => boolean}, generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:boolean(#{})) end) end, membership_criteria = #{value => {in, [true, false]}}, cardinality = {finite, 2} }, %% Natural numbers - infinite, immutable #domain{ name = natural, schema = #{value => natural}, generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:natural(#{})) end) end, membership_criteria = #{value => {'and', {gte, 0}, is_integer}}, cardinality = aleph_zero }, %% Integers - infinite, immutable #domain{ name = integer, schema = #{value => integer}, generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:integer(#{})) end) end, membership_criteria = #{value => is_integer}, cardinality = aleph_zero }, %% Rationals - infinite, immutable #domain{ name = rational, schema = #{numerator => integer, denominator => integer}, generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:rational(#{})) end) end, membership_criteria = #{denominator => {neq, 0}}, cardinality = aleph_zero } %% Reals - TODO: implement %% #domain{ %% name = reals, %% schema = #{value => reals}, %% generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:reals(#{})) end) end, %% membership_criteria = #{value => is_float}, %% cardinality = continuum %% } ], Folder = fun (Elem, AccDB) -> {NextDB, _} = operations:create_immutable_relation(AccDB, Elem), NextDB end, StandardDB = lists:foldl(Folder, DB, StandardTypeRelations), {atomic, ok} = mnesia:transaction(fun () -> mnesia:write(database_state, StandardDB, write) end), StandardDB. %% @doc Create an infinite relation with a generator. %% %% Creates a relation with infinite cardinality (aleph_zero or continuum). %% The relation is defined by a generator function that produces tuples %% lazily based on constraints. %% %% == Primitive Infinite Relations == %% %%
%% %% Create immutable relation of natural numbers
%% {DB1, Naturals} = operations:create_immutable_relation(DB, #domain{
%% name = naturals,
%% schema = #{value => naturals},
%% generator = {generators, naturals},
%% membership_criteria = #{value => {'and', {gte, 0}, is_integer}},
%% cardinality = aleph_zero
%% }).
%%
%% %% Query with bounds
%% Iterator = operations:get_tuples_iterator(DB1, naturals,
%% #{value => {range, 0, 100}}).
%%
%%
%% @param Database `#database_state{}' record
%% @param Specification `#domain{}' record
%% @returns `{UpdatedDatabase, NewRelation}'
%%
%% @see create_relation/3
%% @see get_tuples_iterator/3
create_immutable_relation(Database, Specification)
when is_record(Database, database_state)
andalso is_record(Specification, domain) ->
Name = Specification#domain.name,
Schema = Specification#domain.schema,
Generator = Specification#domain.generator,
MembershipCriteria = Specification#domain.membership_criteria,
Cardinality = Specification#domain.cardinality,
F = fun() ->
RelationHash = hash({Name, Schema, undefined}),
NewRelation = #relation{
hash = RelationHash,
name = Name,
tree = undefined, % Immutable relations have no merkle tree
schema = Schema,
constraints = constraint:empty_constraints(), % No constraints by default
cardinality = Cardinality,
generator = Generator,
membership_criteria = MembershipCriteria,
provenance = build_base_provenance(Schema, Name), % Base relation provenance
lineage = {base, Name} % Base relation lineage
},
mnesia:write(relation, NewRelation, write),
%% Update database relations map
NewRelations = maps:put(Name, RelationHash, Database#database_state.relations),
%% Update database tree
NewTree = merklet:insert({atom_to_binary(Name), RelationHash}, Database#database_state.tree),
{_,NewHash,_,_} = NewTree,
UpdatedDatabase = #database_state{
name = Database#database_state.name,
hash = NewHash,
tree = NewTree,
relations = NewRelations,
timestamp = erlang:timestamp()
},
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, NewRelation}
end,
{atomic, Result} = mnesia:transaction(F),
Result.
%% @doc Get relation hash by name.
%%
%% Looks up a relation's hash in the database's relation map. This hash
%% can be used to retrieve the full relation record from Mnesia.
%%
%% @param Database `#database_state{}' record
%% @param RelationName Atom naming the relation
%% @returns `{ok, Hash}' on success, `{error, not_found}' if relation doesn't exist
%%
%% @see get_relations/1
%% @see hashes_from_tuple/1
get_relation_hash(Database, RelationName) ->
case maps:find(RelationName, Database#database_state.relations) of
error -> {error, not_found};
X -> X
end.
%% @doc Get list of all relation names in the database.
%%
%% Returns a list of all relation names (atoms) currently stored in the database.
%% This includes both user-defined relations and built-in infinite relations.
%%
%% @param Database `#database_state{}' record
%% @returns List of relation names (atoms)
-spec get_relations(#database_state{}) -> [atom()].
get_relations(Database) ->
maps:keys(Database#database_state.relations).
%% @doc Resolve a tuple hash to actual values.
%%
%% This is an internal function that materializes a tuple by:
%%
%% #{constraints => #{age => {gt, 30}}}
%%
%%
%% Flat format (backward compatible):
%%
%% #{age => {gt, 30}}
%%
%%
%% == Constraints ==
%%
%% Constraints bound the iteration over infinite relations:
%%
%% %% 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
%% @returns Pid of iterator process
%%
%% @see get_tuples_iterator/2
%% @see take/3
get_tuples_iterator(Database, RelationName) when is_record(Database, database_state) ->
case maps:get(RelationName, Database#database_state.relations, error) of
error ->
erlang:error({error_tuple_iterator_init, relation_not_found, RelationName});
RelationHash ->
[Relation] = mnesia:dirty_read(relation, RelationHash),
case Relation#relation.cardinality of
{finite, _} when Relation#relation.tree =/= undefined ->
TupleHashes = hashes_from_tuple(RelationHash),
spawn(fun() -> tuple_iterator_loop(TupleHashes) end);
_ ->
get_iterator_from_generator(Relation#relation.name, Relation#relation.generator)
end
end.
%% @doc Get an iterator out of a generator.
get_iterator_from_generator(RelationName, GeneratorFun) ->
case GeneratorFun of
undefined -> spawn(fun() -> tuple_iterator_loop([]) end);
Generator when is_function(Generator) -> Generator();
_ -> erlang:error({error_tuple_iterator_init, non_function_generator, RelationName})
end.
%% @doc Get next tuple from iterator.
%%
%% Sends a `{next, self()}' message to the iterator process and waits for
%% the response. The iterator will materialize the tuple on demand.
%%
%% Timeout is set to 5 seconds to prevent indefinite blocking.
%%
%% @param IteratorPid Pid returned from `get_tuples_iterator/2'
%% @returns `{ok, Tuple}' with materialized tuple map,
%% `done' when no more tuples remain, or
%% `{error, timeout}' if iterator doesn't respond
%%
%% @see get_tuples_iterator/2
%% @see close_iterator/1
next_tuple(IteratorPid) ->
IteratorPid ! {next, self()},
receive
{tuple, Tuple} -> {ok, Tuple};
done -> done;
{error, Reason} -> {error, Reason}
after 5000 ->
{error, timeout}
end.
%% @doc Close iterator process.
%%
%% Sends a `stop' message to terminate the iterator process. This should be
%% called when you're done consuming tuples to avoid leaving processes running.
%%
%% @param IteratorPid Pid returned from `get_tuples_iterator/2'
%% @returns `ok'
%%
%% @see get_tuples_iterator/2
%% @see collect_all/1
close_iterator(IteratorPid) ->
IteratorPid ! stop,
ok.
%% @private
%% Iterator loop implementation (internal).
%% This is the process loop for the Volcano Iterator Model. It maintains a list
%% of tuple hashes and resolves them lazily when requested.
%%
%% Protocol:
%% %% Iterator = operations:get_tuples_iterator(DB, users), %% AllTuples = operations:collect_all(Iterator). %% %% Iterator is automatically closed %%%% %% @param IteratorPid Pid returned from `get_tuples_iterator/2' %% @returns List of tuple maps, `{error, Reason, PartialResults}' on timeout %% %% @see get_tuples_iterator/2 %% @see next_tuple/1 collect_all(IteratorPid) -> collect_all(IteratorPid, []). collect_all(IteratorPid, Acc) -> case next_tuple(IteratorPid) of {ok, Tuple} -> collect_all(IteratorPid, [Tuple | Acc]); done -> close_iterator(IteratorPid), lists:reverse(Acc); % Preserve order {error, Reason} -> close_iterator(IteratorPid), {error, Reason, lists:reverse(Acc)} end. %% @doc Select iterator: Filters tuples based on a predicate. %% %% Lazy operator that wraps a child iterator and only emits tuples that %% satisfy the predicate function. %% %% == Example == %%
%% BaseIter = get_tuples_iterator(DB, employees, #{}),
%% FilteredIter = select_iterator(BaseIter, fun(E) -> maps:get(age, E) > 30 end),
%% Tuples = collect_all(FilteredIter).
%%
%%
%% @param ChildIterator Iterator process to filter
%% @param Predicate Function: tuple() -> boolean()
%% @returns Iterator process (Pid)
select_iterator(ChildIterator, Predicate) when is_function(Predicate, 1) ->
spawn(fun() -> select_loop(ChildIterator, Predicate) end).
select_loop(ChildPid, Predicate) ->
receive
{next, Caller} ->
case next_tuple(ChildPid) of
{ok, Tuple} ->
case Predicate(Tuple) of
true ->
Caller ! {tuple, Tuple},
select_loop(ChildPid, Predicate);
false ->
self() ! {next, Caller},
select_loop(ChildPid, Predicate)
end;
done ->
close_iterator(ChildPid),
Caller ! done;
{error, Reason} ->
close_iterator(ChildPid),
Caller ! {error, Reason}
end;
stop ->
close_iterator(ChildPid),
ok
end.
%% @doc Project iterator: Projects tuples onto specified attributes.
%%
%% Lazy operator that transforms tuples by keeping only the specified
%% attributes. This is the relational projection operator (π).
%%
%% == Example ==
%%
%% BaseIter = get_tuples_iterator(DB, employees, #{}),
%% ProjectedIter = project_iterator(BaseIter, [name, age]),
%% Tuples = collect_all(ProjectedIter).
%%
%%
%% @param ChildIterator Iterator process to project
%% @param Attributes List of attribute names to keep
%% @returns Iterator process (Pid)
project_iterator(ChildIterator, Attributes) when is_list(Attributes) ->
spawn(fun() -> project_loop(ChildIterator, Attributes) end).
project_loop(ChildPid, Attributes) ->
receive
{next, Caller} ->
case next_tuple(ChildPid) of
{ok, Tuple} ->
% Project data attributes only - no metadata handling
ProjectedTuple = maps:with(Attributes, Tuple),
Caller ! {tuple, ProjectedTuple},
project_loop(ChildPid, Attributes);
done ->
close_iterator(ChildPid),
Caller ! done;
{error, Reason} ->
close_iterator(ChildPid),
Caller ! {error, Reason}
end;
stop ->
close_iterator(ChildPid),
ok
end.
%% @doc Sort iterator: Sorts all tuples from child iterator.
%%
%% Blocking operator - must consume ALL tuples from child before
%% emitting any output. Materializes and sorts tuples in memory.
%%
%% Warning: Do not use on unbounded streams! Ensure child iterator
%% produces finite tuples (via constraints or take).
%%
%% == Example ==
%%
%% BaseIter = get_tuples_iterator(DB, employees, #{}),
%% SortedIter = sort_iterator(BaseIter, fun(A, B) ->
%% maps:get(age, A) =< maps:get(age, B)
%% end),
%% Tuples = collect_all(SortedIter).
%%
%%
%% @param ChildIterator Iterator process to sort
%% @param CompareFun Function: (tuple(), tuple()) -> boolean()
%% @returns Iterator process (Pid)
sort_iterator(ChildIterator, CompareFun) when is_function(CompareFun, 2) ->
spawn(fun() ->
%% Sync all state blocking the pipeline
AllTuples = collect_all(ChildIterator),
%% Sort in memory
SortedTuples = lists:sort(CompareFun, AllTuples),
sorted_emit_loop(SortedTuples)
end).
sorted_emit_loop([]) ->
receive
{next, Caller} -> Caller ! done;
stop -> ok
end;
sorted_emit_loop([Tuple | Rest]) ->
receive
{next, Caller} ->
Caller ! {tuple, Tuple},
sorted_emit_loop(Rest);
stop -> ok
end.
%% @doc Take iterator: Limits output to N tuples.
%%
%% Lazy operator that emits at most N tuples from the child iterator,
%% then closes the child and signals done.
%%
%% == Example ==
%%
%% BaseIter = get_tuples_iterator(DB, naturals, #{value => {range, 0, 1000}}),
%% LimitedIter = take_iterator(BaseIter, 10),
%% Tuples = collect_all(LimitedIter). % Gets exactly 10 tuples
%%
%%
%% @param ChildIterator Iterator process to limit
%% @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, N) end).
take_iter_loop(ChildPid, Limit, Count, OriginalN) when Count < Limit ->
receive
{next, Caller} ->
case next_tuple(ChildPid) of
{ok, Tuple} ->
%% Pass through tuple without metadata injection
Caller ! {tuple, Tuple},
take_iter_loop(ChildPid, Limit, Count + 1, OriginalN);
done ->
close_iterator(ChildPid),
Caller ! done;
{error, Reason} ->
close_iterator(ChildPid),
Caller ! {error, Reason}
end;
stop ->
close_iterator(ChildPid),
ok
end;
take_iter_loop(ChildPid, _Limit, _Count, _OriginalN) ->
%% Reached limit - close child and signal done
close_iterator(ChildPid),
receive
{next, Caller} -> Caller ! done;
stop -> ok
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) %%% ============================================================================ %% @doc Materialize an iterator pipeline into a new relation. %% %% This is the boundary between lazy evaluation (iterator pipeline) and %% eager materialization (stored relation). Consumes all tuples from the %% iterator and stores them as a new finite relation in the database. %% %% == Example == %%
%% %% Build lazy pipeline
%% Pipeline = get_tuples_iterator(DB, employees, #{})
%% |> select_iterator(fun(E) -> maps:get(age, E) > 30 end)
%% |> sort_iterator(fun(A, B) -> maps:get(age, A) =< maps:get(age, B) end)
%% |> take_iterator(10)
%% |> project_iterator([name, age]),
%%
%% %% Materialize into new relation
%% {DB1, SeniorEmployees} = materialize(DB, Pipeline, senior_employees).
%%
%%
%% @param Database Current database state
%% @param SourceIterator Iterator process (pipeline endpoint)
%% @param ResultName Atom naming the new relation
%% @returns `{UpdatedDatabase, NewRelation}' tuple
materialize(Database, SourceIterator, ResultName) when is_record(Database, database_state) ->
%% 1. Pull all tuples from pipeline
AllTuples = collect_all(SourceIterator),
case AllTuples of
[] ->
%% Empty result - create empty relation
Schema = #{}, % Empty schema
create_relation(Database, ResultName, Schema);
[FirstTuple | _] ->
%% 2. Infer schema from first tuple
Schema = infer_schema_from_tuple(FirstTuple),
%% 3. Create new relation
{DB1, _Relation} = create_relation(Database, ResultName, Schema),
%% 4. Insert all tuples
{FinalDB, FinalRelation} = create_all_tuples(DB1, ResultName, AllTuples),
{FinalDB, FinalRelation}
end.
%% @private
%% Infer schema from a tuple.
infer_schema_from_tuple(Tuple) when is_map(Tuple) ->
maps:map(fun(_AttrName, Value) -> infer_type(Value) end, Tuple).
%% @private
%% Infer type from value.
infer_type(Value) when is_integer(Value) -> integer;
infer_type(Value) when is_float(Value) -> float;
infer_type(Value) when is_binary(Value) -> binary;
infer_type(Value) when is_list(Value) -> string;
infer_type(Value) when is_atom(Value) -> atom;
infer_type(Value) when is_boolean(Value) -> boolean;
infer_type(_) -> term.
%% @private
%% Insert multiple tuples into a relation.
create_all_tuples(DB, RelationName, Tuples) ->
lists:foldl(
fun(Tuple, {DBacc, _RelAcc}) ->
create_tuple(DBacc, RelationName, Tuple)
end,
{DB, undefined},
Tuples
).
%%% ============================================================================
%%% Generator Support Functions
%%% ============================================================================
%% @private
%% Generator iterator loop for infinite relations.
%% Similar to tuple_iterator_loop but works with generator functions instead
%% of tuple hashes.
generator_iterator_loop(Generator) ->
receive
{next, Caller} ->
case Generator(next) of
done ->
Caller ! done;
{value, Tuple, NextGenerator} ->
Caller ! {tuple, Tuple},
generator_iterator_loop(NextGenerator);
{error, Reason} ->
Caller ! {error, Reason}
end;
stop ->
ok
end.
%% @private
%% 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)
|| 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} ->
close_iterator(LeftIter),
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
%% 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)
|| 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} ->
close_iterator(LeftIter),
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 Simple tuple merge for joins
%% Equally named and typed attributes are not repeated.
%%
%% @param LeftTuple Left side tuple
%% @param RightTuple Right side tuple
merge_tuples(LeftTuple, RightTuple) ->
maps:fold(
fun(RightKey, RightValue, LeftTupleAcc) ->
case maps:get(RightKey, LeftTupleAcc, undefined) of
LeftValue when LeftValue =:= RightValue ->
% Equal attributes, keep single value
LeftTupleAcc;
undefined ->
% No conflict, add right attribute
maps:put(RightKey, RightValue, LeftTupleAcc);
_ ->
% Different attributes
maps:put(RightKey, RightValue, LeftTupleAcc)
end
end,
LeftTuple,
RightTuple
).
%% @private
%% Build base provenance mapping for a relation.
%% Maps each attribute to its source relation and attribute name.
%%
%% @param Schema Map of attribute names to types
%% @param RelationName Name of the source relation
%% @returns Provenance map #{attr => {relation, attr}}
build_base_provenance(Schema, RelationName) ->
maps:fold(
fun(AttrName, _Type, Acc) ->
maps:put(AttrName, {RelationName, AttrName}, Acc)
end,
#{},
Schema
).
%% The issue here is that the creation of a generator on the create_relation is wrong.
%% I assume it was because of the fun(Constraints), which I believe should receive an
%% optional database version, or just build from the current relation attributes as a factory.
%% @private
%% @doc Create a generator function from a list of tuple hashes.
%%
%% Converts a list of tuple hashes into a generator function compatible
%% with the generator protocol used by infinite relations. The generator
%% reads tuples from Mnesia on demand.
%%
%% @param TupleHashes List of tuple hashes to iterate over
%% @returns Generator function that accepts 'next' and returns {value, Tuple, NextGen} or done
make_finite_generator([]) ->
fun(_) -> done end;
make_finite_generator([Hash | Rest]) ->
fun(next) ->
ResolvedTuple = resolve_tuple(Hash),
NextGen = make_finite_generator(Rest),
{value, ResolvedTuple, NextGen}
end.