Generic Aggregate Node

### Initialize Node

The following is how one might define the COUNT aggregate function.

    IdentityFun = fun ([A]) -> A end.
    CountFun = fun (List) -> erlang:length(List) end.
    SumFun = fun (List) -> lists:foldl(fun (Elem, Sum) -> Sum + Elem end, 0, List) end.

    {ok, Pid} = ydb_aggr_node:start_link(
        [
            {history_size, 4}
          , {columns, [a]}
          , {result_name, count}
          , {result_type, int}
          , {eval_fun, IdentityFun}
          , {pr_fun, CountFun}
          , {aggr_fun, SumFun}
        ]
      , [{schema, [{a, {1, int}}]}]
    ).

    ydb_plan_node:add_listener(Pid, erlang:self()).
    ydb_plan_node:relegate(Pid, {get_schema, [{a, {1, int}}]}).

* * *

### Send Tuple Messages

    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 100, {1}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 200, {2}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 300, {3}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 400, {4}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 500, {5}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 600, {6}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 700, {7}}]}).
    ydb_plan_node:relegate(Pid, {tuples, [{ydb_tuple, 800, {8}}]}).

* * *

### Send Diff Messages

    {ok, Tid} = ydb_ets_utils:create_diff_table(diff).

    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 100, {1}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 200, {2}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 300, {3}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 400, {4}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 500, {5}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 600, {6}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 700, {7}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).

    ets:delete_all_objects(Tid).
    ydb_ets_utils:add_diffs(Tid, '+', shell, {ydb_tuple, 800, {8}}).
    ydb_plan_node:relegate(Pid, {diffs, [Tid]}).