visemet/yocto-db — pull request #1 by visemet (2013-05-10 02:22:55+00:00)
Title: 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]}).