WIP plumbing

6 files changed, 53 insertions, 24 deletions

diff --git a/src/couch_mrview/include/couch_mrview.hrl b/src/couch_mrview/include/couch_mrview.hrl
index 67b3cd987..09faf51de 100644
--- a/src/couch_mrview/include/couch_mrview.hrl
+++ b/src/couch_mrview/include/couch_mrview.hrl
@@ -32,7 +32,7 @@
     doc_queue,
     write_queue,
     qserver=nil,
-    partitioned=false
+    partitioned
 }).
 
 
@@ -88,6 +88,7 @@
     conflicts,
     callback,
     sorted = true,
+    partitioned,
     partition_key,
     extra = []
 }).
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index 35554c4c8..aa67c8805 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_and_update_args/1, validate_and_update_args/2]).
+-export([validate_args/1, validate_and_update_args/1]).
 -export([maybe_load_doc/3, maybe_load_doc/4]).
 -export([maybe_update_index_file/1]).
 -export([extract_view/4, extract_view_reduce/1]).
@@ -33,6 +33,7 @@
 -export([changes_key_opts/2]).
 -export([fold_changes/4]).
 -export([to_key_seq/1]).
+-export([set_view_options/3]).
 -export([partition_key/2, unpartition_key/1]).
 
 -define(MOD, couch_mrview_index).
