Current section
Files
Jump to
Current section
Files
src/barrel_bql_exec.erl
%% @doc BQL post-stage executor for collection sources: drives
%% barrel_docdb:find/3 with a compiled plan and applies the per-row
%% stages (unnest expansion, frame-residual predicates, order,
%% offset/limit, projection).
%%
%% The frame helpers (frames_for/2, residual_match/2, finalize/3) are
%% also the reuse surface for the barrel module, which feeds them
%% table-function hits instead of find rows.
%%
%% A frame is #{doc := map() with id merged in, elem := term() |
%% undefined, score := map()}; score carries the synthetic _score /
%% _distance columns of table-function sources (empty here).
-module(barrel_bql_exec).
-export([
run/3,
fold/5
]).
%% Frame surface reused by the barrel module for table functions.
-export([
frames_for/2,
residual_match/2,
finalize/3
]).
-type frame() :: #{doc := map(), elem := term(), score := map()}.
%%====================================================================
%% Entry points (collection sources)
%%====================================================================
%% @doc Run a compiled plan. Streamable plans (no ORDER BY, no UNNEST)
%% keep find/3's chunk semantics: one chunk per call, meta carries the
%% continuation. Materializing plans paginate internally and return
%% everything.
-spec run(binary(), barrel_bql_lower:plan(), map()) ->
{ok, [map()], map()} | {error, term()}.
run(_DbName, #{post := #{empty := true}}, _Opts) ->
{ok, [], #{has_more => false, count => 0}};
run(DbName, #{streamable := true} = Plan, Opts) ->
#{spec := Spec, post := #{project := Project}} = Plan,
case barrel_docdb:find(DbName, Spec, find_opts(Opts)) of
{ok, Rows, Meta} ->
{ok, project_rows(Rows, Project), Meta};
{error, _} = Error ->
Error
end;
run(DbName, Plan, Opts) ->
case maps:is_key(continuation, Opts) of
true ->
{error, {unsupported, continuation}};
false ->
#{post := Post, unnest := Unnest} = Plan,
case collect_frames(DbName, Plan, Opts) of
{ok, Frames, LastSeq} ->
Rows = finalize(Frames, Post, Unnest),
{ok, Rows, #{has_more => false,
count => length(Rows),
last_seq => LastSeq}};
{error, _} = Error ->
Error
end
end.
%% @doc Fold rows without materializing the full result (streamable
%% plans paginate internally; materializing plans fold the built list).
%% Fun(Row, Acc) -> {ok, Acc} | {stop, Acc}.
-spec fold(binary(), barrel_bql_lower:plan(), map(),
fun((map(), term()) -> {ok, term()} | {stop, term()}),
term()) ->
{ok, term(), map()} | {error, term()}.
fold(_DbName, #{post := #{empty := true}}, _Opts, _Fun, Acc) ->
{ok, Acc, #{has_more => false, count => 0}};
fold(DbName, #{streamable := true} = Plan, Opts, Fun, Acc0) ->
#{spec := Spec, post := #{project := Project}} = Plan,
ChunkSize = maps:get(chunk_size, Opts, 100),
fold_chunks(DbName, Spec, Project,
maps:get(continuation, Opts, undefined),
ChunkSize, Fun, Acc0);
fold(DbName, Plan, Opts, Fun, Acc0) ->
case run(DbName, Plan, Opts) of
{ok, Rows, Meta} ->
{_, Acc} = fold_rows(Rows, Fun, Acc0),
{ok, Acc, Meta};
{error, _} = Error ->
Error
end.
%%====================================================================
%% Frame surface (shared with barrel)
%%====================================================================
%% @doc Expand a document into frames: one per unnested array element,
%% or a single element-less frame. Missing / non-array unnest values
%% produce no frames (inner-join UNNEST).
-spec frames_for(map(), undefined | #{path := list(), alias := binary()}) ->
[frame()].
frames_for(Doc, undefined) ->
[#{doc => Doc, elem => undefined, score => #{}}];
frames_for(Doc, #{path := Path}) ->
case get_path(Doc, Path) of
{ok, Elements} when is_list(Elements) ->
[#{doc => Doc, elem => Element, score => #{}}
|| Element <- Elements];
_ ->
[]
end.
%% @doc Evaluate the frame-residual conjuncts (tagged paths) against a
%% frame. Mirrors the engine's exists-and-satisfies leaf semantics.
-spec residual_match([term()], frame()) -> boolean().
residual_match(Conds, Frame) ->
lists:all(fun(Cond) -> eval_cond(Cond, Frame) end, Conds).
%% @doc Order, offset/limit and project a list of frames.
-spec finalize([frame()], map(),
undefined | #{path := list(), alias := binary()}) ->
[map()].
finalize(Frames0, Post, Unnest) ->
#{order := Order, offset := Offset, limit := Limit,
project := Project} = Post,
Frames1 = sort_frames(Frames0, Order),
Frames2 = case Offset of
0 -> Frames1;
_ when Offset >= length(Frames1) -> [];
_ -> lists:nthtail(Offset, Frames1)
end,
Frames = case Limit of
undefined -> Frames2;
_ -> lists:sublist(Frames2, Limit)
end,
[project_frame(Project, Frame, Unnest) || Frame <- Frames].
%%====================================================================
%% Find pagination
%%====================================================================
find_opts(Opts) ->
maps:with([chunk_size, continuation], Opts).
collect_frames(DbName, Plan, Opts) ->
#{spec := Spec, unnest := Unnest,
post := #{residual := Residual, order := Order, offset := Offset,
limit := Limit}} = Plan,
ChunkSize = maps:get(chunk_size, Opts, 1000),
%% Bounded queries without ORDER BY can stop scanning once enough
%% surviving frames are collected; a sort needs everything.
Need = case {Order, Limit} of
{undefined, undefined} -> infinity;
{undefined, _} -> Offset + Limit;
{_, _} -> infinity
end,
collect_frames_loop(DbName, Spec, Unnest, Residual, ChunkSize,
Need, undefined, []).
collect_frames_loop(DbName, Spec, Unnest, Residual, ChunkSize, Need,
Cont, Acc) ->
FindOpts = case Cont of
undefined -> #{chunk_size => ChunkSize};
_ -> #{chunk_size => ChunkSize, continuation => Cont}
end,
case barrel_docdb:find(DbName, Spec, FindOpts) of
{ok, Rows, Meta} ->
Frames = [Frame || Row <- Rows,
Frame <- row_frames(Row, Unnest),
residual_match(Residual, Frame)],
Acc1 = Acc ++ Frames,
Done = Need =/= infinity andalso length(Acc1) >= Need,
case Meta of
#{has_more := true, continuation := Cont1} when not Done ->
collect_frames_loop(DbName, Spec, Unnest, Residual,
ChunkSize, Need, Cont1, Acc1);
_ ->
{ok, Acc1, maps:get(last_seq, Meta, first)}
end;
{error, _} = Error ->
Error
end.
row_frames(#{<<"id">> := Id, <<"doc">> := Doc}, Unnest) ->
frames_for(Doc#{<<"id">> => Id}, Unnest);
row_frames(_DeletedRow, _Unnest) ->
[].
project_rows(Rows, Project) ->
[project_frame(Project, Frame, undefined)
|| Row <- Rows, Frame <- row_frames(Row, undefined)].
fold_chunks(DbName, Spec, Project, Cont, ChunkSize, Fun, Acc0) ->
FindOpts = case Cont of
undefined -> #{chunk_size => ChunkSize};
_ -> #{chunk_size => ChunkSize, continuation => Cont}
end,
case barrel_docdb:find(DbName, Spec, FindOpts) of
{ok, Rows, Meta} ->
ProjRows = project_rows(Rows, Project),
case fold_rows(ProjRows, Fun, Acc0) of
{stopped, Acc} ->
{ok, Acc, Meta};
{ok, Acc} ->
case Meta of
#{has_more := true, continuation := Cont1} ->
fold_chunks(DbName, Spec, Project, Cont1,
ChunkSize, Fun, Acc);
_ ->
{ok, Acc, Meta}
end
end;
{error, _} = Error ->
Error
end.
fold_rows([], _Fun, Acc) ->
{ok, Acc};
fold_rows([Row | Rest], Fun, Acc0) ->
case Fun(Row, Acc0) of
{ok, Acc} -> fold_rows(Rest, Fun, Acc);
{stop, Acc} -> {stopped, Acc}
end.
%%====================================================================
%% Residual evaluation (tagged paths, engine leaf semantics)
%%====================================================================
eval_cond({path, TaggedPath, Value}, Frame) ->
case frame_value(TaggedPath, Frame) of
{ok, Found} -> Found =:= Value;
not_found -> false
end;
eval_cond({compare, TaggedPath, Op, Value}, Frame) ->
case frame_value(TaggedPath, Frame) of
{ok, Found} -> compare_values(Found, Op, Value);
not_found -> false
end;
eval_cond({exists, TaggedPath}, Frame) ->
frame_value(TaggedPath, Frame) =/= not_found;
eval_cond({missing, TaggedPath}, Frame) ->
frame_value(TaggedPath, Frame) =:= not_found;
eval_cond({in, TaggedPath, Values}, Frame) ->
case frame_value(TaggedPath, Frame) of
{ok, Found} -> lists:member(Found, Values);
not_found -> false
end;
eval_cond({contains, TaggedPath, Value}, Frame) ->
case frame_value(TaggedPath, Frame) of
{ok, List} when is_list(List) -> lists:member(Value, List);
_ -> false
end;
eval_cond({prefix, TaggedPath, Prefix}, Frame) ->
case frame_value(TaggedPath, Frame) of
{ok, Found} when is_binary(Found) ->
Size = byte_size(Prefix),
case Found of
<<Prefix:Size/binary, _/binary>> -> true;
_ -> false
end;
_ ->
false
end;
eval_cond({regex, TaggedPath, Pattern}, Frame) ->
case frame_value(TaggedPath, Frame) of
{ok, Found} when is_binary(Found) ->
re:run(Found, Pattern, [{capture, none}]) =:= match;
_ ->
false
end;
eval_cond({'and', Conds}, Frame) ->
lists:all(fun(Cond) -> eval_cond(Cond, Frame) end, Conds);
eval_cond({'or', Conds}, Frame) ->
lists:any(fun(Cond) -> eval_cond(Cond, Frame) end, Conds);
eval_cond({'not', Cond}, Frame) ->
not eval_cond(Cond, Frame).
%% Same-type-only comparisons, mirroring barrel_query:compare_values/3.
compare_values(A, '=/=', B) -> A =/= B;
compare_values(A, '==', B) -> A =:= B;
compare_values(A, Op, B) when is_number(A), is_number(B);
is_binary(A), is_binary(B) ->
case Op of
'<' -> A < B;
'=<' -> A =< B;
'>' -> A > B;
'>=' -> A >= B
end;
compare_values(_, _, _) ->
false.
frame_value({b, Comps}, #{doc := Doc}) ->
get_path(Doc, Comps);
frame_value({u, Comps}, #{elem := Elem}) ->
case Elem of
undefined -> not_found;
_ -> get_path(Elem, Comps)
end;
frame_value({score, Key}, #{score := Score}) ->
case maps:find(Key, Score) of
{ok, Value} -> {ok, Value};
error -> not_found
end.
get_path(Value, []) ->
{ok, Value};
get_path(Map, [Key | Rest]) when is_map(Map), is_binary(Key) ->
case maps:find(Key, Map) of
{ok, Value} -> get_path(Value, Rest);
error -> not_found
end;
get_path(List, [Index | Rest]) when is_list(List), is_integer(Index),
Index >= 0 ->
case Index < length(List) of
true -> get_path(lists:nth(Index + 1, List), Rest);
false -> not_found
end;
get_path(_Value, _Path) ->
not_found.
%%====================================================================
%% Ordering (single key, stable, missing sorts as null)
%%====================================================================
sort_frames(Frames, undefined) ->
Frames;
sort_frames(Frames, {Key, Direction}) ->
Decorated = decorate(Frames, Key, 0, []),
Sorted = lists:sort(
fun({KA, IA, _}, {KB, IB, _}) ->
case KA =:= KB of
true -> IA =< IB;
false when Direction =:= asc -> KA < KB;
false -> KB < KA
end
end,
Decorated),
[Frame || {_, _, Frame} <- Sorted].
decorate([], _Key, _Index, Acc) ->
lists:reverse(Acc);
decorate([Frame | Rest], Key, Index, Acc) ->
SortValue = case frame_value(Key, Frame) of
{ok, Value} -> Value;
not_found -> null
end,
decorate(Rest, Key, Index + 1, [{SortValue, Index, Frame} | Acc]).
%%====================================================================
%% Projection
%%====================================================================
project_frame(star, #{doc := Doc, elem := Elem, score := Score},
Unnest) ->
Row0 = maps:merge(maps:remove(<<"_rev">>, Doc), Score),
case Unnest of
undefined -> Row0;
#{alias := Alias} -> Row0#{Alias => Elem}
end;
project_frame(Projections, #{doc := Doc} = Frame, _Unnest) ->
Row0 = case maps:find(<<"id">>, Doc) of
{ok, Id} -> #{<<"id">> => Id};
error -> #{}
end,
lists:foldl(
fun({Tag, Key, Name}, Row) ->
case frame_value({Tag, Key}, Frame) of
{ok, Value} -> Row#{Name => Value};
not_found -> Row
end
end,
Row0,
Projections).