%% @doc: Registers stored in one large binary
%%
%% This backend uses one plain Erlang binary to store registers. The
%% cost of rebuilding the binary is amortized by keeping a buffer of
%% inserts to perform in the future.

-module(hyper_binary).

-behaviour(hyper_register).

%%-compile(native).

-export([
    new/1,
    set/3,
    compact/1,
    max_merge/1, max_merge/2,
    reduce_precision/2,
    bytes/1,
    register_sum/1,
    register_histogram/1,
    zero_count/1,
    encode_registers/1,
    decode_registers/2,
    empty_binary/1,
    max_registers/1
]).

-define(VALUE_SIZE, 6).
-define(MERGE_THRESHOLD, 0.05).

-record(buffer, {buf, buf_size, p, convert_threshold}).
-record(dense, {b, buf, buf_size, p, merge_threshold}).

new(P) ->
    new_buffer(P).

new_buffer(P) ->
    M = hyper_utils:m(P),
    % 5 words for each entry
    ConvertThreshold = M div (5 * 8),
    #buffer{
        buf = [],
        buf_size = 0,
        p = P,
        convert_threshold = ConvertThreshold
    }.

new_dense(P) ->
    M = hyper_utils:m(P),
    T = max(trunc(M * ?MERGE_THRESHOLD), 16),
    #dense{
        b = empty_binary(M),
        buf = [],
        buf_size = 0,
        p = P,
        merge_threshold = T
    }.

set(Index, Value, #buffer{buf = [{Index, OldValue} | Rest]} = Buffer) ->
    NewVal = hyper_utils:run_of_zeroes(Value),
    Buffer#buffer{buf = [{Index, max(NewVal, OldValue)} | Rest]};
set(Index, Value, #buffer{buf = Buf, buf_size = BufSize} = Buffer) ->
    NewVal = hyper_utils:run_of_zeroes(Value),
    NewBuffer = Buffer#buffer{buf = [{Index, NewVal} | Buf], buf_size = BufSize + 1},
    case NewBuffer#buffer.buf_size < NewBuffer#buffer.convert_threshold of
        true ->
            NewBuffer;
        false ->
            buffer2dense(NewBuffer)
    end;
set(Index, Value, #dense{buf = Buf, buf_size = BufSize} = Dense) ->
    LeftOffset = Index * ?VALUE_SIZE,
    <<_:LeftOffset/bitstring, R:?VALUE_SIZE/integer, _/bitstring>> = Dense#dense.b,

    NewVal = hyper_utils:run_of_zeroes(Value),

    if
        R < NewVal ->
            New = Dense#dense{buf = [{Index, NewVal} | Buf], buf_size = BufSize + 1},
            case New#dense.buf_size < Dense#dense.merge_threshold of
                true ->
                    New;
                false ->
                    compact(New)
            end;
        true ->
            Dense
    end.

compact(#buffer{buf_size = BufSize, convert_threshold = ConvertThreshold} = Buffer) ->
    case BufSize < ConvertThreshold of
        true ->
            Buffer;
        false ->
            buffer2dense(Buffer)
    end;
compact(#dense{b = B, buf = Buf} = Dense) ->
    NewB = merge_buf(B, max_registers(Buf)),
    Dense#dense{
        b = NewB,
        buf = [],
        buf_size = 0
    }.

max_merge([First | Rest]) ->
    lists:foldl(fun max_merge/2, First, Rest).

max_merge(
    #dense{
        b = SmallB,
        buf = SmallBuf,
        buf_size = SmallBufSize
    },
    #dense{
        b = BigB,
        buf = BigBuf,
        buf_size = BigBufSize
    } =
        Big
) ->
    case SmallBufSize + BigBufSize < Big#dense.merge_threshold of
        true ->
            Merged = do_merge(SmallB, BigB, <<>>),
            Big#dense{
                b = Merged,
                buf = SmallBuf ++ BigBuf,
                buf_size = SmallBufSize + BigBufSize
            };
        false ->
            BigWithBuf = merge_buf(BigB, max_registers(SmallBuf ++ BigBuf)),
            Merged = do_merge(SmallB, BigWithBuf, <<>>),
            Big#dense{
                b = Merged,
                buf = [],
                buf_size = 0
            }
    end;
