<commit_before>-module(metric_vnode).
-behaviour(riak_core_vnode).

-include_lib("riak_core/include/riak_core_vnode.hrl").
-include_lib("mmath/include/mmath.hrl").


-export([start_vnode/1,
         init/1,
         terminate/2,
         handle_command/3,
         is_empty/1,
         delete/1,
         repair/3,
         handle_handoff_command/3,
         handoff_starting/2,
         handoff_cancelled/1,
         handoff_finished/2,
         handle_handoff_data/2,
         encode_handoff_item/2,
         handle_coverage/4,
         handle_info/2,
         handle_exit/3]).

-export([mput/3, put/5, get/4]).

-ignore_xref([
              start_vnode/1,
              put/5,
              mput/3,
              get/4,
              repair/4,
              handle_info/2,
              repair/3
             ]).

-record(state, {
          partition,
          node,
          tbl,
          ct,
          io,
          now,
          resolutions = btrie:new(),
          lifetimes  = btrie:new()
         }).

-define(MASTER, metric_vnode_master).
-define(MAX_Q_LEN, 20).
-define(VAC, 1000*60*60*2).

%% API
start_vnode(I) ->
    riak_core_vnode_master:get_vnode_pid(I, ?MODULE).

repair(IdxNode, {Bucket, Metric}, {Time, Obj}) ->
    riak_core_vnode_master:command(
      IdxNode,
      {repair, Bucket, Metric, Time, Obj},
      ignore,
      ?MASTER).


put(Preflist, ReqID, Bucket, Metric, {Time, Values}) when is_list(Values) ->
    put(Preflist, ReqID, Bucket, Metric,
        {Time, << <<?INT:?TYPE_SIZE, V:?BITS/?INT_TYPE>> || V <- Values >>});

put(Preflist, ReqID, Bucket, Metric, {Time, Value}) when is_integer(Value) ->
    put(Preflist, ReqID, Bucket, Metric,
        {Time, <<?INT:?TYPE_SIZE, Value:?BITS/?INT_TYPE>>});

put(Preflist, ReqID, Bucket, Metric, {Time, Value}) ->
    riak_core_vnode_master:command(Preflist,
                                   {put, Bucket, Metric, {Time, Value}},
                                   {raw, ReqID, self()},
                                   ?MASTER).

mput(Preflist, ReqID, Data) ->
    riak_core_vnode_master:command(Preflist,
                                   {mput, Data},
                                   {raw, ReqID, self()},
                                   ?MASTER).

get(Preflist, ReqID, {Bucket, Metric}, {Time, Count}) ->
    riak_core_vnode_master:command(Preflist,
                                   {get, ReqID, Bucket, Metric, {Time, Count}},
                                   {fsm, undefined, self()},
                                   ?MASTER).


init([Partition]) ->
    ok = dalmatiner_vacuum:register(),
    process_flag(trap_exit, true),
    random:seed(erlang:phash2([node()]),
                erlang:monotonic_time(),
                erlang:unique_integer()),
    P = list_to_atom(integer_to_list(Partition)),
    CT = case application:get_env(metric_vnode, cache_points) of
             {ok, V} ->
                 V;
             _ ->
                 10
         end,
    WorkerPoolSize = case application:get_env(metric_vnode, async_workers) of
                         {ok, Val} ->
                             Val;
                         undefined ->
                             5
                     end,
    FoldWorkerPool = {pool, metric_worker, WorkerPoolSize, []},
    {ok, IO} = metric_io:start_link(Partition),
    {ok, #state{
            partition = Partition,
            node = node(),
            tbl = ets:new(P, [public, ordered_set]),
            io = IO,
            now = timestamp(),
            ct = CT
           },
     [FoldWorkerPool]}.

