Initial checkin of _changes filters. The prime weak-spot for this approach is that it maintains an OS-process per connected filtered _changes consumer. I'm pretty sure we'll be able to work around this without changing the API, but it'll involve a lot of OS-process bookkeeping. Those enhancements should generally improve show & list performance as well. Punting on them for now, first wanted to get _changes filters implemented so people could give feedback.

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@795687 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 106 insertions, 51 deletions

diff --git a/src/couchdb/couch_db.erl b/src/couchdb/couch_db.erl
index 21bb1e5e..4e64846c 100644
--- a/src/couchdb/couch_db.erl
+++ b/src/couchdb/couch_db.erl
@@ -279,11 +279,9 @@ validate_doc_update(#db{validate_doc_funs=[]}, _Doc, _GetDiskDocFun) ->
     ok;
 validate_doc_update(_Db, #doc{id= <<"_local/",_/binary>>}, _GetDiskDocFun) ->
     ok;
-validate_doc_update(#db{name=DbName,user_ctx=Ctx}=Db, Doc, GetDiskDocFun) ->
+validate_doc_update(Db, Doc, GetDiskDocFun) ->
     DiskDoc = GetDiskDocFun(),
-    JsonCtx =  {[{<<"db">>, DbName},
-            {<<"name">>,Ctx#user_ctx.name},
-            {<<"roles">>,Ctx#user_ctx.roles}]},
+    JsonCtx = couch_util:json_user_ctx(Db),
     try [case Fun(Doc, DiskDoc, JsonCtx) of
             ok -> ok;
             Error -> throw(Error)
diff --git a/src/couchdb/couch_httpd_db.erl b/src/couchdb/couch_httpd_db.erl
index 0ec1cbda..9d7dda6b 100644
--- a/src/couchdb/couch_httpd_db.erl
+++ b/src/couchdb/couch_httpd_db.erl
@@ -66,33 +66,38 @@ get_changes_timeout(Req, Resp) ->
 
 handle_changes_req(#httpd{method='GET',path_parts=[DbName|_]}=Req, Db) ->
     StartSeq = list_to_integer(couch_httpd:qs_value(Req, "since", "0")),
+    Filter = couch_httpd:qs_value(Req, "filter", nil),
+    {ok, FilterFun, EndFilterFun} = make_filter_funs(Req, Db, Filter),
+    try
+        {ok, Resp} = start_json_response(Req, 200),
+        send_chunk(Resp, "{\"results\":[\n"),
+        case couch_httpd:qs_value(Req, "continuous", "false") of
+        "true" ->
+            Self = self(),
+            {ok, Notify} = couch_db_update_notifier:start_link(
+                fun({_, DbName0}) when DbName0 == DbName ->
+                    Self ! db_updated;
+                (_) ->
+                    ok
+                end),
+            {Timeout, TimeoutFun} = get_changes_timeout(Req, Resp),
+            couch_stats_collector:track_process_count(Self,
+                                {httpd, clients_requesting_changes}),
+            try
+                keep_sending_changes(Req, Resp, Db, StartSeq, <<"">>, Timeout, TimeoutFun, FilterFun)
+            after
+                couch_db_update_notifier:stop(Notify),
+                get_rest_db_updated() % clean out any remaining update messages
+            end;
 
-    {ok, Resp} = start_json_response(Req, 200),
-    send_chunk(Resp, "{\"results\":[\n"),
-    case couch_httpd:qs_value(Req, "continuous", "false") of
-    "true" ->
-        Self = self(),
-        {ok, Notify} = couch_db_update_notifier:start_link(
-            fun({_, DbName0}) when DbName0 == DbName ->
-                Self ! db_updated;
-            (_) ->
-                ok
-            end),
-        {Timeout, TimeoutFun} = get_changes_timeout(Req, Resp),
-        couch_stats_collector:track_process_count(Self,
-                            {httpd, clients_requesting_changes}),
-        try
-            keep_sending_changes(Req, Resp, Db, StartSeq, <<"">>, Timeout, TimeoutFun)
-        after
-            couch_db_update_notifier:stop(Notify),
-            get_rest_db_updated() % clean out any remaining update messages
-        end;
-
-    "false" ->
-        {ok, {LastSeq, _Prepend}} =
-                send_changes(Req, Resp, Db, StartSeq, <<"">>),
-        send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [LastSeq])),
-        send_chunk(Resp, "")
+        "false" ->
+            {ok, {LastSeq, _Prepend}} =
+                    send_changes(Req, Resp, Db, StartSeq, <<"">>, FilterFun),
+            send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [LastSeq])),
+            send_chunk(Resp, "")
+        end
+    after
+        EndFilterFun()
     end;
 
 handle_changes_req(#httpd{path_parts=[_,<<"_changes">>]}=Req, _Db) ->
@@ -113,27 +118,23 @@ get_rest_db_updated() ->
     after 0 -> updated
     end.
 
-keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, Db, StartSeq, Prepend, Timeout, TimeoutFun) ->
-    {ok, {EndSeq, Prepend2}} = send_changes(Req, Resp, Db, StartSeq, Prepend),
+keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, Db, StartSeq, Prepend, Timeout, TimeoutFun, FilterFun) ->
+    {ok, {EndSeq, Prepend2}} = send_changes(Req, Resp, Db, StartSeq, Prepend, FilterFun),
     couch_db:close(Db),
     case wait_db_updated(Timeout, TimeoutFun) of
     updated ->
         {ok, Db2} = couch_db:open(DbName, [{user_ctx, UserCtx}]),
-        keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, TimeoutFun);
+        keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, TimeoutFun, FilterFun);
     stop ->
         send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [EndSeq])),
         send_chunk(Resp, "")
     end.
 
