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.
%%
%% Constraints are propagated: source constraints are preserved, and new
%% constraints may be inferred from declarative predicates.
%%
%% Predicate formats:
%% - fun((map()) -> boolean()) - opaque function, no inference
%% - {attr, AttrName, Op, Value} - declarative, infers constraints
%% - {'and', [Pred1, ...]} - conjunction of declarative predicates
%%
%% @param SourceRelation Source #relation{} record
%% @param Predicate Function or declarative term
%% @returns Ephemeral #relation{} with generator and propagated constraints
-spec select(#relation{}, fun((map()) -> boolean()) | term()) -> #relation{}.
select(SourceRelation, Predicate) when is_record(SourceRelation, relation) ->
    Name = generate_ephemeral_name(),

    %% Build actual filter function from predicate
    FilterFun = predicate_to_function(Predicate),

    SourceGen = SourceRelation#relation.generator,
    GeneratorFun = fun() ->
        SourceIter = SourceGen(),
        operations:select_iterator(SourceIter, FilterFun)
    end,

    %% Infer constraints from predicate and merge with source constraints
    Schema = SourceRelation#relation.schema,
    InferredConstraints = constraint:infer_constraints_from_pred(Predicate, Schema),
    MergedConstraints = constraint:merge_constraints(
        SourceRelation#relation.constraints,
        InferredConstraints
    ),

    #relation{
        hash = <<>>,
        name = Name,
        tree = undefined,
        schema = Schema,
        constraints = MergedConstraints,
        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.
%%
%% Constraints are filtered to only include those for projected attributes.
%% Constraints on attributes that are projected away are dropped.
%%
%% @param SourceRelation Source #relation{} record
%% @param Attributes List of attribute names to keep
%% @returns Ephemeral #relation{} with generator and filtered constraints
-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),

    %% Filter constraints to only projected attributes
    FilteredConstraints = constraint:filter_constraints(
        SourceRelation#relation.constraints,
        Attributes
    ),

    SourceGen = SourceRelation#relation.generator,
    GeneratorFun = fun() ->
        SourceIter = SourceGen(),
        operations:project_iterator(SourceIter, Attributes)
    end,

    #relation{
        hash = <<>>,
        name = Name,
        tree = undefined,
        schema = ProjectedSchema,
        constraints = FilteredConstraints,
        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.
%%
%% Constraints from both relations are merged with AND semantics.
%% If both relations have constraints on the same attribute, both are kept.
%%
%% @param LeftRelation Left #relation{} record
%% @param RightRelation Right #relation{} record
%% @param JoinAttribute Attribute name to join on
%% @returns Ephemeral #relation{} with generator and merged constraints
-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),

    %% Merge constraints from both relations (AND semantics)
    MergedConstraints = constraint:merge_constraints(
        LeftRelation#relation.constraints,
        RightRelation#relation.constraints
    ),

    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 = MergedConstraints,
        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.
%%
%% Constraints are merged from both relations, and additional constraints may
%% be inferred from declarative predicates.
%%
%% @param LeftRelation Left #relation{} record
%% @param RightRelation Right #relation{} record
%% @param Predicate Function or declarative term
%% @returns Ephemeral #relation{} with generator and merged constraints
-spec theta_join(#relation{}, #relation{}, fun((map(), map()) -> boolean()) | term()) -> #relation{}.
theta_join(LeftRelation, RightRelation, Predicate)
    when is_record(LeftRelation, relation),
         is_record(RightRelation, relation) ->

    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),

    %% Merge constraints from both relations
    MergedConstraints = constraint:merge_constraints(
        LeftRelation#relation.constraints,
        RightRelation#relation.constraints
    ),

    %% Infer additional constraints from predicate
    InferredConstraints = constraint:infer_constraints_from_pred(Predicate, MergedSchema),
    FinalConstraints = constraint:merge_constraints(MergedConstraints, InferredConstraints),

    %% Build filter function from predicate (for theta join, needs 2-arity)
    %% The theta_join_iterator expects a 2-arity function
    FilterFun = theta_predicate_to_function(Predicate),

    LeftGen = LeftRelation#relation.generator,
    RightGen = RightRelation#relation.generator,
    GeneratorFun = fun() ->
        LeftIter = LeftGen(),
        RightIter = RightGen(),
        operations:theta_join_iterator(LeftIter, RightIter, FilterFun)
    end,

    #relation{
        hash = <<>>,
        name = Name,
        tree = undefined,
        schema = MergedSchema,
        constraints = FinalConstraints,
        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.