repair_update_cache(Bucket, Metric, Time, Count, Value,
                    #state{tbl=T} = State) ->
    case ets:lookup(T, {Bucket, Metric}) of
        %% ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
        %% │Cache│     │     │     │     │     │Start│ ... │St+Si│     │
        %% ├─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┤
        %% │Rep. │     │Time │ ... │Ti+Co│     │     │     │     │     │
        %% └─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
        %%
        %% If we repair on a place before the metric, we will just
        %% write it!
        [{{Bucket, Metric}, Start, _Size, _Time, _Array}]
          when Time + Count < Start ->
            metric_io:write(State#state.io, Bucket, Metric, Time,
                            Value),
            State;
        %% ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
        %% │Cache│     │Start│ ... │St+Si│     │     │     │     │     │
        %% ├─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┤
        %% │Rep. │     │     │     │     │     │Time │ ... │Ti+Co│     │
        %% └─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
        %%
        %% The data is entirely ahead of the cache, so we flush the
        %% cache and use the repair request as new the cache.
        [{{Bucket, Metric}, Start, Size, _Time, Array}]
          when Start + Size < Time ->
            Bin = k6_bytea:get(Array, 0, Size * ?DATA_SIZE),
            k6_bytea:delete(Array),
            ets:delete(T, {Bucket, Metric}),
            metric_io:write(State#state.io, Bucket, Metric, Start,
                            Bin),
            do_put(Bucket, Metric, Time, Value, State);
        %% ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
        %% │Cache│     │     │     │Start│ ... │St+Si│     │     │     │
        %% ├─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┤
        %% │Rep. │     │     │     │     │Time │ ... │Ti+Co│     │     │
        %% └─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
        %%
        %% Now it gets tricky. The repair intersects with the
        %% cache - this should never happen but it probably will, so
        %% it sucks!  There is no sane way to merge the values if
        %% they intersect, however we know the following:
        %% 1) A repair request, based on how it is built, will never
        %%    contain unset(empty) values.
        %% 2) A cache can be an entirely empty value or contain
        %%    empty segments.
        %% Based on that, the best approach is to flush the cache and
        %% then also to flush the repair request. This ensures that
        %% nothing will be overwritten with a empty value.
        %%
        %% - Flushing the repair once could lead to empties in the
        %%   cache overwriting non-empties from the repair.
        %% - Flushing the cache and caching the repair could lead to
        %%   new empties in the new cache from the repair now
        %%   overwriting non-empties from the previous cache.
        [{{Bucket, Metric}, Start, Size, _Time, Array}] ->
            Bin = k6_bytea:get(Array, 0, Size * ?DATA_SIZE),
            k6_bytea:delete(Array),
            ets:delete(T, {Bucket, Metric}),
            metric_io:write(State#state.io, Bucket, Metric, Start,
                            Bin),
            metric_io:write(State#state.io, Bucket, Metric, Time,
                            Value),
            State;
        %% If we had no privious cache we can safely cache the
        %% repair.
        [] ->
            do_put(Bucket, Metric, Time, Value, State)
    end.

%% Repair request are always full values not including non set values!
handle_command({repair, Bucket, Metric, Time, Value}, _Sender, State)
  when is_binary(Bucket), is_binary(Metric), is_integer(Time) ->
    Count = mmath_bin:length(Value),
    case valid_ts(Time + Count, Bucket, State) of
        {true, State1} ->
            State2 = repair_update_cache(Bucket, Metric, Time, Count, Value,
                                         State1),
            {noreply, State2};
        {false, State1} ->
            {noreply, State1}
    end;

handle_command({mput, Data}, _Sender, State) ->
    State1 = lists:foldl(fun({Bucket, Metric, Time, Value}, StateAcc)
                               when is_binary(Bucket), is_binary(Metric),
                                    is_integer(Time) ->
                                 do_put(Bucket, Metric, Time, Value, StateAcc)
                         end, State, Data),
    {reply, ok, State1};

handle_command({put, Bucket, Metric, {Time, Value}}, _Sender, State)
  when is_binary(Bucket), is_binary(Metric), is_integer(Time) ->
    State1=  do_put(Bucket, Metric, Time, Value, State),
    {reply, ok, State1};