-send_changes(Req, Resp, Db, StartSeq, Prepend0) ->
+send_changes(Req, Resp, Db, StartSeq, Prepend0, FilterFun) ->
     Style = list_to_existing_atom(
             couch_httpd:qs_value(Req, "style", "main_only")),
     couch_db:changes_since(Db, Style, StartSeq,
         fun([#doc_info{id=Id, high_seq=Seq}|_]=DocInfos, {_, Prepend}) ->
-            FilterFun =
-            fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
-                {[{rev, couch_doc:rev_to_str(Rev)}]}
-            end,
             Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos],
             Results = [Result || Result <- Results0, Result /= null],
             case Results of
@@ -147,6 +148,39 @@ send_changes(Req, Resp, Db, StartSeq, Prepend0) ->
             end
         end, {StartSeq, Prepend0}).
 
+make_filter_funs(_Req, _Db, nil) -> 
+    {ok, fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
+        {[{rev, couch_doc:rev_to_str(Rev)}]}
+    end,
+    fun() -> ok end};
+make_filter_funs(Req, Db, Filter) -> 
+    case [list_to_binary(couch_httpd:unquote(Part))
+            || Part <- string:tokens(Filter, "/")] of
+    [DName, FName] ->
+        DesignId = <<"_design/", DName/binary>>,
+        #doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []),
+        Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>),
+        FilterSrc = couch_util:get_nested_json_value({Props}, [<<"filters">>, FName]),
+        {ok, Pid} = couch_query_servers:start_filter(Lang, FilterSrc),
+        FilterFun = fun(DInfo = #doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
+            {ok, Doc} = couch_db:open_doc(Db, DInfo),
+            {ok, Pass} = couch_query_servers:filter_doc(Pid, Doc, Req, Db),
+            case Pass of
+            true ->
+                {[{rev, couch_doc:rev_to_str(Rev)}]};
+            false ->
+                null
+            end
+        end,
+        EndFilterFun = fun() ->
+            couch_query_servers:end_filter(Pid)
+        end,
+        {ok, FilterFun, EndFilterFun};
+    _Else ->
+        throw({bad_request, 
+            "filter parameter must be of the form `designname/filtername`"})
+    end.  
+
 handle_compact_req(#httpd{method='POST',path_parts=[DbName,_,Id|_]}=Req, _Db) ->
     ok = couch_view_compactor:start_compact(DbName, Id),
     send_json(Req, 202, {[{ok, true}]});
@@ -747,7 +781,7 @@ couch_doc_from_req(Req, DocId, Json) ->
 
 % Useful for debugging
 % couch_doc_open(Db, DocId) ->
-%   couch_doc_open(Db, DocId, [], []).
+%   couch_doc_open(Db, DocId, nil, []).
 
 couch_doc_open(Db, DocId, Rev, Options) ->
     case Rev of
diff --git a/src/couchdb/couch_httpd_show.erl b/src/couchdb/couch_httpd_show.erl
index 692ea23d..a73105b4 100644
--- a/src/couchdb/couch_httpd_show.erl
+++ b/src/couchdb/couch_httpd_show.erl
@@ -45,7 +45,7 @@ handle_doc_show(Req, DesignName, ShowName, DocId, Db) ->
     DesignId = <<"_design/", DesignName/binary>>,
     #doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []),
     Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>),
-    ShowSrc = get_nested_json_value({Props}, [<<"shows">>, ShowName]),
+    ShowSrc = couch_util:get_nested_json_value({Props}, [<<"shows">>, ShowName]),
     Doc = case DocId of
         nil -> nil;
         _ ->
@@ -78,18 +78,10 @@ handle_view_list(Req, DesignName, ListName, ViewName, Db, Keys) ->
     DesignId = <<"_design/", DesignName/binary>>,
     #doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []),
     Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>),
