diff options
Diffstat (limited to '')
| -rw-r--r-- | src/relational_operators.erl | 366 |
1 files changed, 366 insertions, 0 deletions
diff --git a/src/relational_operators.erl b/src/relational_operators.erl new file mode 100644 index 0000000..82e1a47 --- /dev/null +++ b/src/relational_operators.erl @@ -0,0 +1,366 @@ +%%% @doc Pure Relational Operators +%%% +%%% This module provides pure relational algebra operators that take relations +%%% as input and return new ephemeral relations as output. No database state +%%% is modified. Each operator returns a relation with a generator closure that +%%% can spawn fresh iterators on demand, allowing reproducible query results. +%%% +%%% == Architecture == +%%% +%%% Relational Operators (Pure Functions): +%%% - Input: One or more #relation{} records +%%% - Output: New ephemeral #relation{} with generator closure +%%% - No side effects, no DB modification +%%% - Generator can be called multiple times for same results +%%% +%%% Iterator Operators (Data Transport): +%%% - Input: Iterator PIDs +%%% - Output: Iterator PID +%%% - Used inside generator closures +%%% - Examples: select_iterator/2, project_iterator/2, take_iterator/2 +%%% +%%% @author Nekoma Team +%%% @copyright 2025 + +-module(relational_operators). + +-include("../include/operations.hrl"). + +-export([ + take/2, + select/2, + project/2, + join/3, + theta_join/3, + rename/2, + sort/2 +]). + +%%% ============================================================================ +%%% Internal Helper Functions +%%% ============================================================================ + +%% @private +%% Generate a random name for ephemeral relations. +generate_ephemeral_name() -> + {UUID, _} = uuid:get_v1(uuid:new(self(), erlang)), + [Base64UUID] = io_lib:format("~s", [base64:encode(UUID)]), + Base64UUID. + +%%% ============================================================================ +%%% Relational Operators +%%% ============================================================================ + +%% @doc Take operator (τ): Returns ephemeral relation with first N tuples. +%% +%% Creates an ephemeral relation containing N arbitrary elements from the +%% source relation. The generator closure captures the source relation and N, +%% spawning fresh iterators on each call. +%% +%% == Example == +%% <pre> +%% NaturalRel = get_relation(DB, natural), +%% Some100 = relational_operators:take(NaturalRel, 100), +%% +%% %% Can call generator multiple times +%% Iter1 = operations:get_iterator_from_generator(ephemeral, Some100#relation.generator), +%% Results1 = operations:collect_all(Iter1), +%% +%% Iter2 = operations:get_iterator_from_generator(ephemeral, Some100#relation.generator), +%% Results2 = operations:collect_all(Iter2), +%% %% Results1 =:= Results2 +%% </pre> +%% +%% @param SourceRelation Source #relation{} record +%% @param N Number of tuples to take +%% @returns Ephemeral #relation{} with generator +-spec take(#relation{}, pos_integer()) -> #relation{}. +take(SourceRelation, N) when is_record(SourceRelation, relation), is_integer(N), N > 0 -> + %% Compute result cardinality + ResultCardinality = case SourceRelation#relation.cardinality of + {finite, M} when M < N -> {finite, M}; + {finite, _} -> {finite, N}; + _ -> {finite, N} + end, + Name = generate_ephemeral_name(), + SourceGen = SourceRelation#relation.generator, + GeneratorFun = fun() -> + SourceIter = SourceGen(), + operations:take_iterator(SourceIter, N) + end, + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = SourceRelation#relation.schema, + constraints = SourceRelation#relation.constraints, + cardinality = ResultCardinality, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = SourceRelation#relation.provenance, + lineage = {take, N, SourceRelation#relation.lineage} + }. + +%% @doc Select operator (σ): Returns ephemeral relation filtered by predicate. +%% +%% Creates an ephemeral relation containing only tuples that satisfy the +%% predicate. The generator closure captures the source relation and predicate. +%% +%% @param SourceRelation Source #relation{} record +%% @param Predicate Function that takes a tuple and returns boolean +%% @returns Ephemeral #relation{} with generator +-spec select(#relation{}, fun((map()) -> boolean())) -> #relation{}. +select(SourceRelation, Predicate) when is_record(SourceRelation, relation), is_function(Predicate, 1) -> + Name = generate_ephemeral_name(), + + SourceGen = SourceRelation#relation.generator, + GeneratorFun = fun() -> + SourceIter = SourceGen(), + operations:select_iterator(SourceIter, Predicate) + end, + + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = SourceRelation#relation.schema, + constraints = SourceRelation#relation.constraints, + cardinality = unknown, % Cannot determine without evaluation + generator = GeneratorFun, + membership_criteria = #{}, + provenance = SourceRelation#relation.provenance, + lineage = {select, Predicate, SourceRelation#relation.lineage} + }. + +%% @doc Project operator (π): Returns ephemeral relation with selected attributes. +%% +%% Creates an ephemeral relation containing only the specified attributes. +%% The generator closure captures the source relation and attribute list. +%% +%% @param SourceRelation Source #relation{} record +%% @param Attributes List of attribute names to keep +%% @returns Ephemeral #relation{} with generator +-spec project(#relation{}, [atom()]) -> #relation{}. +project(SourceRelation, Attributes) when is_record(SourceRelation, relation), is_list(Attributes) -> + Name = generate_ephemeral_name(), + + %% Compute projected schema + ProjectedSchema = maps:with(Attributes, SourceRelation#relation.schema), + + SourceGen = SourceRelation#relation.generator, + GeneratorFun = fun() -> + SourceIter = SourceGen(), + operations:project_iterator(SourceIter, Attributes) + end, + + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = ProjectedSchema, + constraints = SourceRelation#relation.constraints, + cardinality = SourceRelation#relation.cardinality, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = SourceRelation#relation.provenance, + lineage = {project, Attributes, SourceRelation#relation.lineage} + }. + +%% @doc Join operator (⋈): Returns ephemeral relation from equijoin. +%% +%% Creates an ephemeral relation by joining two relations on a common attribute. +%% The generator closure captures both source relations and the join attribute. +%% +%% @param LeftRelation Left #relation{} record +%% @param RightRelation Right #relation{} record +%% @param JoinAttribute Attribute name to join on +%% @returns Ephemeral #relation{} with generator +-spec join(#relation{}, #relation{}, atom()) -> #relation{}. +join(LeftRelation, RightRelation, JoinAttribute) + when is_record(LeftRelation, relation), + is_record(RightRelation, relation), + is_atom(JoinAttribute) -> + + Name = list_to_atom( + "join_" ++ atom_to_list(LeftRelation#relation.name) ++ "_" ++ + atom_to_list(RightRelation#relation.name) ++ "_" ++ + integer_to_list(erlang:unique_integer([positive])) + ), + + %% Merge schemas + MergedSchema = maps:merge(LeftRelation#relation.schema, RightRelation#relation.schema), + + LeftGen = LeftRelation#relation.generator, + RightGen = RightRelation#relation.generator, + GeneratorFun = fun() -> + LeftIter = LeftGen(), + RightIter = RightGen(), + operations:equijoin_iterator(LeftIter, RightIter, JoinAttribute) + end, + + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = MergedSchema, + constraints = #{}, + cardinality = unknown, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = {join, LeftRelation#relation.provenance, RightRelation#relation.provenance}, + lineage = {join, JoinAttribute, LeftRelation#relation.lineage, RightRelation#relation.lineage} + }. + +%% @doc Theta Join operator (⋈θ): Returns ephemeral relation from theta join. +%% +%% Creates an ephemeral relation by joining two relations with an arbitrary +%% predicate. The generator closure captures both source relations and predicate. +%% +%% @param LeftRelation Left #relation{} record +%% @param RightRelation Right #relation{} record +%% @param Predicate Function that takes two tuples and returns boolean +%% @returns Ephemeral #relation{} with generator +-spec theta_join(#relation{}, #relation{}, fun((map(), map()) -> boolean())) -> #relation{}. +theta_join(LeftRelation, RightRelation, Predicate) + when is_record(LeftRelation, relation), + is_record(RightRelation, relation), + is_function(Predicate, 2) -> + + Name = list_to_atom( + "theta_join_" ++ atom_to_list(LeftRelation#relation.name) ++ "_" ++ + atom_to_list(RightRelation#relation.name) ++ "_" ++ + integer_to_list(erlang:unique_integer([positive])) + ), + + MergedSchema = maps:merge(LeftRelation#relation.schema, RightRelation#relation.schema), + + LeftGen = LeftRelation#relation.generator, + RightGen = RightRelation#relation.generator, + GeneratorFun = fun() -> + LeftIter = LeftGen(), + RightIter = RightGen(), + operations:theta_join_iterator(LeftIter, RightIter, Predicate) + end, + + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = MergedSchema, + constraints = #{}, + cardinality = unknown, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = {join, LeftRelation#relation.provenance, RightRelation#relation.provenance}, + lineage = {theta_join, Predicate, LeftRelation#relation.lineage, RightRelation#relation.lineage} + }. + +%% @doc Rename operator (ρ): Returns ephemeral relation with renamed attributes. +%% +%% Creates an ephemeral relation with attributes renamed according to the mapping. +%% The generator closure captures the source relation and rename mappings. +%% +%% @param SourceRelation Source #relation{} record +%% @param RenameMappings Map of old_name => new_name (atoms or strings) +%% @returns Ephemeral #relation{} with generator +-spec rename(#relation{}, #{atom() => atom()}) -> #relation{}. +rename(SourceRelation, RenameMappings) when is_record(SourceRelation, relation), is_map(RenameMappings) -> + Name = generate_ephemeral_name(), + + %% Compute renamed schema + RenamedSchema = maps:fold( + fun(OldAttr, NewAttr, SchemaAcc) -> + case maps:get(OldAttr, SchemaAcc, undefined) of + undefined -> SchemaAcc; + Type -> + SchemaAcc1 = maps:remove(OldAttr, SchemaAcc), + maps:put(NewAttr, Type, SchemaAcc1) + end + end, + SourceRelation#relation.schema, + RenameMappings + ), + + SourceGen = SourceRelation#relation.generator, + GeneratorFun = fun() -> + SourceIter = SourceGen(), + rename_iterator(SourceIter, RenameMappings) + end, + + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = RenamedSchema, + constraints = SourceRelation#relation.constraints, + cardinality = SourceRelation#relation.cardinality, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = SourceRelation#relation.provenance, + lineage = {rename, RenameMappings, SourceRelation#relation.lineage} + }. + +%% @doc Sort operator: Returns ephemeral relation with sorted tuples. +%% +%% Creates an ephemeral relation with tuples sorted by the comparator. +%% WARNING: This is a blocking operation - it materializes all tuples. +%% +%% @param SourceRelation Source #relation{} record +%% @param Comparator Function that compares two tuples +%% @returns Ephemeral #relation{} with generator +-spec sort(#relation{}, fun((map(), map()) -> boolean())) -> #relation{}. +sort(SourceRelation, Comparator) when is_record(SourceRelation, relation), is_function(Comparator, 2) -> + Name = generate_ephemeral_name(), + + SourceGen = SourceRelation#relation.generator, + GeneratorFun = fun() -> + SourceIter = SourceGen(), + operations:sort_iterator(SourceIter, Comparator) + end, + + #relation{ + hash = <<>>, + name = Name, + tree = undefined, + schema = SourceRelation#relation.schema, + constraints = SourceRelation#relation.constraints, + cardinality = SourceRelation#relation.cardinality, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = SourceRelation#relation.provenance, + lineage = {sort, Comparator, SourceRelation#relation.lineage} + }. + +%%% Helper Functions + +%% @private +%% @doc Rename iterator - applies rename mapping to each tuple +rename_iterator(SourceIter, RenameMappings) -> + spawn(fun() -> rename_loop(SourceIter, RenameMappings) end). + +rename_loop(SourceIter, RenameMappings) -> + receive + {next, Caller} -> + case operations:next_tuple(SourceIter) of + done -> + Caller ! done; + {ok, Tuple} -> + RenamedTuple = maps:fold( + fun(OldAttr, NewAttr, TupleAcc) -> + case maps:get(OldAttr, TupleAcc, undefined) of + undefined -> TupleAcc; + Value -> + TupleAcc1 = maps:remove(OldAttr, TupleAcc), + maps:put(NewAttr, Value, TupleAcc1) + end + end, + Tuple, + RenameMappings + ), + Caller ! {tuple, RenamedTuple}, + rename_loop(SourceIter, RenameMappings) + end; + {close, Caller} -> + operations:close_iterator(SourceIter), + Caller ! ok + end. |