path: root/src/relational_operators.erl

%%% @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.