-    ListSrc = get_nested_json_value({Props}, [<<"lists">>, ListName]),
+    ListSrc = couch_util:get_nested_json_value({Props}, [<<"lists">>, ListName]),
     send_view_list_response(Lang, ListSrc, ViewName, DesignId, Req, Db, Keys).
 
-get_nested_json_value({Props}, [Key|Keys]) ->
-    case proplists:get_value(Key, Props, nil) of
-    nil -> throw({not_found, <<"missing json key: ", Key/binary>>});
-    Value -> get_nested_json_value(Value, Keys)
-    end;
-get_nested_json_value(Value, []) ->
-    Value;
-get_nested_json_value(_NotJSONObj, _) ->
-    throw({not_found, json_mismatch}).
+
 
 send_view_list_response(Lang, ListSrc, ViewName, DesignId, Req, Db, Keys) ->
     Stale = couch_httpd_view:get_stale_type(Req),
diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl
index 02018e33..1d452796 100644
--- a/src/couchdb/couch_query_servers.erl
+++ b/src/couchdb/couch_query_servers.erl
@@ -20,6 +20,7 @@
 -export([reduce/3, rereduce/3,validate_doc_update/5]).
 -export([render_doc_show/6, start_view_list/2,
         render_list_head/4, render_list_row/3, render_list_tail/1]).
+-export([start_filter/2, filter_doc/4, end_filter/1]).
 % -export([test/0]).
 
 -include("couch_db.hrl").
@@ -211,8 +212,22 @@ render_list_tail({Lang, Pid}) ->
     ok = ret_os_process(Lang, Pid),
     JsonResp.
 
+start_filter(Lang, FilterSrc) ->
+    Pid = get_os_process(Lang),
+    true = couch_os_process:prompt(Pid, [<<"add_fun">>, FilterSrc]),
+    {ok, {Lang, Pid}}.
 
+filter_doc({_Lang, Pid}, Doc, Req, Db) ->
+    JsonReq = couch_httpd_external:json_req_obj(Req, Db),
+    JsonDoc = couch_doc:to_json_obj(Doc, [revs]),
+    JsonCtx = couch_util:json_user_ctx(Db),
+    [true, [Pass]] = couch_os_process:prompt(Pid,
+        [<<"filter">>, [JsonDoc], JsonReq, JsonCtx]),
+    {ok, Pass}.
 
+end_filter({Lang, Pid}) ->
+    ok = ret_os_process(Lang, Pid).
+    
 
 init([]) ->
 
diff --git a/src/couchdb/couch_util.erl b/src/couchdb/couch_util.erl
index 817572bb..db9f937c 100644
--- a/src/couchdb/couch_util.erl
+++ b/src/couchdb/couch_util.erl
@@ -17,7 +17,7 @@
 -export([new_uuid/0, rand32/0, implode/2, collate/2, collate/3]).
 -export([abs_pathname/1,abs_pathname/2, trim/1, ascii_lower/1]).
 -export([encodeBase64/1, decodeBase64/1, to_hex/1,parse_term/1,dict_find/3]).
--export([file_read_size/1]).
+-export([file_read_size/1, get_nested_json_value/2, json_user_ctx/1]).
 -export([to_binary/1, to_list/1]).
 
 -include("couch_db.hrl").
@@ -75,6 +75,22 @@ parse_term(List) ->
     erl_parse:parse_term(Tokens).
 
 
+get_nested_json_value({Props}, [Key|Keys]) ->
+    case proplists:get_value(Key, Props, nil) of
+    nil -> throw({not_found, <<"missing json key: ", Key/binary>>});
+    Value -> get_nested_json_value(Value, Keys)
+    end;
+get_nested_json_value(Value, []) ->
+    Value;
+get_nested_json_value(_NotJSONObj, _) ->
+    throw({not_found, json_mismatch}).
+
+json_user_ctx(#db{name=DbName, user_ctx=Ctx}) ->
+    {[{<<"db">>, DbName},
+            {<<"name">>,Ctx#user_ctx.name},
+            {<<"roles">>,Ctx#user_ctx.roles}]}.
+    
+
 % returns a random integer
 rand32() ->
     crypto:rand_uniform(0, 16#100000000).