diff options
author | Damien F. Katz <damien@apache.org> | 2008-05-15 21:51:22 +0000 |
---|---|---|
committer | Damien F. Katz <damien@apache.org> | 2008-05-15 21:51:22 +0000 |
commit | a684f95cbcee7f2568a2ce04e7dc2bbb605a27b3 (patch) | |
tree | 3b2fafecde418e10834ec2e896efffb64b16a33a /src/couchdb/couch_query_servers.erl | |
parent | 41eeac05bff1b0a8b05f1110270c0beecd62991a (diff) |
Incremental reduce first checkin. Warning! Disk format change.
git-svn-id: https://svn.apache.org/repos/asf/incubator/couchdb/trunk@656861 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'src/couchdb/couch_query_servers.erl')
-rw-r--r-- | src/couchdb/couch_query_servers.erl | 111 |
1 files changed, 76 insertions, 35 deletions
diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl index a6e1750a..1fb2462c 100644 --- a/src/couchdb/couch_query_servers.erl +++ b/src/couchdb/couch_query_servers.erl @@ -17,7 +17,7 @@ -export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3,stop/0]). -export([start_doc_map/2, map_docs/2, stop_doc_map/1]). - +-export([reduce/3, combine/3]). -export([test/0, test/1]). -include("couch_db.hrl"). @@ -65,32 +65,21 @@ writeline(Port, String) -> % send command and get a response. prompt(Port, Json) -> writeline(Port, cjson:encode(Json)), - read_json(Port). + case read_json(Port) of + {obj, [{"error", Id}, {"reason", Reason}]} -> + throw({list_to_atom(Id),Reason}); + {obj, [{"reason", Reason}, {"error", Id}]} -> + throw({list_to_atom(Id),Reason}); + Result -> + Result + end. start_doc_map(Lang, Functions) -> - Port = - case gen_server:call(couch_query_servers, {get_port, Lang}) of - {ok, Port0} -> - link(Port0), - Port0; - {empty, Cmd} -> - ?LOG_INFO("Spawning new ~s instance.", [Lang]), - open_port({spawn, Cmd}, [stream, - {line, 1000}, - exit_status, - hide]); - Error -> - throw(Error) - end, - true = prompt(Port, {"reset"}), + Port = get_linked_port(Lang), % send the functions as json strings lists:foreach(fun(FunctionSource) -> - case prompt(Port, {"add_fun", FunctionSource}) of - true -> ok; - {obj, [{"error", Id}, {"reason", Reason}]} -> - throw({Id, Reason}) - end + true = prompt(Port, {"add_fun", FunctionSource}) end, Functions), {ok, {Lang, Port}}. @@ -100,19 +89,13 @@ map_docs({_Lang, Port}, Docs) -> Results = lists:map( fun(Doc) -> Json = couch_doc:to_json_obj(Doc, []), - case prompt(Port, {"map_doc", Json}) of - {obj, [{"error", Id}, {"reason", Reason}]} -> - throw({list_to_atom(Id),Reason}); - {obj, [{"reason", Reason}, {"error", Id}]} -> - throw({list_to_atom(Id),Reason}); - Results when is_tuple(Results) -> - % the results are a json array of function map yields like this: - % {FunResults1, FunResults2 ...} - % where funresults is are json arrays of key value pairs: - % {{Key1, Value1}, {Key2, Value2}} - % Convert to real lists, execept the key, value pairs - [tuple_to_list(FunResult) || FunResult <- tuple_to_list(Results)] - end + Results = prompt(Port, {"map_doc", Json}), + % the results are a json array of function map yields like this: + % {FunResults1, FunResults2 ...} + % where funresults is are json arrays of key value pairs: + % {{Key1, Value1}, {Key2, Value2}} + % Convert to real lists, execept the key, value pairs + [tuple_to_list(FunResult) || FunResult <- tuple_to_list(Results)] end, Docs), {ok, Results}. @@ -121,10 +104,68 @@ map_docs({_Lang, Port}, Docs) -> stop_doc_map(nil) -> ok; stop_doc_map({Lang, Port}) -> + return_linked_port(Lang, Port). + +get_linked_port(Lang) -> + case gen_server:call(couch_query_servers, {get_port, Lang}) of + {ok, Port0} -> + link(Port0), + true = prompt(Port0, {"reset"}), + Port0; + {empty, Cmd} -> + ?LOG_INFO("Spawning new ~s instance.", [Lang]), + open_port({spawn, Cmd}, [stream, + {line, 1000}, + exit_status, + hide]); + Error -> + throw(Error) + end. + +return_linked_port(Lang, Port) -> ok = gen_server:call(couch_query_servers, {return_port, {Lang, Port}}), true = unlink(Port), ok. +group_reductions_results([]) -> + []; +group_reductions_results(List) -> + {Heads, Tails} = lists:foldl( + fun([H|T], {HAcc,TAcc}) -> + {[H|HAcc], [T|TAcc]} + end, {[], []}, List), + case Tails of + [[]|_] -> % no tails left + [Heads]; + _ -> + [Heads | group_reductions_results(Tails)] + end. + +combine(_Lang, [], _ReducedValues) -> + {ok, []}; +combine(Lang, RedSrcs, ReducedValues) -> + Port = get_linked_port(Lang), + Grouped = group_reductions_results(ReducedValues), + Results = lists:zipwith( + fun(FunSrc, Values) -> + {true, {Result}} = + prompt(Port, {"combine", {FunSrc}, list_to_tuple(Values)}), + Result + end, RedSrcs, Grouped), + + return_linked_port(Lang, Port), + {ok, Results}. + +reduce(_Lang, [], _KVs) -> + {ok, []}; +reduce(Lang, RedSrcs, KVs) -> + Port = get_linked_port(Lang), + {true, Results} = prompt(Port, + {"reduce", list_to_tuple(RedSrcs), list_to_tuple(KVs)}), + return_linked_port(Lang, Port), + {ok, tuple_to_list(Results)}. + + init(QueryServerList) -> {ok, {QueryServerList, []}}. |