You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2018/07/10 15:06:31 UTC
[couchdb] branch user-partitioned-dbs-wip updated (03f2cc3 ->
e89c39a)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
discard 03f2cc3 unpartition key in view responses
discard 03956d7 ?partition=foo
discard 28e0dfa WIP plumbing
discard 785213d WIP support user-partitioned views
discard 67dc469 add ?partitioned=true parameter when creating a database
discard 6e27629 fixup! bccd6accfabc4fa02c75183d652f4a3790f729b8
discard 7452c38 Extract side-effect free part of validate_args
discard bccd6ac introduce mem3_util:docid_hash/1 and docid_hash/2
new 2b63513 introduce mem3_util:docid_hash/1 and docid_hash/2
new 38245dd Extract side-effect free part of validate_args
new 28bb00b add ?partitioned=true parameter when creating a database
new 9c8e4af WIP support user-partitioned views
new 67de500 WIP plumbing
new 002e04c ?partition=foo
new 388df94 unpartition key in view responses
new 0ab80ec cleanup
new c36edd5 Only consult the shards holding the partition in question
new e89c39a map documents to shards by their partition
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (03f2cc3)
\
N -- N -- N refs/heads/user-partitioned-dbs-wip (e89c39a)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/couch_mrview/src/couch_mrview_util.erl | 7 -------
src/fabric/src/fabric_util.erl | 1 -
src/fabric/src/fabric_view.erl | 11 ++++++++---
src/mem3/src/mem3.erl | 8 +++++++-
src/mem3/src/mem3_shards.erl | 23 ++++++++++++++---------
5 files changed, 29 insertions(+), 21 deletions(-)
[couchdb] 05/10: WIP plumbing
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 67de50068795da8d44855e12e5af03392e292132
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Jul 9 22:39:16 2018 +0100
WIP plumbing
---
src/couch_mrview/include/couch_mrview.hrl | 3 ++-
src/couch_mrview/src/couch_mrview_util.erl | 29 +++++++++++++++--------------
src/fabric/src/fabric.erl | 14 +++++++++-----
src/fabric/src/fabric_db_create.erl | 6 +++++-
src/mem3/src/mem3.erl | 10 +++++++++-
src/mem3/src/mem3_util.erl | 15 +++++++++++++--
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 67b3cd9..09faf51 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 35554c4..aa67c88 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 1b761ab..59fbb2b 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 94ffd56..35e38cb 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 ca083c4..f4bed76 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 fc6123d..5af8c95 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) ->
[couchdb] 07/10: unpartition key in view responses
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 388df9406bc9374663c10026b68375da820012a8
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Jul 10 11:50:37 2018 +0100
unpartition key in view responses
---
src/fabric/include/fabric.hrl | 3 ++-
src/fabric/src/fabric_view.erl | 12 ++++++++++--
src/fabric/src/fabric_view_map.erl | 3 ++-
src/fabric/src/fabric_view_reduce.erl | 3 ++-
4 files changed, 16 insertions(+), 5 deletions(-)
diff --git a/src/fabric/include/fabric.hrl b/src/fabric/include/fabric.hrl
index be1d639..8ceab22 100644
--- a/src/fabric/include/fabric.hrl
+++ b/src/fabric/include/fabric.hrl
@@ -31,7 +31,8 @@
lang,
sorted,
user_acc,
- update_seq
+ update_seq,
+ partitioned=false
}).
-record(stream_acc, {
diff --git a/src/fabric/src/fabric_view.erl b/src/fabric/src/fabric_view.erl
index 69f4290..bf4d64e 100644
--- a/src/fabric/src/fabric_view.erl
+++ b/src/fabric/src/fabric_view.erl
@@ -128,8 +128,11 @@ maybe_send_row(State) ->
try get_next_row(State) of
{_, NewState} when Skip > 0 ->
maybe_send_row(NewState#collector{skip=Skip-1});
- {Row, NewState} ->
- case Callback(transform_row(possibly_embed_doc(NewState,Row)), AccIn) of
+ {Row0, NewState} ->
+ Row1 = possibly_embed_doc(NewState, Row0),
+ Row2 = unpartition_row(NewState, Row1),
+ Row3 = transform_row(Row2),
+ case Callback(Row3, AccIn) of
{stop, Acc} ->
{stop, NewState#collector{user_acc=Acc, limit=Limit-1}};
{ok, Acc} ->
@@ -272,6 +275,11 @@ transform_row(#view_row{key=Key, id=_Id, value=_Value, doc={error,Reason}}) ->
transform_row(#view_row{key=Key, id=Id, value=Value, doc=Doc}) ->
{row, [{id,Id}, {key,Key}, {value,Value}, {doc,Doc}]}.
+unpartition_row(#collector{partitioned=true}, #view_row{key=[_Partition, Key]} = Row) ->
+ Row#view_row{key = Key};
+unpartition_row(#collector{partitioned=false}, Row) ->
+ Row.
+
compare(_, _, A, A) -> true;
compare(fwd, <<"raw">>, A, B) -> A < B;
compare(rev, <<"raw">>, A, B) -> B < A;
diff --git a/src/fabric/src/fabric_view_map.erl b/src/fabric/src/fabric_view_map.erl
index b6a3d6f..75b71f1 100644
--- a/src/fabric/src/fabric_view_map.erl
+++ b/src/fabric/src/fabric_view_map.erl
@@ -76,7 +76,8 @@ go(DbName, Workers, {map, View, _}, Args, Callback, Acc0) ->
sorted = Args#mrargs.sorted,
collation = Collation,
user_acc = Acc0,
- update_seq = case UpdateSeq of true -> []; false -> nil end
+ update_seq = case UpdateSeq of true -> []; false -> nil end,
+ partitioned = Args#mrargs.partitioned
},
case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
State, infinity, 1000 * 60 * 60) of
diff --git a/src/fabric/src/fabric_view_reduce.erl b/src/fabric/src/fabric_view_reduce.erl
index a74be10..c58d3aa 100644
--- a/src/fabric/src/fabric_view_reduce.erl
+++ b/src/fabric/src/fabric_view_reduce.erl
@@ -82,7 +82,8 @@ go2(DbName, Workers, {red, {_, Lang, View}, _}=VInfo, Args, Callback, Acc0) ->
collation = couch_util:get_value(<<"collation">>, View#mrview.options),
rows = dict:new(),
user_acc = Acc0,
- update_seq = case UpdateSeq of true -> []; false -> nil end
+ update_seq = case UpdateSeq of true -> []; false -> nil end,
+ partitioned = Args#mrargs.partitioned
},
try rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
State, infinity, 1000 * 60 * 60) of
[couchdb] 01/10: introduce mem3_util:docid_hash/1 and docid_hash/2
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 2b63513f50a364a196e3e5ba9ceb8b325cfd4517
Author: Robert Newson <rn...@apache.org>
AuthorDate: Wed Jul 4 17:15:16 2018 +0100
introduce mem3_util:docid_hash/1 and docid_hash/2
mem3_util:docid_hash/1 is identical to mem3_util:hash/1
mem3_util:docid_hash/2 allows the user to control the result with
specially-formatted doc ids.
---
src/mem3/src/mem3.erl | 2 +-
src/mem3/src/mem3_shards.erl | 4 ++--
src/mem3/src/mem3_util.erl | 15 +++++++++++++++
3 files changed, 18 insertions(+), 3 deletions(-)
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index 0e5eabf..ca083c4 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -239,7 +239,7 @@ belongs(DbName, DocId) when is_binary(DbName), is_binary(DocId) ->
true.
belongs(Begin, End, DocId) ->
- HashKey = mem3_util:hash(DocId),
+ HashKey = mem3_util:docid_hash(DocId),
Begin =< HashKey andalso HashKey =< End.
range(#shard{range = Range}) ->
diff --git a/src/mem3/src/mem3_shards.erl b/src/mem3/src/mem3_shards.erl
index da3b69a..0559149 100644
--- a/src/mem3/src/mem3_shards.erl
+++ b/src/mem3/src/mem3_shards.erl
@@ -67,7 +67,7 @@ for_docid(DbName, DocId) ->
for_docid(DbName, DocId, []).
for_docid(DbName, DocId, Options) ->
- HashKey = mem3_util:hash(DocId),
+ HashKey = mem3_util:docid_hash(DocId),
ShardHead = #shard{
dbname = DbName,
range = ['$1', '$2'],
@@ -397,7 +397,7 @@ load_shards_from_db(ShardDb, DbName) ->
load_shards_from_disk(DbName, DocId)->
Shards = load_shards_from_disk(DbName),
- HashKey = mem3_util:hash(DocId),
+ HashKey = mem3_util:docid_hash(DocId),
[S || S <- Shards, in_range(S, HashKey)].
in_range(Shard, HashKey) ->
diff --git a/src/mem3/src/mem3_util.erl b/src/mem3/src/mem3_util.erl
index 0b69d79..fc6123d 100644
--- a/src/mem3/src/mem3_util.erl
+++ b/src/mem3/src/mem3_util.erl
@@ -16,6 +16,7 @@
n_val/2, to_atom/1, to_integer/1, write_db_doc/1, delete_db_doc/1,
shard_info/1, ensure_exists/1, open_db_doc/1]).
-export([is_deleted/1, rotate_list/2]).
+-export([docid_hash/1, docid_hash/2]).
%% do not use outside mem3.
-export([build_ordered_shards/2, downcast/1]).
@@ -34,6 +35,20 @@ hash(Item) when is_binary(Item) ->
hash(Item) ->
erlang:crc32(term_to_binary(Item)).
+
+docid_hash(DocId) when is_binary(DocId) ->
+ docid_hash(DocId, []).
+
+docid_hash(DocId, Options) when is_binary(DocId), is_list(Options) ->
+ Data = case lists:member(partitioned, Options) of
+ true ->
+ hd(binary:split(DocId, <<":">>));
+ false ->
+ DocId
+ end,
+ erlang:crc32(Data).
+
+
name_shard(Shard) ->
name_shard(Shard, "").
[couchdb] 03/10: add ?partitioned=true parameter when creating a
database
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 28bb00bc2677b7cc63cd947b9114152ba9e2e24a
Author: Robert Newson <rn...@apache.org>
AuthorDate: Thu Jul 5 14:34:29 2018 +0100
add ?partitioned=true parameter when creating a database
---
src/chttpd/src/chttpd_db.erl | 13 ++++++++++++-
src/couch/src/couch_db_updater.erl | 4 +++-
2 files changed, 15 insertions(+), 2 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index bf1fe8c..51ee74a 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -284,12 +284,13 @@ create_db_req(#httpd{}=Req, DbName) ->
N = chttpd:qs_value(Req, "n", config:get("cluster", "n", "3")),
Q = chttpd:qs_value(Req, "q", config:get("cluster", "q", "8")),
P = chttpd:qs_value(Req, "placement", config:get("cluster", "placement")),
+ PartitionedOpt = parse_partitioned_opt(Req),
EngineOpt = parse_engine_opt(Req),
Options = [
{n, N},
{q, Q},
{placement, P}
- ] ++ EngineOpt,
+ ] ++ EngineOpt ++ PartitionedOpt,
DocUrl = absolute_uri(Req, "/" ++ couch_util:url_encode(DbName)),
case fabric:create_db(DbName, Options) of
ok ->
@@ -1413,6 +1414,16 @@ parse_engine_opt(Req) ->
end
end.
+parse_partitioned_opt(Req) ->
+ case chttpd:qs_value(Req, "partitioned") of
+ undefined ->
+ [];
+ "true" ->
+ [partitioned];
+ _ ->
+ throw({bad_request, <<"`partitioned` parameter can only be set to true.">>})
+ end.
+
parse_doc_query({Key, Value}, Args) ->
case {Key, Value} of
{"attachments", "true"} ->
diff --git a/src/couch/src/couch_db_updater.erl b/src/couch/src/couch_db_updater.erl
index fba99a7..0713582 100644
--- a/src/couch/src/couch_db_updater.erl
+++ b/src/couch/src/couch_db_updater.erl
@@ -27,7 +27,9 @@ init({Engine, DbName, FilePath, Options0}) ->
erlang:put(io_priority, {db_update, DbName}),
update_idle_limit_from_config(),
DefaultSecObj = default_security_object(DbName),
- Options = [{default_security_object, DefaultSecObj} | Options0],
+ Partitioned = lists:member(partitioned, Options0),
+ Options1 = [{default_security_object, DefaultSecObj} | Options0],
+ Options = [{default_props, [{partitioned, Partitioned}]} | Options1],
try
{ok, EngineState} = couch_db_engine:init(Engine, FilePath, Options),
Db = init_db(DbName, FilePath, EngineState, Options),
[couchdb] 10/10: map documents to shards by their partition
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit e89c39a89ae345cecbbbcc100e9209b030d3d833
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Jul 10 16:06:20 2018 +0100
map documents to shards by their partition
---
src/mem3/src/mem3_shards.erl | 23 ++++++++++++++---------
1 file changed, 14 insertions(+), 9 deletions(-)
diff --git a/src/mem3/src/mem3_shards.erl b/src/mem3/src/mem3_shards.erl
index 0559149..18ed82e 100644
--- a/src/mem3/src/mem3_shards.erl
+++ b/src/mem3/src/mem3_shards.erl
@@ -67,26 +67,27 @@ for_docid(DbName, DocId) ->
for_docid(DbName, DocId, []).
for_docid(DbName, DocId, Options) ->
- HashKey = mem3_util:docid_hash(DocId),
ShardHead = #shard{
dbname = DbName,
- range = ['$1', '$2'],
_ = '_'
},
OrderedShardHead = #ordered_shard{
dbname = DbName,
- range = ['$1', '$2'],
_ = '_'
},
- Conditions = [{'=<', '$1', HashKey}, {'=<', HashKey, '$2'}],
- ShardSpec = {ShardHead, Conditions, ['$_']},
- OrderedShardSpec = {OrderedShardHead, Conditions, ['$_']},
+ ShardSpec = {ShardHead, [], ['$_']},
+ OrderedShardSpec = {OrderedShardHead, [], ['$_']},
Shards = try ets:select(?SHARDS, [ShardSpec, OrderedShardSpec]) of
[] ->
load_shards_from_disk(DbName, DocId);
- Else ->
+ Shards0 ->
gen_server:cast(?MODULE, {cache_hit, DbName}),
- Else
+ Options1 = case mem3:is_partitioned(hd(Shards0)) of
+ true -> [partitioned];
+ false -> []
+ end,
+ HashKey = mem3_util:docid_hash(DocId, Options1),
+ [S || S <- Shards0, in_range(S, HashKey)]
catch error:badarg ->
load_shards_from_disk(DbName, DocId)
end,
@@ -397,7 +398,11 @@ load_shards_from_db(ShardDb, DbName) ->
load_shards_from_disk(DbName, DocId)->
Shards = load_shards_from_disk(DbName),
- HashKey = mem3_util:docid_hash(DocId),
+ Options = case mem3:is_partitioned(hd(Shards)) of
+ true -> [partitioned];
+ false -> []
+ end,
+ HashKey = mem3_util:docid_hash(DocId, Options),
[S || S <- Shards, in_range(S, HashKey)].
in_range(Shard, HashKey) ->
[couchdb] 08/10: cleanup
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 0ab80ecee516001a6734304a89507ca8bba85ada
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Jul 10 13:10:15 2018 +0100
cleanup
---
src/couch_mrview/src/couch_mrview_util.erl | 7 -------
1 file changed, 7 deletions(-)
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index a5e3840..ce8f4eb 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -34,7 +34,6 @@
-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).
-define(GET_VIEW_RETRY_COUNT, 1).
@@ -1260,9 +1259,3 @@ 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].
-
-unpartition_key([_Partition, Key]) ->
- Key.
[couchdb] 02/10: Extract side-effect free part of validate_args
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 38245dd55c830c7a436a1b6abad4e1d2d27f69d8
Author: Robert Newson <rn...@apache.org>
AuthorDate: Wed Jul 4 17:27:10 2018 +0100
Extract side-effect free part of validate_args
validate_args/1 also modified the arguments. By separating the
validation from the update we are able to add updates that are not
idempotent.
---
src/chttpd/src/chttpd_db.erl | 4 ++--
src/chttpd/src/chttpd_view.erl | 2 +-
src/couch_mrview/src/couch_mrview.erl | 2 +-
src/couch_mrview/src/couch_mrview_http.erl | 2 +-
src/couch_mrview/src/couch_mrview_util.erl | 25 ++++++++++++++++------
src/couch_mrview/test/couch_mrview_util_tests.erl | 2 +-
.../src/couch_replicator_httpd.erl | 2 +-
src/fabric/src/fabric.erl | 2 +-
8 files changed, 26 insertions(+), 15 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index ed0adea..bf1fe8c 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -663,7 +663,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_args(QueryArg1),
+ QueryArgs2 = couch_mrview_util:validate_and_update_args(QueryArg1),
set_namespace(OP, QueryArgs2)
end, Queries),
Options = [{user_ctx, Req#httpd.user_ctx}],
@@ -683,7 +683,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_args(Args1),
+ Args2 = couch_mrview_util:validate_and_update_args(Args1),
Args3 = set_namespace(OP, Args2),
Options = [{user_ctx, Req#httpd.user_ctx}],
Max = chttpd:chunked_response_buffer_size(),
diff --git a/src/chttpd/src/chttpd_view.erl b/src/chttpd/src/chttpd_view.erl
index 3c05c64..799df61 100644
--- a/src/chttpd/src/chttpd_view.erl
+++ b/src/chttpd/src/chttpd_view.erl
@@ -24,7 +24,7 @@ multi_query_view(Req, Db, DDoc, ViewName, Queries) ->
QueryArg = couch_mrview_http:parse_params(Query, undefined,
Args1, [decoded]),
QueryArg1 = couch_mrview_util:set_view_type(QueryArg, ViewName, Views),
- couch_mrview_util:validate_args(QueryArg1)
+ couch_mrview_util:validate_and_update_args(QueryArg1)
end, Queries),
Options = [{user_ctx, Req#httpd.user_ctx}],
VAcc0 = #vacc{db=Db, req=Req, prepend="\r\n"},
diff --git a/src/couch_mrview/src/couch_mrview.erl b/src/couch_mrview/src/couch_mrview.erl
index 82bbd79..a9d5d0a 100644
--- a/src/couch_mrview/src/couch_mrview.erl
+++ b/src/couch_mrview/src/couch_mrview.erl
@@ -223,7 +223,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_args(Args1),
+ Args2 = couch_mrview_util:validate_and_update_args(Args1),
{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_http.erl b/src/couch_mrview/src/couch_mrview_http.erl
index 004caef..9dae1d8 100644
--- a/src/couch_mrview/src/couch_mrview_http.erl
+++ b/src/couch_mrview/src/couch_mrview_http.erl
@@ -296,7 +296,7 @@ multi_query_view(Req, Db, DDoc, ViewName, Queries) ->
{ok, _, _, Args1} = couch_mrview_util:get_view(Db, DDoc, ViewName, Args0),
ArgQueries = lists:map(fun({Query}) ->
QueryArg = parse_params(Query, undefined, Args1),
- couch_mrview_util:validate_args(QueryArg)
+ couch_mrview_util:validate_and_update_args(QueryArg)
end, Queries),
{ok, Resp2} = couch_httpd:etag_maybe(Req, fun() ->
Max = chttpd:chunked_response_buffer_size(),
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index eb461d0..086bf9b 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]).
+-export([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]).
@@ -59,7 +59,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_args(Args1)}
+ {ok, validate_and_update_args(Args1)}
end,
couch_index_server:get_index(?MOD, Db, DDoc, ArgCheck).
@@ -546,6 +546,17 @@ validate_args(Args) ->
{red, _} -> mrverror(<<"`conflicts` is invalid for reduce views.">>)
end,
+ case is_boolean(Args#mrargs.sorted) of
+ true -> ok;
+ _ -> mrverror(<<"Invalid value for `sorted`.">>)
+ end,
+
+ true.
+
+
+update_args(#mrargs{} = Args) ->
+ GroupLevel = determine_group_level(Args),
+
SKDocId = case {Args#mrargs.direction, Args#mrargs.start_key_docid} of
{fwd, undefined} -> <<>>;
{rev, undefined} -> <<255>>;
@@ -558,11 +569,6 @@ validate_args(Args) ->
{_, EKDocId1} -> EKDocId1
end,
- case is_boolean(Args#mrargs.sorted) of
- true -> ok;
- _ -> mrverror(<<"Invalid value for `sorted`.">>)
- end,
-
Args#mrargs{
start_key_docid=SKDocId,
end_key_docid=EKDocId,
@@ -570,6 +576,11 @@ validate_args(Args) ->
}.
+validate_and_update_args(#mrargs{} = Args) ->
+ true = validate_args(Args),
+ update_args(Args).
+
+
determine_group_level(#mrargs{group=undefined, group_level=undefined}) ->
0;
determine_group_level(#mrargs{group=false, group_level=undefined}) ->
diff --git a/src/couch_mrview/test/couch_mrview_util_tests.erl b/src/couch_mrview/test/couch_mrview_util_tests.erl
index 7046c9b..df50187 100644
--- a/src/couch_mrview/test/couch_mrview_util_tests.erl
+++ b/src/couch_mrview/test/couch_mrview_util_tests.erl
@@ -34,6 +34,6 @@ couch_mrview_util_test_() ->
validate_group_level(Group, GroupLevel) ->
Args0 = #mrargs{group=Group, group_level=GroupLevel, view_type=red},
- Args1 = couch_mrview_util:validate_args(Args0),
+ Args1 = couch_mrview_util:validate_and_update_args(Args0),
Args1#mrargs.group_level.
diff --git a/src/couch_replicator/src/couch_replicator_httpd.erl b/src/couch_replicator/src/couch_replicator_httpd.erl
index abd9f7f..f37d436 100644
--- a/src/couch_replicator/src/couch_replicator_httpd.erl
+++ b/src/couch_replicator/src/couch_replicator_httpd.erl
@@ -124,7 +124,7 @@ handle_scheduler_docs(Db, Req) when is_binary(Db) ->
reduce = false,
extra = [{filter_states, States}]
},
- VArgs2 = couch_mrview_util:validate_args(VArgs1),
+ VArgs2 = couch_mrview_util:validate_and_update_args(VArgs1),
Opts = [{user_ctx, Req#httpd.user_ctx}],
Max = chttpd:chunked_response_buffer_size(),
Acc = couch_replicator_httpd_util:docs_acc_new(Req, Db, Max),
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 4a07271..1b761ab 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -355,7 +355,7 @@ query_view(DbName, Options, DDoc, ViewName, Callback, Acc0, QueryArgs0) ->
{ok, #mrst{views=Views, language=Lang}} =
couch_mrview_util:ddoc_to_mrst(Db, DDoc),
QueryArgs1 = couch_mrview_util:set_view_type(QueryArgs0, View, Views),
- QueryArgs2 = couch_mrview_util:validate_args(QueryArgs1),
+ 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
true ->
[couchdb] 06/10: ?partition=foo
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 002e04c063c1bfe142dae9dcaa33d0adae2d7ae4
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Jul 9 23:16:45 2018 +0100
?partition=foo
---
src/couch_mrview/include/couch_mrview.hrl | 2 +-
src/couch_mrview/src/couch_mrview_http.erl | 4 ++--
src/couch_mrview/src/couch_mrview_util.erl | 32 +++++++++++++++---------------
3 files changed, 19 insertions(+), 19 deletions(-)
diff --git a/src/couch_mrview/include/couch_mrview.hrl b/src/couch_mrview/include/couch_mrview.hrl
index 09faf51..094b1d5 100644
--- a/src/couch_mrview/include/couch_mrview.hrl
+++ b/src/couch_mrview/include/couch_mrview.hrl
@@ -89,7 +89,7 @@
callback,
sorted = true,
partitioned,
- partition_key,
+ partition,
extra = []
}).
diff --git a/src/couch_mrview/src/couch_mrview_http.erl b/src/couch_mrview/src/couch_mrview_http.erl
index 5ff7285..fac1d42 100644
--- a/src/couch_mrview/src/couch_mrview_http.erl
+++ b/src/couch_mrview/src/couch_mrview_http.erl
@@ -582,8 +582,8 @@ parse_param(Key, Val, Args, IsDecoded) ->
Args#mrargs{callback=couch_util:to_binary(Val)};
"sorted" ->
Args#mrargs{sorted=parse_boolean(Val)};
- "partition_key" ->
- Args#mrargs{partition_key=couch_util:to_binary(Val)};
+ "partition" ->
+ Args#mrargs{partition=couch_util:to_binary(Val)};
_ ->
BKey = couch_util:to_binary(Key),
BVal = couch_util:to_binary(Val),
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index aa67c88..a5e3840 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -560,15 +560,15 @@ validate_args(Args) ->
_ -> mrverror(<<"Invalid value for `sorted`.">>)
end,
- case {Args#mrargs.partitioned, Args#mrargs.partition_key} of
+ case {Args#mrargs.partitioned, Args#mrargs.partition} of
{true, undefined} ->
- mrverror(<<"`partition_key` parameter is mandatory for queries to this database.">>);
- {true, _PartitionKey} ->
+ mrverror(<<"`partition` parameter is mandatory for queries to this database.">>);
+ {true, _Partition} ->
ok;
{undefined, undefined} ->
ok;
- {undefined, _PartitionKey} ->
- mrverror(<<"`partition_key` parameter is not supported in this database.">>)
+ {undefined, _Partition} ->
+ mrverror(<<"`partition` parameter is not supported in this database.">>)
end,
Args.
@@ -592,23 +592,23 @@ update_args(#mrargs{} = Args) ->
HighestKey = {[{<<239, 191, 176>>, null}]}, % \ufff0
{StartKey, EndKey} = case Args of
- #mrargs{partition_key=undefined} ->
+ #mrargs{partition=undefined} ->
{Args#mrargs.start_key, Args#mrargs.end_key};
- #mrargs{partition_key=PKey0} when not is_binary(PKey0) ->
- mrverror(<<"`partition_key` must be a string.">>);
+ #mrargs{partition=P0} when not is_binary(P0) ->
+ mrverror(<<"`partition` must be a string.">>);
- #mrargs{partition_key=PKey0, start_key=undefined, end_key=undefined} ->
- {[PKey0, LowestKey], [PKey0, HighestKey]};
+ #mrargs{partition=P0, start_key=undefined, end_key=undefined} ->
+ {[P0, LowestKey], [P0, HighestKey]};
- #mrargs{partition_key=PKey0, start_key=SK0, end_key=undefined} ->
- {[PKey0, SK0], [PKey0, HighestKey]};
+ #mrargs{partition=P0, start_key=SK0, end_key=undefined} ->
+ {[P0, SK0], [P0, HighestKey]};
- #mrargs{partition_key=PKey0, start_key=undefined, end_key=EK0} ->
- {[PKey0, LowestKey], [PKey0, EK0]};
+ #mrargs{partition=P0, start_key=undefined, end_key=EK0} ->
+ {[P0, LowestKey], [P0, EK0]};
- #mrargs{partition_key=PKey0, start_key=SK0, end_key=EK0} ->
- {[PKey0, SK0], [PKey0, EK0]}
+ #mrargs{partition=P0, start_key=SK0, end_key=EK0} ->
+ {[P0, SK0], [P0, EK0]}
end,
Args#mrargs{
[couchdb] 09/10: Only consult the shards holding the partition in
question
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit c36edd52f509dcccd9482e7c3e093ee00ba1bbce
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Jul 10 15:40:19 2018 +0100
Only consult the shards holding the partition in question
---
src/fabric/src/fabric_util.erl | 1 -
src/fabric/src/fabric_view.erl | 11 ++++++++---
src/mem3/src/mem3.erl | 8 +++++++-
3 files changed, 15 insertions(+), 5 deletions(-)
diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index dd4b80d..4d2f2c7 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 bf4d64e..32ad422 100644
--- a/src/fabric/src/fabric_view.erl
+++ b/src/fabric/src/fabric_view.erl
@@ -317,10 +317,15 @@ 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{partitioned=false,stable=true}) ->
mem3:ushards(DbName);
-get_shards(DbName, #mrargs{stable=false}) ->
- mem3:shards(DbName).
+get_shards(DbName, #mrargs{partitioned=false,stable=false}) ->
+ mem3:shards(DbName);
+get_shards(DbName, #mrargs{partitioned=true,stable=true}=Args) ->
+ mem3:ushards(DbName, Args#mrargs.partition);
+get_shards(DbName, #mrargs{partitioned=true,stable=false}=Args) ->
+ mem3:shards(DbName, Args#mrargs.partition).
+
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 f4bed76..7cdadba 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,
[couchdb] 04/10: WIP support user-partitioned views
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 9c8e4afe7fff33d5a0ace8568fa330961e75b67f
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Jul 9 18:08:57 2018 +0100
WIP support user-partitioned views
---
src/couch_mrview/include/couch_mrview.hrl | 4 +-
src/couch_mrview/src/couch_mrview_http.erl | 2 +
src/couch_mrview/src/couch_mrview_updater.erl | 15 +++++-
src/couch_mrview/src/couch_mrview_util.erl | 66 ++++++++++++++++++++++++---
4 files changed, 78 insertions(+), 9 deletions(-)
diff --git a/src/couch_mrview/include/couch_mrview.hrl b/src/couch_mrview/include/couch_mrview.hrl
index a341e30..67b3cd9 100644
--- a/src/couch_mrview/include/couch_mrview.hrl
+++ b/src/couch_mrview/include/couch_mrview.hrl
@@ -31,7 +31,8 @@
doc_acc,
doc_queue,
write_queue,
- qserver=nil
+ qserver=nil,
+ partitioned=false
}).
@@ -87,6 +88,7 @@
conflicts,
callback,
sorted = true,
+ partition_key,
extra = []
}).
diff --git a/src/couch_mrview/src/couch_mrview_http.erl b/src/couch_mrview/src/couch_mrview_http.erl
index 9dae1d8..5ff7285 100644
--- a/src/couch_mrview/src/couch_mrview_http.erl
+++ b/src/couch_mrview/src/couch_mrview_http.erl
@@ -582,6 +582,8 @@ parse_param(Key, Val, Args, IsDecoded) ->
Args#mrargs{callback=couch_util:to_binary(Val)};
"sorted" ->
Args#mrargs{sorted=parse_boolean(Val)};
+ "partition_key" ->
+ Args#mrargs{partition_key=couch_util:to_binary(Val)};
_ ->
BKey = couch_util:to_binary(Key),
BVal = couch_util:to_binary(Val),
diff --git a/src/couch_mrview/src/couch_mrview_updater.erl b/src/couch_mrview/src/couch_mrview_updater.erl
index 214f487..e1cc280 100644
--- a/src/couch_mrview/src/couch_mrview_updater.erl
+++ b/src/couch_mrview/src/couch_mrview_updater.erl
@@ -311,7 +311,8 @@ write_kvs(State, UpdateSeq, ViewKVs, DocIdKeys, Seqs, Log0) ->
#mrst{
id_btree=IdBtree,
log_btree=LogBtree,
- first_build=FirstBuild
+ first_build=FirstBuild,
+ partitioned=Partitioned
} = State,
Revs = dict:from_list(dict:fetch_keys(Log0)),
@@ -328,8 +329,15 @@ write_kvs(State, UpdateSeq, ViewKVs, DocIdKeys, Seqs, Log0) ->
_ -> update_log(LogBtree, Log, Revs, Seqs, FirstBuild)
end,
- UpdateView = fun(#mrview{id_num=ViewId}=View, {ViewId, {KVs, SKVs}}) ->
+ UpdateView = fun(#mrview{id_num=ViewId}=View, {ViewId, {KVs0, SKVs}}) ->
#mrview{seq_indexed=SIndexed, keyseq_indexed=KSIndexed} = View,
+ KVs = case Partitioned of
+ true ->
+ [{{[partition(D), K], D}, V} || {{K, D}, V} <- KVs0];
+ false ->
+ KVs0
+ end,
+
ToRem = couch_util:dict_find(ViewId, ToRemByView, []),
{ok, VBtree2} = couch_btree:add_remove(View#mrview.btree, KVs, ToRem),
NewUpdateSeq = case VBtree2 =/= View#mrview.btree of
@@ -484,3 +492,6 @@ maybe_notify(State, View, KVs, ToRem) ->
[Key || {Key, _DocId} <- ToRem]
end,
couch_index_plugin:index_update(State, View, Updated, Removed).
+
+partition(DocId) ->
+ hd(binary:split(DocId, <<":">>)).
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index 086bf9b..35554c4 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]).
+-export([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]).
@@ -33,6 +33,7 @@
-export([changes_key_opts/2]).
-export([fold_changes/4]).
-export([to_key_seq/1]).
+-export([partition_key/2, unpartition_key/1]).
-define(MOD, couch_mrview_index).
-define(GET_VIEW_RETRY_COUNT, 1).
@@ -281,6 +282,12 @@ init_state(Db, Fd, State, Header) ->
OpenViewFun = fun(St, View) -> open_view(Db, Fd, Lang, St, View) end,
Views2 = lists:zipwith(OpenViewFun, ViewStates, Views),
+ Partitioned = case couch_db_engine:get_prop(Db, partitioned) of
+ {ok, true} -> true;
+ {ok, false} -> false;
+ {error, no_value} -> false
+ end,
+
State#mrst{
fd=Fd,
fd_monitor=erlang:monitor(process, Fd),
@@ -288,7 +295,8 @@ init_state(Db, Fd, State, Header) ->
purge_seq=PurgeSeq,
id_btree=IdBtree,
log_btree=LogBtree,
- views=Views2
+ views=Views2,
+ partitioned=Partitioned
}.
open_view(_Db, Fd, Lang, ViewState, View) ->
@@ -441,7 +449,7 @@ fold_reduce({NthRed, Lang, View}, Fun, Acc, Options) ->
couch_btree:fold_reduce(Bt, WrapperFun, Acc, Options).
-validate_args(Args) ->
+validate_args(Args, Options) ->
GroupLevel = determine_group_level(Args),
Reduce = Args#mrargs.reduce,
case Reduce == undefined orelse is_boolean(Reduce) of
@@ -551,10 +559,21 @@ validate_args(Args) ->
_ -> mrverror(<<"Invalid value for `sorted`.">>)
end,
+ case {lists:member(partitioned, Options), Args#mrargs.partition_key} of
+ {true, undefined} ->
+ mrverror(<<"`partition_key` parameter is mandatory for queries to this database.">>);
+ {true, _PartitionKey} ->
+ ok;
+ {false, undefined} ->
+ ok;
+ {false, _PartitionKey} ->
+ mrverror(<<"`partition_key` parameter is not supported in this database.">>)
+ end,
+
true.
-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
@@ -569,7 +588,32 @@ update_args(#mrargs{} = Args) ->
{_, EKDocId1} -> EKDocId1
end,
+ LowestKey = null,
+ HighestKey = {[{<<239, 191, 176>>, null}]}, % \ufff0
+
+ {StartKey, EndKey} = case Args of
+ #mrargs{partition_key=undefined} ->
+ {Args#mrargs.start_key, Args#mrargs.end_key};
+
+ #mrargs{partition_key=PKey0} when not is_binary(PKey0) ->
+ mrverror(<<"`partition_key` must be a string.">>);
+
+ #mrargs{partition_key=PKey0, start_key=undefined, end_key=undefined} ->
+ {[PKey0, LowestKey], [PKey0, HighestKey]};
+
+ #mrargs{partition_key=PKey0, start_key=SK0, end_key=undefined} ->
+ {[PKey0, SK0], [PKey0, HighestKey]};
+
+ #mrargs{partition_key=PKey0, start_key=undefined, end_key=EK0} ->
+ {[PKey0, LowestKey], [PKey0, EK0]};
+
+ #mrargs{partition_key=PKey0, start_key=SK0, end_key=EK0} ->
+ {[PKey0, SK0], [PKey0, EK0]}
+ end,
+
Args#mrargs{
+ start_key=StartKey,
+ end_key=EndKey,
start_key_docid=SKDocId,
end_key_docid=EKDocId,
group_level=GroupLevel
@@ -577,8 +621,11 @@ update_args(#mrargs{} = Args) ->
validate_and_update_args(#mrargs{} = Args) ->
- true = validate_args(Args),
- update_args(Args).
+ validate_and_update_args(Args, []).
+
+validate_and_update_args(#mrargs{} = Args, Options) ->
+ true = validate_args(Args, Options),
+ update_args(Args, Options).
determine_group_level(#mrargs{group=undefined, group_level=undefined}) ->
@@ -1211,3 +1258,10 @@ kv_external_size(KVList, Reduction) ->
lists:foldl(fun([[Key, _], Value], Acc) ->
?term_size(Key) + ?term_size(Value) + Acc
end, ?term_size(Reduction), KVList).
+
+
+partition_key(Key, DocId) ->
+ [hd(binary:split(DocId, <<":">>)), Key].
+
+unpartition_key([_Partition, Key]) ->
+ Key.