diff options
Diffstat (limited to 'src/query_planner.erl')
| -rw-r--r-- | src/query_planner.erl | 356 |
1 files changed, 295 insertions, 61 deletions
diff --git a/src/query_planner.erl b/src/query_planner.erl index 9d9d2bc..52bffda 100644 --- a/src/query_planner.erl +++ b/src/query_planner.erl @@ -8,43 +8,37 @@ %%% %%% The DSL uses nested tuples to represent relational operations: %%% <pre> -%%% % Scan relation (base case) -%%% {scan, employees} -%%% -%%% % Reference relation (synonym for scan) -%%% {relation, employees} +%%% % Base relation (just use the relation name) +%%% employees %%% %%% % Selection (filter) -%%% {select, {scan, employees}, fun(T) -> maps:get(age, T) > 30 end} +%%% {select, employees, fun(T) -> maps:get(age, T) > 30 end} %%% %%% % Projection -%%% {project, {scan, employees}, [name, age]} +%%% {project, employees, [name, age]} %%% %%% % Equijoin (natural join on common attribute) -%%% {join, {scan, employees}, {scan, departments}, dept_id} +%%% {join, employees, departments, dept_id} %%% %%% % Theta join (join with arbitrary predicate) %%% {theta_join, -%%% {scan, employees}, -%%% {scan, departments}, +%%% employees, +%%% departments, %%% fun(L, R) -> maps:get(dept_id, L) =:= maps:get(id, R) end} %%% %%% % Sort -%%% {sort, {scan, employees}, fun(A, B) -> maps:get(age, A) =< maps:get(age, B) end} +%%% {sort, employees, fun(A, B) -> maps:get(age, A) =< maps:get(age, B) end} %%% %%% % Take (limit) -%%% {take, {scan, employees}, 10} +%%% {take, employees, 10} %%% %%% % Rename attribute -%%% {rename, old_name, new_name, {scan, employees}} +%%% {rename, old_name, new_name, employees} %%% %%% % Composite query %%% {project, %%% {select, -%%% {join, -%%% {scan, employees}, -%%% {scan, departments}, -%%% dept_id}, +%%% {join, employees, departments, dept_id}, %%% fun(T) -> maps:get(age, T) > 30 end}, %%% [name, dept_name]} %%% </pre> @@ -52,7 +46,7 @@ %%% == Usage == %%% <pre> %%% % Prepare a query plan -%%% Plan = query_planner:prepare({select, {scan, employees}, +%%% Plan = query_planner:prepare({select, employees, %%% fun(T) -> maps:get(age, T) > 30 end}). %%% %%% % Execute and get iterator @@ -77,8 +71,7 @@ %%% Types -type query_plan() :: - {scan, atom()} % Scan relation - | {relation, atom()} % Reference relation + atom() % Relation name | {select, query_plan(), fun((map()) -> boolean())} % Filter | {project, query_plan(), [atom()]} % Projection | {join, query_plan(), query_plan(), atom()} % Equijoin @@ -111,21 +104,22 @@ prepare(Plan) -> %% @doc Execute a prepared query plan against a database. %% -%% Compiles the query plan into an iterator pipeline and returns -%% the iterator handle. The caller must consume the iterator using -%% operations:next_tuple/1 and close it with operations:close_iterator/1. +%% Returns an ephemeral relation for top-level operations that can spawn +%% multiple independent iterators. The relation's generator creates fresh +%% iterator pipelines on demand, allowing reproducible results. %% %% @param DB The database state %% @param Plan The prepared query plan -%% @returns Iterator process PID --spec execute(term(), query_plan()) -> pid(). +%% @returns Ephemeral relation record +-spec execute(term(), query_plan()) -> #relation{}. execute(DB, Plan) -> - compile_to_iterator(DB, Plan). + create_ephemeral_relation(DB, Plan). %% @doc Execute a query plan and collect all results. %% %% Convenience function that executes the plan and collects all -%% tuples into a list. Automatically closes the iterator. +%% tuples into a list. Uses the ephemeral relation to create an +%% iterator and automatically closes it. %% %% If the iterator times out (e.g., when sorting an infinite relation), %% returns partial results with a warning. @@ -135,7 +129,8 @@ execute(DB, Plan) -> %% @returns List of result tuples, or partial results on timeout -spec collect(term(), query_plan()) -> [map()]. collect(DB, Plan) -> - Iter = execute(DB, Plan), + EphemeralRelation = execute(DB, Plan), + Iter = (EphemeralRelation#relation.generator)(), Results = operations:collect_all(Iter), operations:close_iterator(Iter), case Results of @@ -165,11 +160,188 @@ explain(Plan) -> %%% Internal Functions +%% @doc Create an ephemeral relation for the top-level query plan. +%% +%% Uses pure relational operators from relational_operators module to construct +%% the query result. Returns an ephemeral relation with a generator closure that +%% can spawn fresh iterators, allowing reproducible query results. +%% +%% @param DB The database state (for looking up base relations only) +%% @param Plan The query plan +%% @returns Ephemeral relation record +-spec create_ephemeral_relation(term(), query_plan()) -> #relation{}. +create_ephemeral_relation(DB, Plan) -> + % Compile plan to ephemeral relation using pure operators + compile_to_relation(DB, Plan). + +%% @doc Compute the schema for a query plan. +%% +%% Recursively analyzes the query plan structure to determine the output schema. +%% The schema maps attribute names to their domain/type information. +%% +%% @param DB The database state +%% @param Plan The query plan +%% @returns Schema map +-spec compute_schema(term(), query_plan()) -> map(). +compute_schema(DB, RelationName) when is_atom(RelationName) -> + case maps:get(RelationName, DB#database_state.relations, undefined) of + undefined -> + % Relation not found, return empty schema + #{}; + RelationHash -> + [Relation] = mnesia:dirty_read(relation, RelationHash), + Relation#relation.schema + end; + +compute_schema(DB, {select, SubPlan, _Predicate}) -> + % Select doesn't change schema, just filters tuples + compute_schema(DB, SubPlan); + +compute_schema(DB, {project, SubPlan, Attributes}) -> + % Project filters the schema to only include selected attributes + SubSchema = compute_schema(DB, SubPlan), + maps:with(Attributes, SubSchema); + +compute_schema(DB, {join, LeftPlan, RightPlan, _Attribute}) -> + % Join combines schemas from both sides, handling attribute conflicts + LeftSchema = compute_schema(DB, LeftPlan), + RightSchema = compute_schema(DB, RightPlan), + merge_join_schemas(LeftSchema, RightSchema); + +compute_schema(DB, {theta_join, LeftPlan, RightPlan, _Predicate}) -> + % Theta join also combines schemas from both sides, handling attribute conflicts + LeftSchema = compute_schema(DB, LeftPlan), + RightSchema = compute_schema(DB, RightPlan), + merge_join_schemas(LeftSchema, RightSchema); + +compute_schema(DB, {sort, SubPlan, _Comparator}) -> + % Sort doesn't change schema + compute_schema(DB, SubPlan); + +compute_schema(DB, {take, SubPlan, _N}) -> + % Take doesn't change schema, just limits tuples + compute_schema(DB, SubPlan); + +compute_schema(DB, {rename, OldAttr, NewAttr, SubPlan}) -> + % Rename changes one attribute name in the schema + SubSchema = compute_schema(DB, SubPlan), + case maps:is_key(OldAttr, SubSchema) of + true -> + AttrType = maps:get(OldAttr, SubSchema), + SubSchema1 = maps:remove(OldAttr, SubSchema), + maps:put(NewAttr, AttrType, SubSchema1); + false -> + % Attribute doesn't exist, return unchanged schema + SubSchema + end; + +compute_schema(DB, {materialize, SubPlan}) -> + % Materialize doesn't change schema + compute_schema(DB, SubPlan); + +compute_schema(DB, {materialize, SubPlan, _N}) -> + % Materialize with limit doesn't change schema + compute_schema(DB, SubPlan); + +compute_schema(_DB, _UnknownPlan) -> + % Unknown plan type, return empty schema + #{}. + +%% @doc Merge schemas for join operations, handling attribute conflicts elegantly. +%% +%% Instead of prefixing conflicting attribute names, this approach maintains +%% both provenances in the schema value. For attributes that exist in both +%% relations, the schema value becomes a list of the original values. +%% This is especially important for natural joins where the join attribute +%% logically represents the same entity from both sides. +%% +%% @param LeftSchema Schema from left relation +%% @param RightSchema Schema from right relation +%% @returns Merged schema map with provenance lists for conflicts +-spec merge_join_schemas(map(), map()) -> map(). +merge_join_schemas(LeftSchema, RightSchema) -> + maps:fold( + fun(Key, RightValue, Acc) -> + case maps:get(Key, Acc, undefined) of + undefined -> + % No conflict, add the right attribute + maps:put(Key, RightValue, Acc); + LeftValue when LeftValue =:= RightValue -> + % Same type/domain, keep single value (common in natural joins) + maps:put(Key, LeftValue, Acc); + LeftValue when is_list(LeftValue) -> + % Left value is already a list, append if not already present + case lists:member(RightValue, LeftValue) of + true -> maps:put(Key, LeftValue, Acc); + false -> maps:put(Key, LeftValue ++ [RightValue], Acc) + end; + LeftValue -> + % Conflict with different types - create provenance list + maps:put(Key, [LeftValue, RightValue], Acc) + end + end, + LeftSchema, + RightSchema + ). + +%% @doc Compute provenance for a query plan. +%% +%% Recursively analyzes the query plan structure to determine the provenance, +%% following the provenance types defined in operations.hrl. +%% +%% @param Plan The query plan +%% @returns Provenance term +-spec compute_provenance(query_plan()) -> term(). +compute_provenance(RelationName) when is_atom(RelationName) -> + {base, RelationName}; + +compute_provenance({select, SubPlan, _Predicate}) -> + % For select, we'd need to represent the predicate constraints + % For now, simplified without constraint details + SubProvenance = compute_provenance(SubPlan), + {select, SubProvenance, #{}}; + +compute_provenance({project, SubPlan, Attributes}) -> + SubProvenance = compute_provenance(SubPlan), + {project, SubProvenance, Attributes}; + +compute_provenance({join, LeftPlan, RightPlan, _Attribute}) -> + LeftProvenance = compute_provenance(LeftPlan), + RightProvenance = compute_provenance(RightPlan), + {join, LeftProvenance, RightProvenance}; + +compute_provenance({theta_join, LeftPlan, RightPlan, _Predicate}) -> + LeftProvenance = compute_provenance(LeftPlan), + RightProvenance = compute_provenance(RightPlan), + {join, LeftProvenance, RightProvenance}; + +compute_provenance({sort, SubPlan, _Comparator}) -> + % Sort doesn't change provenance + compute_provenance(SubPlan); + +compute_provenance({take, SubPlan, N}) -> + SubProvenance = compute_provenance(SubPlan), + {take, SubProvenance, N}; + +compute_provenance({rename, _OldAttr, _NewAttr, SubPlan}) -> + % Rename preserves underlying provenance + compute_provenance(SubPlan); + +compute_provenance({materialize, SubPlan}) -> + % Materialize doesn't change provenance + compute_provenance(SubPlan); + +compute_provenance({materialize, SubPlan, _N}) -> + % Materialize with limit doesn't change provenance + compute_provenance(SubPlan); + +compute_provenance(_UnknownPlan) -> + % Unknown plan type + undefined. + %% Validate query plan structure -spec validate_plan(query_plan()) -> ok | {error, term()}. -validate_plan({scan, Name}) when is_atom(Name) -> - ok; -validate_plan({relation, Name}) when is_atom(Name) -> +validate_plan(Name) when is_atom(Name) -> ok; validate_plan({select, SubPlan, Pred}) when is_function(Pred, 1) -> validate_plan(SubPlan); @@ -205,10 +377,7 @@ validate_plan(Plan) -> %% Compile query plan to iterator pipeline -spec compile_to_iterator(term(), query_plan()) -> pid(). -compile_to_iterator(DB, {scan, Name}) -> - operations:get_tuples_iterator(DB, Name); - -compile_to_iterator(DB, {relation, Name}) -> +compile_to_iterator(DB, Name) when is_atom(Name) -> operations:get_tuples_iterator(DB, Name); compile_to_iterator(DB, {select, SubPlan, Predicate}) -> @@ -234,22 +403,16 @@ compile_to_iterator(DB, {sort, SubPlan, Comparator}) -> operations:sort_iterator(SubIter, Comparator); compile_to_iterator(DB, {take, SubPlan, N}) -> - %% Use relational take operator when possible (for scans) - case SubPlan of - {scan, RelationName} -> - %% Direct scan: use relational take operator - {DB2, TakeRelation} = operations:take(DB, RelationName, N), - %% Get iterator from the take relation (now properly registered) - operations:get_tuples_iterator(DB2, TakeRelation#relation.name, #{}); - _ -> - %% Complex plan: fall back to iterator-based take for composition - SubIter = compile_to_iterator(DB, SubPlan), - operations:take_iterator(SubIter, N) - end; + %% For iterator-based compilation, use take_iterator + SubIter = compile_to_iterator(DB, SubPlan), + operations:take_iterator(SubIter, N); compile_to_iterator(DB, {rename, OldAttr, NewAttr, SubPlan}) -> SubIter = compile_to_iterator(DB, SubPlan), - rename_iterator(SubIter, OldAttr, NewAttr); + %% Inline simple rename - no need for separate function + spawn(fun() -> + rename_loop_inline(SubIter, OldAttr, NewAttr) + end); compile_to_iterator(DB, {materialize, SubPlan}) -> % Materialize returns all tuples as a "static" iterator @@ -286,18 +449,14 @@ static_loop(Tuples) -> Caller ! ok end. -%% Rename iterator - renames an attribute in each tuple --spec rename_iterator(pid(), atom(), atom()) -> pid(). -rename_iterator(SourceIter, OldAttr, NewAttr) -> - spawn(fun() -> rename_loop(SourceIter, OldAttr, NewAttr) end). - -rename_loop(SourceIter, OldAttr, NewAttr) -> +%% @private +%% Inline helper for compile_to_iterator rename case +rename_loop_inline(SourceIter, OldAttr, NewAttr) -> receive {next, Caller} -> case operations:next_tuple(SourceIter) of done -> - Caller ! done, - rename_loop(SourceIter, OldAttr, NewAttr); + Caller ! done; {ok, Tuple} -> RenamedTuple = case maps:is_key(OldAttr, Tuple) of true -> @@ -308,7 +467,7 @@ rename_loop(SourceIter, OldAttr, NewAttr) -> Tuple end, Caller ! {tuple, RenamedTuple}, - rename_loop(SourceIter, OldAttr, NewAttr) + rename_loop_inline(SourceIter, OldAttr, NewAttr) end; {close, Caller} -> operations:close_iterator(SourceIter), @@ -317,10 +476,7 @@ rename_loop(SourceIter, OldAttr, NewAttr) -> %% Pretty-print query plan -spec explain_plan(query_plan(), non_neg_integer()) -> string(). -explain_plan({scan, Name}, Indent) -> - indent(Indent) ++ "Scan: " ++ atom_to_list(Name) ++ "\n"; - -explain_plan({relation, Name}, Indent) -> +explain_plan(Name, Indent) when is_atom(Name) -> indent(Indent) ++ "Relation: " ++ atom_to_list(Name) ++ "\n"; explain_plan({select, SubPlan, _Pred}, Indent) -> @@ -365,3 +521,81 @@ explain_plan({materialize, SubPlan, N}, Indent) -> %% Helper: indent string -spec indent(non_neg_integer()) -> string(). indent(N) -> lists:duplicate(N, $\s). + +%% Helper: get relation from database +-spec get_relation_from_db(term(), atom()) -> #relation{}. +get_relation_from_db(DB, RelationName) -> + case maps:get(RelationName, DB#database_state.relations, undefined) of + undefined -> + erlang:error({relation_not_found, RelationName}); + RelationHash -> + [Relation] = mnesia:dirty_read(relation, RelationHash), + Relation + end. + +%% @doc Compile query plan to ephemeral relation using pure relational operators. +%% +%% This is the new approach: build a relation tree using pure operators, +%% then return the final ephemeral relation. No iterators are created until +%% the generator is called. +-spec compile_to_relation(term(), query_plan()) -> #relation{}. +compile_to_relation(DB, RelationName) when is_atom(RelationName) -> + BaseRelation = get_relation_from_db(DB, RelationName), + GeneratorFun = fun() -> + operations:get_tuples_iterator(DB, RelationName) + end, + BaseRelation#relation{generator = GeneratorFun}; + +compile_to_relation(DB, {select, SubPlan, Predicate}) -> + SubRelation = compile_to_relation(DB, SubPlan), + relational_operators:select(SubRelation, Predicate); + +compile_to_relation(DB, {project, SubPlan, Attributes}) -> + SubRelation = compile_to_relation(DB, SubPlan), + relational_operators:project(SubRelation, Attributes); + +compile_to_relation(DB, {join, LeftPlan, RightPlan, Attribute}) -> + LeftRelation = compile_to_relation(DB, LeftPlan), + RightRelation = compile_to_relation(DB, RightPlan), + relational_operators:join(LeftRelation, RightRelation, Attribute); + +compile_to_relation(DB, {theta_join, LeftPlan, RightPlan, Predicate}) -> + LeftRelation = compile_to_relation(DB, LeftPlan), + RightRelation = compile_to_relation(DB, RightPlan), + relational_operators:theta_join(LeftRelation, RightRelation, Predicate); + +compile_to_relation(DB, {sort, SubPlan, Comparator}) -> + SubRelation = compile_to_relation(DB, SubPlan), + relational_operators:sort(SubRelation, Comparator); + +compile_to_relation(DB, {take, SubPlan, N}) -> + SubRelation = compile_to_relation(DB, SubPlan), + relational_operators:take(SubRelation, N); + +compile_to_relation(DB, {rename, OldAttr, NewAttr, SubPlan}) -> + SubRelation = compile_to_relation(DB, SubPlan), + relational_operators:rename(SubRelation, #{OldAttr => NewAttr}); + +compile_to_relation(DB, {materialize, SubPlan}) -> + GeneratorFun = fun() -> + compile_to_iterator(DB, SubPlan) + end, + EphemeralName = list_to_atom("materialized_" ++ integer_to_list(erlang:unique_integer([positive]))), + #relation{ + hash = crypto:hash(sha256, term_to_binary({materialize, SubPlan})), + name = EphemeralName, + tree = undefined, + schema = compute_schema(DB, SubPlan), + constraints = #{}, + cardinality = unknown, + generator = GeneratorFun, + membership_criteria = #{}, + provenance = compute_provenance(SubPlan) + }; + +compile_to_relation(DB, {materialize, SubPlan, N}) -> + % Materialize with limit + SubRelation = compile_to_relation(DB, SubPlan), + TakeRelation = relational_operators:take(SubRelation, N), + % Then wrap in a materialized form (for now just return take) + TakeRelation. |