WIP shard-local queryarchive/shard-local-queryshard-local-query

5 files changed, 19 insertions, 5 deletions

diff --git a/src/couch_mrview/include/couch_mrview.hrl b/src/couch_mrview/include/couch_mrview.hrl
index a341e30db..a369c8e79 100644
--- a/src/couch_mrview/include/couch_mrview.hrl
+++ b/src/couch_mrview/include/couch_mrview.hrl
@@ -87,6 +87,7 @@
     conflicts,
     callback,
     sorted = true,
+    shard_key,
     extra = []
 }).
 
diff --git a/src/couch_mrview/src/couch_mrview_http.erl b/src/couch_mrview/src/couch_mrview_http.erl
index 004caef09..96f5a8d9d 100644
--- a/src/couch_mrview/src/couch_mrview_http.erl
+++ b/src/couch_mrview/src/couch_mrview_http.erl
@@ -543,6 +543,10 @@ parse_param(Key, Val, Args, IsDecoded) ->
             Args#mrargs{update=lazy};
         "update" ->
             throw({query_parse_error, <<"Invalid value for `update`.">>});
+        "shardkey" ->
+            Args#mrargs{shard_key=couch_util:to_binary(Val)};
+        "shard_key" ->
+            Args#mrargs{shard_key=couch_util:to_binary(Val)};
         "descending" ->
             case parse_boolean(Val) of
                 true -> Args#mrargs{direction=rev};
diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index dd4b80da6..4d2f2c700 100644
--- a/src/fabric/src/fabric_util.erl
+++ b/src/fabric/src/fabric_util.erl
@@ -64,7 +64,6 @@ stream_start(Workers0, Keypos, StartFun, Replacements) ->
     Timeout = request_timeout(),
     case rexi_utils:recv(Workers0, Keypos, Fun, Acc, Timeout, infinity) of
         {ok, #stream_acc{workers=Workers}} ->
-            true = fabric_view:is_progress_possible(Workers),
             AckedWorkers = fabric_dict:fold(fun(Worker, From, WorkerAcc) ->
                 rexi:stream_start(From),
                 [Worker | WorkerAcc]
diff --git a/src/fabric/src/fabric_view.erl b/src/fabric/src/fabric_view.erl
index dd0fcfd8b..6aec0ad15 100644
--- a/src/fabric/src/fabric_view.erl
+++ b/src/fabric/src/fabric_view.erl
@@ -308,10 +308,14 @@ index_of(X, [X|_Rest], I) ->
 index_of(X, [_|Rest], I) ->
     index_of(X, Rest, I+1).
 
-get_shards(DbName, #mrargs{stable=true}) ->
+get_shards(DbName, #mrargs{stable=true, shard_key=undefined}) ->
     mem3:ushards(DbName);
-get_shards(DbName, #mrargs{stable=false}) ->
-    mem3:shards(DbName).
+get_shards(DbName, #mrargs{stable=true, shard_key=ShardKey}) ->
+    mem3:ushards(DbName, ShardKey);
+get_shards(DbName, #mrargs{stable=false, shard_key=undefined}) ->
+    mem3:shards(DbName);
+get_shards(DbName, #mrargs{stable=false, shard_key=ShardKey}) ->
+    mem3:shards(DbName, ShardKey).
 
 maybe_update_others(DbName, DDoc, ShardsInvolved, ViewName,
     #mrargs{update=lazy} = Args) ->
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index 0e5eabfe3..e8f192b56 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -13,7 +13,7 @@
 -module(mem3).
 
 -export([start/0, stop/0, restart/0, nodes/0, node_info/2, shards/1, shards/2,
-    choose_shards/2, n/1, n/2, dbname/1, ushards/1]).
+    choose_shards/2, n/1, n/2, dbname/1, ushards/1, ushards/2]).
 -export([get_shard/3, local_shards/1, shard_suffix/1, fold_shards/2]).
 -export([sync_security/0, sync_security/1]).
 -export([compare_nodelists/0, compare_shards/1]).
@@ -133,6 +133,12 @@ ushards(DbName) ->
     Shards = ushards(DbName, live_shards(DbName, Nodes, [ordered]), ZoneMap),
     mem3_util:downcast(Shards).
 
+-spec ushards(DbName::iodata(), DocId::binary()) -> [#shard{}].
+ushards(DbName, DocId) ->
+    Shards = shards_int(DbName, DocId, [ordered]),
+    Shard = hd(Shards),
+    mem3_util:downcast([Shard]).
+
 ushards(DbName, Shards0, ZoneMap) ->
     {L,S,D} = group_by_proximity(Shards0, ZoneMap),
     % Prefer shards in the local zone over shards in a different zone,