%%
%% Constraints are updated: both the map keys and any {var, AttrName} references
%% inside constraint terms are renamed according to the mapping.
%%
%% @param SourceRelation Source #relation{} record
%% @param RenameMappings Map of old_name => new_name (atoms or strings)
%% @returns Ephemeral #relation{} with generator and renamed constraints
-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
    ),

    %% Rename constraint attributes (both keys and internal variable references)
    RenamedConstraints = constraint:rename_constraint_attrs(
        SourceRelation#relation.constraints,
        RenameMappings
    ),

    SourceGen = SourceRelation#relation.generator,
    GeneratorFun = fun() ->
        SourceIter = SourceGen(),
        rename_iterator(SourceIter, RenameMappings)
    end,

    #relation{
        hash = <<>>,
        name = Name,
        tree = undefined,
        schema = RenamedSchema,
        constraints = RenamedConstraints,
        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.

%% @private
%% @doc Convert a predicate (function or declarative term) to a filter function.
%%
%% Supports:
%% - fun((map()) -> boolean()) - returns as-is
%% - {attr, AttrName, Op, Value} - single attribute comparison
%% - {'and', [Pred1, ...]} - conjunction
%% - {'or', [Pred1, ...]} - disjunction
predicate_to_function(Predicate) when is_function(Predicate, 1) ->
    Predicate;
predicate_to_function({attr, AttrName, Op, Value}) ->
    fun(Tuple) ->
        AttrValue = maps:get(AttrName, Tuple, undefined),
        apply_comparison(Op, AttrValue, Value)
    end;
predicate_to_function({'and', Predicates}) ->
    Funs = [predicate_to_function(P) || P <- Predicates],
    fun(Tuple) ->
        lists:all(fun(F) -> F(Tuple) end, Funs)
    end;
predicate_to_function({'or', Predicates}) ->
    Funs = [predicate_to_function(P) || P <- Predicates],
    fun(Tuple) ->
        lists:any(fun(F) -> F(Tuple) end, Funs)
    end;
predicate_to_function(_) ->
    %% Unknown format - accept all
    fun(_) -> true end.

%% @private
%% @doc Apply a comparison operation.
apply_comparison(lt, A, B) -> A < B;
apply_comparison(lte, A, B) -> A =< B;
apply_comparison(gt, A, B) -> A > B;
apply_comparison(gte, A, B) -> A >= B;
apply_comparison(eq, A, B) -> A =:= B;
apply_comparison(neq, A, B) -> A =/= B;
apply_comparison(between, A, {Min, Max}) -> A >= Min andalso A < Max;
apply_comparison(_, _, _) -> true.

%% @private
%% @doc Convert a theta predicate to a 2-arity filter function.
%%
%% Theta join predicates operate on merged tuples from both relations.
%% For declarative predicates, we convert to a function that works on
%% the merged tuple (left and right tuples combined).
theta_predicate_to_function(Predicate) when is_function(Predicate, 2) ->
    Predicate;
theta_predicate_to_function(Predicate) when is_function(Predicate, 1) ->
    %% Convert 1-arity function to 2-arity by merging tuples
    fun(LeftTuple, RightTuple) ->
        MergedTuple = maps:merge(LeftTuple, RightTuple),
        Predicate(MergedTuple)
    end;
theta_predicate_to_function(DeclarativePred) ->
    %% Convert declarative predicate to 1-arity first, then wrap
    OneTupleFun = predicate_to_function(DeclarativePred),
    fun(LeftTuple, RightTuple) ->
        MergedTuple = maps:merge(LeftTuple, RightTuple),
        OneTupleFun(MergedTuple)
    end.