handle_command({get, ReqID, Bucket, Metric, {Time, Count}}, Sender,
               #state{tbl=T, io=IO, node=N, partition=P} = State) ->
    BM = {Bucket, Metric},
    case ets:lookup(T, BM) of
        %% If our request is entirely cache we don't need to bother the IO node
        [{BM, Start, Size, _Time, Array}]
          when Start =< Time,
               (Start + Size) >= (Time + Count)
               ->
            %% How many bytes can we skip?
            SkipBytes = (Time - Start) * ?DATA_SIZE,
            Data = k6_bytea:get(Array, SkipBytes, (Count * ?DATA_SIZE)),
            {Resolution, State1} = get_resolution(Bucket, State),
            {reply, {ok, ReqID, {P, N}, {Resolution, Data}}, State1};
        %% The request is neither before, after nor entirely inside the cache
        %% we have to read data, but apply cached part on top of it.
        [{BM, Start, Size, _Time, Array}]
          when
              %% The window starts before the cache but ends in it
              (Time < Start andalso Time + Count >= Start)

              %% The window starts inside the cahce and ends afterwards
              orelse (Time =< Start + Size andalso Time + Count > Start + Size)
              ->
            PartStart = max(Time, Start),
            PartCount = min(Time + Count, Start + Size) - PartStart,
            SkipBytes = (PartStart - Start) * ?DATA_SIZE,
            Bin = k6_bytea:get(Array, SkipBytes, (PartCount * ?DATA_SIZE)),
            Offset = PartStart - Time,
            Part = {Offset, PartCount, Bin},
            metric_io:read_rest(
              IO, Bucket, Metric, Time, Count, Part, ReqID, Sender),
            {noreply, State};
        %% If we are here we know that there is either no cahce or the requested
        %% window and the cache do not overlap, so we can simply serve it from
        %% the io servers
        _ ->
            metric_io:read(IO, Bucket, Metric, Time, Count, ReqID, Sender),
            {noreply, State}
    end.

handle_handoff_command(?FOLD_REQ{foldfun=Fun, acc0=Acc0}, Sender,
                       State=#state{tbl=T, io = IO}) ->
    ets:foldl(fun({{Bucket, Metric}, Start, Size, _, Array}, _) ->
                      Bin = k6_bytea:get(Array, 0, Size * ?DATA_SIZE),
                      k6_bytea:delete(Array),
                      metric_io:write(IO, Bucket, Metric, Start, Bin)
              end, ok, T),
    ets:delete_all_objects(T),
    FinishFun =
        fun(Acc) ->
                riak_core_vnode:reply(Sender, Acc)
        end,
    case metric_io:fold(IO, Fun, Acc0) of
        {ok, AsyncWork} ->
            {async, {fold, AsyncWork, FinishFun}, Sender, State};
        empty ->
            {async, {fold, fun() -> Acc0 end, FinishFun}, Sender, State}
    end;

%% We want to forward all the other handoff commands
handle_handoff_command(_Message, _Sender, State) ->
    {forward, State}.

handoff_starting(_TargetNode, State) ->
    {true, State}.

handoff_cancelled(State) ->
    {ok, State}.

handoff_finished(_TargetNode, State) ->
    {ok, State}.

handle_handoff_data(Data, State) ->
    {{Bucket, Metric}, ValList} = binary_to_term(Data),
    true = is_binary(Bucket),
    true = is_binary(Metric),
    State1 = lists:foldl(fun ({T, Bin}, StateAcc) ->
                                 do_put(Bucket, Metric, T, Bin, StateAcc, 2)
                         end, State, ValList),
    {reply, ok, State1}.

encode_handoff_item(Key, Value) ->
    term_to_binary({Key, Value}).

is_empty(State = #state{tbl = T, io=IO}) ->
    case ets:first(T) == '$end_of_table' andalso metric_io:empty(IO) of
        true ->
            {true, State};
        false ->
            Count = metric_io:count(IO),
            {false, {Count, objects}, State}
    end.

delete(State = #state{io = IO, tbl=T}) ->
    ets:delete_all_objects(T),
    ok = metric_io:delete(IO),
    {ok, State}.

handle_coverage({metrics, Bucket}, _KS, Sender, State = #state{io = IO}) ->
    AsyncWork = fun() ->
                        {ok, Ms} = metric_io:metrics(IO, Bucket),
                        Ms
                end,<commit_msg>Use compression for handoffs.<commit_after>-module(metric_vnode).
-behaviour(riak_core_vnode).

-include_lib("riak_core/include/riak_core_vnode.hrl").
-include_lib("mmath/include/mmath.hrl").


-export([start_vnode/1,
         init/1,
         terminate/2,
         handle_command/3,
         is_empty/1,
         delete/1,
         repair/3,
         handle_handoff_command/3,
         handoff_starting/2,
         handoff_cancelled/1,
         handoff_finished/2,
         handle_handoff_data/2,
         encode_handoff_item/2,
         handle_coverage/4,
         handle_info/2,
         handle_exit/3]).

-export([mput/3, put/5, get/4]).

-ignore_xref([
              start_vnode/1,
              put/5,
              mput/3,
              get/4,
              repair/4,
              handle_info/2,
              repair/3
             ]).

-record(state, {
          partition,
          node,
          tbl,
          ct,
          io,
          now,
          resolutions = btrie:new(),
          lifetimes  = btrie:new()
         }).

-define(MASTER, metric_vnode_master).
-define(MAX_Q_LEN, 20).
-define(VAC, 1000*60*60*2).

%% API
start_vnode(I) ->
    riak_core_vnode_master:get_vnode_pid(I, ?MODULE).

repair(IdxNode, {Bucket, Metric}, {Time, Obj}) ->
    riak_core_vnode_master:command(
      IdxNode,
      {repair, Bucket, Metric, Time, Obj},
      ignore,
      ?MASTER).


put(Preflist, ReqID, Bucket, Metric, {Time, Values}) when is_list(Values) ->
    put(Preflist, ReqID, Bucket, Metric,
        {Time, << <<?INT:?TYPE_SIZE, V:?BITS/?INT_TYPE>> || V <- Values >>});

put(Preflist, ReqID, Bucket, Metric, {Time, Value}) when is_integer(Value) ->
    put(Preflist, ReqID, Bucket, Metric,
        {Time, <<?INT:?TYPE_SIZE, Value:?BITS/?INT_TYPE>>});

put(Preflist, ReqID, Bucket, Metric, {Time, Value}) ->
    riak_core_vnode_master:command(Preflist,
                                   {put, Bucket, Metric, {Time, Value}},
                                   {raw, ReqID, self()},
                                   ?MASTER).

mput(Preflist, ReqID, Data) ->
    riak_core_vnode_master:command(Preflist,
                                   {mput, Data},
                                   {raw, ReqID, self()},
                                   ?MASTER).

get(Preflist, ReqID, {Bucket, Metric}, {Time, Count}) ->
    riak_core_vnode_master:command(Preflist,
                                   {get, ReqID, Bucket, Metric, {Time, Count}},
                                   {fsm, undefined, self()},
                                   ?MASTER).


init([Partition]) ->
    ok = dalmatiner_vacuum:register(),
    process_flag(trap_exit, true),
    random:seed(erlang:phash2([node()]),
                erlang:monotonic_time(),
                erlang:unique_integer()),
    P = list_to_atom(integer_to_list(Partition)),
    CT = case application:get_env(metric_vnode, cache_points) of
             {ok, V} ->
                 V;
             _ ->
                 10
         end,
    WorkerPoolSize = case application:get_env(metric_vnode, async_workers) of
                         {ok, Val} ->
                             Val;
                         undefined ->
                             5
                     end,
    FoldWorkerPool = {pool, metric_worker, WorkerPoolSize, []},
    {ok, IO} = metric_io:start_link(Partition),
    {ok, #state{
            partition = Partition,
            node = node(),
            tbl = ets:new(P, [public, ordered_set]),
            io = IO,
            now = timestamp(),
            ct = CT
           },
     [FoldWorkerPool]}.

repair_update_cache(Bucket, Metric, Time, Count, Value,
                    #state{tbl=T} = State) ->
    case ets:lookup(T, {Bucket, Metric}) of
        %% ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
        %% │Cache│     │     │     │     │     │Start│ ... │St+Si│     │
        %% ├─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┤
        %% │Rep. │     │Time │ ... │Ti+Co│     │     │     │     │     │
        %% └─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
        %%
        %% If we repair on a place before the metric, we will just
        %% write it!
        [{{Bucket, Metric}, Start, _Size, _Time, _Array}]
          when Time + Count < Start ->
            metric_io:write(State#state.io, Bucket, Metric, Time,
                            Value),
            State;
        %% ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
        %% │Cache│     │Start│ ... │St+Si│     │     │     │     │     │
        %% ├─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┤
        %% │Rep. │     │     │     │     │     │Time │ ... │Ti+Co│     │
        %% └─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
        %%
        %% The data is entirely ahead of the cache, so we flush the
        %% cache and use the repair request as new the cache.
        [{{Bucket, Metric}, Start, Size, _Time, Array}]
          when Start + Size < Time ->
            Bin = k6_bytea:get(Array, 0, Size * ?DATA_SIZE),
            k6_bytea:delete(Array),
            ets:delete(T, {Bucket, Metric}),
            metric_io:write(State#state.io, Bucket, Metric, Start,
                            Bin),
            do_put(Bucket, Metric, Time, Value, State);
        %% ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
        %% │Cache│     │     │     │Start│ ... │St+Si│     │     │     │
        %% ├─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┼─────┤
        %% │Rep. │     │     │     │     │Time │ ... │Ti+Co│     │     │
        %% └─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
        %%
        %% Now it gets tricky. The repair intersects with the
        %% cache - this should never happen but it probably will, so
        %% it sucks!  There is no sane way to merge the values if
        %% they intersect, however we know the following:
        %% 1) A repair request, based on how it is built, will never
        %%    contain unset(empty) values.
        %% 2) A cache can be an entirely empty value or contain
        %%    empty segments.
        %% Based on that, the best approach is to flush the cache and
        %% then also to flush the repair request. This ensures that
        %% nothing will be overwritten with a empty value.
        %%
        %% - Flushing the repair once could lead to empties in the
        %%   cache overwriting non-empties from the repair.
        %% - Flushing the cache and caching the repair could lead to
        %%   new empties in the new cache from the repair now
        %%   overwriting non-empties from the previous cache.
        [{{Bucket, Metric}, Start, Size, _Time, Array}] ->
            Bin = k6_bytea:get(Array, 0, Size * ?DATA_SIZE),
            k6_bytea:delete(Array),
            ets:delete(T, {Bucket, Metric}),
            metric_io:write(State#state.io, Bucket, Metric, Start,
                            Bin),
            metric_io:write(State#state.io, Bucket, Metric, Time,
                            Value),
            State;
        %% If we had no privious cache we can safely cache the
        %% repair.
        [] ->
            do_put(Bucket, Metric, Time, Value, State)
    end.

%% Repair request are always full values not including non set values!
handle_command({repair, Bucket, Metric, Time, Value}, _Sender, State)
  when is_binary(Bucket), is_binary(Metric), is_integer(Time) ->
    Count = mmath_bin:length(Value),
    case valid_ts(Time + Count, Bucket, State) of
        {true, State1} ->
            State2 = repair_update_cache(Bucket, Metric, Time, Count, Value,
                                         State1),
            {noreply, State2};
        {false, State1} ->
            {noreply, State1}
    end;

handle_command({mput, Data}, _Sender, State) ->
    State1 = lists:foldl(fun({Bucket, Metric, Time, Value}, StateAcc)
                               when is_binary(Bucket), is_binary(Metric),
                                    is_integer(Time) ->
                                 do_put(Bucket, Metric, Time, Value, StateAcc)
                         end, State, Data),
    {reply, ok, State1};

handle_command({put, Bucket, Metric, {Time, Value}}, _Sender, State)
  when is_binary(Bucket), is_binary(Metric), is_integer(Time) ->
    State1=  do_put(Bucket, Metric, Time, Value, State),
    {reply, ok, State1};

handle_command({get, ReqID, Bucket, Metric, {Time, Count}}, Sender,
               #state{tbl=T, io=IO, node=N, partition=P} = State) ->
    BM = {Bucket, Metric},
    case ets:lookup(T, BM) of
        %% If our request is entirely cache we don't need to bother the IO node
        [{BM, Start, Size, _Time, Array}]
          when Start =< Time,
               (Start + Size) >= (Time + Count)
               ->
            %% How many bytes can we skip?
            SkipBytes = (Time - Start) * ?DATA_SIZE,
            Data = k6_bytea:get(Array, SkipBytes, (Count * ?DATA_SIZE)),
            {Resolution, State1} = get_resolution(Bucket, State),
            {reply, {ok, ReqID, {P, N}, {Resolution, Data}}, State1};
        %% The request is neither before, after nor entirely inside the cache
        %% we have to read data, but apply cached part on top of it.
        [{BM, Start, Size, _Time, Array}]
          when
              %% The window starts before the cache but ends in it
              (Time < Start andalso Time + Count >= Start)

              %% The window starts inside the cahce and ends afterwards
              orelse (Time =< Start + Size andalso Time + Count > Start + Size)
              ->
            PartStart = max(Time, Start),
            PartCount = min(Time + Count, Start + Size) - PartStart,
            SkipBytes = (PartStart - Start) * ?DATA_SIZE,
            Bin = k6_bytea:get(Array, SkipBytes, (PartCount * ?DATA_SIZE)),
            Offset = PartStart - Time,
            Part = {Offset, PartCount, Bin},
            metric_io:read_rest(
              IO, Bucket, Metric, Time, Count, Part, ReqID, Sender),
            {noreply, State};
        %% If we are here we know that there is either no cahce or the requested
        %% window and the cache do not overlap, so we can simply serve it from
        %% the io servers
        _ ->
            metric_io:read(IO, Bucket, Metric, Time, Count, ReqID, Sender),
            {noreply, State}
    end.

handle_handoff_command(?FOLD_REQ{foldfun=Fun, acc0=Acc0}, Sender,
                       State=#state{tbl=T, io = IO}) ->
    ets:foldl(fun({{Bucket, Metric}, Start, Size, _, Array}, _) ->
                      Bin = k6_bytea:get(Array, 0, Size * ?DATA_SIZE),
                      k6_bytea:delete(Array),
                      metric_io:write(IO, Bucket, Metric, Start, Bin)
              end, ok, T),
    ets:delete_all_objects(T),
    FinishFun =
        fun(Acc) ->
                riak_core_vnode:reply(Sender, Acc)
        end,
    case metric_io:fold(IO, Fun, Acc0) of
        {ok, AsyncWork} ->
            {async, {fold, AsyncWork, FinishFun}, Sender, State};
        empty ->
            {async, {fold, fun() -> Acc0 end, FinishFun}, Sender, State}
    end;

%% We want to forward all the other handoff commands
handle_handoff_command(_Message, _Sender, State) ->
    {forward, State}.

handoff_starting(_TargetNode, State) ->
    {true, State}.

handoff_cancelled(State) ->
    {ok, State}.

handoff_finished(_TargetNode, State) ->
    {ok, State}.

-dialyzer({no_return, handle_handoff_data/2}).
handle_handoff_data(Compressed, State) ->
    {ok, Data} = snappy:decompress(Compressed),
    {{Bucket, Metric}, ValList} = binary_to_term(Data),
    true = is_binary(Bucket),
    true = is_binary(Metric),
    State1 = lists:foldl(fun ({T, Bin}, StateAcc) ->
                                 do_put(Bucket, Metric, T, Bin, StateAcc, 2)
                         end, State, ValList),
    {reply, ok, State1}.

-dialyzer({no_return, encode_handoff_item/2}).
encode_handoff_item(Key, Value) ->
    {ok, R} = snappy:compress(term_to_binary({Key, Value})),
    R.

is_empty(State = #state{tbl = T, io=IO}) ->
    case ets:first(T) == '$end_of_table' andalso metric_io:empty(IO) of
        true ->
            {true, State};
        false ->
            Count = metric_io:count(IO),
            {false, {Count, objects}, State}
    end.

delete(State = #state{io = IO, tbl=T}) ->
    ets:delete_all_objects(T),
    ok = metric_io:delete(IO),
    {ok, State}.

handle_coverage({metrics, Bucket}, _KS, Sender, State = #state{io = IO}) ->
    AsyncWork = fun() ->
                        {ok, Ms} = metric_io:metrics(IO, Bucket),
                        Ms
                end,
{'fraction_non_alphanumeric': 0.16335847558554983, 'fraction_numerical': 0.005875347360063518, 'mean_word_length': 1.849338310145911, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 113, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2123278', 'n_tokens_mistral': 8489, 'n_tokens_neox': 8150, 'n_words': 2000}