Packages
dqe
0.4.1
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.25
0.3.24
0.3.23
0.3.22
0.3.21
0.3.20
0.3.19
0.3.18
0.3.17
0.3.16
0.3.15
0.3.14
0.3.13
0.3.12
0.3.11
0.3.10
0.3.9
0.3.8
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.2.2
0.2.1
0.2.0
0.1.36
0.1.35
0.1.34
0.1.33
0.1.22
0.1.21
0.1.20
0.1.10
0.1.9
DalmatinerDB query engine
Current section
Files
Jump to
Current section
Files
src/dqe_funnel.erl
-module(dqe_funnel).
-behaviour(dflow).
-export([init/1, describe/1, start/2, emit/3, done/2]).
-record(state, {buffer = [], refs = [], limit}).
init([Limit, SubQs]) ->
SubQs1 = [{make_ref(), Q} || Q <- SubQs],
Refs = [R || {R, _} <- SubQs1],
{ok, #state{refs = Refs, limit = Limit}, SubQs1}.
describe(_) ->
"funnel".
start(_, State) ->
{ok, State}.
emit(Child, Data, State = #state{buffer = Buffer}) ->
State1 = State#state{buffer = orddict:store(Child, Data, Buffer)},
{ok, State1}.
done({last, _}, State = #state{buffer = []}) ->
{done, {error, no_results}, State};
done({last, _}, State = #state{buffer = B, refs = Rs, limit = Limit}) ->
Data = [case orddict:find(R, B) of
error -> [];
{ok, V} -> V
end || R <- Rs],
{done, apply_limit(Data, Limit), State};
done(_O, State) ->
{ok, State}.
apply_limit(Data, undefined) ->
Data;
apply_limit(Data, {Direction, N, Mod}) ->
case [E || E = #{type := metrics} <- Data] of
_Metrics when length(_Metrics) =< N ->
Data;
Metrics ->
Events = [E || E = #{type := events} <- Data],
Metrics1 = [calculate_limit_value(E, Mod) ||
E <- Metrics],
%% We don't need the ranking after we sort,
Metrics2 = [E || {_, E} <- lists:sort(Metrics1)],
take(Metrics2 ++ Events, Direction, N)
end.
%% When we use top we want the 'biggest' elements that are
%% last in the list so we reverse the list and treat it as
%% bottom
take(Data, top, N) ->
take(lists:reverse(Data), bottom, N);
take(Data, bottom, N) ->
lists:sublist(Data, N).
calculate_limit_value(#{data := Data} = M,
#{args := #{constants := Cs,
%%inputs => [#{op => dummy,return => metric}],
mod := Mod},
op := fcall}) ->
Cs1 = [to_v(C) || C <- Cs],
Cs2 = [C || C <- Cs1, C =/= undefined],
Count = mmath_bin:length_r(Data),
S0 = Mod:init(Cs2 ++ [Count]),
{_, S1} = Mod:resolution(1, S0),
{V, _} = Mod:run([Data], S1),
[V0] = mmath_bin:to_list(mmath_bin:derealize(V)),
{V0, M}.
to_v(#{op := float, args := [V]}) ->
V;
to_v(#{op := integer, args := [V]}) ->
V;
to_v(_) ->
undefined.