ddskerl

-module(ddskerl_std).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
?MODULEDOC("""
DDSketch implementation in Erlang.

This implements an unbounded bucket count, that is, on a degenerate case,
the memory consumption will be unbounded logarithmic on the number of different elements.

The underlying data structure is a map. Note however, that, if measuring picoseconds,
a map with 2184 buckets would keep track of all values all the way to 107 days.

Because the data structure cannot be accessed concurretly (share-nothing semantics),
a good strategy would be to keep many running in parallel, and running queries over a merge.

> #### When to use {: .tip}
> This is a good choice when the summary is tracked by a process
> and updates the state through its mailbox.
>
> For example, the following could be used as a template:
> ```erlang
> ...
> -behaviour(gen_server).
>
> -spec start_link(ddskerl_std:opts()) -> gen_server:start_ret().
start_link(Opts) ->
>    gen_server:start_link(?MODULE, Opts, [{spawn_opt, [{message_queue_data, off_heap}]}]).
>
> -spec init(ddskerl_std:opts()) -> {ok, ddskerl_std:ddsketch()}.
> init(Opts) ->
>    {ok, ddskerl_std:new(Opts)}.
>
> handle_call({get_quantile, Q}, _From, Sketch) ->
>    {reply, ddskerl_std:quantile(Sketch, Q), Sketch}.
>
> handle_cast({new_value, Value}, Sketch) ->
>    {noreply, ddskerl_std:insert(Sketch, Value)}.
> ...
> ```
""").

-include("./ddskerl.hrl").

-behaviour(ddskerl).

-export([new/1, total/1, sum/1, insert/2, merge/2, quantile/2]).

?DOC("Options for the DDSketch.").
-type opts() :: #{error := float(), _ => _}.

?DOC("DDSketch instance.").
-opaque ddsketch() :: #ddskerl_std{}.

-export_type([ddsketch/0, opts/0]).

?DOC("Create a new DDSketch instance.").
-spec new(opts()) -> ddsketch().
new(#{error := Err}) ->
    Gamma = (1 + Err) / (1 - Err),
    InvLogGamma = 1.0 / math:log2(Gamma),
    #ddskerl_std{gamma = Gamma, inv_log_gamma = InvLogGamma}.

?DOC("Get the total number of elements in the DDSketch.").
-spec total(ddsketch()) -> non_neg_integer().
total(#ddskerl_std{total = Total}) ->
    Total.

?DOC("Get the sum of elements in the DDSketch.").
-spec sum(ddsketch()) -> number().
sum(#ddskerl_std{sum = Sum}) ->
    Sum.

?DOC("Insert a value into the DDSketch.").
-spec insert(ddsketch(), number()) -> ddsketch().
insert(
    #ddskerl_std{
        data = Data, total = Total, sum = Sum, min = Min, max = Max, inv_log_gamma = InvLogGamma
    } = S,
    Val
) when
    Val > 0
->
    Bin = ceil(math:log2(Val) * InvLogGamma),
    NewData = maps:update_with(Bin, fun(X) -> X + 1 end, 1, Data),
    S#ddskerl_std{
        data = NewData, total = Total + 1, sum = Sum + Val, min = min(Min, Val), max = max(Max, Val)
    }.

?DOC("Calculate the quantile of a DDSketch.").
-spec quantile(ddsketch(), float()) -> float() | undefined.
quantile(#ddskerl_std{min = Min}, +0.0) ->
    Min;
quantile(#ddskerl_std{max = Max}, 1.0) ->
    Max;
quantile(#ddskerl_std{data = Data, gamma = Gamma, total = Total}, Quantile) when
    0.0 < Quantile, Quantile < 1.0
->
    AccumulatedRank = 0,
    TotalQuantile = Total * Quantile,
    case TotalQuantile =< AccumulatedRank of
        true ->
            result(Gamma, 0);
        false ->
            get_quantile(Data, TotalQuantile, AccumulatedRank, Data, Gamma)
    end.

get_quantile(Data, TotalQuantile, AccumulatedRank, Data, Gamma) ->
    SortedData = lists:sort(fun({Pos1, _}, {Pos2, _}) -> Pos1 =< Pos2 end, maps:to_list(Data)),
    try
        lists:foldl(
            fun({Pos, Val}, Acc) ->
                NewAccumulatedRank = Acc + Val,
                case TotalQuantile =< NewAccumulatedRank of
                    true ->
                        throw({found, Pos});
                    false ->
                        NewAccumulatedRank
                end
            end,
            AccumulatedRank,
            SortedData
        )
    of
        _ -> undefined
    catch
        throw:{found, Pos} ->
            result(Gamma, Pos)
    end.

?DOC("Merge two DDSketch instances.").
-spec merge(ddsketch(), ddsketch()) -> ddsketch().
merge(
    #ddskerl_std{gamma = G, data = Data1, total = Total1, sum = Sum1, min = Min1, max = Max1} = S1,
    #ddskerl_std{gamma = G, data = Data2, total = Total2, sum = Sum2, min = Min2, max = Max2}
) ->
    Data = maps:merge_with(fun(_K, Val1, Val2) -> Val1 + Val2 end, Data1, Data2),
    S1#ddskerl_std{
        data = Data,
        total = Total1 + Total2,
        sum = Sum1 + Sum2,
        min = min(Min1, Min2),
        max = max(Max1, Max2)
    }.

-compile({inline, [result/2]}).
-spec result(number(), non_neg_integer()) -> float().
result(_, 0) ->
    0.0;
result(Gamma, Pos) ->
    2 * math:pow(Gamma, Pos) / (Gamma + 1).