@@ -60,7 +61,7 @@ get_view(Db, DDoc, ViewName, Args0) ->
 get_view_index_pid(Db, DDoc, ViewName, Args0) ->
     ArgCheck = fun(InitState) ->
         Args1 = set_view_type(Args0, ViewName, InitState#mrst.views),
-        {ok, validate_and_update_args(Args1)}
+        {ok, validate_args(Args1)}
     end,
     couch_index_server:get_index(?MOD, Db, DDoc, ArgCheck).
 
@@ -449,7 +450,7 @@ fold_reduce({NthRed, Lang, View}, Fun,  Acc, Options) ->
     couch_btree:fold_reduce(Bt, WrapperFun, Acc, Options).
 
 
-validate_args(Args, Options) ->
+validate_args(Args) ->
     GroupLevel = determine_group_level(Args),
     Reduce = Args#mrargs.reduce,
     case Reduce == undefined orelse is_boolean(Reduce) of
@@ -559,21 +560,20 @@ validate_args(Args, Options) ->
         _ -> mrverror(<<"Invalid value for `sorted`.">>)
     end,
 
-    case {lists:member(partitioned, Options), Args#mrargs.partition_key} of
+    case {Args#mrargs.partitioned, Args#mrargs.partition_key} of
         {true, undefined} ->
             mrverror(<<"`partition_key` parameter is mandatory for queries to this database.">>);
         {true, _PartitionKey} ->
             ok;
-        {false, undefined} ->
+        {undefined, undefined} ->
             ok;
-        {false, _PartitionKey} ->
+        {undefined, _PartitionKey} ->
             mrverror(<<"`partition_key` parameter is not supported in this database.">>)
     end,
+    Args.
 
-    true.
 
-
-update_args(#mrargs{} = Args, _Options) ->
+update_args(#mrargs{} = Args) ->
     GroupLevel = determine_group_level(Args),
 
     SKDocId = case {Args#mrargs.direction, Args#mrargs.start_key_docid} of
@@ -621,11 +621,8 @@ update_args(#mrargs{} = Args, _Options) ->
 
 
 validate_and_update_args(#mrargs{} = Args) ->
-    validate_and_update_args(Args, []).
-
-validate_and_update_args(#mrargs{} = Args, Options) ->
-    true = validate_args(Args, Options),
-    update_args(Args, Options).
+    Args = validate_args(Args),
+    update_args(Args).
 
 
 determine_group_level(#mrargs{group=undefined, group_level=undefined}) ->
@@ -1259,6 +1256,10 @@ kv_external_size(KVList, Reduction) ->
         ?term_size(Key) + ?term_size(Value) + Acc
     end, ?term_size(Reduction), KVList).
 
+set_view_options(#mrargs{} = Args, partitioned, true) ->
+    Args#mrargs{partitioned=true};
+set_view_options(#mrargs{} = Args, partitioned, false) ->
+    Args#mrargs{partitioned=false}.
 
 partition_key(Key, DocId) ->
     [hd(binary:split(DocId, <<":">>)), Key].
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 1b761abf3..59fbb2bb7 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -354,16 +354,20 @@ query_view(DbName, Options, DDoc, ViewName, Callback, Acc0, QueryArgs0) ->
     end,
     {ok, #mrst{views=Views, language=Lang}} =
         couch_mrview_util:ddoc_to_mrst(Db, DDoc),
+
+    Partitioned = mem3:is_partitioned(hd(mem3:shards(Db))), % hideous
+
     QueryArgs1 = couch_mrview_util:set_view_type(QueryArgs0, View, Views),
-    QueryArgs2 = couch_mrview_util:validate_and_update_args(QueryArgs1),
-    VInfo = couch_mrview_util:extract_view(Lang, QueryArgs2, View, Views),
-    case is_reduce_view(QueryArgs2) of
+    QueryArgs2 = couch_mrview_util:set_view_options(QueryArgs1, partitioned, Partitioned),
+    QueryArgs3 = couch_mrview_util:validate_and_update_args(QueryArgs2),
+    VInfo = couch_mrview_util:extract_view(Lang, QueryArgs3, View, Views),
+    case is_reduce_view(QueryArgs3) of
         true ->
             fabric_view_reduce:go(
                 Db,
                 DDoc,
                 View,
-                QueryArgs2,
+                QueryArgs3,
                 Callback,
                 Acc0,
                 VInfo
@@ -374,7 +378,7 @@ query_view(DbName, Options, DDoc, ViewName, Callback, Acc0, QueryArgs0) ->
                 Options,
                 DDoc,
                 View,
-                QueryArgs2,
+                QueryArgs3,
                 Callback,
                 Acc0,
                 VInfo
diff --git a/src/fabric/src/fabric_db_create.erl b/src/fabric/src/fabric_db_create.erl
index 94ffd5643..35e38cb43 100644
--- a/src/fabric/src/fabric_db_create.erl
+++ b/src/fabric/src/fabric_db_create.erl
@@ -168,6 +168,10 @@ make_document([#shard{dbname=DbName}|_] = Shards, Suffix, Options) ->
         E when is_binary(E) -> [{<<"engine">>, E}];
         _ -> []
     end,
+    PartitionedProp = case lists:member(partitioned, Options) of
+        true  -> [{<<"partitioned">>, true}];
+        false -> []
+    end,
     #doc{
         id = DbName,
         body = {[
@@ -175,7 +179,7 @@ make_document([#shard{dbname=DbName}|_] = Shards, Suffix, Options) ->
             {<<"changelog">>, lists:sort(RawOut)},
             {<<"by_node">>, {[{K,lists:sort(V)} || {K,V} <- ByNodeOut]}},
             {<<"by_range">>, {[{K,lists:sort(V)} || {K,V} <- ByRangeOut]}}
-        ] ++ EngineProp}
+        ] ++ EngineProp ++ PartitionedProp}
     }.
 
 db_exists(DbName) -> is_list(catch mem3:shards(DbName)).
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index ca083c4d4..f4bed76f2 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -23,7 +23,7 @@
 -export([get_placement/1]).
 
 %% For mem3 use only.
--export([name/1, node/1, range/1, engine/1]).
+-export([name/1, node/1, range/1, engine/1, is_partitioned/1]).
 
 -include_lib("mem3/include/mem3.hrl").
 -include_lib("couch/include/couch_db.hrl").
@@ -331,6 +331,14 @@ engine(Opts) when is_list(Opts) ->
             []
     end.
 
+is_partitioned(#shard{opts=Opts}) ->
+    is_partitioned(Opts);
+is_partitioned(#ordered_shard{opts=Opts}) ->
+    is_partitioned(Opts);
+is_partitioned(Opts) when is_list(Opts) ->
+    lists:member(partitioned, Opts).
+
+
 -ifdef(TEST).
 
 -include_lib("eunit/include/eunit.hrl").
diff --git a/src/mem3/src/mem3_util.erl b/src/mem3/src/mem3_util.erl
index fc6123d39..5af8c9584 100644
--- a/src/mem3/src/mem3_util.erl
+++ b/src/mem3/src/mem3_util.erl
@@ -177,7 +177,7 @@ build_shards_by_node(DbName, DocProps) ->
                 dbname = DbName,
                 node = to_atom(Node),
                 range = [Beg, End],
-                opts = get_engine_opt(DocProps)
+                opts = get_opts(DocProps)
             }, Suffix)
         end, Ranges)
     end, ByNode).
@@ -195,7 +195,7 @@ build_shards_by_range(DbName, DocProps) ->
                 node = to_atom(Node),
                 range = [Beg, End],
                 order = Order,
-                opts = get_engine_opt(DocProps)
+                opts = get_opts(DocProps)
             }, Suffix)
         end, lists:zip(Nodes, lists:seq(1, length(Nodes))))
     end, ByRange).
@@ -212,6 +212,9 @@ to_integer(N) when is_binary(N) ->
 to_integer(N) when is_list(N) ->
     list_to_integer(N).
 
+get_opts(DocProps) ->
+    get_engine_opt(DocProps) ++ get_partitioned_opt(DocProps).
+
 get_engine_opt(DocProps) ->
     case couch_util:get_value(<<"engine">>, DocProps) of
         Engine when is_binary(Engine) ->
@@ -220,6 +223,14 @@ get_engine_opt(DocProps) ->
             []
     end.
 
+get_partitioned_opt(DocProps) ->
+    case couch_util:get_value(<<"partitioned">>, DocProps) of
+        true ->
+            [partitioned];
+        _ ->
+            []
+    end.
+
 n_val(undefined, NodeCount) ->
     n_val(config:get("cluster", "n", "3"), NodeCount);
 n_val(N, NodeCount) when is_list(N) ->