max_merge(
    #buffer{buf = Buf, buf_size = BufferSize},
    #dense{buf = DenseBuf, buf_size = DenseSize} = Dense
) ->
    case BufferSize + DenseSize < Dense#dense.merge_threshold of
        true ->
            Dense#dense{buf = Buf ++ DenseBuf, buf_size = BufferSize + DenseSize};
        false ->
            Merged = max_registers(DenseBuf ++ Buf),
            Dense#dense{buf = Merged, buf_size = length(Merged)}
    end;
max_merge(#dense{} = Dense, #buffer{} = Buffer) ->
    max_merge(Buffer, Dense);
max_merge(
    #buffer{buf = LeftBuf, buf_size = LeftBufSize},
    #buffer{buf = RightBuf, buf_size = RightBufSize} = Right
) ->
    case LeftBufSize + RightBufSize < Right#buffer.convert_threshold of
        true ->
            Right#buffer{buf = LeftBuf ++ RightBuf, buf_size = LeftBufSize + RightBufSize};
        false ->
            Merged = max_registers(LeftBuf ++ RightBuf),
            NewRight = Right#buffer{buf = Merged, buf_size = length(Merged)},
            case NewRight#buffer.buf_size < NewRight#buffer.convert_threshold of
                true ->
                    NewRight;
                false ->
                    buffer2dense(NewRight)
            end
    end.

reduce_precision(NewP, #dense{p = OldP} = Dense) ->
    ChangeP = OldP - NewP,
    Buf = register_fold(ChangeP, Dense),
    Empty = new_dense(NewP),
    compact(Empty#dense{buf = Buf});
reduce_precision(NewP, #buffer{p = OldP} = Buffer) ->
    ChangeP = OldP - NewP,
    Buf = register_fold(ChangeP, Buffer),
    Empty = new_dense(NewP),
    compact(Empty#dense{buf = Buf}).

%% fold an HLL to a lower number of registers `NewM` by projecting the values
%% onto their new index, see
%% http://research.neustar.biz/2012/09/12/set-operations-on-hlls-of-different-sizes/
%% NOTE: this function does not perform the max_registers step
register_fold(ChangeP, B) ->
    ChangeM = hyper_utils:m(ChangeP),
    element(
        3,
        fold(
            fun
                (I, V, {_Index, CurrentList, Acc}) when CurrentList == [] ->
                    {I, [V], Acc};
                (I, V, {Index, CurrentList, Acc}) when I - Index < ChangeM ->
                    {Index, [V | CurrentList], Acc};
                (I, V, {_Index, CurrentList, Acc}) ->
                    {I, [V], [
                        {I bsr ChangeP, hyper_utils:changeV(lists:reverse(CurrentList), ChangeP)}
                        | Acc
                    ]}
            end,
            {0, [], []},
            B
        )
    ).

register_sum(B) ->
    fold(
        fun
            (_, 0, Acc) ->
                Acc + 1.0;
            (_, 1, Acc) ->
                Acc + 0.5;
            (_, 2, Acc) ->
                Acc + 0.25;
            (_, 3, Acc) ->
                Acc + 0.125;
            (_, 4, Acc) ->
                Acc + 0.0625;
            (_, 5, Acc) ->
                Acc + 0.03125;
            (_, 6, Acc) ->
                Acc + 0.015625;
            (_, 7, Acc) ->
                Acc + 0.0078125;
            (_, 8, Acc) ->
                Acc + 0.00390625;
            (_, 9, Acc) ->
                Acc + 0.001953125;
            (_, V, Acc) ->
                Acc + math:pow(2, -V)
        end,
        0,
        B
    ).

register_histogram(#buffer{p = P} = B) ->
    register_histogram(B, P);
register_histogram(#dense{p = P} = B) ->
    register_histogram(B, P).

register_histogram(B, P) ->
    % todo use from_keys once we are at otp 26
    fold(
        fun(_, Value, Acc) -> maps:update_with(Value, fun(V) -> V + 1 end, 0, Acc) end,
        maps:from_list(
            lists:map(fun(I) -> {I, 0} end, lists:seq(0, 65 - P))
        ),
        B
    ).

zero_count(B) ->
    fold(
        fun
            (_, 0, Acc) ->
                Acc + 1;
            (_, _, Acc) ->
                Acc
        end,
        0,
        B
    ).

encode_registers(#buffer{} = Buffer) ->
    encode_registers(buffer2dense(Buffer));
encode_registers(#dense{b = B}) ->
    <<<<I:8/integer>> || <<I:?VALUE_SIZE/integer>> <= B>>.

decode_registers(AllBytes, P) ->
    M = hyper_utils:m(P),
    Bytes =
        case AllBytes of
            <<B:M/binary>> ->
                B;
            <<B:M/binary, 0>> ->
                B
        end,
    Dense = new_dense(P),
    Dense#dense{b = <<<<I:?VALUE_SIZE/integer>> || <<I:8>> <= Bytes>>}.

bytes(#dense{b = B}) ->
    erlang:byte_size(B);
bytes(#buffer{} = Buffer) ->
    erts_debug:flat_size(Buffer) * 8.

%%
%% INTERNALS
%%

empty_binary(M) ->
    list_to_bitstring([<<0:?VALUE_SIZE/integer>> || _ <- lists:seq(0, M - 1)]).

max_registers(Buf) ->
    lists:keysort(
        1,
        maps:to_list(
            lists:foldl(
                fun({I, V}, Acc) ->
                    case maps:find(I, Acc) of
                        {ok, R} when R >= V -> Acc;
                        _ -> maps:put(I, V, Acc)
                    end
                end,
                #{},
                Buf
            )
        )
    ).

buffer2dense(#buffer{buf = Buf, p = P}) ->
    Dense = new_dense(P),
    Merged = merge_buf(Dense#dense.b, max_registers(Buf)),
    Dense#dense{b = Merged}.

do_merge(<<>>, <<>>, Acc) ->
    Acc;
do_merge(
    <<Left:?VALUE_SIZE/integer, SmallRest/bitstring>>,
    <<Right:?VALUE_SIZE/integer, BigRest/bitstring>>,
    Acc
) ->
    do_merge(SmallRest, BigRest, <<Acc/bits, (max(Left, Right)):?VALUE_SIZE>>).

fold(F, Acc, #buffer{} = Buffer) ->
    % TODO fix this to not need to move to a dense, it is not technically neeeded
    % We can just fold on the list i think
    fold(F, Acc, buffer2dense(Buffer));
fold(F, Acc, #dense{b = B, buf = Buf}) ->
    do_fold(F, Acc, merge_buf(B, max_registers(Buf)), 0).

do_fold(_, Acc, <<>>, _) ->
    Acc;
do_fold(F, Acc, <<Value:?VALUE_SIZE/integer, Rest/bitstring>>, Index) ->
    do_fold(F, F(Index, Value, Acc), Rest, Index + 1).

merge_buf(B, L) ->
    merge_buf(B, L, -1, <<>>).

merge_buf(B, [], _PrevIndex, Acc) ->
    <<Acc/bitstring, B/bitstring>>;
merge_buf(B, [{Index, Value} | Rest], PrevIndex, Acc) ->
    I = (Index - PrevIndex - 1) * ?VALUE_SIZE,
    case B of
        <<Left:I/bitstring, OldValue:?VALUE_SIZE/integer, Right/bitstring>> ->
            case OldValue < Value of
                true ->
                    NewAcc = <<Acc/bitstring, Left/bitstring, Value:?VALUE_SIZE/integer>>,
                    merge_buf(Right, Rest, Index, NewAcc);
                false ->
                    NewAcc = <<Acc/bitstring, Left/bitstring, OldValue:?VALUE_SIZE/integer>>,
                    merge_buf(Right, Rest, Index, NewAcc)
            end;
        <<Left:I/bitstring>> ->
            <<Acc/bitstring, Left/bitstring, Value:?VALUE_SIZE/integer>>
    end.