support _find (WIP)

6 files changed, 137 insertions, 50 deletions

diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 51ee74a18..d8925b7d1 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -664,7 +664,7 @@ multi_all_docs_view(Req, Db, OP, Queries) ->
     ArgQueries = lists:map(fun({Query}) ->
         QueryArg1 = couch_mrview_http:parse_params(Query, undefined,
             Args1, [decoded]),
-        QueryArgs2 = couch_mrview_util:validate_and_update_args(QueryArg1),
+        QueryArgs2 = couch_mrview_util:validate_and_update_args(QueryArg1, [all_docs]),
         set_namespace(OP, QueryArgs2)
     end, Queries),
     Options = [{user_ctx, Req#httpd.user_ctx}],
@@ -684,7 +684,7 @@ multi_all_docs_view(Req, Db, OP, Queries) ->
 all_docs_view(Req, Db, Keys, OP) ->
     Args0 = couch_mrview_http:parse_params(Req, Keys),
     Args1 = Args0#mrargs{view_type=map},
-    Args2 = couch_mrview_util:validate_and_update_args(Args1),
+    Args2 = couch_mrview_util:validate_and_update_args(Args1, [all_docs]),
     Args3 = set_namespace(OP, Args2),
     Options = [{user_ctx, Req#httpd.user_ctx}],
     Max = chttpd:chunked_response_buffer_size(),
diff --git a/src/couch_mrview/src/couch_mrview.erl b/src/couch_mrview/src/couch_mrview.erl
index 7219d1f44..e09483afe 100644
--- a/src/couch_mrview/src/couch_mrview.erl
+++ b/src/couch_mrview/src/couch_mrview.erl
@@ -231,7 +231,7 @@ query_all_docs(Db, Args0, Callback, Acc) ->
         couch_index_util:hexsig(crypto:hash(md5, term_to_binary(Info)))
     end),
     Args1 = Args0#mrargs{view_type=map},
-    Args2 = couch_mrview_util:validate_and_update_args(Args1),
+    Args2 = couch_mrview_util:validate_and_update_args(Args1, [all_docs]),
     {ok, Acc1} = case Args2#mrargs.preflight_fun of
         PFFun when is_function(PFFun, 2) -> PFFun(Sig, Acc);
         _ -> {ok, Acc}
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index 3f454cf4f..57168bae8 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -24,7 +24,7 @@
 -export([temp_view_to_ddoc/1]).
 -export([calculate_external_size/1]).
 -export([calculate_active_size/1]).
--export([validate_args/1, validate_and_update_args/1]).
+-export([validate_args/1, validate_and_update_args/1, validate_and_update_args/2]).
 -export([maybe_load_doc/3, maybe_load_doc/4]).
 -export([maybe_update_index_file/1]).
 -export([extract_view/4, extract_view_reduce/1]).
@@ -545,10 +545,9 @@ validate_args(Args) ->
         _ -> mrverror(<<"Invalid value for `inclusive_end`.">>)
     end,
 
-    case {Args#mrargs.partitioned, Args#mrargs.view_type, Args#mrargs.include_docs} of
-        {true, _, true} -> mrverror(<<"`include_docs` is invalid for partitioned views">>);
-        {_, red, true} -> mrverror(<<"`include_docs` is invalid for reduce">>);
-        {_, _, ID} when is_boolean(ID) -> ok;
+    case {Args#mrargs.view_type, Args#mrargs.include_docs} of
+        {red, true} -> mrverror(<<"`include_docs` is invalid for reduce">>);
+        {_, ID} when is_boolean(ID) -> ok;
         _ -> mrverror(<<"Invalid value for `include_docs`">>)
     end,
 
@@ -573,12 +572,12 @@ validate_args(Args) ->
         {false, undefined} ->
             ok;
         {false, _Partition} ->
-            mrverror(<<"`partition` parameter is not supported in this view.">>)
+            ok % mrverror(<<"`partition` parameter is not supported in this view.">>)
     end,
     Args.
 
 
-update_args(#mrargs{} = Args) ->
+update_args(#mrargs{} = Args, Options) ->
     GroupLevel = determine_group_level(Args),
 
     SKDocId = case {Args#mrargs.direction, Args#mrargs.start_key_docid} of
@@ -593,50 +592,20 @@ update_args(#mrargs{} = Args) ->
         {_, EKDocId1} -> EKDocId1
     end,
 
-    LowestKey = null,
-    HighestKey = {[{<<239, 191, 176>>, null}]}, % \ufff0
-
-    {StartKey, EndKey} = case Args of
-        #mrargs{partition=undefined} ->
-            {Args#mrargs.start_key, Args#mrargs.end_key};
-
-        #mrargs{partition=P0} when not is_binary(P0) ->
-            mrverror(<<"`partition` must be a string.">>);
-
-        #mrargs{partition=P0, direction=fwd, start_key=undefined, end_key=undefined} ->
-            {[P0, LowestKey], [P0, HighestKey]};
-
-        #mrargs{partition=P0, direction=rev, start_key=undefined, end_key=undefined} ->
-            {[P0, HighestKey], [P0, LowestKey]};
-
-        #mrargs{partition=P0, direction=fwd, start_key=SK0, end_key=undefined} ->
-            {[P0, SK0], [P0, HighestKey]};
-
-        #mrargs{partition=P0, direction=rev, start_key=SK0, end_key=undefined} ->
-            {[P0, SK0], [P0, LowestKey]};
-
-        #mrargs{partition=P0, direction=fwd, start_key=undefined, end_key=EK0} ->
-            {[P0, LowestKey], [P0, EK0]};
-
-        #mrargs{partition=P0, direction=rev, start_key=undefined, end_key=EK0} ->
-            {[P0, HighestKey], [P0, EK0]};
-
-        #mrargs{partition=P0, start_key=SK0, end_key=EK0} ->
-            {[P0, SK0], [P0, EK0]}
-    end,
-
-    Args#mrargs{
-        start_key=StartKey,
-        end_key=EndKey,
+    Args1 = Args#mrargs{
         start_key_docid=SKDocId,
         end_key_docid=EKDocId,
         group_level=GroupLevel
-    }.
+    },
+    partition_mrargs(Args1, Options).
 
 
 validate_and_update_args(#mrargs{} = Args) ->
+    validate_and_update_args(Args, []).
+
+validate_and_update_args(#mrargs{} = Args, Options) ->
     Args = validate_args(Args),
-    update_args(Args).
+    update_args(Args, Options).
 
 
 determine_group_level(#mrargs{group=undefined, group_level=undefined}) ->
@@ -1274,3 +1243,108 @@ set_view_options(#mrargs{} = Args, partitioned, true) ->
     Args#mrargs{partitioned=true};
 set_view_options(#mrargs{} = Args, partitioned, false) ->
     Args#mrargs{partitioned=false}.
+
+
+partition_mrargs(#mrargs{} = Args, Options) ->
+couch_log:notice("partition ~p ~p", [Args, Options]),
+    case {Args, lists:member(all_docs, Options)} of
+        {#mrargs{partition=undefined}, _} ->
+            Args;
+
+        {#mrargs{partition=P0}, _} when not is_binary(P0) ->
+            mrverror(<<"`partition` must be a string.">>);
+
+        {#mrargs{}, false} ->
+            partition_as_lists(Args);
+
+        {#mrargs{}, true} ->
+            partition_as_strings(Args)
+    end.
+
+
+partition_as_strings(Args) ->
+    case Args of
+        #mrargs{partition=P0, direction=fwd, start_key=undefined, end_key=undefined} ->
+            Args#mrargs{start_key = <<P0/binary, $:>>, end_key = <<P0/binary, $;>>};
+
+        #mrargs{partition=P0, direction=rev, start_key=undefined, end_key=undefined} ->
+            Args#mrargs{start_key = <<P0/binary, $;>>, end_key = <<P0/binary, $:>>};
+
+        #mrargs{partition=P0, direction=fwd, start_key=SK0, end_key=undefined} ->
+            case prefix(SK0, <<P0/binary, $:>>) of
+                true ->
+                    Args#mrargs{start_key = SK0, end_key = <<P0/binary, $;>>};
+                false ->
+                    mrverror(<<"`start_key` must be prefixed with selected partition.">>)
+            end;
+
+        #mrargs{partition=P0, direction=rev, start_key=SK0, end_key=undefined} ->
+            case prefix(SK0, <<P0/binary, $:>>) of
+                true ->
+                    Args#mrargs{start_key = SK0, end_key = <<P0/binary, $:>>};
+                false ->
+                    mrverror(<<"`start_key` must be prefixed with selected partition.">>)
+            end;
+
+        #mrargs{partition=P0, direction=fwd, start_key=undefined, end_key=EK0} ->
+            case prefix(EK0, <<P0/binary, $:>>) of
+                true ->
+                    Args#mrargs{start_key = <<P0/binary, $:>>, end_key = EK0};
+                false ->
+                    mrverror(<<"`end_key` must be prefixed with selected partition.">>)
+            end;
+
+        #mrargs{partition=P0, direction=rev, start_key=undefined, end_key=EK0} ->
+            case prefix(EK0, <<P0/binary, $:>>) of
+                true ->
+                    Args#mrargs{start_key = EK0, end_key = <<P0/binary, $:>>};
+                false ->
+                    mrverror(<<"`end_key` must be prefixed with selected partition.">>)
+            end;
+
+        #mrargs{partition=P0, start_key=SK0, end_key=EK0} ->
+            case {prefix(SK0, <<P0/binary, $:>>), prefix(EK0, <<P0/binary, $:>>)} of
+                {false, false} ->
+                    mrverror(<<"`start_key` and `end_key` must be prefixed with selected partition.">>);
+                {false, true} ->
+                    mrverror(<<"`start_key` must be prefixed with selected partition.">>);
+                {true, false} ->
+                    mrverror(<<"`end_key` must be prefixed with selected partition.">>);
+                {true, true} ->
+                    Args
+            end
+    end.
+
+
+partition_as_lists(Args) ->
+    LowestKey = null,
+    HighestKey = {[{<<239, 191, 176>>, null}]}, % \ufff0
+
+    case Args of
+        #mrargs{partition=P0, direction=fwd, start_key=undefined, end_key=undefined} ->
+            Args#mrargs{start_key=[P0, LowestKey], end_key=[P0, HighestKey]};
+
+        #mrargs{partition=P0, direction=rev, start_key=undefined, end_key=undefined} ->
+            Args#mrargs{start_key=[P0, HighestKey], end_key=[P0, LowestKey]};
+
+        #mrargs{partition=P0, direction=fwd, start_key=SK0, end_key=undefined} ->
+            Args#mrargs{start_key=[P0, SK0], end_key=[P0, HighestKey]};
+
+        #mrargs{partition=P0, direction=rev, start_key=SK0, end_key=undefined} ->
+            Args#mrargs{start_key=[P0, SK0], end_key=[P0, LowestKey]};
+
+        #mrargs{partition=P0, direction=fwd, start_key=undefined, end_key=EK0} ->
+            Args#mrargs{start_key=[P0, LowestKey], end_key=[P0, EK0]};
+
+        #mrargs{partition=P0, direction=rev, start_key=undefined, end_key=EK0} ->
+            Args#mrargs{start_key=[P0, HighestKey], end_key=[P0, EK0]};
+
+        #mrargs{partition=P0, start_key=SK0, end_key=EK0} ->
+            Args#mrargs{start_key=[P0, SK0], end_key=[P0, EK0]}
+    end.
+
+prefix(Subject, Prefix) ->
+    case binary:match(Subject, Prefix) of
+        {0, _} -> true;
+        _      -> false
+    end.
diff --git a/src/fabric/src/fabric_view_all_docs.erl b/src/fabric/src/fabric_view_all_docs.erl
index ac16dac52..eba3b694b 100644
--- a/src/fabric/src/fabric_view_all_docs.erl
+++ b/src/fabric/src/fabric_view_all_docs.erl
@@ -20,8 +20,14 @@
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("couch_mrview/include/couch_mrview.hrl").
 
-go(DbName, Options, #mrargs{keys=undefined} = QueryArgs, Callback, Acc) ->
-    Shards = mem3:shards(DbName),
+go(DbName, Options, #mrargs{keys=undefined} = QueryArgs0, Callback, Acc) ->
+    case QueryArgs0#mrargs.partitioned of
+        true ->
+            Shards = mem3:shards(DbName, QueryArgs0#mrargs.partition);
+        _ ->
+            Shards = mem3:shards(DbName)
+    end,
+    QueryArgs = QueryArgs0#mrargs{partitioned = true},
     Workers0 = fabric_util:submit_jobs(
             Shards, fabric_rpc, all_docs, [Options, QueryArgs]),
     RexiMon = fabric_util:create_monitors(Workers0),
diff --git a/src/mango/src/mango_cursor_view.erl b/src/mango/src/mango_cursor_view.erl
index 1e2108b7d..0b99265a6 100644
--- a/src/mango/src/mango_cursor_view.erl
+++ b/src/mango/src/mango_cursor_view.erl
@@ -92,10 +92,11 @@ maybe_replace_max_json([H | T] = EndKey) when is_list(EndKey) ->
 maybe_replace_max_json(EndKey) ->
     EndKey.
 
-base_args(#cursor{index = Idx} = Cursor) ->
+base_args(#cursor{index = Idx, opts = Opts} = Cursor) ->
     #mrargs{
         view_type = map,
         reduce = false,
+        partition = couch_util:get_value(partition, Opts),
         start_key = mango_idx:start_key(Idx, Cursor#cursor.ranges),
         end_key = mango_idx:end_key(Idx, Cursor#cursor.ranges),
         include_docs = true
diff --git a/src/mango/src/mango_opts.erl b/src/mango/src/mango_opts.erl
index 7bae9c90d..87d876a3c 100644
--- a/src/mango/src/mango_opts.erl
+++ b/src/mango/src/mango_opts.erl
@@ -81,6 +81,12 @@ validate_find({Props}) ->
             {tag, selector},
             {validator, fun validate_selector/1}
         ]},
+        {<<"partition">>, [
+            {tag, partition},
+            {optional, true},
+            {default, <<>>},
+            {validator, fun is_string/1}
+        ]},
         {<<"use_index">>, [
             {tag, use_index},
             {optional, true},