diff options
Diffstat (limited to '')
| -rw-r--r-- | src/operations.erl | 1484 |
1 files changed, 0 insertions, 1484 deletions
diff --git a/src/operations.erl b/src/operations.erl deleted file mode 100644 index de61fd8..0000000 --- a/src/operations.erl +++ /dev/null @@ -1,1484 +0,0 @@ -%%% @doc Extended Relational Database Operations (RM/T) -%%% -%%% This module implements core database operations for Sakura, an extended -%%% relational database engine based on E. F. Codd's RM/T (1979). -%%% -%%% == Architecture == -%%% -%%% The system uses: -%%% <ul> -%%% <li><b>Content-Addressed Storage</b> - All entities identified by SHA-256 hashes</li> -%%% <li><b>Immutable Data Structures</b> - Operations create new versions</li> -%%% <li><b>Merkle Trees</b> - Efficient versioning and diffing via merklet</li> -%%% <li><b>Mnesia Backend</b> - ACID transactions and persistent storage</li> -%%% <li><b>Volcano Iterator Model</b> - Lazy evaluation with process-based iterators</li> -%%% </ul> -%%% -%%% == Data Model == -%%% -%%% The system maintains four tables in Mnesia: -%%% <ul> -%%% <li>`attribute' - Stores atomic values with their hashes</li> -%%% <li>`tuple' - Stores tuples as maps from attribute names to value hashes</li> -%%% <li>`relation' - Stores relations with merkle trees of tuple hashes</li> -%%% <li>`database_state' - Stores database snapshots with relation merkle trees</li> -%%% </ul> -%%% -%%% == RM/T Integrity Rules == -%%% -%%% <ul> -%%% <li><b>Rule 3 (Entity Integrity)</b> - E-relations accept insertions/deletions, not updates</li> -%%% <li><b>Rule 4 (Property Integrity)</b> - Tuples require entity existence in E-relation</li> -%%% </ul> -%%% -%%% @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, - value_hash/1, - create_relation/3, - create_immutable_relation/2, - update_relation_constraints/3, - 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 Hash a value using Sakura's canonical content hash. -%% -%% Public wrapper around the internal hash function, used by other modules -%% that need to compare values against stored attribute hashes. -%% -%% @param Value Any Erlang term -%% @returns Binary SHA-256 hash --spec value_hash(term()) -> binary(). -value_hash(Value) -> - hash(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: -%% <ol> -%% <li>Stores each attribute value with its hash (automatic deduplication)</li> -%% <li>Creates a tuple record mapping attribute names to value hashes</li> -%% <li>Inserts the tuple hash into the relation's merkle tree</li> -%% <li>Computes new relation hash from updated tree</li> -%% <li>Updates database merkle tree with new relation hash</li> -%% <li>Computes new database hash</li> -%% </ol> -%% -%% All operations execute within a Mnesia transaction for ACID guarantees. -%% -%% == Example == -%% <pre> -%% 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}). -%% </pre> -%% -%% @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(Database, 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 - }, - 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. -%% -%% <b>Important:</b> 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 == -%% <pre> -%% DB = operations:create_database(my_db), -%% Schema = #{name => string, age => integer, email => string}, -%% {DB1, _Rel} = operations:create_relation(DB, users, Schema). -%% </pre> -%% -%% @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) -> - 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(Database, 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 Update a relation's constraints. -%% -%% This function updates the constraints on an existing relation and returns -%% an updated database state with the new relation version. The merklet tree -%% automatically replaces the old relation hash with the new one. -%% -%% == Example == -%% <pre> -%% %% Create constraints -%% AgeConstraint = constraint:create_1op(age, integer, [constraint:gte(18), constraint:lt(70)]), -%% SalaryConstraint = constraint:create_1op(salary, integer, [constraint:between(30000, 200000)]), -%% C1 = constraint:add_attribute_constraint(undefined, age, AgeConstraint), -%% C2 = constraint:add_attribute_constraint(C1, salary, SalaryConstraint), -%% -%% %% Update relation with constraints -%% {DB2, UpdatedRel} = operations:update_relation_constraints(DB1, employees, C2). -%% </pre> -%% -%% @param Database Current database state -%% @param RelationName Name of the relation to update -%% @param Constraints The #relation_constraints{} record to set -%% @returns {UpdatedDatabase, UpdatedRelation} -update_relation_constraints(Database, RelationName, Constraints) - when is_record(Database, database_state), - is_atom(RelationName), - is_record(Constraints, relation_constraints) -> - F = fun() -> - %% Get current relation hash from database state - CurrentRelationHash = maps:get(RelationName, Database#database_state.relations), - - %% Read current relation from mnesia - case mnesia:read(relation, CurrentRelationHash) of - [] -> - %% Relation not found, might be using wrong database version - mnesia:abort({relation_not_found, RelationName, CurrentRelationHash}); - [CurrentRelation] -> - update_relation_with_constraints(CurrentRelation, RelationName, Constraints, Database) - end - end, - {atomic, Result} = mnesia:transaction(F), - Result. - -%% @private -update_relation_with_constraints(CurrentRelation, RelationName, Constraints, Database) -> - %% Create updated relation with new constraints - %% The hash changes because constraints are part of the relation definition - NewRelationHash = hash({CurrentRelation#relation.name, - CurrentRelation#relation.schema, - Constraints}), - - UpdatedRelation = CurrentRelation#relation{ - hash = NewRelationHash, - constraints = Constraints - }, - - %% Write updated relation - mnesia:write(relation, UpdatedRelation, write), - - %% Update database relations map - NewRelations = maps:put(RelationName, NewRelationHash, Database#database_state.relations), - - %% Update database tree (merklet:insert automatically replaces old value) - NewTree = merklet:insert({atom_to_binary(RelationName), NewRelationHash}, - 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, UpdatedRelation}. - -%% @doc Create a new database state. -%% -%% Initializes an empty database with no relations. The database starts with: -%% <ul> -%% <li>Empty relation map</li> -%% <li>Undefined merkle tree (will be created when first relation is added)</li> -%% <li>Empty hash</li> -%% <li>Timestamp of creation</li> -%% </ul> -%% -%% All changes are committed to Mnesia within a transaction. -%% -%% == Example == -%% <pre> -%% DB = operations:create_database(my_app). -%% </pre> -%% -%% @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) - %% TODO: Since 1OP constraints were added, the constraints passed to the base models now must be supported. - %% The idea is to have the generators be able to produce the constrained subsets - StandardTypeRelations = [ - #domain{ - name = integer, - schema = #{value => integer}, - generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:integer(#{})) end) end, - membership_criteria = #{test => fun(#{value := V}) -> is_integer(V) end}, - cardinality = aleph_zero - }, - #domain{ - name = natural, - schema = #{value => natural}, - generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:natural(#{})) end) end, - membership_criteria = #{test => fun(#{value := V}) -> is_integer(V) andalso V >= 0 end}, - cardinality = aleph_zero - }, - #domain{ - name = rational, - schema = #{numerator => integer, denominator => integer}, - generator = fun() -> spawn(fun() -> generator_iterator_loop(generators:rational(#{})) end) end, - membership_criteria = #{test => fun(#{numerator := N, denominator := D}) -> - is_integer(N) andalso is_integer(D) andalso D /= 0 - end - }, - cardinality = aleph_zero - }, - #domain{ - name = string, - schema = #{value => string}, - generator = undefined, - membership_criteria = #{test => fun(#{value := V}) -> is_list(V) orelse is_binary(V) end}, - cardinality = aleph_zero - }, - #domain{ - name = atom, - schema = #{value => atom}, - generator = undefined, - membership_criteria = #{test => fun(#{value := V}) -> is_atom(V) end}, - cardinality = aleph_zero - }, - #domain{ - name = term, - schema = #{value => term}, - generator = undefined, - membership_criteria = #{test => fun(_) -> true end}, - cardinality = aleph_zero - }, - constraint:less_than(term, aleph_zero), - constraint:less_than_or_equal(term, aleph_zero), - constraint:greater_than(term, aleph_zero), - constraint:greater_than_or_equal(term, aleph_zero), - constraint:equal(term, aleph_zero), - constraint:not_equal(term, aleph_zero), - constraint:plus(integer, aleph_zero), - constraint:times(integer, aleph_zero), - constraint:minus(integer, aleph_zero), - constraint:divide(integer, aleph_zero) - ], - 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 == -%% -%% <ul> -%% <li>`naturals' - Natural numbers: {0, 1, 2, 3, ...}</li> -%% <li>`integers' - Integers: {0, 1, -1, 2, -2, ...}</li> -%% <li>`rationals' - Rationals via Stern-Brocot tree</li> -%% </ul> -%% -%% == Example == -%% <pre> -%% %% 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}}). -%% </pre> -%% -%% @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(), % TODO: No constraints unless by domain extension - 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), - NewRelations = maps:put(Name, RelationHash, Database#database_state.relations), - 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; - -create_immutable_relation(Database, Specification) - when is_record(Database, database_state) - andalso is_record(Specification, relation) -> - F = fun() -> - RelationHash = hash({Specification#relation.name, Specification#relation.schema, undefined}), - UpdatedRelation = Specification#relation{hash = RelationHash}, - mnesia:write(relation, UpdatedRelation, write), - NewRelations = maps:put(Specification#relation.name, RelationHash, Database#database_state.relations), - NewTree = merklet:insert({atom_to_binary(Specification#relation.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, Specification} - 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: -%% <ol> -%% <li>Reading the tuple record from Mnesia (contains attribute_map)</li> -%% <li>For each attribute, reading the actual value using its hash</li> -%% <li>Building a map with attribute names and materialized values</li> -%% </ol> -%% -%% 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. -%% -%% <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): -%% <pre> -%% #{constraints => #{age => {gt, 30}}} -%% </pre> -%% -%% Flat format (backward compatible): -%% <pre> -%% #{age => {gt, 30}} -%% </pre> -%% -%% == Constraints == -%% -%% Constraints bound the iteration over infinite relations: -%% <ul> -%% <li>`{range, Min, Max}' - Range constraint</li> -%% <li>`{eq, Value}' - Equality constraint</li> -%% <li>`{lt, Value}', `{lte, Value}' - Upper bounds</li> -%% <li>`{gt, Value}', `{gte, Value}' - Lower bounds</li> -%% <li>`{in, List}' - Membership constraint</li> -%% </ul> -%% -%% == Example == -%% <pre> -%% %% 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}} -%% %% }} -%% </pre> -%% -%% @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: -%% <ul> -%% <li>`{next, Caller}' - Resolve next tuple and send `{tuple, ResolvedTuple}' or `done'</li> -%% <li>`stop' - Terminate the process</li> -%% </ul> -%% -%% @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. -%% -%% <b>Warning:</b> This materializes all tuples in memory, defeating the purpose -%% of lazy evaluation. Only use for small relations or testing. -%% -%% == Example == -%% <pre> -%% Iterator = operations:get_tuples_iterator(DB, users), -%% AllTuples = operations:collect_all(Iterator). -%% %% Iterator is automatically closed -%% </pre> -%% -%% @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 == -%% <pre> -%% BaseIter = get_tuples_iterator(DB, employees, #{}), -%% FilteredIter = select_iterator(BaseIter, fun(E) -> maps:get(age, E) > 30 end), -%% Tuples = collect_all(FilteredIter). -%% </pre> -%% -%% @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 == -%% <pre> -%% BaseIter = get_tuples_iterator(DB, employees, #{}), -%% ProjectedIter = project_iterator(BaseIter, [name, age]), -%% Tuples = collect_all(ProjectedIter). -%% </pre> -%% -%% @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. -%% -%% <b>Blocking operator</b> - must consume ALL tuples from child before -%% emitting any output. Materializes and sorts tuples in memory. -%% -%% <b>Warning:</b> Do not use on unbounded streams! Ensure child iterator -%% produces finite tuples (via constraints or take). -%% -%% == Example == -%% <pre> -%% 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). -%% </pre> -%% -%% @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 == -%% <pre> -%% BaseIter = get_tuples_iterator(DB, naturals, #{value => {range, 0, 1000}}), -%% LimitedIter = take_iterator(BaseIter, 10), -%% Tuples = collect_all(LimitedIter). % Gets exactly 10 tuples -%% </pre> -%% -%% @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 == -%% <pre> -%% %% 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) -%% </pre> -%% -%% @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 == -%% <pre> -%% %% 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) -%% </pre> -%% -%% @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 == -%% <pre> -%% %% 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). -%% </pre> -%% -%% @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. |