%%% @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: %%% %%% %%% == Data Model == %%% %%% The system maintains four tables in Mnesia: %%% %%% %%% == RM/T Integrity Rules == %%% %%% %%% %%% @author Nekoma Team %%% @copyright 2025 %%% @version 0.1.0 %%% @reference E. F. Codd (1979). "Extending the Database Relational Model to Capture More Meaning" %%% @reference E. F. Codd (1970). "A Relational Model of Data for Large Shared Data Banks" -module(operations). -include("../include/operations.hrl"). -export([ create_database/1, create_relation/3, create_immutable_relation/2, create_tuple/3, create_all_tuples/3, retract_tuple/3, retract_tuple/4, clear_relation/2, clear_relation/3, retract_relation/2, retract_relation/3, get_relation_hash/2, get_relations/1, hashes_from_tuple/1, get_tuples_iterator/2, get_iterator_from_generator/2, next_tuple/1, close_iterator/1, collect_all/1, %% Iterator operators (lazy) select_iterator/2, project_iterator/2, sort_iterator/2, take_iterator/2, %% Join operators equijoin_iterator/3, theta_join_iterator/3, %% Materialization (eager) materialize/3 ]). %% @private %% @doc Compute SHA-256 hash of an Erlang term. %% %% This is the foundational hashing function used throughout the system for %% content addressing. All entities (attributes, tuples, relations, databases) %% are identified by hashes computed with this function. %% %% @param Value Any Erlang term to hash %% @returns Binary SHA-256 hash (32 bytes) %% %% @see create_tuple/3 %% @see create_relation/3 hash(Value) -> crypto:hash(sha256, term_to_binary(Value)). %% @doc Insert a tuple into a relation and update the database state. %% %% This is the primary operation for adding data to the database. It: %%
    %%
  1. Stores each attribute value with its hash (automatic deduplication)
  2. %%
  3. Creates a tuple record mapping attribute names to value hashes
  4. %%
  5. Inserts the tuple hash into the relation's merkle tree
  6. %%
  7. Computes new relation hash from updated tree
  8. %%
  9. Updates database merkle tree with new relation hash
  10. %%
  11. Computes new database hash
  12. %%
%% %% All operations execute within a Mnesia transaction for ACID guarantees. %% %% == Example == %%
%% 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: %% %% %% All changes are committed to Mnesia within a transaction. %% %% == Example == %%
%% 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 == %% %% %% %% == Example == %%
%% %% 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: %%
    %%
  1. Reading the tuple record from Mnesia (contains attribute_map)
  2. %%
  3. For each attribute, reading the actual value using its hash
  4. %%
  5. Building a map with attribute names and materialized values
  6. %%
%% %% Uses dirty reads since all data is immutable. %% %% @param TupleHash Binary hash identifying the tuple %% @returns Map with attribute names and actual values (e.g., `#{name => "Alice", age => 30}') %% %% @see get_tuples_iterator/2 %% @see create_tuple/3 resolve_tuple(TupleHash) -> [#tuple{attribute_map = AttributeMap}] = mnesia:dirty_read(tuple, TupleHash), maps:map(fun(_AttrName, ValueHash) -> [#attribute{value = Value}] = mnesia:dirty_read(attribute, ValueHash), Value end, AttributeMap). %% @doc Extract all tuple hashes from a relation. %% %% Retrieves the list of all tuple hashes stored in the relation's merkle tree. %% Returns an empty list if the relation has no tuples (tree is `undefined'). %% %% Uses dirty reads since the relation data is immutable. %% %% @param RelationHash Binary hash identifying the relation %% @returns List of binary tuple hashes, or empty list `[]' if relation is empty %% %% @see get_tuples_iterator/2 %% @see create_tuple/3 hashes_from_tuple(RelationHash) -> [Relation] = mnesia:dirty_read(relation, RelationHash), case Relation#relation.tree of undefined -> []; Tree -> merklet:keys(Tree) end. %% @doc Create an iterator for streaming tuples from a relation with constraints. %% %% Extended version of get_tuples_iterator/2 that accepts boundedness constraints. %% For infinite relations, constraints are mandatory to ensure termination. %% %% All tuples include metadata tracking with a nested `meta' field containing: %% %% %% == Options Format == %% %% Nested format (preferred): %%
%% #{constraints => #{age => {gt, 30}}}
%% 
%% %% Flat format (backward compatible): %%
%% #{age => {gt, 30}}
%% 
%% %% == Constraints == %% %% Constraints bound the iteration over infinite relations: %% %% %% == Example == %%
%% %% 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: %% %% %% @param TupleHashes List of binary tuple hashes to stream tuple_iterator_loop([]) -> receive {next, Caller} -> Caller ! done; stop -> ok end; tuple_iterator_loop([TupleHash | Rest]) -> receive {next, Caller} -> %% Resolve tuple on demand ResolvedTuple = resolve_tuple(TupleHash), Caller ! {tuple, ResolvedTuple}, tuple_iterator_loop(Rest); stop -> ok end. %% @doc Helper to collect all tuples from an iterator into a list. %% %% This is a convenience function primarily used for testing. It exhausts the %% iterator by repeatedly calling `next_tuple/1' until `done' is received, %% accumulating all tuples into a list. %% %% Warning: This materializes all tuples in memory, defeating the purpose %% of lazy evaluation. Only use for small relations or testing. %% %% == Example == %%
%% 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.